repair functions

worker-dev/app.py

@@ -0,0 +1,44 @@
+import pika
+import sys
+import os
+import json
+import datetime
+import re
+import config
+from pymongo import MongoClient
+from msg import msg
+import requests
+import time
+
+def main():
+    connection = pika.BlockingConnection(pika.ConnectionParameters(
+        "192.168.0.20", 5672, "mkt", pika.PlainCredentials(
+            "rabbit", "mrl2X0jwnYuCCiKFTshG7WKyOAhfDo")
+    ))
+    channel = connection.channel()
+
+    def callback(ch, method, properties, body: bytearray):
+        srcJson = json.loads(str(body.decode('utf-8')).replace("\'", "\""))
+#        if srcJson['state'] == 'HANGUP' and srcJson['direction'] == 'incoming' and srcJson['from'] == '89135853246':
+        try:
+            if srcJson['callstatus'] != 'ANSWER':
+                print(srcJson)
+        except:
+            pass
+
+    channel.basic_consume(
+        queue='copy_incoming', on_message_callback=callback, auto_ack=False)
+
+    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-dev/config.py

@@ -0,0 +1,16 @@
+
+IMPORTANT_LINE = ['83912051045']
+
+IGNORE_LIST = ['83912051045',
+               '83912051046',
+               '83919865589',
+               '83919865589',
+               '84950213944',
+               '84951183750',
+               '84951252791',
+               '89919237009',
+               '89919241441',
+               '89919398228',
+               '89919500798',
+               '89919505445',
+               '89919863883']

worker-dev/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 *.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-dev/getmsg.py

@@ -0,0 +1,94 @@
+import pika
+import sys
+import os
+import json
+import datetime
+import re
+import config
+from pymongo import MongoClient
+from msg import msg
+import requests
+import time
+
+CONNECTION_STRING = "mongodb://mongodb:Cc03Wz5XX3iI3uY3@192.168.0.20"
+db_connection = MongoClient(CONNECTION_STRING)
+
+db_base = db_connection["phone"]
+coll_call = db_base["phone_dev"]
+from_api = db_base["from_api"]
+
+
+def main():
+    connection = pika.BlockingConnection(pika.ConnectionParameters(
+        "192.168.0.20", 5672, "mkt", pika.PlainCredentials(
+            "rabbit", "mrl2X0jwnYuCCiKFTshG7WKyOAhfDo")
+    ))
+    channel = connection.channel()
+
+    def callback(ch, method, properties, body: bytearray):
+        m = msg(body)
+
+        # Разбираем входящие звонки
+        if m.isIncoming():
+            # Если статус START
+            if m.getState() == "START":
+                insert_value = {
+                    "uuid": m.getUUID(),
+                    "status": -1,
+                    "canClose": 0
+                }
+
+                coll_call.update_one(
+                    filter={"client": m.getClient(), "status": -1},
+                    update={"$set": insert_value},
+                    upsert=True
+                )
+                #print("НАЧАТ:", m.dict)
+
+            # Если Статус END
+            if m.getState() == "END":
+                coll_call.update_one(
+                    filter={"uuid": m.getUUID()},
+                    update={"$set": {"canClose": 1}}
+                )
+                #print("ЗАКОНЧЕН:", m.dict)
+                time.sleep(5)
+                resp = requests.get("https://callinfo.services.mobilon.ru/api/call/info/1e86a98e026578eb5f6bf8c092c0c4a2/" + m.getUUID())
+                j = json.loads(resp.content.decode('utf-8'))
+                d = dict(j)
+                from_api.insert_one(d)
+
+            # Если статус HANGUP
+            if m.getState() == "HANGUP":
+                if m.isAnswered():
+                    insert_value = {
+                        "status": 0,
+                        "duration": m.getDuration(),
+                        "recordUrl": m.getRecordUrl()
+                    }
+                else:
+                    insert_value = {
+                        "status": 1
+                    }
+                coll_call.update_one(
+                    filter={"uuid": m.getUUID(), "canClose": 1},
+                    update={"$set": insert_value}
+                )
+                #print("Положена трубка:", m.dict)
+
+    channel.basic_consume(
+        queue='test', 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-dev/msg.py

@@ -0,0 +1,68 @@
+import os
+import json
+import datetime
+import re
+import config
+
+
+class msg:
+    def __init__(self, body):
+        self.body: bytearray = body
+        self.decode: str = self.body.decode('utf-8')
+        self.json: json = json.loads(self.decode.replace("\'", "\""))
+        self.dict: dict = dict(self.json)
+
+    def getDirection(self):
+        try:
+            return self.dict["direction"]
+        except:
+            pass
+
+    def isIncoming(self) -> bool:
+        return True if self.dict["direction"] == "incoming" else False
+
+    def isExternal(self) -> bool:
+        return not self.isIncoming()
+
+    def getState(self) -> str:
+        try:
+            return self.dict["state"]
+        except:
+            pass
+
+    def getClient(self) -> str:
+        return self.dict["from"] if self.isIncoming() else self.dict["to"]
+
+    def getTime(self) -> str:
+        try:
+            return datetime.datetime.fromtimestamp(self.dict["time"]).strftime('%Y-%m-%d %H:%M:%S')
+        except:
+            return None
+
+    def getStatus(self) -> int:
+        pass
+
+    def getRecordUrl(self) -> str:
+        return self.dict["recordUrl"]
+
+    def getDuration(self) -> int:
+        return self.dict["duration"]
+
+    def getUUID(self) -> str:
+        try:
+            return self.dict["uuid"]
+        except:
+            pass
+
+    def isAnswered(self) -> bool:
+        try:
+            self.dict["callstatus"]
+            return True
+        except:
+            return False
+
+    def isImportant(self):
+        try:
+            return True if self.getState() == "START" and self.isIncoming() and self.dict["to"] in config.IMPORTANT_LINE else False
+        except:
+            pass

worker-dev/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