feat: celery

2025-05-03 08:25:15 +03:00
parent 166ad0ba2f
commit 44ce626a1b
16 changed files with 456 additions and 200 deletions
@@ -0,0 +1,97 @@
+from bs4 import BeautifulSoup
+import os
+import psycopg2
+from dotenv import load_dotenv
+from celery import Celery
+from utils.logg import LoggerSingleton
+from driver.driver_creator import DriverCreator
+from selenium.webdriver.common.by import By
+from time import sleep
+from celery_app import app
+
+load_dotenv()
+
+logger = LoggerSingleton.get_logger()
+
+class ForumCrawler:
+    def __init__(self, forum_url, proxy_list):
+        self.forum_url = forum_url
+        self.proxy_list = proxy_list
+        self.db_config = {
+            'host': os.getenv('DB_HOST'),
+            'port': os.getenv('DB_PORT'),
+            'database': os.getenv('DB_NAME'),
+            'user': os.getenv('DB_USER'),
+            'password': os.getenv('DB_PASSWORD')
+        }
+        self.conn = None
+        self._connect()
+
+    def _connect(self):
+        try:
+            self.conn = psycopg2.connect(**self.db_config)
+            logger.info("Database connection established")
+        except Exception as e:
+            logger.error(f"Database connection error: {e}")
+            raise
+
+    def _close(self):
+        if self.conn and not self.conn.closed:
+            try:
+                self.conn.close()
+                logger.info("Database connection closed")
+            except Exception as e:
+                logger.error(f"Error closing database connection: {e}")
+
+    def _crawl_leaks(self):
+        driver_creator = DriverCreator(self.proxy_list)
+        driver = driver_creator.get_driver()
+
+        try:
+            logger.info(f"Starting forum crawl: {self.forum_url}")
+            driver.get(self.forum_url)
+            sleep(2)
+
+            posts = driver.find_elements(By.CSS_SELECTOR, 'a.story-link h2.home-title')
+            if not posts:
+                logger.info("No posts found on the page")
+                return
+            last_post = posts[0]
+            title = last_post.text.strip()
+            link = last_post.get_attribute('href')
+            post_content = f"{title} - {link}"
+
+            if 'data breach' in title.lower() or 'leak' in title.lower():
+                self._save_to_db('Hacker News', post_content)
+                logger.info(f"New leak found: {title} - {link}")
+            else:
+                logger.info("Last post is not about leaks")
+
+        except Exception as e:
+            logger.error(f"Error during forum crawling: {e}")
+            raise
+        finally:
+            driver.quit()
+            logger.info("WebDriver session closed")
+        
+    def _save_to_db(self, source, message):
+        if not self.conn or self.conn.closed:
+            self._connect()
+        try:
+            with self.conn.cursor() as cur:
+                cur.execute(
+                    "INSERT INTO leaks (resource_name, message) VALUES (%s, %s)",
+                    (source, message)
+                )
+            self.conn.commit()
+            logger.info(f"Leak from {source} saved in the database")
+        except Exception as e:
+            logger.error(f"Error writing to the database: {e}")
+
+@app.task(name='forum_crawler.crawl_forum_task')
+def crawl_forum_task(forum_url, proxy_list):
+    crawler = ForumCrawler(forum_url, proxy_list)
+    try:
+        crawler._crawl_leaks()
+    finally:
+        crawler._close()
@@ -0,0 +1,121 @@
+import logging
+import os
+import asyncio
+import psycopg2
+from datetime import datetime
+from telethon.sync import TelegramClient
+from telethon.tl.functions.messages import GetHistoryRequest
+from dotenv import load_dotenv
+from celery import Celery
+from utils.logg import LoggerSingleton
+from celery_app import app
+
+load_dotenv()
+
+logger = LoggerSingleton.get_logger()
+
+class TelegramMonitor:
+    def __init__(self):
+        self.db_config = {
+            'host': os.getenv('DB_HOST'),
+            'port': os.getenv('DB_PORT'),
+            'database': os.getenv('DB_NAME'),
+            'user': os.getenv('DB_USER'),
+            'password': os.getenv('DB_PASSWORD')
+        }
+        self.conn = None
+        self._connect()
+
+    def _connect(self):
+        try:
+            self.conn = psycopg2.connect(**self.db_config)
+            logger.info("Database connection established")
+        except Exception as e:
+            logger.error(f"Database connection error: {e}")
+            raise
+
+    def _close(self):
+        if self.conn and not self.conn.closed:
+            try:
+                self.conn.close()
+                logger.info("Database connection closed")
+            except Exception as e:
+                logger.error(f"Error closing database connection: {e}")
+
+    def _fetch_post(self, channel_username, source_name):
+        try:
+            with TelegramClient('session', os.getenv('API_ID'), os.getenv('API_HASH')) as client:
+                entity = client.get_entity(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:
+                    self._save_to_db(source_name, history.messages[0].message)
+        except Exception as e:
+            logger.error(f"Error fetching post from {source_name}: {e}")
+            raise
+
+    def _save_to_db(self, source, message):
+        if not self.conn or self.conn.closed:
+            self._connect()
+            
+        try:
+            with self.conn.cursor() as cur:
+                cur.execute(
+                    "INSERT INTO leaks (resource_name, message) VALUES (%s, %s)",
+                    (source, message)
+                )
+            self.conn.commit()
+            logger.info(f"Data from {source} saved successfully")
+        except Exception as e:
+            logger.error(f"Database save error for {source}: {e}")
+            self.conn.rollback()
+            raise
+
+@app.task(bind=True, name='tasks.tg_crawler.monitor_channels')
+def monitor_channels(self):
+    channels = [
+        ('trueosint', 'trueosint'),
+        ('dataleak', 'dataleak'),
+        ('Vaultofdataleaksss', 'Vaultofdataleaksss')
+    ]
+    
+    logger.info("Starting Telegram channels monitoring")
+    
+    for channel, source in channels:
+        try:
+            monitor_channel.apply_async(
+                args=(channel, source),
+                queue='telegram',
+                retry=True,
+                retry_policy={
+                    'max_retries': 3,
+                    'interval_start': 2,
+                    'interval_step': 5,
+                    'interval_max': 20
+                }
+            )
+            logger.info(f"Sent task for channel: {channel}")
+        except Exception as e:
+            logger.error(f"Failed to send task for {channel}: {e}")
+            raise self.retry(exc=e)
+
+@app.task(bind=True, name='tasks.tg_crawler.monitor_channel')
+def monitor_channel(self, channel, source):
+    logger.info(f"Starting monitoring channel: {channel}")
+    monitor = TelegramMonitor()
+    try:
+        monitor._fetch_post(channel, source)
+        logger.info(f"Successfully monitored channel: {channel}")
+    except Exception as e:
+        logger.error(f"Channel monitoring failed for {channel}: {e}")
+        raise self.retry(exc=e)
+    finally:
+        monitor._close()