""" 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. """ 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 # @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 # Keywords that signal interesting content 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", "产品", "产品发布", ] # Groups to always ignore (IDs or title substrings) 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 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 return score def should_ignore(chat_title: str) -> bool: """Check if a chat should be ignored entirely.""" if not chat_title: return False for pat in IGNORE_PATTERNS: 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.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])) 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 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 if not dialog.unread_messages_count: continue # 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 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) # 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()), ) else: total_skipped += 1 # Mark as processed 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") if not dry_run and interesting: for item in interesting: _print_item(item) 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']}") 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))