SYNC

dev/worker/app.py

@@ -0,0 +1,107 @@
+import pika
+import sys
+import os
+import json
+import datetime
+from pymongo import MongoClient
+import time
+from phone import phone
+import requests
+
+db_connection = MongoClient("mongodb://mongodb:Cc03Wz5XX3iI3uY3@mongo")
+db_base = db_connection["phone-dev"]
+coll = db_base["phone"]
+coll_all = db_base["phone-all"]
+err = db_base["err"]
+blacklist = db_base["blacklist"]
+phoneDebug = db_base["phone-debug"]
+coll_userkey = db_base['userkey']
+
+mkt_phones = ['83912051046', '83912051045']
+
+mobilon_wait = 5  # Время ожидания необходимо чтобы на стороне мобилона все улеглось
+min_recall_len_in_sec = 8
+
+
+def isBlack(number: str) -> bool:
+    return True if blacklist.count_documents({"_id": number}) > 0 else False
+
+
+# Stats:
+# 0 - ответили
+# 1 - не приняли
+# 2 - перезвонили успешно
+# 4 - Перезвонили неуспешно
+
+
+# LOG
+# 0 - выходящий вызов закончен
+# 1 - Вызов не принят
+# 2 - Перезвонили и дозвонились
+def main():
+
+    connection = pika.BlockingConnection(pika.ConnectionParameters(
+        'rabbitmq', 5672, 'mkt', pika.PlainCredentials('rabbit', 'mrl2X0jwnYuCCiKFTshG7WKyOAhfDo')))
+    channel = connection.channel()
+
+    channel.queue_declare(queue='dev-incoming')
+
+    def callback(ch, method, properties, body: bytearray):
+        # Парсим строку
+        ph = phone(body)
+        # Определяем тип соединения
+        # Если входящий
+        print(ph.ConvertToJSON())
+
+        phoneDebug.insert_one(ph.ConvertToJSON())
+
+        if ph.GetState() == "START":
+            ph.addStart()
+
+        if ph.GetState() == "END":
+            ph.addEnd()
+
+        if ph.GetState() == "HANGUP" and ph.isCanClose() == 1:
+            # Ждем
+            time.sleep(mobilon_wait)
+
+            t_req = "https://callinfo.services.mobilon.ru/api/call/info/1e86a98e026578eb5f6bf8c092c0c4a2/" + ph.GetUUID()
+            try:
+                res: dict = json.loads(requests.get(t_req).content)
+            except:
+                print("Get data from Mobilon", t_req)
+
+            try:
+                coll_all.insert_one(res)
+            except:
+                print("coll_all:", res)
+
+            # В этом месте описывается логика работы программы исходя основываясь на данных мобилона
+            if res["direction"] == "incoming" and isBlack(res["from"]) == False:
+                # Подмена полей
+                res["client"] = res.pop("from")
+                res["mkt_phone"] = res.pop("to")
+
+                # 2. Удаляем пропущенные если есть
+                coll.delete_many(
+                    {"client": {'$regex': res["client"]}, "status": {"$in": ["NOT_ANSWERED", "RECALL_FALSE"]}})
+                coll.insert_one(res)
+            # -------------------------------
+            res.clear()
+
+    channel.basic_consume(
+        queue='incoming-dev', 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)

dev/worker/dockerfile

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

dev/worker/phone.py

@@ -0,0 +1,60 @@
+import json
+import redis
+
+
+r = redis.Redis(host="redis")
+
+
+class phone:
+    def __init__(self, message: bytearray):
+        self.msg = message
+        self.dict = self.ConvertToJSON()
+
+    def ConvertToJSON(self):
+        try:
+            return json.loads(str(self.msg.decode('utf-8')).replace("\'", "\""))
+        except:
+            return {}
+
+    # def GetDirection(self) -> str:
+    #     return self.dict["direction"]
+
+    def GetUUID(self) -> str:
+        return self.dict["uuid"]
+
+    # def GetClient(self) -> str:
+    #     return self.dict["from"] if self.GetDirection() == "incoming" else self.dict["to"]
+
+    # def GetOperator(self) -> str:
+    #     return self.dict["to"] if self.GetDirection() == "incoming" else self.dict["from"]
+
+    # def GetTimestamp(self) -> int:
+    #     """Return int timestamp"""
+    #     return self.dict["time"]
+
+    def GetState(self) -> str:
+        return self.dict["state"]
+
+    # def isIncoming(self) -> bool:
+    #     """True Если входящий, иначе False"""
+    #     return True if self.GetDirection() == "incoming" else False
+
+    # def isExternal(self) -> bool:
+    #     """True если исходящий, иначе False"""
+    #     return True if self.GetDirection() == "external" else False
+
+    def isCanClose(self) -> int:
+        try:
+            return int(r.hget(self.GetUUID(), "canClose"))
+        except:
+            return 0
+
+    def addStart(self):
+        r.hset(self.GetUUID(), mapping={
+            "canClose": "0"
+        })
+
+    def addEnd(self):
+        r.hset(self.GetUUID(), mapping={
+            "canClose": "1"
+        })

dev/worker/requirements.txt

@@ -0,0 +1,18 @@
+async-timeout==4.0.3
+click==8.1.3
+dnspython==2.3.0
+Flask==2.2.3
+importlib-metadata==6.2.0
+itsdangerous==2.1.2
+Jinja2==3.1.2
+MarkupSafe==2.1.2
+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
+redis==5.0.1
+Werkzeug==2.2.3
+zipp==3.15.0