feat(collector): implement historical channel scraper command and ui guide

2026-08-27 21:05:06 +03:30
parent 00aca80180
commit 963d7f365f
2 changed files with 114 additions and 5 deletions
@@ -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 = (
+                "📥 <b>Historical Channel Scraper:</b>\n\n"
+                "Import previous/past posts from any channel:\n"
+                "<code>/scrape_history &lt;channel_id&gt; [number_of_posts]</code>\n\n"
+                "<i>Example (Scrape last 30 posts):</i>\n"
+                "<code>/scrape_history -1001234567890 30</code>\n\n"
+                "<i>(Default is 20 posts if count is omitted).</i>"
+            )
+            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 <b>{limit}</b> posts from <code>{ch_id}</code> 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 = (
                "📖 <b>Copykar Bot Quick Help</b>\n\n"
                "1. <b>Authentication:</b> Click <code>🔑 Request Login Code</code> and submit via <code>/code &lt;12345&gt;</code>.\n"
-                "2. <b>Monitored Sources:</b> Add channels the userbot should listen to via <code>/add_source</code>.\n"
-                "3. <b>Review Flow:</b> AI scans posts, checks duplicates, and sends drafts to the review channel with inline approval buttons.\n"
-                "4. <b>Publishing:</b> Approved posts are published to your target channels strictly according to their interval minutes."
+                "2. <b>Monitored Sources:</b> Add channels via <code>/add_source</code>.\n"
+                "3. <b>Historical Posts:</b> Backfill existing posts using <code>/scrape_history &lt;channel_id&gt; [limit]</code>.\n"
+                "4. <b>Review Flow:</b> AI scans posts, checks duplicates, and sends drafts to the review channel with inline approval buttons.\n"
+                "5. <b>Publishing:</b> 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())

@@ -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 <b>{collected_count}</b> new historical posts from <code>{channel_id}</code> (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 <code>{channel_id}</code>: {e}")
+            return collected_count
+
    async def stop(self):
        if self.client.is_connected():
            await self.client.disconnect()