ngfchl
357 lines
18 KiB
Python
357 lines
18 KiB
Python
"""中文垃圾广告检测引擎
|
||
|
||
策略：
|
||
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}"
|