这是文本自动审核服务包含敏感词检测、规则识别、风险打分和人工复审逻辑。python# -*- coding: utf-8 -*-文本自动审核引擎- 敏感词 DFA 多模匹配(分级加权)- 正则规则(联系方式/广告/灌水)- 抗绕过:全角、零宽字符、干扰符- 输出:pass / review / reject 命中明细 打码文本from __future__ import annotationsimport reimport unicodedatafrom dataclasses import dataclass, field, asdictfrom typing import Iterable# ---------------- 常量 ----------------PASS pass # 通过REVIEW review # 转人工复审REJECT reject # 拒绝MAX_SCORE 100DEFAULT_REJECT_THRESHOLD 80DEFAULT_REVIEW_THRESHOLD 30# 敏感词等级 - 风险分SCORE_BY_LEVEL {1: 15, 2: 45, 3: 100}# ---------------- 词库 ----------------# ⚠️ 生产环境请替换成你自己的词库,并从 DB / Redis 加载、支持热更新DEFAULT_WORDS: dict[str, int] {# level 3 —— 严重违规,命中即拒绝赌博: 3, 博彩: 3, 私彩: 3, 六合彩: 3,毒品: 3, 冰毒: 3, 大麻: 3, 迷药: 3,枪支: 3, 弹药: 3, 假币: 3, 假钞: 3,代开发票: 3, 办证: 3, 假证: 3,色情: 3, 成人网站: 3, 一夜情: 3, 约炮: 3,黑客: 3, 代考: 3, 包过: 3,# level 2 —— 广告 / 引流刷单: 2, 兼职刷单: 2, 刷信誉: 2,加微信: 2, 私聊我: 2, 私信我: 2,贷款: 2, 套现: 2, 花呗套现: 2, 无抵押: 2,高薪招聘: 2, 日结: 2, 返利: 2,# level 1 —— 轻度傻逼: 1, 去死: 1, 滚蛋: 1,}# ---------------- 正则规则 ----------------dataclass(frozenTrue)class RegexRule:name: strpattern: re.Patternscore: intmax_hits: int 3 # 同一规则最多计分次数,避免刷分DEFAULT_REGEX_RULES: list[RegexRule] [RegexRule(手机号, re.compile(r(?!\d)1[3-9]\d{9}(?!\d)), 40),RegexRule(QQ号, re.compile(r(?:qq|扣扣|企鹅)\s*[:]?\s*\d{5,12}, re.I), 35),RegexRule(微信号, re.compile(r(?:微信|weixin|vx|v信|薇信|威信)\s*[:]?\s*[a-zA-Z][a-zA-Z0-9_-]{4,19}), 35),RegexRule(网址, re.compile(r(?:https?://|www\.)[\w\-.]\.[a-zA-Z]{2,}(?:/\S*)?, re.I), 30),RegexRule(邮箱, re.compile(r[\w.-][\w-]\.[a-zA-Z]{2,}), 25),RegexRule(银行卡号, re.compile(r(?!\d)\d{16,19}(?!\d)), 50),RegexRule(灌水重复, re.compile(r(.)\1{7,}), 20, 1),]# ---------------- 文本归一化 ----------------_ZERO_WIDTH re.compile(r[\u200b-\u200f\u202a-\u202e\u2060\ufeff])# 常见用于绕过检测的干扰符号_INTERFERENCE re.compile(r[\s\*\.\-_~·、,。!?#$%^|/\\\[\]【】()《》\“”‘’:;])def normalize(text: str) - str:全角转半角 去掉零宽字符。注意:可能改变字符串长度。text unicodedata.normalize(NFKC, text)return _ZERO_WIDTH.sub(, text)def deobfuscate(text: str) - str:去掉空格、星号等干扰符,用于检测 赌 博 / 赌*博 这类变体。return _INTERFERENCE.sub(, text)# ---------------- 结果模型 ----------------dataclassclass Hit:kind: str # 命中类型:sensitive_word / rule:xxx / remote:xxxdetail: str # 命中的具体内容score: int # 风险分start: int -1 # 在归一化文本中的位置(仅本地规则有效)end: int -1dataclassclass AuditResult:action: strscore: inthits: list[Hit] field(default_factorylist)normalized_text: str filtered_text: str # 命中部分打码后的文本def to_dict(self) - dict:return {action: self.action,score: self.score,hits: [asdict(h) for h in self.hits],filtered_text: self.filtered_text,}def mask_spans(text: str, spans: Iterable[tuple[int, int]]) - str:把命中的位置替换成 *chars list(text)n len(chars)for s, e in spans:for k in range(max(0, s), min(n, e)):chars[k] *return .join(chars)# ---------------- DFA 敏感词匹配 ----------------class DFAFilter:基于 Trie 的多模匹配,最长匹配优先。时间复杂度 O(n * 最长词长)。__slots__ (root,)def __init__(self) - None:self.root: dict {}def add(self, word: str) - None:node self.rootfor ch in word:node node.setdefault(ch, {})node[\x00] True # 结束标记def build(self, words: Iterable[str]) - DFAFilter:for w in words:w normalize(w).strip()if w:self.add(w)return selfdef match(self, text: str) - list[tuple[int, int, str]]:返回 [(start, end, word)]hits: list[tuple[int, int, str]] []n len(text)i 0while i n:node self.rootj ilast_end -1while j n:nxt node.get(text[j])if nxt is None:breaknode nxtj 1if \x00 in node:last_end j # 记录最近一次完整匹配if last_end i:hits.append((i, last_end, text[i:last_end]))i last_end # 跳过已匹配部分else:i 1return hits# ---------------- 审核引擎 ----------------class TextModerator:def __init__(self,words: dict[str, int] | None None,regex_rules: list[RegexRule] | None None,reject_threshold: int DEFAULT_REJECT_THRESHOLD,review_threshold: int DEFAULT_REVIEW_THRESHOLD,) - None:self.words self._normalize_words(DEFAULT_WORDS if words is None else words)self.dfa DFAFilter().build(self.words)self.regex_rules DEFAULT_REGEX_RULES if regex_rules is None else regex_rulesself.reject_threshold reject_thresholdself.review_threshold review_thresholdstaticmethoddef _normalize_words(words: dict[str, int]) - dict[str, int]:out: dict[str, int] {}for k, v in words.items():k normalize(k).strip()if k:out[k] vreturn outdef reload_words(self, words: dict[str, int]) - None:热更新词库(可对接 Redis 订阅 / 定时拉取)self.words self._normalize_words(words)self.dfa DFAFilter().build(self.words)# ---------- 决策 ----------def decide(self, score: int, hits: list[Hit]) - str:if any(h.score MAX_SCORE for h in hits):return REJECTif score self.reject_threshold:return REJECTif score self.review_threshold:return REVIEWreturn PASS# ---------- 主流程 ----------def audit(self, text: str) - AuditResult:if not text or not text.strip():return AuditResult(actionPASS, score0)norm normalize(text)hits: list[Hit] []spans: list[tuple[int, int]] []score 0# 1) 敏感词for start, end, word in self.dfa.match(norm):level self.words.get(word, 2)s SCORE_BY_LEVEL.get(level, 45)hits.append(Hit(sensitive_word, word, s, start, end))spans.append((start, end))score s# 2) 抗绕过:去掉干扰符后再匹配一次(位置无法映射回原文,故不打码)stripped deobfuscate(norm)if stripped ! norm:for _, _, word in self.dfa.match(stripped):hits.append(Hit(sensitive_word_obfuscated, word, 60))score 60# 3) 正则规则for rule in self.regex_rules:count 0for m in rule.pattern.finditer(norm):if count rule.max_hits:breakcount 1hits.append(Hit(frule:{rule.name}, m.group(), rule.score, m.start(), m.end()))spans.append((m.start(), m.end()))score rule.scorescore min(score, MAX_SCORE)return AuditResult(actionself.decide(score, hits),scorescore,hitshits,normalized_textnorm,filtered_textmask_spans(norm, spans),)# 单例,供接口层复用moderator TextModerator()审核流程与操作说明这块代码把自动审核拆成了三道关卡下面具体说说每一步在干什么。· 敏感词分级检测命中敏感词时按等级加权扣分3 级词汇直接触发拒绝2 级和 1 级累积风险分。· 格式规则识别内置手机号、QQ号、微信号、网址等正则同一规则最多计分 3 次避免刷分。· 抗绕过处理先做全角转半角、去掉零宽字符再把文字里的空格、星号等干扰符去掉后二次匹配防止“赌 博”“赌*博”这类变体。· 自动决策根据最终得分落在 [0,30)、[30,80)、[80,100] 区间分别给出通过、人工复审、拒绝三种结果并输出命中明细和打码后的文本。---优化建议 建议您将 DEFAULT_WORDS 里的示例词替换为自己的词库并接入数据库或 Redis 支持热更新也可按需调整拒绝和复审的分数阈值。文章仅供参考用。