Kratt / app /rules.py
Kratt Deploy
Deploy Kratt backend
f66cd72
Raw
History Blame Contribute Delete
2.17 kB
"""
The "Rules" half of the flowchart's "Rules + trained BERT" step. BERT decides
genuine-vs-bot; these rules subdivide the bot comments into ads_spam / copy_paste,
and provide a bot heuristic for the fallback path when BERT isn't loaded.
"""
import re
from collections import Counter
# ads/spam: links, contact handles, crypto/earnings pitches, self-promo.
_SPAM_PATTERNS = [
r"https?://",
r"\bwww\.",
r"\bt\.me/",
r"\b(telegram|whats?app|onlyfans)\b",
r"\b(free\s+crypto|crypto\s+giveaway|bitcoin|forex|binary\s+option)\b",
r"\b(earn|make|win)\s*\$?\s*\d",
r"\b(promo|coupon|discount)\s*code\b",
r"\bcheck\s+(out\s+)?my\s+(page|channel|profile|bio|link)\b",
r"\b(dm|inbox|message)\s+me\b",
r"\bsub(scribe)?\s+(to\s+)?my\b",
]
SPAM_RE = re.compile("|".join(_SPAM_PATTERNS), re.IGNORECASE)
# generic/low-effort filler used only in the BERT-unavailable fallback.
_LOW_EFFORT_PATTERNS = [
r"^\W*first\W*$",
r"who'?s\s+watching",
r"^\W*(nice|great|good|love|amazing|wow|lol|lmao|cool)\W*$",
]
LOW_EFFORT_RE = re.compile("|".join(_LOW_EFFORT_PATTERNS), re.IGNORECASE)
_EMOJI_OR_SYMBOL = re.compile(r"[\w]", re.UNICODE) # any word char -> has "content"
def is_ads_spam(raw_text: str) -> bool:
return bool(SPAM_RE.search(str(raw_text)))
def normalize_for_dup(text: str) -> str:
"""Collapse whitespace + lowercase so near-identical comments cluster."""
return re.sub(r"\s+", " ", str(text)).strip().lower()
def duplicate_counts(texts) -> Counter:
return Counter(normalize_for_dup(t) for t in texts)
def is_copy_paste(text: str, counts: Counter, threshold: int = 2) -> bool:
# near-duplicate WITHIN this video's fetched comments (the cross-video signal
# from training isn't available at request time — we only have one video).
return counts[normalize_for_dup(text)] >= threshold
def is_low_effort_fallback(text: str) -> bool:
s = str(text).strip()
if not _EMOJI_OR_SYMBOL.search(s):
return True # emoji/punctuation only
if len(s) <= 4:
return True # extremely short
return bool(LOW_EFFORT_RE.search(s))