upload

main.py

@@ -0,0 +1,108 @@
+import logging
+import logging.config
+import os
+import asyncio
+from datetime import datetime
+import socket
+import aiofiles
+from dotenv import load_dotenv
+
+load_dotenv()
+
+AMI_HOST = os.environ.get("AMI_HOST")
+AMI_PORT = int(os.environ.get("AMI_PORT"))
+AMI_USER = os.environ.get("AMI_USER")
+AMI_PASSWORD = os.environ.get("AMI_PASSWORD")
+
+
+def send_action(s, action):
+    s.send(action.encode())
+    response = ""
+    while True:
+        chunk = s.recv(4096)
+        try:
+            response += chunk.decode()
+        except UnicodeDecodeError:
+            pass
+        if "\r\n\r\n" in response:
+            break
+    response_parts = response.split("\r\n\r\n", 1)
+    body = response_parts[1]
+    body_dict = {
+        line.split(":")[0].strip(): line.split(":")[1].strip()
+        for line in body.split("\n")
+        if len(line.split(":")) > 1
+    }
+    return body_dict
+
+
+async def full_log(event):
+    file_name = f"log/{datetime.now().strftime('%Y-%m-%d')}.log"
+    async with aiofiles.open(file_name, "a") as f:
+        await f.write("\n\n" + datetime.now().strftime("%H:%M:%S") + "\n\n")
+        for key, value in event.items():
+            await f.write(f"{key}: {value}\n")
+
+
+async def ami_listening():
+    # процедура регулярной проверки соединения
+    async def check_connection():
+        while True:
+            try:
+                await asyncio.sleep(3600)
+                s.send(b"Action: Ping\r\n\r\n")
+                await asyncio.sleep(1)
+            except (ConnectionError, ConnectionResetError):
+                logging.warning("Connection lost. Restarting...")
+                await asyncio.sleep(1)
+                return True  # возвращаемся в начало функции ami_listening
+
+    # проверка соединения каждые 5 секунд
+    conn_check_task = asyncio.create_task(check_connection())
+    conn_check_task.add_done_callback(lambda t: ami_listening() if t.result() else None)
+
+    while True:
+        try:
+            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            logging.info("Connecting to Asterisk...")
+            s.connect((AMI_HOST, AMI_PORT))
+            break
+        except ConnectionRefusedError:
+            logging.warning("Connection refused. Retrying...")
+            await asyncio.sleep(1)
+
+    logging.info("Connected to Asterisk")
+    logging.info("Logging in...")
+
+    login_action = f"Action: Login\r\nUsername: {AMI_USER}\r\nSecret: {AMI_PASSWORD}\r\n\r\n"
+    send_action(s, login_action)
+
+    logging.info("Logged in")
+    logging.info("Listening for events...")
+
+    events_action = "Action: Events\r\nEventMask: on\r\n\r\n"
+    send_action(s, events_action)
+
+    try:
+        while True:
+            event = send_action(s, "")
+            if event:
+                await full_log(event)
+    except (KeyboardInterrupt, SystemExit):
+        logging.info("Exiting...")
+        s.close()
+    except ConnectionResetError:
+        logging.warning("Connection reset. Restarting...")
+        await ami_listening()
+
+
+
+async def main():
+    if not os.path.exists("log"):
+        os.makedirs("log")
+    logging.config.fileConfig("logging.conf")
+    await ami_listening()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())