sweeper.py

#!/usr/bin/env python
"""Telegram spam sweeper.

Scans recent unread dialogs; stranger DMs and stranger-added groups that
score as spam (see heuristics.py) are reported to Telegram, blocked, and
deleted. Every flagged chat is fully logged to sweeper.log BEFORE any
destructive action.

Usage:
  python sweeper.py --login      one-time interactive login (run by hand)
  python sweeper.py --dry-run    log what would be removed, touch nothing
  python sweeper.py              live run (for cron)
"""
import argparse
import asyncio
import logging
import os
import sys
import time

from dotenv import load_dotenv
from telethon import TelegramClient, functions
from telethon.errors import FloodWaitError

from heuristics import SPAM_THRESHOLD, classify_dm, classify_group

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
LOCK_PATH = os.path.join(BASE_DIR, "sweeper.lock")
LOG_PATH = os.path.join(BASE_DIR, "sweeper.log")
SESSION = os.path.join(BASE_DIR, "sweeper")
LOCK_STALE_SECONDS = 600
DIALOG_SCAN_LIMIT = 100   # spam lands in recent dialogs; keeps runs fast
MESSAGE_FETCH_LIMIT = 20
TELEGRAM_SERVICE_IDS = {777000, 42777}  # official Telegram service accounts

log = logging.getLogger("sweeper")


def setup_logging():
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
        handlers=[logging.FileHandler(LOG_PATH), logging.StreamHandler()],
    )


def acquire_lock():
    """Best-effort single-instance lock; returns False if another run is live."""
    if os.path.exists(LOCK_PATH):
        if time.time() - os.path.getmtime(LOCK_PATH) < LOCK_STALE_SECONDS:
            return False
        os.unlink(LOCK_PATH)  # stale lock from a crashed run
    fd = os.open(LOCK_PATH, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
    os.write(fd, str(os.getpid()).encode())
    os.close(fd)
    return True


def release_lock():
    try:
        os.unlink(LOCK_PATH)
    except FileNotFoundError:
        pass


def log_flagged(kind, name, ident, verdict, texts):
    log.info(
        "FLAGGED %s: %r (id=%s) score=%s reasons=%s",
        kind, name, ident, verdict.score, "; ".join(verdict.reasons),
    )
    for text in texts:
        if text:
            log.info("  message: %r", text[:1000])


async def handle_dm(client, dialog, dry_run):
    user = dialog.entity
    if user.is_self or user.id in TELEGRAM_SERVICE_IDS:
        log.debug("  clear: self/telegram-service account")
        return False

    messages = await client.get_messages(user, limit=MESSAGE_FETCH_LIMIT)
    has_outgoing = any(m.out for m in messages)
    incoming = [m for m in messages if not m.out]
    texts = [m.message for m in incoming if m.message]
    display_name = " ".join(filter(None, [user.first_name, user.last_name]))

    verdict = classify_dm(
        is_contact=bool(user.contact),
        has_outgoing=has_outgoing,
        texts=texts,
        sender_verified=bool(user.verified),
        sender_has_username=bool(user.username),
        sender_name=display_name,
        media_only=bool(incoming) and not texts,
    )
    if not verdict.is_spam:
        log.debug(
            "  clear: score=%s (threshold %s) reasons=[%s], %s msg(s) checked",
            verdict.score, SPAM_THRESHOLD, "; ".join(verdict.reasons), len(texts),
        )
        for text in texts:
            log.debug("    msg: %r", text[:200])
        return False

    name = display_name or user.username
    log_flagged("DM", name, user.id, verdict, texts)
    if dry_run:
        log.info("  dry-run: would report+block+delete")
        return True

    await client(functions.messages.ReportSpamRequest(peer=user))
    await client(functions.contacts.BlockRequest(id=user))
    await client.delete_dialog(user)
    log.info("  reported, blocked, deleted")
    return True


async def inviter_is_contact(client, entity):
    """True if the person who added us to this channel/group is a contact."""
    try:
        result = await client(
            functions.channels.GetParticipantRequest(entity, "me")
        )
        inviter_id = getattr(result.participant, "inviter_id", None)
        if inviter_id:
            inviter = await client.get_entity(inviter_id)
            return bool(getattr(inviter, "contact", False))
    except Exception:
        pass  # small legacy groups / missing perms: fall through to scoring
    return False


async def handle_group(client, dialog, dry_run):
    entity = dialog.entity
    if getattr(entity, "creator", False):
        log.debug("  clear: group created by me")
        return False

    messages = await client.get_messages(entity, limit=MESSAGE_FETCH_LIMIT)
    texts = [m.message for m in messages if m.message]

    verdict = classify_group(
        title=dialog.name,
        texts=texts,
        inviter_is_contact=await inviter_is_contact(client, entity),
        i_am_creator=False,
    )
    if not verdict.is_spam:
        log.debug(
            "  clear: score=%s (threshold %s) reasons=[%s], %s msg(s) checked",
            verdict.score, SPAM_THRESHOLD, "; ".join(verdict.reasons), len(texts),
        )
        return False

    log_flagged("GROUP", dialog.name, entity.id, verdict, texts)
    if dry_run:
        log.info("  dry-run: would report+leave+delete")
        return True

    try:
        await client(functions.messages.ReportSpamRequest(peer=entity))
    except Exception as e:
        log.warning("  report failed (%s), still leaving", e)
    await client.delete_dialog(entity)  # leaves and removes the dialog
    log.info("  reported and left")
    return True


def dialog_is_unread(dialog):
    """Telethon's Dialog wrapper exposes unread_count but not unread_mark;
    the manual 'marked unread' flag lives on the raw TL object underneath."""
    return bool(
        dialog.unread_count or getattr(dialog.dialog, "unread_mark", False)
    )


def should_scan(dialog, include_read):
    return include_read or dialog_is_unread(dialog)


async def sweep(client, dry_run, include_read=False):
    scanned = eligible = swept = 0
    async for dialog in client.iter_dialogs(limit=DIALOG_SCAN_LIMIT):
        scanned += 1
        kind = "dm" if dialog.is_user else "group/channel"
        if not should_scan(dialog, include_read):
            log.debug("skip (already read): %s %r", kind, dialog.name)
            continue
        eligible += 1
        log.debug(
            "scanning %s: %r (id=%s, unread=%s)",
            kind, dialog.name, dialog.id, dialog.unread_count,
        )
        try:
            if dialog.is_user:
                swept += await handle_dm(client, dialog, dry_run)
            elif dialog.is_group or dialog.is_channel:
                swept += await handle_group(client, dialog, dry_run)
        except FloodWaitError as e:
            log.warning(
                "FloodWait %ss from Telegram; stopping, next run resumes", e.seconds
            )
            break
    log.info(
        "run complete: %s dialog(s) seen, %s scanned%s, %s flagged%s",
        scanned, eligible,
        " (including read)" if include_read else " (unread only)",
        swept, " [dry-run]" if dry_run else "",
    )
    return swept


async def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--login", action="store_true", help="interactive login")
    parser.add_argument("--dry-run", action="store_true", help="log only, no actions")
    parser.add_argument(
        "--verbose", "-v", action="store_true",
        help="log every dialog examined and why it was skipped or cleared",
    )
    parser.add_argument(
        "--include-read", action="store_true",
        help="also scan chats already marked read (for one-off backlog sweeps)",
    )
    args = parser.parse_args()

    if args.verbose:
        log.setLevel(logging.DEBUG)

    load_dotenv(os.path.join(BASE_DIR, ".env"))
    api_id, api_hash = os.getenv("TG_API_ID"), os.getenv("TG_API_HASH")
    if not api_id or not api_hash:
        log.error("TG_API_ID / TG_API_HASH missing — copy .env.example to .env")
        sys.exit(1)

    client = TelegramClient(SESSION, int(api_id), api_hash)

    if args.login:
        await client.start()  # prompts for phone + code interactively
        me = await client.get_me()
        log.info("logged in as %s (id=%s); session saved", me.first_name, me.id)
        await client.disconnect()
        return

    await client.connect()
    if not await client.is_user_authorized():
        log.error("not logged in — run: python sweeper.py --login")
        sys.exit(1)

    await sweep(client, args.dry_run, include_read=args.include_read)
    await client.disconnect()


if __name__ == "__main__":
    setup_logging()
    if not acquire_lock():
        sys.exit(0)  # previous run still going; this minute's tick skips
    try:
        asyncio.run(main())
    finally:
        release_lock()