From 535b8195efbc81ffc0ef6964cf572209d4e088f2 Mon Sep 17 00:00:00 2001 From: mr-forust Date: Sat, 25 Apr 2026 02:19:43 +0200 Subject: [PATCH] Prepare scraper for shared data and server mode --- telegram_scraper_with_forwarding.py | 98 ++++++++++++++++++----------- 1 file changed, 60 insertions(+), 38 deletions(-) diff --git a/telegram_scraper_with_forwarding.py b/telegram_scraper_with_forwarding.py index f070365..4704064 100644 --- a/telegram_scraper_with_forwarding.py +++ b/telegram_scraper_with_forwarding.py @@ -61,7 +61,11 @@ class ForwardingRule: class OptimizedTelegramScraper: def __init__(self): - self.STATE_FILE = 'state.json' + self.DATA_DIR = Path('data') + self.SESSION_DIR = Path('session') + self.DATA_DIR.mkdir(exist_ok=True) + self.SESSION_DIR.mkdir(exist_ok=True) + self.STATE_FILE = str(self.DATA_DIR / 'state.json') self.state = self.load_state() self.client = None self.continuous_scraping_active = False @@ -147,8 +151,8 @@ class OptimizedTelegramScraper: def get_db_connection(self, channel: str) -> sqlite3.Connection: if channel not in self.db_connections: - channel_dir = Path(channel) - channel_dir.mkdir(exist_ok=True) + channel_dir = self.DATA_DIR / channel + channel_dir.mkdir(parents=True, exist_ok=True) db_file = channel_dir / f'{channel}.db' conn = sqlite3.connect(str(db_file), check_same_thread=False, timeout=30) @@ -223,7 +227,7 @@ class OptimizedTelegramScraper: return None try: - channel_dir = Path(channel) + channel_dir = self.DATA_DIR / channel media_folder = channel_dir / 'media' media_folder.mkdir(exist_ok=True) @@ -299,10 +303,7 @@ class OptimizedTelegramScraper: async def forward_message(self, message, rule: ForwardingRule, source_channel_id: int = None): try: - if rule.destination_channel.lstrip('-').isdigit(): - dest_entity = await self.client.get_entity(PeerChannel(int(rule.destination_channel))) - else: - dest_entity = await self.client.get_entity(rule.destination_channel) + dest_entity = await self._resolve_entity(rule.destination_channel) if rule.forward_mode == "forward": await self.client.forward_messages(dest_entity, message) @@ -527,9 +528,10 @@ class OptimizedTelegramScraper: test_id = int(dest_input) # Try to resolve the channel via Telethon to verify it exists try: - entity = await self.client.get_entity(PeerChannel(test_id)) + entity = await self._resolve_entity(dest_input) dest_channel = dest_input - print(f"āœ… Found channel: {getattr(entity, 'title', dest_input)}") + name = getattr(entity, 'title', None) or ' '.join(filter(None, [getattr(entity, 'first_name', None), getattr(entity, 'last_name', None)])) or dest_input + print(f"āœ… Found: {name}") except Exception: print(f"āŒ Could not access channel ID {dest_input}. Make sure this account is a member.") return @@ -692,12 +694,25 @@ class OptimizedTelegramScraper: else: print("āŒ Failed to delete rule") + async def _resolve_entity(self, channel: str): + """Resolve a channel/chat/user entity from an ID string or username.""" + if channel.lstrip('-').isdigit(): + cid = int(channel) + # Channel/supergroup IDs are negative and large; user IDs are positive or small negative + try: + return await self.client.get_entity(PeerChannel(cid)) + except Exception: + # Fallback: let Telethon figure it out (works for users and basic groups) + return await self.client.get_entity(cid) + else: + return await self.client.get_entity(channel) + async def scrape_channel(self, channel: str, offset_id: int): try: if not self.client.is_connected(): await self.client.connect() - entity = await self.client.get_entity(PeerChannel(int(channel)) if channel.startswith('-') else channel) + entity = await self._resolve_entity(channel) result = await self.client.get_messages(entity, offset_id=offset_id, reverse=True, limit=0) total_messages = result.total @@ -834,14 +849,11 @@ class OptimizedTelegramScraper: print(f"šŸ“„ Reprocessing {len(message_ids)} media files for {channel_name} (ID: {channel})") try: - if channel.lstrip('-').isdigit(): - entity = await self.client.get_entity(PeerChannel(int(channel))) - else: - entity = await self.client.get_entity(channel) + entity = await self._resolve_entity(channel) semaphore = asyncio.Semaphore(self.max_concurrent_downloads) completed_media = 0 successful_downloads = 0 - + async def download_single_media(message): async with semaphore: return await self.download_media(channel, message) @@ -850,7 +862,7 @@ class OptimizedTelegramScraper: for i in range(0, len(message_ids), batch_size): batch_ids = message_ids[i:i + batch_size] messages = await self.client.get_messages(entity, ids=batch_ids) - + valid_messages = [msg for msg in messages if msg and msg.media and not isinstance(msg.media, MessageMediaWebPage)] tasks = [asyncio.create_task(download_single_media(msg)) for msg in valid_messages] @@ -862,13 +874,13 @@ class OptimizedTelegramScraper: successful_downloads += 1 except Exception: pass - + completed_media += 1 progress = (completed_media / len(message_ids)) * 100 bar_length = 30 filled_length = int(bar_length * completed_media // len(message_ids)) bar = 'ā–ˆ' * filled_length + 'ā–‘' * (bar_length - filled_length) - + sys.stdout.write(f"\ršŸ”„ Rescrape: [{bar}] {progress:.1f}% ({completed_media}/{len(message_ids)})") sys.stdout.flush() @@ -909,14 +921,11 @@ class OptimizedTelegramScraper: print(f"\nšŸ”§ Attempting to download {len(missing_media)} missing media files...") try: - if channel.lstrip('-').isdigit(): - entity = await self.client.get_entity(PeerChannel(int(channel))) - else: - entity = await self.client.get_entity(channel) + entity = await self._resolve_entity(channel) semaphore = asyncio.Semaphore(self.max_concurrent_downloads) completed_media = 0 successful_downloads = 0 - + async def download_single_media(message): async with semaphore: return await self.download_media(channel, message) @@ -1063,7 +1072,7 @@ class OptimizedTelegramScraper: def export_to_csv(self, channel: str): conn = self.get_db_connection(channel) filename = self.get_export_filename(channel) - csv_file = Path(channel) / f'{filename}.csv' + csv_file = self.DATA_DIR / channel / f'{filename}.csv' cursor = conn.cursor() cursor.execute('SELECT * FROM messages ORDER BY date') @@ -1082,7 +1091,7 @@ class OptimizedTelegramScraper: def export_to_json(self, channel: str): conn = self.get_db_connection(channel) filename = self.get_export_filename(channel) - json_file = Path(channel) / f'{filename}.json' + json_file = self.DATA_DIR / channel / f'{filename}.json' cursor = conn.cursor() cursor.execute('SELECT * FROM messages ORDER BY date') @@ -1137,11 +1146,11 @@ class OptimizedTelegramScraper: if not channel_name or channel_name in ['Unknown', 'no_username']: try: - if channel.lstrip('-').isdigit(): - entity = await self.client.get_entity(PeerChannel(int(channel))) - else: - entity = await self.client.get_entity(channel) - channel_name = getattr(entity, 'title', None) or getattr(entity, 'username', None) or 'Unknown' + entity = await self._resolve_entity(channel) + channel_name = getattr(entity, 'title', None) or getattr(entity, 'first_name', None) or getattr(entity, 'username', None) or 'Unknown' + # For users, combine first+last name + if isinstance(entity, User) and entity.first_name: + channel_name = ' '.join(filter(None, [entity.first_name, entity.last_name])) if 'channel_names' not in self.state: self.state['channel_names'] = {} self.state['channel_names'][channel] = channel_name @@ -1163,15 +1172,21 @@ class OptimizedTelegramScraper: if not self.client.is_connected(): await self.client.connect() - print("\nList of channels and groups joined by account:") + print("\nList of channels, groups and private chats joined by account:") count = 1 channels_data = [] async for dialog in self.client.iter_dialogs(): entity = dialog.entity - if dialog.id != 777000 and (isinstance(entity, Channel) or isinstance(entity, Chat)): - channel_type = "Channel" if isinstance(entity, Channel) and entity.broadcast else "Group" + if dialog.id != 777000 and (isinstance(entity, (Channel, Chat, User))): + if isinstance(entity, Channel): + channel_type = "Channel" if entity.broadcast else "Group" + elif isinstance(entity, Chat): + channel_type = "Group" + else: + channel_type = "Private" username = getattr(entity, 'username', None) or 'no_username' - print(f"[{count}] {dialog.title} (ID: {dialog.id}, Type: {channel_type}, Username: @{username})") + display_username = f"@{username}" if username != 'no_username' else '(no username)' + print(f"[{count}] {dialog.title} (ID: {dialog.id}, Type: {channel_type}, Username: {display_username})") channels_data.append({ 'number': count, 'channel_name': dialog.title, @@ -1182,7 +1197,7 @@ class OptimizedTelegramScraper: count += 1 if channels_data: - csv_file = Path('channels_list.csv') + csv_file = self.DATA_DIR / 'channels_list.csv' with open(csv_file, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=['number', 'channel_name', 'channel_id', 'username', 'type']) writer.writeheader() @@ -1246,8 +1261,11 @@ class OptimizedTelegramScraper: print(f"\nāŒ Phone authentication failed: {e}") return False - async def initialize_client(self): + async def initialize_client(self, interactive: bool = True): if not all([self.state.get('api_id'), self.state.get('api_hash')]): + if not interactive: + print("API credentials are missing in state.json") + return False print("\n=== API Configuration Required ===") print("You need to provide API credentials from https://my.telegram.org") try: @@ -1258,7 +1276,7 @@ class OptimizedTelegramScraper: print("Invalid API ID. Must be a number.") return False - self.client = TelegramClient('session', self.state['api_id'], self.state['api_hash']) + self.client = TelegramClient(str(self.SESSION_DIR / 'session'), self.state['api_id'], self.state['api_hash']) try: await self.client.connect() @@ -1267,6 +1285,10 @@ class OptimizedTelegramScraper: return False if not await self.client.is_user_authorized(): + if not interactive: + print("Telegram session is not authorized. Run the CLI once to sign in.") + await self.client.disconnect() + return False print("\n=== Choose Authentication Method ===") print("[1] QR Code (Recommended - No phone number needed)") print("[2] Phone Number (Traditional method)")