heuristics.py

"""Pure spam-scoring heuristics for the Telegram spam sweeper.

No Telethon or I/O here — everything is a pure function so it can be
unit-tested and tuned from the sweeper.log without touching the wiring.
"""
import re
from dataclasses import dataclass, field

SPAM_THRESHOLD = 3

URL_RE = re.compile(
    r"(https?://\S+|www\.\S+|t\.me/\S+|telegram\.me/\S+)", re.IGNORECASE
)
PHONE_RE = re.compile(r"\+?\d[\d\s().-]{7,}\d")

KEYWORD_CATEGORIES = {
    "keyword:crypto-investment": [
        "usdt", "crypto", "bitcoin", "btc", "eth ", "forex", "trading signal",
        "passive income", "guaranteed profit", "daily profit", "invest",
        "investment", "returns", "profit", "earn money", "make money",
        "financial freedom", "binary option", "stock tips", "airdrop",
    ],
    "keyword:premium-gift-bait": [
        "telegram premium", "premium gift", "free premium", "gift subscription",
        "you have received a gift", "claim your", "you won", "congratulations",
        "lucky winner", "prize",
    ],
    "keyword:platform-redirect": [
        "whatsapp", "signal app", "text me on", "message me on", "add me on",
        "reach me on",
    ],
    "keyword:generic-spam": [
        "dear friend", "hello dear", "job offer", "work from home",
        "part-time job", "salary per day", "casino", "betting", "18+",
        "adult content", "hot photos",
    ],
}

_EMOJI_RE = re.compile(
    "[\U0001F300-\U0001FAFF\U00002600-\U000027BF\U0001F900-\U0001F9FF]"
)


@dataclass
class Verdict:
    is_spam: bool
    score: int
    reasons: list = field(default_factory=list)


def score_text(text):
    """Score one message body. Returns (score, reasons)."""
    if not text:
        return 0, []
    score = 0
    reasons = []
    lower = text.lower()

    if URL_RE.search(text):
        score += 2
        reasons.append("url/invite-link")

    for category, keywords in KEYWORD_CATEGORIES.items():
        hits = [kw for kw in keywords if kw in lower]
        if hits:
            score += 2
            reasons.append(f"{category} ({', '.join(hits[:3])})")

    if PHONE_RE.search(text):
        score += 1
        reasons.append("phone-number")

    emoji_count = len(_EMOJI_RE.findall(text))
    if emoji_count >= 5:
        score += 1
        reasons.append(f"emoji-blast ({emoji_count})")

    return score, reasons


def _score_texts(texts):
    """Combined score for a chat: union of signals across its messages."""
    total = 0
    all_reasons = []
    seen = set()
    for text in texts or []:
        score, reasons = score_text(text)
        for reason in reasons:
            key = reason.split(" ")[0]
            if key not in seen:
                seen.add(key)
                all_reasons.append(reason)
        total = max(total, score)
    # Re-derive the union score from distinct signal kinds so spam split
    # across several short messages scores like one combined blast.
    union_score = 0
    for reason in all_reasons:
        if reason.startswith("url") or reason.startswith("keyword"):
            union_score += 2
        else:
            union_score += 1
    return max(total, union_score), all_reasons


def classify_dm(
    is_contact,
    has_outgoing,
    texts,
    sender_verified=False,
    sender_has_username=True,
    sender_name=None,
    media_only=False,
):
    """Classify a private chat with a sender. Trusted senders short-circuit.

    Policy: a non-contact, unverified sender with no reply history is spam
    on its own — the stranger baseline alone meets the threshold. Content
    and profile signals still accumulate so the log shows how spammy a
    flagged chat looked beyond mere strangerhood. `sender_name` is the
    display name; pass None when unknown to skip the name checks (empty
    string means an actually blank name).
    """
    if is_contact or has_outgoing or sender_verified:
        return Verdict(False, 0, ["trusted: contact/mutual-history/verified"])

    score, reasons = _score_texts(texts)

    score += SPAM_THRESHOLD
    reasons.append("stranger (not a contact, unverified, no history)")

    if not sender_has_username:
        score += 1
        reasons.append("no-username")

    if sender_name is not None:
        if not sender_name.strip():
            score += 1
            reasons.append("empty name")
        elif _EMOJI_RE.search(sender_name):
            score += 1
            reasons.append("emoji-decorated name")

    if media_only:
        score += 1
        reasons.append("media-only opener")

    return Verdict(score >= SPAM_THRESHOLD, score, reasons)


def classify_group(title, texts, inviter_is_contact, i_am_creator):
    """Classify a group/channel the account is in."""
    if i_am_creator or inviter_is_contact:
        return Verdict(False, 0, ["trusted: own-group/contact-invite"])

    score, reasons = _score_texts([title] + list(texts or []))
    return Verdict(score >= SPAM_THRESHOLD, score, reasons)