feat: init tg nodes, init web view

tg/tg_crawler.py

@@ -0,0 +1,73 @@
+import logging
+import psycopg2
+from datetime import datetime
+from telethon.sync import TelegramClient
+from telethon.tl.functions.messages import GetHistoryRequest
+
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(levelname)s - %(message)s',
+    handlers=[
+        logging.FileHandler('tg_nodes.log'),
+        logging.StreamHandler()
+    ]
+)
+
+class TelegramChannelMonitor:
+    db_config = None
+
+    def __init__(self, session_name, api_id, api_hash, channel_username, source_name):
+        self.session_name = session_name
+        self.api_id = api_id
+        self.api_hash = api_hash
+        self.channel_username = channel_username
+        self.source_name = source_name
+
+    @classmethod
+    def set_db_config(cls, config):
+        cls.db_config = config
+
+    def fetch_last_post(self):
+        logging.info(f"[{self.source_name}] checking a new post...")
+
+        try:
+            with TelegramClient(self.session_name, self.api_id, self.api_hash) as client:
+                entity = client.get_entity(self.channel_username)
+                history = client(GetHistoryRequest(
+                    peer=entity,
+                    limit=1,
+                    offset_date=None,
+                    offset_id=0,
+                    max_id=0,
+                    min_id=0,
+                    add_offset=0,
+                    hash=0
+                ))
+
+                if history.messages:
+                    msg = history.messages[0]
+                    logging.info(f"[{self.source_name}] received a post: {msg.message[:60]}...")
+                    self.save_to_db(self.source_name, msg.message)
+                else:
+                    logging.info(f"[{self.source_name}] there is no new messages")
+        except Exception as e:
+            logging.error(f"[{self.source_name}] error when receiving a post: {e}")
+
+    def save_to_db(self, source, message):
+        if not self.db_config:
+            logging.error("DB config is not set")
+            return
+
+        try:
+            conn = psycopg2.connect(**self.db_config)
+            cur = conn.cursor()
+            cur.execute(
+                "INSERT INTO leaks (source, message) VALUES (%s, %s)",
+                (source, message)
+            )
+            conn.commit()
+            cur.close()
+            conn.close()
+            logging.info(f"[{self.source_name}] message is recorded in the database")
+        except Exception as e:
+            logging.error(f"[{self.source_name}] error when writing to the database: {e}")

tg/tg_node_0.py

@@ -0,0 +1,34 @@
+import asyncio
+import os
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from pytz import timezone
+from tg_crawler import TelegramChannelMonitor
+from dotenv import load_dotenv
+
+load_dotenv()
+
+TelegramChannelMonitor.set_db_config({
+    'host': os.getenv("HOST"),
+    'port': os.getenv("PORT"),
+    'database': os.getenv("DBNAME"),
+    'user': os.getenv("USER"),
+    'password': os.getenv("PASSWORD")
+})
+
+monitor = TelegramChannelMonitor(
+    session_name='session_trueosint',
+    api_id=os.getenv("TELETHON_API_ID"),
+    api_hash=os.getenv("TELETHON_API_HASH"),
+    channel_username='trueosint',
+    source_name='trueosint'
+)
+
+def main():
+    scheduler = AsyncIOScheduler()
+    scheduler.add_job(monitor.fetch_last_post, "cron", hour=9, minute=0, timezone=timezone("Europe/Moscow"))
+    scheduler.start()
+
+    asyncio.get_event_loop().run_forever()
+
+if __name__ == '__main__':
+    main()

tg/tg_node_1.py

@@ -0,0 +1,34 @@
+import asyncio
+import os
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from pytz import timezone
+from tg_crawler import TelegramChannelMonitor
+from dotenv import load_dotenv
+
+load_dotenv()
+
+TelegramChannelMonitor.set_db_config({
+    'host': os.getenv("HOST"),
+    'port': os.getenv("PORT"),
+    'database': os.getenv("DBNAME"),
+    'user': os.getenv("USER"),
+    'password': os.getenv("PASSWORD")
+})
+
+monitor = TelegramChannelMonitor(
+    session_name='session_dataleak',
+    api_id=os.getenv("TELETHON_API_ID"),
+    api_hash=os.getenv("TELETHON_API_HASH"),
+    channel_username='dataleak',
+    source_name='dataleak'
+)
+
+def main():
+    scheduler = AsyncIOScheduler()
+    scheduler.add_job(monitor.fetch_last_post, "cron", hour=9, minute=0, timezone=timezone("Europe/Moscow"))
+    scheduler.start()
+
+    asyncio.get_event_loop().run_forever()
+
+if __name__ == '__main__':
+    main()