diff --git a/backend/pyro_session.session b/backend/pyro_session.session new file mode 100644 index 0000000..435ec40 Binary files /dev/null and b/backend/pyro_session.session differ diff --git a/backend/tg_filter.py b/backend/tg_filter.py index 5aa9818..2d3d128 100644 --- a/backend/tg_filter.py +++ b/backend/tg_filter.py @@ -1,9 +1,7 @@ """ Telegram Group Filter — scours 100+ groups for interesting messages. -Runs as: t-youtube venv + Pyrogram session. - -Scoring: mentions, links, code blocks, media, keywords → score ≥ threshold → saved. -Delivers to: hermes_telegram_bot or stdout. +Rotates through all groups across cron cycles to avoid rate limits. +Resets processed tracking per batch so each cycle gets fresh messages. """ 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_KEYWORD = 15 # Interesting keyword match -SCORE_LINK = 10 # Contains URL -SCORE_MEDIA = 8 # Photo/video/file -SCORE_CODE = 12 # Code block -SCORE_LONG_TEXT = 5 # ≥200 chars of substance -THRESHOLD = 20 # Minimum score to forward +SCORE_MENTION = 30 +SCORE_KEYWORD = 15 +SCORE_LINK = 10 +SCORE_MEDIA = 8 +SCORE_CODE = 12 +SCORE_LONG_TEXT = 5 +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"): - score += SCORE_MENTION - if e.type == "url": - score += SCORE_LINK - if e.type == "code" or e.type == "pre": - 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 + 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 - - # Long substantive text - 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 - + 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: - """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: - return True + if pat in chat_title: 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(""" - 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 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 -async def scan_and_report(dry_run: bool = False) -> list: - """Scan dialogs for interesting messages. Returns list of interesting items.""" - db = init_db() - processed_ids = set() - for row in db.execute("SELECT chat_id, message_id FROM processed"): - processed_ids.add((row[0], row[1])) +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() @@ -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: - continue + log.info(f"Total eligible groups: {len(candidates)}") - # Read recent messages (up to 50 per chat) - async for msg in c.get_chat_history(chat_id, limit=50): - if (chat_id, msg.id) in processed_ids: - continue + # 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, + "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) - - # Save to DB + 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, datetime.now(timezone.utc).isoformat()), + (chat_id, chat_title, msg.id, sender, text, score, now), ) else: total_skipped += 1 - # Mark as processed - db.execute("INSERT OR IGNORE INTO processed (chat_id, message_id) VALUES (?, ?)", (chat_id, msg.id)) - + update_group_stats(db, chat_id, chat_title, group_found, now) 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']: - print(f"👤 {item['sender']}") - if item['has_media']: - print(f"📎 [has media]") - if item['url']: - print(f"🔗 {item['url']}") + 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}")