★ FEATURED ARTICLE 在日常社区运营和内容审核的对接中一类高频出现的文本总是让人头疼账号在评论区、商品问答区、私信入口批量投放“哈喽我们是源头工厂有需要就找我呀”这类揽客话术。它不算明显的违规广告单看每个字都很正常但架不住量大、变体多、人工审不完。本文就以此为切入点完整实现一个可运行的营销引流文本检测小工具从文本归一化、规则设计、权重打分到配置外置、灰度上线全流程拆解。读者不需要机器学习基础只要会写 Python 函数就能跟着跑通已经做过内容风控的同学也可以直接拿去对照自己的规则库查漏补缺。1. 从一个真实场景说起营销引流文本为什么需要工程化治理先把问题定义清楚。所谓“营销引流文本”指的是那些目标明确、但表述上刻意模糊的商业推广内容自称厂家、批发、一手货源引导用户离开当前平台去私下沟通制造限时紧迫感等等。这类内容在电商评论区、二手交易区、内容社区、在线客服会话里都非常常见。它和传统意义上的垃圾广告有几个明显区别也正因为这些区别简单的关键词黑名单很难奏效。1.1 这类文本的三个技术特征第一个特征是词汇本身是正常词。“源头工厂”“批发价”“有需要找我”这些词单独拎出来在正规的商品描述、售后说明、用户提问里都会出现。你把“源头工厂”整个拉黑第二天就会收到一堆误伤投诉。所以规则必须建立在词组组合上而不是单词上。第二个特征是变体成本极低。加个全角字母、塞几个零宽字符、把“微信”写成“薇信”、把汉字重复几遍都能在不改变人眼阅读效果的前提下绕过字面匹配。这一点在后文会专门用一节讲文本归一化因为这是所有文本风控的第一道地基。第三个特征是分级困难。一句“我是源头工厂欢迎咨询”到底该直接拦截还是人工复核如果全部拦截误伤率会高到业务方无法接受如果全部放过等于没做。实际工程里必须有中间档位也就是“疑似”状态把判断权交给人。1.2 几种常见方案的边界对比在动手写代码之前先看清楚几种技术路线的适用边界避免一上来就选错工具。方案实现成本可解释性变体对抗能力适合阶段纯关键词黑名单极低强很弱冷启动、兜底归一化 组合正则低强中等规则库主力词向量/文本分类模型高弱强有标注数据后大模型语义判别中中强增量补充、复核辅助需要强调的是这几条路线不是替代关系而是叠加关系。一个成熟的文本风控系统通常是归一化打底规则引擎跑主力模型负责兜底和兜住长尾人工复核处理争议样本最后把复核结果回流成新的规则和训练数据。本文聚焦第二条路线因为它是最容易在两周内上线、并且效果立刻可量化的方案。对于日均评论量在几万到几十万量级的业务一套设计良好的规则引擎能得到超过八成的问题召回误伤也完全可控。1.3 本文要实现的系统能力具体来说我们要做的东西包含这几个能力点输入一条原始文本输出一个 0 到 100 的分数以及block/review/pass三档结论。支持解释不仅告诉你“这条有问题”还要告诉你“命中了哪条规则、命中了哪几个字”。支持配置化规则库不写死在代码里能通过文件修改并生效。支持批量与单条两种调用方式便于本地调试和接入服务。附带单元测试保证改了规则之后不会悄悄把已有能力改坏。下面进入环境准备。2. 环境准备与项目结构2.1 运行环境与依赖选择这个项目刻意只使用 Python 标准库原因很实际内容审核往往要嵌到已有的服务里多引一个第三方包就多一份版本冲突风险和供应链风险。re、unicodedata、json、argparse、unittest全部内置装好 Python 就能跑。环境说明Python 版本建议使用 3.8 及以上。本文用到的dataclasses在 3.7 之后是标准库f-string 的格式说明符在 3.6 之后可用。具体小版本请按你自己的项目统一这里不写死。操作系统Windows、macOS、Linux 均可代码中没有平台相关调用。编辑器PyCharm、VS Code 都行建议开启保存时自动格式化。不需要数据库、不需要 Web 框架全部逻辑跑在单进程内存里。如果你后续要接到 Flask / FastAPI 的接口里直接把RuleEngine实例挂在应用启动阶段初始化一次即可规则编译是一次性的运行期只有正则匹配开销。2.2 目录结构设计项目结构如下保持扁平方便直接复制text_guard_demo/ ├── text_guard/ │ ├── __init__.py │ ├── normalize.py │ ├── rules.py │ ├── engine.py │ └── cli.py ├── config/ │ └── rules.json ├── samples.txt └── tests/ └── test_engine.py这里的分层是按职责切的normalize.py只负责把文本洗干净rules.py只放规则数据engine.py负责匹配和打分cli.py只做输入输出。这样切的好处是将来你想把规则换成模型只动engine.py想把规则换成数据库下发只动rules.py的加载逻辑。很多人在写这类小工具时喜欢全塞在一个文件里调试时舒服上线后维护会很痛苦。2.3 安装与运行方式创建目录并初始化mkdir -p text_guard_demo/text_guard text_guard_demo/config text_guard_demo/tests cd text_guard_demo touch text_guard/__init__.py tests/__init__.py__init__.py可以为空它的作用是让text_guard成为一个包这样python -m text_guard.cli才能正常找到模块。tests/__init__.py同理方便测试发现。运行单元测试python -m unittest discover -s tests -v运行命令行工具python -m text_guard.cli -t 哈喽我们是源头工厂有需要就找我呀下面开始写真正的逻辑。3. 文本归一化把变体拉回可比较的形态3.1 为什么归一化是第一步先看一组例子。下面这五条文本人一眼就能看出是同一句话但字符串比较的结果全是“不相等”哈喽我们是源头工厂有需要就找我呀 哈喽,我们是源头工厂,有需要就找我呀 哈喽我们是源头工厂有需\u200b要就找我呀 哈喽我们是源头工厂有需要要要找我呀 哈喽我们是源头工厂有需要就找我丫第一条用的是全角逗号第二条用半角逗号第三条中间插了一个零宽空格U200B第四条把“要”重复了三遍第五条把“呀”换成了字形接近的“丫”。如果你把规则写成re.search(有需要就找我, text)只有第一条能命中。这不是规则写得不对而是输入没有对齐。所以归一化的定位不是“美化文本”而是把所有等价写法映射到同一个可比较形态把对抗成本从规则方转移回发布方。3.2 三类必须处理的字符噪声第一类全角与兼容字符。全角字母到、全角数字到、全角标点、带圈数字、罗马数字都属于 Unicode 的兼容字符区。Python 的unicodedata.normalize(NFKC, text)可以把它们统一折叠成常见形态。这个函数不需要额外依赖一次调用就能把大量“换皮”写法拉平。需要注意的是NFKC 是幂等的重复调用结果不变所以在整条流水线里调一次就够不必每步都调。第二类不可见字符。零宽空格U200B、零宽非连接符U200C、零宽连接符U200D、左右方向标记U200E、U200F、字节序标记UFEFF、软连字符U00AD这些字符在终端里看不见却真实存在于字符串中。用正则把这些区间整体删掉即可。同时把 ASCII 控制字符除\t、\n、\r外也一并清理避免日志里出现乱码。第三类夸张重复。把某个汉字重复两三次是极常见的规避手法。处理方式是用反向引用压缩连续三个及以上相同字符折叠成一个。这里有一个细节必须注意——数字不能压缩。如果你无脑用(.)\1{2,}那么“最后100件”会被压成“最后10件”价格和型号信息就丢了后续如果要做数量识别会出错。所以正则里要把数字和空白排除掉。3.3 归一化顺序为什么不能随意调换归一化的每一步都有先后依赖顺序错了会留下漏洞。推荐顺序是删除不可见字符和控制字符。NFKC 折叠全角与兼容字符。统一转小写。压缩重复字符。压缩连续空白并去掉首尾空格。为什么“删不可见字符”要放在最前面因为这些字符可能夹在 Unicode 组合序列中间先折叠再做删除有可能产生新的组合结果导致漏删。为什么不转小写就压缩重复大小写混写如vv折叠前后是两种形态先转小写能保证压缩逻辑看到的是同一个字符。另外提醒一点本文的规则库是针对中文话术设计的所以没有做繁简转换。如果你的业务场景里繁体内容占比不低建议在 NFKC 之后、转小写之前插入一步繁简统一用开源的字表映射即可不要自己手写几百对字符。这一步加不加取决于你的实际数据分布不要为了“完整”而堆功能。4. 规则设计与打分模型4.1 用组合词而不是单词这是规则库设计里最容易踩的坑。假设你把“工厂”设为关键词后果是什么用户问“这个工厂的生产资质能看下吗”直接被判成营销。误伤几次之后业务方就会要求关掉整个功能。正确的做法是把词拆成“意图信号”每个信号用词组或短语表达身份宣称类源头工厂、源头厂家、一手货源、厂家直销、工厂直供、原厂直供。注意这里全都是两到四个字的组合单独一个“工厂”或“货源”不构成信号。引导联系类加V、加微信、留联系方式、私信我、私聊我、滴滴我、扣1。这类信号的危险度最高因为它的目的就是把人从平台导走。招呼揽客类有需要就找我、欢迎咨询、欢迎来撩、随时找我。这是最典型的“哈喽体”尾巴。价格诱惑类批发价、出厂价、低价清仓、一件代发、量大从优。紧迫感类限时、秒杀、手慢无、仅限今天、最后 N 件。批量起订类N 件起批、起订量、大量现货。这样拆分以后单条文本的召回和精准度都会好很多。同时规则具备可解释性命中了哪一类审核人员一眼就能看懂原因。4.2 权重怎么定有了信号之后要给权重。权重的意义不是“精确概率”而是相对危险度的排序。本文的取值思路是规则权重理由引导联系25直接触达用户风险最高招呼揽客18典型揽客尾巴但可能出现在客服场景身份宣称15单独出现不足以判定需组合价格诱惑12常见于正常促销紧迫感10广告常用但正常活动也用批量起订8补充信号几个设计要点值得说明。第一同一条规则在一条文本里只计一次分。有人会想“命中三次就乘三”这会让刷分变得极其容易——重复五遍“加V”就能把分数顶到上限。用re.search而不是re.finditer天然做到一条规则一次计分。第二分数要有上限。所有命中权重求和后做一次min(raw, 100)避免后续加规则时分数计算溢出预期区间也方便阈值调整。第三权重是调出来的不是想出来的。上线后一定要跑真实的复核样本把误判的样本和漏判的样本分别列出来看是该调权重还是该补规则。凭感觉调权重两周之后没人记得为什么是 15 而不是 12。4.3 阈值分级与人工复核三档阈值的取值建议score 45→block直接拦截或折叠展示。20 score 45→review进入人工复核队列。score 20→pass正常放行。为什么中间要留这么宽的一档因为营销文本的判定本质上是概率性的两档拦/放会逼着你在“误伤太多”和“漏放太多”之间二选一。有了 review 档你可以先用较宽的规则召回再通过人工复核把样本标出来逐步把规则收紧。这也是冷启动阶段最务实的策略。需要提醒的是阈值不能几个业务线共用一套。客服会话里的“欢迎咨询”是正常的商品评论区里的“欢迎咨询”就很可疑。合理的做法是给不同场景各配一套阈值或者给规则加上场景标签。5. 完整实战一个可运行的文本风控小工具下面把四个模块的代码完整写出来每段都能直接复制进对应文件运行。5.1 归一化模块文件路径text_guard/normalize.py文本归一化把各种变体写法统一成可比较的形态。 import re import unicodedata # 零宽字符、方向控制符、软连字符等不可见字符 _INVISIBLE_RE re.compile(r[\u200b-\u200f\u202a-\u202e\u2060\ufeff\u00ad]) # ASCII 控制字符\t、\n、\r 交给后面的空白压缩处理 _CTRL_RE re.compile(r[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]) # 三个及以上连续相同的「非数字、非空白」字符压缩成一个 _REPEAT_RE re.compile(r([^\d\s])\1{2,}) # 连续空白压缩成一个空格 _SPACE_RE re.compile(r\s) def strip_invisible(text: str) - str: 删除零宽字符和控制字符避免「看起来一样、实际不同」的文本绕过规则。 return _INVISIBLE_RE.sub(, _CTRL_RE.sub(, text)) def to_halfwidth(text: str) - str: 使用 NFKC 归一化把全角字母数字、兼容字符折叠成常见形态。 return unicodedata.normalize(NFKC, text) def collapse_repeats(text: str) - str: 压缩夸张重复有需要要要找我 - 有需要找我。数字不压缩避免破坏价格和型号。 return _REPEAT_RE.sub(r\1, text) def normalize(text: str) - str: 归一化主入口顺序不可随意调换。 if not text: return t strip_invisible(text) t to_halfwidth(t) t t.lower() t collapse_repeats(t) t _SPACE_RE.sub( , t).strip() return t逐段解释一下。strip_invisible做了两层sub嵌套外层删零宽字符内层删控制字符顺序无关因为两类区间不重叠。to_halfwidth里用 NFKC 而不做手工映射是因为全角字母有 26 个、全角数字 10 个、各种标点几十个手写映射表既容易漏又难维护标准库已经把这件事件做好了。collapse_repeats的替换串是r\1配合正则里的捕获组([^\d\s])把整段重复压缩成第一个字符。数字被显式排除是为了保住“最后100件”这类信息。最后normalize函数的顺序就是 3.3 节说的顺序注释里也标明了不可随意调换。5.2 规则定义模块文件路径text_guard/rules.py规则数据定义。这里只放数据不放逻辑。 from dataclasses import dataclass from typing import List dataclass(frozenTrue) class Rule: rule_id: str # 规则唯一标识用于日志和解释 pattern: str # 正则在「已归一化文本」上匹配 weight: int # 命中分值 desc: str # 人话描述给审核人员看 DEFAULT_RULES: List[Rule] [ Rule( rule_idsrc_factory_claim, patternr源头工厂|源头厂家|一手货源|厂家直销|工厂直供|原厂直供, weight15, desc自称源头/厂家直供, ), Rule( rule_idcontact_solicit, patternr加\s*[v威薇微w]|加个?微信|留个?联系方式|私信我|私聊我|滴滴我|扣\s*1, weight25, desc引导私下联系, ), Rule( rule_idsales_greeting, patternr有需要就?找我|需要的?话?[^,。?]{0,6}找我|欢迎咨询|欢迎来撩|随时找我, weight18, desc招呼式揽客, ), Rule( rule_idprice_claim, patternr批发价|出厂价|低价清仓|跳楼价|赚差价|一件代发|量大从优, weight12, desc低价/代发话术, ), Rule( rule_idurgency, patternr限时|秒杀|手慢无|仅限今天|最后\s*\d\s*件, weight10, desc制造紧迫感, ), Rule( rule_idwholesale_qty, patternr\d\s*件起批|起订量|大量现货, weight8, desc批量起订话术, ), ]几个正则细节需要单独说明。加\s*[v威薇微w]里的\s*是为了兼容“加 V”这种中间带空格的写法。归一化已经把连续空白压成一个空格但空格还在所以匹配时必须允许它。需要的?话?[^,。?]{0,6}找我是“需要的资料可以找我”这类长句的兜底。中间允许 0 到 6 个非标点字符超过 6 个字符通常说明这是两个独立语义误伤风险上升。这里的[^,。?]用的是半角标点因为归一化已经通过 NFKC 把中文逗号句号问号转成了半角。最后\s*\d\s*件允许数字两边有空格同样是为了兼容“最后 100 件”这类排版。注意这里和collapse_repeats不冲突因为数字没有被压缩。5.3 规则引擎模块文件路径text_guard/engine.py规则引擎负责匹配、打分、给出结论。 import re from dataclasses import dataclass, field from typing import List, Sequence from .normalize import normalize from .rules import Rule, DEFAULT_RULES VERDICT_BLOCK block VERDICT_REVIEW review VERDICT_PASS pass dataclass class Hit: rule_id: str desc: str weight: int matched: str dataclass class Decision: text: str normalized: str score: int verdict: str hits: List[Hit] field(default_factorylist) def to_dict(self) - dict: return { verdict: self.verdict, score: self.score, normalized: self.normalized, hits: [ { rule_id: h.rule_id, desc: h.desc, weight: h.weight, matched: h.matched, } for h in self.hits ], } class RuleEngine: def __init__( self, rules: Sequence[Rule] DEFAULT_RULES, block_threshold: int 45, review_threshold: int 20, ): if review_threshold block_threshold: raise ValueError(review_threshold 必须小于 block_threshold) # 规则在初始化阶段一次性编译运行期不再重复编译 self._rules [(rule, re.compile(rule.pattern)) for rule in rules] self.block_threshold block_threshold self.review_threshold review_threshold def decide(self, text: str) - Decision: normalized normalize(text) hits: List[Hit] [] for rule, pattern in self._rules: # 用 search 而不是 finditer同一条规则只计一次分 m pattern.search(normalized) if m: hits.append( Hit( rule_idrule.rule_id, descrule.desc, weightrule.weight, matchedm.group(0), ) ) score min(sum(h.weight for h in hits), 100) if score self.block_threshold: verdict VERDICT_BLOCK elif score self.review_threshold: verdict VERDICT_REVIEW else: verdict VERDICT_PASS return Decision( texttext, normalizednormalized, scorescore, verdictverdict, hitshits, )这段代码里有三处设计是刻意的值得单独讲。构造期编译正则。re.compile本身有缓存但显式编译能保证行为确定也避免运行期第一次调用时的抖动。规则库一旦大起来几百条每次请求重复编译的开销不可忽视尤其是高频短文本场景。用search而不是finditer。这是 4.2 节说的“一条规则只计一次分”的实现基础。如果改成finditer并累加恶意用户只要把话术重复几遍就能把分数打满整个阈值体系就失效了。分数封顶 100。这个细节在多规则叠加时很重要。如果不封顶将来加规则容易把分数推到几百阈值就需要跟着重算历史积累的调参经验全部作废。封顶之后阈值区间保持稳定。5.4 命令行入口文件路径text_guard/cli.py命令行入口支持单条、文件和标准输入三种方式。 import argparse import json import sys from .engine import RuleEngine def build_parser() - argparse.ArgumentParser: parser argparse.ArgumentParser( progtext-guard, description营销引流话术检测小工具, ) source parser.add_mutually_exclusive_group() source.add_argument(-t, --text, help检测单条文本) source.add_argument(-f, --file, help检测文本文件每行一条) parser.add_argument(--block, typeint, default45, help拦截阈值默认 45) parser.add_argument(--review, typeint, default20, help复核阈值默认 20) parser.add_argument(--json, actionstore_true, help以 JSON 行格式输出) return parser def main(argvNone) - int: args build_parser().parse_args(argv) engine RuleEngine(block_thresholdargs.block, review_thresholdargs.review) if args.text is not None: items [args.text] elif args.file: with open(args.file, encodingutf-8) as f: items [line.rstrip(\n) for line in f if line.strip()] else: items [line.rstrip(\n) for line in sys.stdin if line.strip()] for idx, item in enumerate(items, start1): decision engine.decide(item) if args.json: payload decision.to_dict() payload[line] idx print(json.dumps(payload, ensure_asciiFalse)) else: print(f[{idx}] {decision.verdict.upper():6} score{decision.score:3} {item}) for hit in decision.hits: print(f - {hit.rule_id}({hit.weight}): {hit.matched}) return 0 if __name__ __main__: raise SystemExit(main())ensure_asciiFalse这一行很关键。默认的json.dumps会把中文转成\uXXXX日志里几乎没法看排查问题时非常痛苦。加上这个参数后输出保持中文原样配合encodingutf-8读文件整条链路不会出现乱码。raise SystemExit(main())是标准写法保证入口函数的返回值能正确传递给操作系统便于在脚本里通过退出码判断成功与否。5.5 单元测试文件路径tests/test_engine.pyimport unittest from text_guard.engine import ( RuleEngine, VERDICT_BLOCK, VERDICT_PASS, VERDICT_REVIEW, ) from text_guard.normalize import normalize class NormalizeTest(unittest.TestCase): def test_fullwidth_to_halfwidth(self): self.assertEqual(normalize(加), 加v) def test_zero_width_removed(self): self.assertEqual(normalize(加\u200bv), 加v) def test_repeat_collapsed(self): self.assertEqual(normalize(有需要要要找我), 有需要找我) def test_digits_kept(self): self.assertEqual(normalize(最后100件), 最后100件) class EngineTest(unittest.TestCase): def setUp(self): self.engine RuleEngine() def test_marketing_greeting_is_review(self): d self.engine.decide(哈喽我们是源头工厂有需要就找我呀) self.assertEqual(d.verdict, VERDICT_REVIEW) self.assertEqual(d.score, 33) def test_hard_sell_is_block(self): d self.engine.decide(源头工厂一手货源加Vabc123一件代发量大从优) self.assertEqual(d.verdict, VERDICT_BLOCK) self.assertEqual(d.score, 52) def test_normal_question_passes(self): d self.engine.decide(这个零件是源头工厂生产的吗我想了解下工艺流程) self.assertEqual(d.verdict, VERDICT_PASS) def test_contact_only_is_review(self): d self.engine.decide(有货的话加个微信聊) self.assertEqual(d.verdict, VERDICT_REVIEW) if __name__ __main__: unittest.main()测试用例的设计思路是“一类风险一条”。归一化层测四个点全角折叠、零宽删除、重复压缩、数字保留。引擎层测四个点打招呼式揽客、硬广、正常提问、仅引导联系。这样改规则时有基线保护不会出现“修了漏判、炸了误伤”还不知道的情况。如果你的团队用 pytest这套用例可以直接跑不需要改任何断言写法因为unittest.TestCase的子类 pytest 也识别。5.6 运行验证与结果解读准备样本文件文件路径samples.txt哈喽我们是源头工厂有需要就找我呀 源头工厂一手货源加Vabc123一件代发量大从优 这个零件是源头工厂生产的吗我想了解下工艺流程 你好请问这个型号还有货吗 限时秒杀最后3件手慢无需要的私聊我执行python -m text_guard.cli -f samples.txt预期输出[1] REVIEW score33 哈喽我们是源头工厂有需要就找我呀 - src_factory_claim(15): 源头工厂 - sales_greeting(18): 有需要就找我 [2] BLOCK score52 源头工厂一手货源加Vabc123一件代发量大从优 - src_factory_claim(15): 源头工厂 - contact_solicit(25): 加v - price_claim(12): 一件代发 [3] PASS score15 这个零件是源头工厂生产的吗我想了解下工艺流程 - src_factory_claim(15): 源头工厂 [4] PASS score0 你好请问这个型号还有货吗 [5] REVIEW score35 限时秒杀最后3件手慢无需要的私聊我 - contact_solicit(25): 私聊我 - urgency(10): 限时逐条解读一下结果这几条正好覆盖了系统的主要行为。第 1 条就是标题里那句话得分 33落在 review 档。它命中了“源头工厂”这个身份宣称和“有需要就找我”这个揽客尾巴但没有引导联系、没有价格诱惑所以不直接拦截。这个判定是合理的——如果你看到一条评论是这么写的通常会选择折叠或者限流而不是直接删除账号。第 2 条得分 52直接 block。注意它的命中项里contact_solicit匹配到的是归一化之后的加v说明全角字母折叠生效了。三条规则叠加刚好越过拦截线这正是“组合信号”的价值——单独任何一条都不足以定罪。第 3 条只命中身份宣称得 15 分判 pass。这条最值得注意它证明了两件事一是组合词设计确实降低了误伤如果规则里放的是单个“工厂”这条会被判 review二是 review 阈值设在 20 是有依据的单信号不应该进复核队列否则复核队列会被大量正常提问淹没。第 4 条零命中正常放行。第 5 条命中了引导联系和紧迫感得 35 分进 review。这里其实可以讨论私聊我权重 25 是否过高如果这条来自客服场景可能是正常回复。这就是为什么 4
阅读完成 · 觉得有帮助？
