refactor: reformat tg_filter.py and clean up

backend/tg_filter.py

@@ -1,9 +1,7 @@
 """
 Telegram Group Filter — scours 100+ groups for interesting messages.
-Runs as: t-youtube venv + Pyrogram session.
+Rotates through all groups across cron cycles to avoid rate limits.
-
+Resets processed tracking per batch so each cycle gets fresh messages.
-Scoring: mentions, links, code blocks, media, keywords → score ≥ threshold → saved.
-Delivers to: hermes_telegram_bot or stdout.
 """
 
 import asyncio
@@ -32,15 +30,19 @@ WORKDIR = "/Users/jimmyg"
 DB_PATH = os.path.join(WORKDIR, ".hermes", "tg_filter.db")
 
 # Scoring
-SCORE_MENTION = 30        # @username or reply to user
+SCORE_MENTION = 30
-SCORE_KEYWORD = 15        # Interesting keyword match
+SCORE_KEYWORD = 15
-SCORE_LINK = 10           # Contains URL
+SCORE_LINK = 10
-SCORE_MEDIA = 8           # Photo/video/file
+SCORE_MEDIA = 8
-SCORE_CODE = 12           # Code block
+SCORE_CODE = 12
-SCORE_LONG_TEXT = 5       # ≥200 chars of substance
+SCORE_LONG_TEXT = 5
-THRESHOLD = 20            # Minimum score to forward
+THRESHOLD = 40
 
-# Keywords that signal interesting content
+# Scan config
+MSGS_PER_GROUP = 500
+GROUPS_PER_CYCLE = 56
+
+# Keywords
 INTERESTING_KEYWORDS = [
     "ai", "llm", "gpt", "deepseek", "claude", "gemini", "qwen", "openai",
     "docker", "kubernetes", "k8s", "tailscale", "wireguard", "vps", "nas",
@@ -53,101 +55,64 @@ INTERESTING_KEYWORDS = [
     "startup", "saas", "mvp", "产品", "产品发布",
 ]
 
-# Groups to always ignore (IDs or title substrings)
+IGNORE_PATTERNS = ["合租社群", "黄油派对"]
-IGNORE_PATTERNS = [
+
-    "合租社群",
-    "黄油派对",
-]
 
 def score_message(msg: Message) -> int:
-    """Score a message for interestingness."""
     score = 0
     text = msg.text or msg.caption or ""
-
-    # Mentions
     if msg.entities:
         for e in msg.entities:
-            if e.type in ("mention", "text_mention"):
+            if e.type in ("mention", "text_mention"): score += SCORE_MENTION
-                score += SCORE_MENTION
+            if e.type == "url": score += SCORE_LINK
-            if e.type == "url":
+            if e.type == "code" or e.type == "pre": score += SCORE_CODE
-                score += SCORE_LINK
+    if msg.photo or msg.video or msg.document or msg.audio: score += SCORE_MEDIA
-            if e.type == "code" or e.type == "pre":
+    if re.search(r'https?://[^\s]+', text): score += SCORE_LINK
-                score += SCORE_CODE
-
-    # Media
-    if msg.photo or msg.video or msg.document or msg.audio:
-        score += SCORE_MEDIA
-
-    # Links in plain text
-    if re.search(r'https?://[^\s]+', text):
-        score += SCORE_LINK
-
-    # Keyword matches
     text_lower = text.lower()
     for kw in INTERESTING_KEYWORDS:
-        if kw in text_lower:
+        if kw in text_lower: score += SCORE_KEYWORD
-            score += SCORE_KEYWORD
+    if len(text.strip()) >= 200: score += SCORE_LONG_TEXT
-
+    if msg.forward_from or msg.forward_from_chat: score += 5
-    # Long substantive text
+    if msg.reply_to_message_id: score += 3
-    if len(text.strip()) >= 200:
-        score += SCORE_LONG_TEXT
-
-    # Forwarded messages are often interesting
-    if msg.forward_from or msg.forward_from_chat:
-        score += 5
-
-    # Reply to user's own messages (engagement)
-    if msg.reply_to_message_id:
-        score += 3
-
     return score
 
 
 def should_ignore(chat_title: str) -> bool:
-    """Check if a chat should be ignored entirely."""
+    if not chat_title: return False
-    if not chat_title:
-        return False
     for pat in IGNORE_PATTERNS:
-        if pat in chat_title:
+        if pat in chat_title: return True
-            return True
     return False
 
 
 def init_db():
-    """Create DB + table for tracking processed messages."""
     os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
     db = sqlite3.connect(DB_PATH)
-    db.execute("""
+    db.execute("CREATE TABLE IF NOT EXISTS processed (chat_id INTEGER, message_id INTEGER, PRIMARY KEY (chat_id, message_id))")
-        CREATE TABLE IF NOT EXISTS processed (
+    db.execute("CREATE TABLE IF NOT EXISTS interesting (id INTEGER PRIMARY KEY AUTOINCREMENT, chat_id INTEGER, chat_title TEXT, message_id INTEGER, sender TEXT, text TEXT, score INTEGER, captured_at TEXT, delivered INTEGER DEFAULT 0)")
-            chat_id INTEGER,
+    db.execute("CREATE TABLE IF NOT EXISTS group_stats (chat_id INTEGER PRIMARY KEY, chat_title TEXT, total_scanned INTEGER DEFAULT 0, interesting_found INTEGER DEFAULT 0, last_interesting_at TEXT, last_scan_at TEXT)")
-            message_id INTEGER,
-            PRIMARY KEY (chat_id, message_id)
-        )
-    """)
-    db.execute("""
-        CREATE TABLE IF NOT EXISTS interesting (
-            id INTEGER PRIMARY KEY AUTOINCREMENT,
-            chat_id INTEGER,
-            chat_title TEXT,
-            message_id INTEGER,
-            sender TEXT,
-            text TEXT,
-            score INTEGER,
-            captured_at TEXT,
-            delivered INTEGER DEFAULT 0
-        )
-    """)
     db.commit()
     return db
 
 
-async def scan_and_report(dry_run: bool = False) -> list:
+def update_group_stats(db, chat_id: int, chat_title: str, found_interesting: int, now: str):
-    """Scan dialogs for interesting messages. Returns list of interesting items."""
+    cur = db.execute("SELECT total_scanned, interesting_found FROM group_stats WHERE chat_id = ?", (chat_id,))
-    db = init_db()
+    row = cur.fetchone()
-    processed_ids = set()
+    if row:
-    for row in db.execute("SELECT chat_id, message_id FROM processed"):
+        db.execute(
-        processed_ids.add((row[0], row[1]))
+            "UPDATE group_stats SET total_scanned = total_scanned + 1, interesting_found = interesting_found + ?, "
+            "last_interesting_at = CASE WHEN ? > 0 THEN ? ELSE last_interesting_at END, last_scan_at = ? WHERE chat_id = ?",
+            (found_interesting, found_interesting, now if found_interesting > 0 else None, now, chat_id)
+        )
+    else:
+        db.execute(
+            "INSERT INTO group_stats (chat_id, chat_title, total_scanned, interesting_found, last_interesting_at, last_scan_at) "
+            "VALUES (?, ?, 1, ?, ?, ?)",
+            (chat_id, chat_title, found_interesting, now if found_interesting > 0 else None, now)
+        )
 
+
+async def scan_and_report(dry_run: bool = False) -> list:
+    db = init_db()
     c = Client("pyro_session", API_ID, API_HASH, workdir=WORKDIR)
     await c.start()
     me = await c.get_me()
@@ -156,78 +121,104 @@ async def scan_and_report(dry_run: bool = False) -> list:
     interesting = []
     total_read = 0
     total_skipped = 0
+    total_groups = 0
+    now = datetime.now(timezone.utc).isoformat()
 
+    # Collect all eligible dialogs
+    candidates = []
     async for dialog in c.get_dialogs():
         chat = dialog.chat
         chat_id = chat.id
         chat_title = chat.title or chat.first_name or "Unknown"
-
         if should_ignore(chat_title):
             continue
+        candidates.append((chat_id, chat_title))
 
-        if not dialog.unread_messages_count:
+    log.info(f"Total eligible groups: {len(candidates)}")
-            continue
 
-        # Read recent messages (up to 50 per chat)
+    # Sort by last_scan_at (NULLs first) to rotate
-        async for msg in c.get_chat_history(chat_id, limit=50):
+    sorted_candidates = []
-            if (chat_id, msg.id) in processed_ids:
+    for cid, title in candidates:
-                continue
+        cur = db.execute("SELECT last_scan_at FROM group_stats WHERE chat_id = ?", (cid,))
+        row = cur.fetchone()
+        if row and row[0]:
+            sorted_candidates.append((cid, title, row[0]))
+        else:
+            sorted_candidates.append((cid, title, ""))
+    sorted_candidates.sort(key=lambda x: (x[2] == "", x[2]))
+    sorted_candidates = sorted_candidates[:GROUPS_PER_CYCLE]
+
+    log.info(f"Scanning {len(sorted_candidates)} groups this cycle")
+
+    # Clear processed entries for these groups so we get fresh messages
+    group_ids = [cid for cid, _, _ in sorted_candidates]
+    db.execute("DELETE FROM processed WHERE chat_id IN ({})".format(
+        ",".join("?" * len(group_ids))
+    ), group_ids)
+    db.commit()
+
+    for chat_id, chat_title, _ in sorted_candidates:
+        total_groups += 1
+        group_found = 0
+
+        async for msg in c.get_chat_history(chat_id, limit=MSGS_PER_GROUP):
             total_read += 1
-
             score = score_message(msg)
+
             if score >= THRESHOLD:
                 sender = None
                 if msg.from_user:
                     sender = msg.from_user.first_name or msg.from_user.username or str(msg.from_user.id)
                 text = (msg.text or msg.caption or "")[:2000]
-                
                 item = {
-                    "chat_id": chat_id,
+                    "chat_id": chat_id, "chat_title": chat_title, "message_id": msg.id,
-                    "chat_title": chat_title,
+                    "sender": sender, "text": text, "score": score,
-                    "message_id": msg.id,
-                    "sender": sender,
-                    "text": text,
-                    "score": score,
                     "has_media": bool(msg.photo or msg.video or msg.document),
                     "url": f"https://t.me/c/{str(chat_id).replace('-100', '')}/{msg.id}" if chat_id < 0 else None,
                 }
                 interesting.append(item)
-
+                group_found += 1
-                # Save to DB
                 db.execute(
                     "INSERT OR IGNORE INTO interesting (chat_id, chat_title, message_id, sender, text, score, captured_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
-                    (chat_id, chat_title, msg.id, sender, text, score, datetime.now(timezone.utc).isoformat()),
+                    (chat_id, chat_title, msg.id, sender, text, score, now),
                 )
             else:
                 total_skipped += 1
 
-            # Mark as processed
+        update_group_stats(db, chat_id, chat_title, group_found, now)
-            db.execute("INSERT OR IGNORE INTO processed (chat_id, message_id) VALUES (?, ?)", (chat_id, msg.id))
-
         db.commit()
 
     await c.stop()
     db.close()
 
-    log.info(f"Read {total_read} messages, skipped {total_skipped}, found {len(interesting)} interesting")
+    log.info(f"Scanned {total_groups} groups, read {total_read} msgs, found {len(interesting)} interesting")
 
     if not dry_run and interesting:
         for item in interesting:
             _print_item(item)
 
+    # Group quality summary
+    print(f"\n{'='*50}")
+    print(f"📊 GROUP QUALITY SUMMARY")
+    print(f"{'='*50}")
+    db2 = sqlite3.connect(DB_PATH)
+    rows = db2.execute(
+        "SELECT chat_title, total_scanned, interesting_found, last_interesting_at FROM group_stats ORDER BY interesting_found DESC LIMIT 50"
+    ).fetchall()
+    for title, scanned, found, last_interest in rows:
+        status = "🟢" if found >= 1 else "🟡" if scanned >= 3 else "⚪"
+        print(f"{status} {title} — {scanned} scans, {found} interesting, last signal: {last_interest or 'never'}")
+    db2.close()
+
     return interesting
 
 
 def _print_item(item: dict):
-    """Print an interesting item for delivery."""
     print(f"\n{'='*50}")
     print(f"📌 {item['chat_title']} [{item['score']}pts]")
-    if item['sender']:
+    if item['sender']: print(f"👤 {item['sender']}")
-        print(f"👤 {item['sender']}")
+    if item['has_media']: print(f"📎 [has media]")
-    if item['has_media']:
+    if item['url']: print(f"🔗 {item['url']}")
-        print(f"📎 [has media]")
-    if item['url']:
-        print(f"🔗 {item['url']}")
     print(f"\n{item['text'][:500]}")
     print(f"{'='*50}")