t-youtube/backend/tg_filter.py

"""
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))