diff --git a/services/admin_bot.py b/services/admin_bot.py index 5bf448d..a408516 100644 --- a/services/admin_bot.py +++ b/services/admin_bot.py @@ -17,7 +17,7 @@ def get_main_menu_keyboard(): [Button.text("🔑 Request Login Code", resize=True), Button.text("📊 Fleet Statistics", resize=True)], [Button.text("📡 Monitored Sources", resize=True), Button.text("🎯 Target Channels", resize=True)], [Button.text("➕ Add Source Guide", resize=True), Button.text("➕ Add Target Guide", resize=True)], - [Button.text("❓ Help & Documentation", resize=True)] + [Button.text("📥 Scrape History Guide", resize=True), Button.text("❓ Help & Documentation", resize=True)] ] class AdminBotService: @@ -137,6 +137,38 @@ class AdminBotService: result = await self.collector.submit_password(pwd) await status_msg.edit(result, parse_mode="html", buttons=get_main_menu_keyboard()) + # --- History Scraper Commands --- + @self.client.on(events.NewMessage(pattern=r"(?i)^(📥 Scrape History Guide)$")) + async def cmd_scrape_guide(event: events.NewMessage.Event): + if not self.is_admin(event.sender_id): + return + guide = ( + "📥 Historical Channel Scraper:\n\n" + "Import previous/past posts from any channel:\n" + "/scrape_history <channel_id> [number_of_posts]\n\n" + "Example (Scrape last 30 posts):\n" + "/scrape_history -1001234567890 30\n\n" + "(Default is 20 posts if count is omitted)." + ) + await event.reply(guide, parse_mode="html") + + @self.client.on(events.NewMessage(pattern=r"^/scrape_history\s+(-?\d+)(?:\s+(\d+))?")) + async def cmd_scrape_history(event: events.NewMessage.Event): + if not self.is_admin(event.sender_id): + return + if not self.collector: + await event.reply("Collector service not linked.") + return + ch_id = int(event.pattern_match.group(1)) + limit = int(event.pattern_match.group(2)) if event.pattern_match.group(2) else 20 + + status_msg = await event.reply(f"⏳ Scraping the last {limit} posts from {ch_id} in the background...", parse_mode="html") + + async def progress_notify(txt: str): + await status_msg.edit(txt, parse_mode="html", buttons=get_main_menu_keyboard()) + + await self.collector.scrape_channel_history(channel_id=ch_id, limit=limit, progress_callback=progress_notify) + # --- Statistics --- @self.client.on(events.NewMessage(pattern=r"(?i)^(/stats|📊 Fleet Statistics)$")) async def cmd_stats(event: events.NewMessage.Event): @@ -259,9 +291,10 @@ class AdminBotService: help_text = ( "📖 Copykar Bot Quick Help\n\n" "1. Authentication: Click 🔑 Request Login Code and submit via /code <12345>.\n" - "2. Monitored Sources: Add channels the userbot should listen to via /add_source.\n" - "3. Review Flow: AI scans posts, checks duplicates, and sends drafts to the review channel with inline approval buttons.\n" - "4. Publishing: Approved posts are published to your target channels strictly according to their interval minutes." + "2. Monitored Sources: Add channels via /add_source.\n" + "3. Historical Posts: Backfill existing posts using /scrape_history <channel_id> [limit].\n" + "4. Review Flow: AI scans posts, checks duplicates, and sends drafts to the review channel with inline approval buttons.\n" + "5. Publishing: Approved posts are published to your target channels strictly according to their interval minutes." ) await event.reply(help_text, parse_mode="html", buttons=get_main_menu_keyboard()) diff --git a/services/collector.py b/services/collector.py index 6d7322b..530964a 100644 --- a/services/collector.py +++ b/services/collector.py @@ -65,7 +65,6 @@ class CollectorService: async def submit_code(self, code: str) -> str: if not self.phone or not self.phone_code_hash: - # Re-request code sent = await self.client.send_code_request(self.phone) self.phone_code_hash = sent.phone_code_hash @@ -144,6 +143,83 @@ class CollectorService: except Exception as e: logger.error(f"Error handling message from {event.chat_id}: {e}", exc_info=True) + async def scrape_channel_history( + self, + channel_id: int, + limit: int = 20, + progress_callback: Optional[Callable[[str], Awaitable[None]]] = None + ) -> int: + """Scrape historical messages from a source channel.""" + if not self.client.is_connected() or not await self.client.is_user_authorized(): + if progress_callback: + await progress_callback("❌ Collector Userbot is not authorized. Please log in first.") + return 0 + + collected_count = 0 + skipped_count = 0 + + try: + entity = await self.client.get_input_entity(channel_id) + messages = [] + async for msg in self.client.iter_messages(entity, limit=limit): + messages.append(msg) + + messages.reverse() + + for message in messages: + raw_text = message.raw_text or "" + if not raw_text and not message.media: + continue + + media_path = None + media_type = None + media_hash = None + + if message.media: + if isinstance(message.media, MessageMediaPhoto): + media_type = "photo" + elif isinstance(message.media, MessageMediaDocument): + media_type = "document" + else: + media_type = "other" + + filename = f"{channel_id}_{message.id}" + download_target = os.path.join(MEDIA_DIR, filename) + downloaded_file = await message.download_media(file=download_target) + if downloaded_file: + media_path = downloaded_file + media_hash = compute_file_hash(downloaded_file) + + content_hash = compute_content_hash(raw_text, media_hash) + + post_id = await self.repo.create_raw_post( + source_channel_id=channel_id, + source_message_id=message.id, + raw_text=raw_text, + media_path=media_path, + media_type=media_type, + content_hash=content_hash, + ) + + if post_id: + collected_count += 1 + COLLECTED_POSTS_TOTAL.labels(source_channel_id=str(channel_id)).inc() + logger.info(f"Backfilled historical post ID {post_id} from {channel_id}") + await self.ai_processor.process_post(post_id) + else: + skipped_count += 1 + + if progress_callback: + await progress_callback( + f"✅ Scraped {collected_count} new historical posts from {channel_id} (Skipped {skipped_count} existing/empty)." + ) + return collected_count + except Exception as e: + logger.error(f"Error scraping history from {channel_id}: {e}", exc_info=True) + if progress_callback: + await progress_callback(f"❌ Error scraping channel {channel_id}: {e}") + return collected_count + async def stop(self): if self.client.is_connected(): await self.client.disconnect()