init project

worker/app.py

@@ -0,0 +1,119 @@
+import pika
+import sys
+import os
+import json
+import datetime
+import requests
+from pymongo import MongoClient
+
+
+db_connection = MongoClient("mongodb://mongodb:Cc03Wz5XX3iI3uY3@mongo")
+db_base = db_connection["phone"]
+coll_phone = db_base["phone"]
+coll_userkey = db_base['userkey']
+
+#def sendMessage(dt, num):
+#    token = "2035324623:AAGACtvZ551m9V--yTYF9cFuegGejylSsLg"
+#    chat_id = "-1001941363918"
+#    message = "*" + num + "* - " + dt
+#    send_text = 'https://api.telegram.org/bot' + token + '/sendMessage?chat_id=' + chat_id + '&parse_mode=Markdown&text=' + message
+#    response = requests.get(send_text)
+#    return response
+
+
+def main():
+    connection = pika.BlockingConnection(pika.ConnectionParameters(
+        'rabbitmq', 5672, 'mkt', pika.PlainCredentials('rabbit', 'mrl2X0jwnYuCCiKFTshG7WKyOAhfDo')))
+    channel = connection.channel()
+
+    channel.queue_declare(queue='incoming')
+    tmpIncoming = {}
+    # tmpExternal = {}
+
+    def callback(ch, method, properties, body: bytearray):
+        try:
+            # Парсим строку
+            srcJson = json.loads(str(body.decode('utf-8')).replace("\'", "\""))
+            srcJson["time"] = datetime.datetime.fromtimestamp(
+                srcJson["time"]).strftime('%Y-%m-%d %H:%M:%S')
+
+            # Определяем направление соединения
+            if srcJson['direction'] == 'incoming':
+                # Определяем начальный статус
+                if srcJson['state'] == 'START':
+                    # Создаем переменную. Ответ = false, можно закрывать = false, Приоритетная линия
+                    if srcJson['to'] == '83912051045':
+                        tmpIncoming[srcJson['uuid']] = [False, False, True]
+                    else:
+                        tmpIncoming[srcJson['uuid']] = [False, False, False]
+                    # Обновление статуса при входящем звонке
+                    coll_phone.delete_one({'$and': [{'client': srcJson['from']}, {'status': 1}]})
+                if srcJson['state'] == 'ANSWER' and srcJson['uuid'] in tmpIncoming:
+                    tmpIncoming[srcJson['uuid']][0] = True
+                if srcJson['state'] == 'END' and srcJson['uuid'] in tmpIncoming:
+                    tmpIncoming[srcJson['uuid']][1] = True
+                if srcJson['state'] == 'HANGUP' and srcJson['uuid'] in tmpIncoming and tmpIncoming[srcJson['uuid']][1] == True:
+                    try:
+                        srcJson['callstatus']
+                        insDict = {"client": srcJson['from'], "time": srcJson['time'], "status": 0,
+                                   "recordUrl": srcJson["recordUrl"], "duration": srcJson["duration"], "important": tmpIncoming[srcJson['uuid']][2]}
+                    except Exception as e:
+                        # print(srcJson)
+                        print(e)
+                        insDict = {
+                            "client": srcJson['from'], "time": srcJson['time'], "status": 1, "important": tmpIncoming[srcJson['uuid']][2]}
+                        #sendMessage(srcJson['time'], srcJson['from'])
+                    finally:
+                        print(insDict)
+                        coll_phone.insert_one(insDict)
+                        tmpIncoming.pop(srcJson['uuid'])
+                        try:
+                            insUserKey = {'userkey': srcJson['userkey']}
+                            print(insUserKey)
+                            print(srcJson['to'])
+                            coll_userkey.update_one(
+                                filter={
+                                    'operator': srcJson['to'],
+                                },
+                                update={
+                                    '$set': insUserKey,
+                                },
+                                upsert=True
+                            )
+                        except Exception as e:
+                            print(e)
+
+            if srcJson['direction'] == 'external':
+                # coll_phone.update_one({'$and': [{'client': srcJson['to']}, {'status': 1}]}, {
+                #                       '$set': {'status': 2}})
+                r = coll_phone.update_one({'$and': [{'client': {'$regex': srcJson['to']}}, {'status': 1}]}, {
+                                      '$set': {'status': 2, 'callid': srcJson['uuid']}})
+                if srcJson['state'] == 'HANGUP':
+                    try:
+                        coll_phone.update_one({'callid':srcJson['uuid']}, {'$set': {'recordUrl': srcJson['recordUrl']}})
+                    except Exception as e:
+                        print(e)
+                print(r)
+
+        except Exception as e:
+            print(e.with_traceback)
+            print(e)
+            # print(None)
+            exit()
+
+    channel.basic_consume(
+        queue='incoming', on_message_callback=callback, auto_ack=True)
+
+    print(' [*] Waiting for messages. To exit press CTRL+C')
+    channel.start_consuming()
+
+
+if __name__ == '__main__':
+    try:
+        main()
+    except KeyboardInterrupt:
+        print('Interrupted')
+        try:
+            sys.exit(0)
+        except SystemExit:
+            os._exit(0)

worker/dockerfile

@@ -0,0 +1,14 @@
+FROM alpine
+
+RUN apk update && apk upgrade && apk add python3 && apk add -U tzdata
+
+WORKDIR /app
+
+COPY requirements.txt requirements.txt
+ADD app.py /app
+RUN python3 -m venv .venv
+RUN /app/.venv/bin/pip3 install -r /app/requirements.txt
+
+EXPOSE 5000
+
+CMD [".venv/bin/python3", "app.py"]

worker/requirements.txt

@@ -0,0 +1,8 @@
+dnspython==2.3.0
+pika==1.3.1
+pymongo==4.3.3
+certifi==2022.12.7
+charset-normalizer==3.1.0
+idna==3.4
+requests==2.28.2
+urllib3==1.26.15