""" Telegram Group Filter — scours 100+ groups for interesting messages. Rotates through all groups across cron cycles to avoid rate limits. Resets processed tracking per batch so each cycle gets fresh messages. """ import asyncio import os import re import json import sqlite3 import logging from datetime import datetime, timezone, timedelta from pyrogram import Client from pyrogram.types import Message from typing import Optional logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S", ) log = logging.getLogger("tg_filter") # === CONFIG === PHONE = open("/tmp/tg_phone.txt").read().strip() API_ID = 2040 API_HASH = "b18441a1ff607e10a989891a5462e627" WORKDIR = "/Users/jimmyg" DB_PATH = os.path.join(WORKDIR, ".hermes", "tg_filter.db") # Scoring SCORE_MENTION = 30 SCORE_KEYWORD = 15 SCORE_LINK = 10 SCORE_MEDIA = 8 SCORE_CODE = 12 SCORE_LONG_TEXT = 5 THRESHOLD = 40 # 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", "self-host", "homelab", "server", "deploy", "ci/cd", "gitea", "git", "api", "automation", "script", "python", "rust", "go", "typescript", "svelte", "react", "nextjs", "fastapi", "sveltekit", "tutorial", "guide", "how to", "free", "open source", "github", "security", "vulnerability", "cve", "patch", "update", "opportunity", "合作", "项目", "招聘", "job", "remote", "startup", "saas", "mvp", "产品", "产品发布", ] IGNORE_PATTERNS = ["合租社群", "黄油派对"] def score_message(msg: Message) -> int: score = 0 text = msg.text or msg.caption or "" if msg.entities: for e in msg.entities: if e.type in ("mention", "text_mention"): score += SCORE_MENTION if e.type == "url": score += SCORE_LINK if e.type == "code" or e.type == "pre": score += SCORE_CODE if msg.photo or msg.video or msg.document or msg.audio: score += SCORE_MEDIA if re.search(r'https?://[^\s]+', text): score += SCORE_LINK text_lower = text.lower() for kw in INTERESTING_KEYWORDS: if kw in text_lower: score += SCORE_KEYWORD if len(text.strip()) >= 200: score += SCORE_LONG_TEXT if msg.forward_from or msg.forward_from_chat: score += 5 if msg.reply_to_message_id: score += 3 return score def should_ignore(chat_title: str) -> bool: if not chat_title: return False for pat in IGNORE_PATTERNS: if pat in chat_title: return True return False def init_db(): os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) db = sqlite3.connect(DB_PATH) db.execute("CREATE TABLE IF NOT EXISTS processed (chat_id INTEGER, 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.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)") db.commit() return db def update_group_stats(db, chat_id: int, chat_title: str, found_interesting: int, now: str): cur = db.execute("SELECT total_scanned, interesting_found FROM group_stats WHERE chat_id = ?", (chat_id,)) row = cur.fetchone() if row: db.execute( "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() log.info(f"Scanned as: {me.first_name}") 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)) log.info(f"Total eligible groups: {len(candidates)}") # Sort by last_scan_at (NULLs first) to rotate sorted_candidates = [] for cid, title in candidates: 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_title": chat_title, "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 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, now), ) else: total_skipped += 1 update_group_stats(db, chat_id, chat_title, group_found, now) db.commit() await c.stop() db.close() 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(f"\n{'='*50}") print(f"📌 {item['chat_title']} [{item['score']}pts]") if item['sender']: print(f"👤 {item['sender']}") if item['has_media']: print(f"📎 [has media]") if item['url']: print(f"🔗 {item['url']}") print(f"\n{item['text'][:500]}") print(f"{'='*50}") if __name__ == "__main__": import sys dry_run = "--dry-run" in sys.argv asyncio.run(scan_and_report(dry_run=dry_run))