"""中文垃圾广告检测引擎 策略： 1. 先做本地规则评分（成熟中文广告/黑灰产话术体系） 2. 高置信垃圾直接拦截 3. 明确正常消息直接放行 4. 低置信/模糊样本交给 AI 兜底 核心原则：短消息不能无条件放行；必须先识别广告意图。 """ import logging import re from dataclasses import dataclass from pathlib import Path import httpx from knowledge import OLLAMA_URL, OLLAMA_MODEL logger = logging.getLogger("spam_guard") @dataclass(frozen=True) class Rule: name: str pattern: str score: int ROOT_DIR = Path(__file__).resolve().parent.parent RULES_DIR = ROOT_DIR / "rules" def load_keyword_rules(path: Path, default_score: int = 45) -> list[Rule]: """加载外部关键词词库：词条 或 词条|分数。""" rules: list[Rule] = [] if not path.exists(): return rules for line in path.read_text(encoding="utf-8").splitlines(): raw = line.strip() if not raw or raw.startswith("#"): continue if "|" in raw: word, score_text = raw.rsplit("|", 1) try: score = int(score_text.strip()) except ValueError: word, score = raw, default_score else: word, score = raw, default_score word = word.strip() if word: rules.append(Rule(f"外部关键词:{word}", re.escape(word), score)) return rules def load_regex_rules(path: Path, default_score: int = 70) -> list[Rule]: """加载外部正则库：正则 或 正则|分数。""" rules: list[Rule] = [] if not path.exists(): return rules for i, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1): raw = line.strip() if not raw or raw.startswith("#"): continue if "|" in raw: pattern, score_text = raw.rsplit("|", 1) try: score = int(score_text.strip()) except ValueError: pattern, score = raw, default_score else: pattern, score = raw, default_score try: re.compile(pattern) rules.append(Rule(f"外部正则:{i}", pattern, score)) except re.error as e: logger.warning(f"⚠️ 跳过无效外部正则 {path}:{i}: {e}") return rules def load_adblock_domains(path: Path) -> set[str]: """加载 Adblock 域名过滤库。仅用于消息中 URL/域名匹配，不参与普通关键词扫描。""" if not path.exists(): return set() domains: set[str] = set() for line in path.read_text(encoding="utf-8", errors="ignore").splitlines(): raw = line.strip().lower() if not raw or raw.startswith("#"): continue domains.add(raw.lstrip(".")) return domains EXTERNAL_KEYWORD_RULES = load_keyword_rules(RULES_DIR / "spam_keywords.txt") EXTERNAL_REGEX_RULES = load_regex_rules(RULES_DIR / "spam_regex.txt") ADBLOCK_DOMAINS = load_adblock_domains(RULES_DIR / "adblock_domains.txt") logger.info(f"🧩 已加载外部垃圾规则：关键词 {len(EXTERNAL_KEYWORD_RULES)} 条，正则 {len(EXTERNAL_REGEX_RULES)} 条，Adblock 域名 {len(ADBLOCK_DOMAINS)} 条") def reload_external_rules() -> dict: """重新加载外部关键词/正则/Adblock 域名规则，用于 Web 写规则或导入后热更新。""" global EXTERNAL_KEYWORD_RULES, EXTERNAL_REGEX_RULES, ADBLOCK_DOMAINS EXTERNAL_KEYWORD_RULES = load_keyword_rules(RULES_DIR / "spam_keywords.txt") EXTERNAL_REGEX_RULES = load_regex_rules(RULES_DIR / "spam_regex.txt") ADBLOCK_DOMAINS = load_adblock_domains(RULES_DIR / "adblock_domains.txt") logger.info(f"🧩 已热加载外部垃圾规则：关键词 {len(EXTERNAL_KEYWORD_RULES)} 条，正则 {len(EXTERNAL_REGEX_RULES)} 条，Adblock 域名 {len(ADBLOCK_DOMAINS)} 条") return {"keywords": len(EXTERNAL_KEYWORD_RULES), "regex": len(EXTERNAL_REGEX_RULES), "adblock_domains": len(ADBLOCK_DOMAINS)} # 高危黑产：单项命中即可高度可疑或直接垃圾 HARD_RULES = [ Rule("博彩赌博", r"菠菜|赌博|博彩|赌场|棋牌|彩票|开奖|中奖|赔率|下注|百家乐|时时彩|六合彩", 100), Rule("色情引流", r"色情|约炮|外围|楼凤|一夜情|裸聊|成人视频|萝莉|上门服务", 100), Rule("诈骗/传销/资金盘", r"刷单|传销|杀猪盘|资金盘|庞氏|返利盘|跑分|洗钱|代收款", 100), Rule("非法交易", r"办证|假证|代开发票|买卖账号|出售账号|卖号|接码|猫池", 100), Rule("Telegram 引流", r"(?:t\.me|telegram\.me|tg://|飞机群).{0,40}(?:join|group|channel|\+|进群|频道)", 90), ] # 广告意图/黑灰产常见话术：评分组合 DOMAIN_TEXT_RE = re.compile(r"(?i)(?:https?://|www\.)?([a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?(?:\.[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)+)") COMMON_TEXT_DOMAINS = {"ptools.fun", "github.com", "gitee.com", "telegram.org", "t.me", "qq.com", "weixin.qq.com"} def extract_domains_from_text(text: str) -> set[str]: domains: set[str] = set() for m in DOMAIN_TEXT_RE.finditer(text or ""): d = m.group(1).lower().strip(".-_") if d and d not in COMMON_TEXT_DOMAINS: domains.add(d) return domains def match_adblock_domains(text: str) -> list[Rule]: """只在文本出现 URL/域名时匹配广告域名，降低普通聊天误杀。""" if not ADBLOCK_DOMAINS or not re.search(r"(?i)(https?://|www\.|[a-z0-9-]+\.[a-z]{2,})", text or ""): return [] hits: list[Rule] = [] for domain in extract_domains_from_text(text): parts = domain.split(".") candidates = [".".join(parts[i:]) for i in range(max(0, len(parts) - 4), len(parts) - 1)] candidates.insert(0, domain) for c in candidates: if c in ADBLOCK_DOMAINS: hits.append(Rule(f"Adblock域名:{c}", re.escape(c), 80)) break return hits SCORE_RULES = [ # 收益承诺 Rule("收益承诺-日赚", r"(?:日赚|日入|每天赚|一天赚|轻松赚|稳赚|躺赚|副业收入).{0,10}(?:\d{2,}|几百|上千|过千|一千|几千|上万|万元?|k|K|w|W)", 75), Rule("收益承诺-月入", r"(?:月入|月赚|月收益|月收入).{0,10}(?:\d{3,}|过万|上万|几万|万元?|w|W)", 70), Rule("收益承诺-夸张", r"(?:暴富|翻倍|稳赚不赔|保本收益|无风险收益|零风险|高回报|高收益)", 55), # 兼职/刷单/项目 Rule("兼职刷单", r"(?:兼职|副业|日结|周结|在家做|手机操作|无需经验|宝妈|学生党).{0,18}(?:赚钱|收入|结算|工资|项目|单)", 55), Rule("项目引流", r"(?:项目|路子|渠道|资源|教程).{0,12}(?:带你|私聊|加我|看主页|看签名|进群)", 55), Rule("带人赚钱", r"(?:带你赚|带你做|限时带人|带人|手把手带|小白可带|包教包会|一对一带)", 65), Rule("黑U灰产", r"(?:黑\s*[uU]|黑油|黑资|黑钱|黑产|灰产|黑钱包|资金盘|交易所).{0,18}(?:项目|路子|带人|联系|兄弟|稳|最稳)", 80), Rule("限时名额诱导", r"(?:限时|名额有限|先到先得|名额|机会不多).{0,20}(?:联系|私聊|看简介|看简界|简介|主页|项目|带人)", 65), Rule("简介引流", r"(?:看我|看下|点我|主页|简介|简界|签名).{0,10}(?:简介|简界|主页|签名|头像)", 55), # 联系/引流动作 Rule("联系方式-微信", r"(?:加微信|微信号|微[信xX]|vx|v信|薇信|扣扣|QQ|q群|私信|私聊|PM|DM|联系我|找我)", 45), Rule("引导动作", r"(?:看主页|看签名|主页有|签名有|点我头像|点头像|进群|拉群|频道|群内|开户链接)", 45), Rule("链接引流", r"(?:https?://|www\.|\.com|\.cn|\.top|\.xyz|\.cc|短链接|复制链接|点击链接)", 40), # 金融/币圈/贷款 Rule("虚拟币搬砖", r"(?:USDT|U\s?S\s?D\s?T|泰达币|虚拟币|币圈|搬砖|套利).{0,18}(?:赚钱|收益|项目|加我|看主页|进群)", 65), Rule("贷款套现", r"(?:贷款|借款|套现|信用卡|花呗|白条|提额|下款|黑户|征信).{0,18}(?:秒批|当天|无视|包过|渠道)", 60), Rule("投资理财", r"(?:投资|理财|股票|期货|外汇|量化|跟单).{0,18}(?:老师|带单|收益|稳赚|回本|翻倍)", 60), # 营销常见词 Rule("免费福利", r"(?:免费领|免费送|限时领|点击领取|扫码领取|福利|红包|返现|佣金|返佣)", 40), Rule("代理加盟", r"(?:代理|加盟|招商|推广返|分润|招募|招代理|合作共赢)", 45), Rule("豪车炫富", r"(?:换|开|提).{0,8}(?:迈巴赫|宝马|奔驰|保时捷|法拉利|兰博基尼)", 65), Rule("月入收益承诺", r"(?:一个月|每月|月入|月赚|下个月).{0,12}(?:\d{1,4}|几万|几十万).{0,4}(?:w|W|万|个w|个W)", 75), Rule("安全收益承诺", r"(?:安全|稳赚|保本|不要本钱|不要本|零成本).{0,18}(?:赚|收益|月入|月赚|开奔驰|提车)", 70), ] # 组合加权：成熟广告识别关键在“组合意图”，不是单词命中 COMBO_RULES = [ Rule("收益+金额", r"(?:赚|收入|收益|工资|佣金|返现|日结|副业|一天|随便|随随便便).{0,12}(?:\d{3,}|几百|上千|过千|几千|上万|万元?|万\s*\+|k|K|w|W)", 55), Rule("收益+引流", r"(?:赚钱|收益|收入|副业|项目|项木|路子|黑\s*[uU]|交易所).{0,24}(?:加我|私聊|看主页|看签名|看简介|看简界|进群|联系我|直接联系|微信|vx|QQ)", 60), Rule("金额+引流", r"(?:\d{3,}|几百|上千|过千|几千|上万|万元?|万\s*\+|k|K|w|W).{0,24}(?:加我|私聊|看主页|看签名|看简介|看简界|进群|联系我|直接联系|微信|vx|QQ)", 55), Rule("灰产收益组合", r"(?:黑\s*[uU]|交易所|项木|项目).{0,30}(?:带人|联系|稳|最稳).{0,40}(?:一天|日赚|日入|\d+\s*万|\d+\s*w|名额有限|先到先得)", 85), ] NORMAL_RULES = [ Rule("项目技术讨论", r"(?:收割机|harvest|签到|站点|cookie|docker|compose|安装|配置|更新|升级|bug|报错|失败|成功|数据库|postgres|sqlite|redis)", 30), Rule("疑问求助", r"(?:怎么|如何|为什么|能不能|请问|求|帮忙|麻烦|谢谢|这个|那个).{0,30}[?？]?$", 25), Rule("常见短回复", r"^(?:ok|OK|yes|YES|no|NO|收到|好的|可以|不行|对|错|嗯|嗯嗯|是的|不是|谢谢|感谢)$", 60), Rule("纯数字", r"^\d{1,8}$", 30), ] SPAM_PROMPT = """判断以下群聊消息是否为中文垃圾广告/黑灰产推广。 消息内容: {text} 发送者用户名: {username} 发送者昵称: {first_name} 垃圾广告包括：赚钱项目、刷单兼职、日赚/月入承诺、加微信/私聊引流、博彩色情、贷款套现、虚拟币搬砖、黑U/交易所/跑分/灰产项目、投资带单、免费福利诱导、名额有限/先到先得/看简介等引流话术。 正常消息包括：技术讨论、问答、简短确认、正常聊天。 只输出一行：垃圾 或 正常，并简短说明原因。""" def normalize(text: str) -> str: """去零宽字符、统一空白和常见混淆字符。""" t = re.sub(r"[\u200b\u200c\u200d\u2060\u200e\u200f\ufeff\u00a0]", "", text or "") t = re.sub(r"\s+", " ", t).strip() return t def compact(text: str) -> str: """去掉空格，处理 日赚 1000 / 日赚1000 这类绕过。""" return re.sub(r"\s+", "", text) def hit_rules(text: str, rules: list[Rule]) -> list[Rule]: return [r for r in rules if re.search(r.pattern, text, re.IGNORECASE)] USERNAME_RULES = [ Rule("用户名-广告收益", r"(?:日赚|月入|赚钱|兼职|副业|返佣|代理|招商|刷单|项目|福利)", 65), Rule("用户名-灰产金融", r"(?:贷款|套现|下款|黑户|USDT|币圈|搬砖|套利|博彩|菠菜|棋牌)", 75), Rule("用户名-色情引流", r"(?:约炮|裸聊|外围|楼凤|上门|看片|成人视频)", 90), Rule("用户名-联系方式", r"(?:加微|微信|vx|v信|QQ|私聊|看主页|频道|群)", 45), Rule("用户名-收益数字", r"(?:赚|入|佣金|返现).{0,8}(?:\d{3,}|过万|上万|w|W|k|K)", 70), ] def classify_username_spam(username: str = "", first_name: str = "", full_name: str = "") -> tuple[bool, str, int]: """新入群用户的用户名/昵称广告检测。只用本地规则，避免 AI 误判。""" text = " ".join(x for x in [username, first_name, full_name] if x).strip() if not text: return False, "无用户名信息", 0 norm = normalize(text) tight = compact(norm) targets = [norm, tight] hits = [] external_hits = [] for target in targets: hits.extend(hit_rules(target, HARD_RULES)) hits.extend(hit_rules(target, USERNAME_RULES)) external_hits.extend(hit_rules(target, EXTERNAL_KEYWORD_RULES)) external_hits.extend(hit_rules(target, EXTERNAL_REGEX_RULES)) dedup = lambda xs: list({x.name: x for x in xs}.values()) hits = dedup(hits) external_hits = dedup(external_hits) score = sum(r.score for r in hits) + sum(r.score for r in external_hits) if any(r.name.startswith(("博彩", "色情", "诈骗")) or r.name in ("用户名-色情引流", "用户名-灰产金融") for r in hits): names = ",".join(r.name for r in (hits + external_hits)[:5]) return True, f"新用户名称硬规则命中: {names}", max(score, 100) if score >= 70: names = ",".join(r.name for r in (hits + external_hits)[:5]) return True, f"新用户名称广告评分 {score}: {names}", score return False, f"新用户名称未命中高风险规则，评分 {score}", score def local_classify(text: str) -> tuple[bool | None, str, int]: """本地规则分类。返回 (是否垃圾/None需AI, 原因, 分数)。""" norm = normalize(text) tight = compact(norm) targets = [norm, tight] hard_hits = [] score_hits = [] combo_hits = [] external_hits = [] normal_hits = [] for target in targets: hard_hits.extend(hit_rules(target, HARD_RULES)) score_hits.extend(hit_rules(target, SCORE_RULES)) combo_hits.extend(hit_rules(target, COMBO_RULES)) external_hits.extend(hit_rules(target, EXTERNAL_KEYWORD_RULES)) external_hits.extend(hit_rules(target, EXTERNAL_REGEX_RULES)) normal_hits.extend(hit_rules(target, NORMAL_RULES)) external_hits.extend(match_adblock_domains(norm)) # 去重 dedup = lambda xs: list({x.name: x for x in xs}.values()) hard_hits, score_hits, combo_hits, external_hits, normal_hits = map(dedup, [hard_hits, score_hits, combo_hits, external_hits, normal_hits]) if hard_hits: names = ",".join(r.name for r in hard_hits[:3]) return True, f"硬规则命中: {names}", 100 spam_score = sum(r.score for r in score_hits) + sum(r.score for r in combo_hits) + sum(r.score for r in external_hits) normal_score = sum(r.score for r in normal_hits) # 广告组合是核心：即使很短也要先判垃圾 if spam_score >= 75: names = ",".join(r.name for r in (score_hits + combo_hits + external_hits)[:5]) return True, f"广告规则评分 {spam_score}: {names}", spam_score # 纯短消息只有在没有广告分时才放行 if len(norm) <= 8 and spam_score == 0: return False, "短正常消息", -normal_score # 明显正常技术/求助消息，且无明显广告分 if normal_score >= 30 and spam_score < 45: names = ",".join(r.name for r in normal_hits[:3]) return False, f"正常规则命中: {names}", spam_score - normal_score # 长文本或命中弱灰产/引流信号时交给 AI，避免 0 分直接放过混淆广告 weak_ai_signal = re.search(r"黑\s*[uU]|交易所|项木|项目|带人|直\s*接\s*联系|联系|名额|先到先得|简介|简界|主页|日赚|日入|一天|万\s*\+|w\s*\+", norm, re.IGNORECASE) if spam_score == 0 and (weak_ai_signal or len(norm) >= 80): return None, "本地无强规则，进入AI兜底", 0 # 低风险且没有广告信号 if spam_score == 0: return False, "无广告信号", 0 # 模糊样本交给 AI names = ",".join(r.name for r in (score_hits + combo_hits + external_hits)[:5]) return None, f"疑似广告评分 {spam_score}: {names}", spam_score async def ai_classify(text: str, username: str = "", first_name: str = "") -> tuple[bool, str]: prompt = SPAM_PROMPT.format(text=text, username=username, first_name=first_name) try: async with httpx.AsyncClient(timeout=30, proxy=None) as client: resp = await client.post( f"{OLLAMA_URL}/api/chat", json={ "model": OLLAMA_MODEL, "messages": [{"role": "user", "content": prompt}], "stream": False, "options": {"num_predict": 80, "temperature": 0.1}, "think": False, }, ) resp.raise_for_status() reply = resp.json()["message"]["content"].strip() logger.info(f"🤖 垃圾检测 AI 回复: {reply[:100]}") if re.search(r"垃圾|广告|诈骗|菠菜|赌博|色情", reply, re.IGNORECASE): return True, f"AI判定: {reply}" if re.search(r"正常|非垃圾|不是垃圾", reply, re.IGNORECASE): return False, f"AI判定: {reply}" return False, f"AI判定不明确: {reply}" except Exception as e: logger.error(f"❌ Ollama 调用失败: {e}") return False, f"AI异常: {e}" async def detect(text: str, username: str = "", first_name: str = "") -> tuple[bool, str]: """返回 (is_spam, reason)。""" decision, reason, score = local_classify(text) if decision is not None: return decision, reason # AI 不可用时，中等以上广告分直接兜底拦截；低分放行 ai_decision, ai_reason = await ai_classify(text, username, first_name) if ai_reason.startswith("AI异常") and score >= 45: return True, f"规则兜底(AI不可用): {reason}" return ai_decision, f"{reason}; {ai_reason}"