Files
yu-zhi-ran/scripts/compliance_checker.py
T
Yuzhiran Dev 23ff63baa9 fix: 三平台内容差异化 + admin敏感词管理表格化
- writer.py: _expand_section() 去除 <100字阈值,始终调用 LLM 平台专属扩写
- prompt_loader.py: 新增 section_expansion_zhihu/wechat/xiaohongshu 三个独立 prompt
- admin.html: 配置管理标签页 + 敏感词/清理规则子标签 + 敏感词表格化管理(编辑/删除)
- config_items.py: PUT /sensitive-words/{id} 支持更新 word/category
- compliance_checker.py: AI 套话从 DB 加载 + 人称规则修正
- initial_data.py: PlatformConfig 字数迁移 + 新种子
- 各前端页面: LLM 配置 rate_limit 字段 + 供应商列表排序
2026-06-08 13:51:35 +08:00

533 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""
内容合规审查模块
检查文章是否符合法律法规、平台规则、品牌规范
"""
import re
from typing import Dict, List, Tuple
SENSITIVE_WORDS = {
"政治敏感": ["国家主席", "政治局", "常委", "军委", "统战部", "颠覆国家", "分裂主义", "台独", "疆独", "藏独"],
"违禁内容": ["赌博", "毒品", "迷药", "枪支", "炸药", "色情", "低俗", "反动", "邪教"],
"不实信息": [" guaranteed 赚钱", "一夜暴富", "100%有效", "包治百病", "绝对正确"],
"领导人相关": ["主席", "总理", "总书记", "国家领导人"]
}
PLATFORM_RULES = {
"zhihu": {
"max_title_len": 100,
"min_word_count": 1000,
"allowed_tags": ["科技", "生活", "职场", "教育", "可持续", "AI", "远程工作", "个人成长"],
"forbidden_patterns": [r"加微信", r"私聊", r"付费咨询", r"点击领取"]
},
"wechat": {
"max_title_len": 32,
"min_word_count": 800,
"allowed_tags": ["科技", "生活", "职场", "教育", "可持续", "AI", "远程工作", "个人成长"],
"forbidden_patterns": [r"诱导分享", r"朋友圈", r"转发群"]
},
"xiaohongshu": {
"max_title_len": 50,
"min_word_count": 400,
"allowed_tags": ["生活方式", "可持续", "AI", "个人成长", "极简", "环保"],
"forbidden_patterns": [r"私信", r"加群", r"导流"]
}
}
# AI 套话检测模式(一旦出现在正文中,说明写作痕迹明显)
AI_TELTALES = [
"说回到",
"一个真实的.*案例很能说明问题",
"这就是.*被.*后的样子",
"如果你也",
"值得注意的是",
"首先其次最后",
"综上所述",
"总的来说",
"说到这里",
"我们来总结一下",
"总而言之",
"我们不难发现",
"我们可以看出",
"从以上分析可以看出",
"无可否认",
"众所周知",
"毋庸置疑",
"不知大家有没有发现",
"不可否认",
"毫无疑义",
"从某种意义上",
"从某种程度上",
"在一定程度上",
"换而言之",
"换言之",
"从本质",
"归根结底",
"说到底",
"这为我们提供了",
"为我们提供了宝贵的",
"引发了我们",
"不得不让人思考",
"引人深思",
"毫无悬念",
"毫无意外",
"毫无争议",
]
_cached_sensitive_words = None
_cached_platform_rules = None
_cached_ai_telltales = None
def _load_sensitive_words():
global _cached_sensitive_words
if _cached_sensitive_words is not None:
return _cached_sensitive_words
try:
from app.core.prompt_loader import _get_session
from app.models import SensitiveWord
session = _get_session()
try:
rows = session.query(SensitiveWord).filter(SensitiveWord.is_active == True).all()
if rows:
result = {}
for r in rows:
cat = r.category or "general"
if cat not in result:
result[cat] = []
result[cat].append(r.word)
_cached_sensitive_words = result
return _cached_sensitive_words
finally:
session.close()
except Exception:
pass
_cached_sensitive_words = SENSITIVE_WORDS
return _cached_sensitive_words
def _load_platform_rules():
global _cached_platform_rules
if _cached_platform_rules is not None:
return _cached_platform_rules
try:
from app.core.prompt_loader import _get_session
from app.models import PlatformConfig
import json
session = _get_session()
try:
rows = session.query(PlatformConfig).all()
if rows:
result = {}
for r in rows:
try:
cfg = json.loads(r.config_data) if r.config_data else {}
except:
cfg = {}
if cfg:
result[r.platform] = cfg
if result:
_cached_platform_rules = result
return _cached_platform_rules
finally:
session.close()
except Exception:
pass
_cached_platform_rules = PLATFORM_RULES
return _cached_platform_rules
def _load_ai_telltales():
"""从 DB ContentCleanRule 加载 AI 套话模式(rule_type='ai_telltale'),无 DB 时回退硬编码列表"""
global _cached_ai_telltales
if _cached_ai_telltales is not None:
return _cached_ai_telltales
try:
from app.core.prompt_loader import _get_session
from app.models import ContentCleanRule
session = _get_session()
try:
rows = session.query(ContentCleanRule).filter(
ContentCleanRule.rule_type == 'ai_telltale',
ContentCleanRule.is_active == True
).order_by(ContentCleanRule.sort_order).all()
if rows:
_cached_ai_telltales = [r.pattern for r in rows]
return _cached_ai_telltales
finally:
session.close()
except Exception:
pass
_cached_ai_telltales = list(AI_TELTALES)
return _cached_ai_telltales
class ComplianceChecker:
"""合规审查器"""
def __init__(self, platform_config: Dict = None):
self.issues = []
self.platform_config = platform_config or {}
def _get_platform_rule(self, key: str, default=None):
"""从 platform_config 读取规则,fallback 到硬编码 PLATFORM_RULES"""
if self.platform_config:
compliance_rules = self.platform_config.get('compliance_rules', {})
if key in compliance_rules:
return compliance_rules[key]
if key == 'min_word_count' and self.platform_config.get('min_words'):
return self.platform_config['min_words']
return default
def check_text(self, text: str, platform: str, topic_data: Dict = None) -> Dict:
"""执行全面合规检查"""
self.issues = []
# 1. 敏感词检查
self._check_sensitive_words(text)
# 2. 平台规则检查
self._check_platform_rules(text, platform)
# 3. 法律法规检查
self._check_legal_compliance(text)
# 4. 品牌调性检查
self._check_brand_guidelines(text)
# 5. 内容事实性检查(如有主题数据)
if topic_data:
self._check_factual_consistency(text, topic_data)
# 6. 最小字数检查
self._check_min_length(text, platform)
# 7. 内容质量检查
self._check_ai_telltales(text)
self._check_pronoun_consistency(text, platform)
self._check_reading_experience(text, platform)
self._check_platform_engagement(text, platform)
self._check_inline_images(text)
self._check_timeliness(text)
hard_types = ('敏感词', '法律法规', '平台规则', '品牌规范', '资源合规')
hard_issues = [i for i in self.issues if i['type'] in hard_types]
return {
"passed": len(hard_issues) == 0,
"issues": self.issues,
"score": max(0, 100 - sum(
10 if i['type'] in hard_types else 5 for i in self.issues
))
}
def _check_sensitive_words(self, text: str):
"""检查敏感词"""
words_map = _load_sensitive_words()
for category, words in words_map.items():
for word in words:
if word in text:
self.issues.append({
"type": "敏感词",
"category": category,
"word": word,
"suggestion": f"删除或替换'{word}'"
})
def _check_platform_rules(self, text: str, platform: str):
"""检查平台特定规则"""
rules_map = _load_platform_rules()
rules = rules_map.get(platform, {})
# 标题长度(从HTML中提取)
max_title_len = self._get_platform_rule('max_title_len', rules.get("max_title_len"))
title_match = re.search(r'<title>([^<]+)</title>', text) or re.search(r'<h1[^>]*>([^<]+)</h1>', text)
if title_match and max_title_len:
title_len = len(title_match.group(1))
if title_len > max_title_len:
self.issues.append({
"type": "平台规则",
"category": "标题长度",
"detail": f"标题{title_len}字,超过{platform}限制{max_title_len}",
"suggestion": "缩短标题"
})
# 禁止的模式匹配
forbidden_patterns = self._get_platform_rule('forbidden_patterns', rules.get("forbidden_patterns", []))
for pattern in forbidden_patterns:
if re.search(pattern, text):
self.issues.append({
"type": "平台规则",
"category": "禁止内容",
"pattern": pattern,
"suggestion": "移除违规内容或联系方式"
})
# 标签检查:仅检查专门的标签容器(避免误伤正文中的话题引用)
tags_container_match = re.search(r'<div class="tags">([^<]+)</div>', text) or re.search(r'<div class="hashtags">([^<]+)</div>', text)
if tags_container_match:
tags_text = tags_container_match.group(1)
tags = re.findall(r'#([A-Za-z0-9一-龥]{2,10})', tags_text)
# 过滤掉纯十六进制颜色码(如 #1a1a1a, #fff
tags = [t for t in tags if not re.fullmatch(r'[0-9a-fA-F]{3,6}', t)]
else:
tags = []
allowed_tags = self._get_platform_rule('allowed_tags', rules.get("allowed_tags", []))
if allowed_tags:
for tag in tags:
if tag not in allowed_tags:
self.issues.append({
"type": "平台规则",
"category": "标签合规",
"tag": tag,
"suggestion": f"使用平台允许的标签,如{', '.join(allowed_tags[:3])}"
})
def _check_legal_compliance(self, text: str):
"""检查法律法规合规性"""
# 检查是否涉及国家秘密、国家安全
if re.search(r'国家机密|军事秘密|绝密|机密', text):
self.issues.append({
"type": "法律法规",
"category": "国家秘密",
"suggestion": "立即删除涉密内容"
})
# 检查是否宣传迷信、邪教
if re.search(r'算命|看相|测八字|跳大神|法轮功', text):
self.issues.append({
"type": "法律法规",
"category": "封建迷信",
"suggestion": "删除迷信内容"
})
# 检查是否赌博相关
if re.search(r'赌|博彩|下注|时时彩|六合彩', text):
self.issues.append({
"type": "法律法规",
"category": "赌博违法",
"suggestion": "删除赌博相关内容"
})
# 检查版权问题(是否使用未授权素材)
if re.search(r'版权声明.*?未经授权|转载请联系|盗用', text, re.IGNORECASE):
self.issues.append({
"type": "法律法规",
"category": "版权风险",
"suggestion": "确保所有引用已标注来源或获得授权"
})
def _check_brand_guidelines(self, text: str):
"""检查品牌调性(宇之然)"""
# 检查是否使用第一人称"我"
first_person_count = len(re.findall(r'^(我|本人|笔者)\b', text, re.MULTILINE))
if first_person_count > 2: # 允许少量情感连接
self.issues.append({
"type": "品牌规范",
"category": "人称使用",
"detail": f"发现{first_person_count}处第一人称,建议使用客观叙事",
"suggestion": "改为'实践者''本专栏'等客观表述"
})
# 检查是否有商业推广倾向
if re.search(r'强烈推荐|必买|最好的|最赚钱|独家', text):
self.issues.append({
"type": "品牌规范",
"category": "过度推广",
"suggestion": "使用更中立的表达,避免绝对化用语"
})
# 检查是否提及具体品牌(需模糊化)
known_brands = ["米家", "花帮主", "园艺助手", "Aerogarden"]
for brand in known_brands:
if brand in text:
self.issues.append({
"type": "品牌规范",
"category": "品牌露出",
"brand": brand,
"suggestion": f"'{brand}'改为'一些第三方工具''智能设备'"
})
def _check_factual_consistency(self, text: str, topic_data: Dict):
"""检查内容与选题的一致性"""
topic = topic_data.get("topic", {})
expected_title = topic.get("title", "")
expected_field = topic.get("field", "")
# 检查标题是否出现在文章中
if expected_title and expected_title[:5] not in text:
self.issues.append({
"type": "内容质量",
"category": "主题一致性",
"detail": f"文章可能偏离选题'{expected_title}'",
"suggestion": "确认内容围绕选题展开"
})
# 检查是否有核心观点
core_concept = topic.get("core_concept", "")
if core_concept and len(core_concept) > 10:
# 核心概念应出现在前1/3内容
first_third = text[:len(text)//3]
if core_concept[:10] not in first_third:
self.issues.append({
"type": "内容质量",
"category": "核心观点",
"suggestion": "在文章前1/3部分明确阐述核心观点"
})
def _check_min_length(self, text: str, platform: str):
"""检查文章最小字数(去除HTML标签)"""
plain = re.sub(r'<[^>]+>', '', text)
word_count = len(plain.strip())
min_words = self._get_platform_rule('min_word_count', 1000)
if word_count < min_words:
self.issues.append({
"type": "内容完整度",
"category": "字数不足",
"detail": f"当前{word_count}字,低于平台要求{min_words}",
"suggestion": "扩写内容至最低要求"
})
def _check_ai_telltales(self, text: str):
"""检查AI套话——正文中出现这些模式说明AI写作痕迹明显"""
plain = re.sub(r'<[^>]+>', '', text)
patterns = _load_ai_telltales()
for pattern in patterns:
if re.search(pattern, plain):
self.issues.append({
"type": "内容质量",
"category": "AI套话",
"detail": f"正文出现AI套话模式: 「{pattern}",
"suggestion": "删除或替换为自然表达,不要让读者感觉是AI写的"
})
def _check_pronoun_consistency(self, text: str, platform: str):
"""检查人称一致性"""
plain = re.sub(r'<[^>]+>', '', text)
has_ni = '' in plain
has_nimen = '你们' in plain
if has_nimen and has_ni:
self.issues.append({
"type": "内容质量",
"category": "人称混用",
"detail": "同时使用「你」和「你们」,建议统一为「你」",
"suggestion": "将「你们」替换为「你」,保持对话感"
})
def _check_reading_experience(self, text: str, platform: str = ""):
"""检查阅读体验(段落长度、配图)"""
paragraphs = re.findall(r'<p>(.*?)</p>', text, re.DOTALL)
long_paras = [p for p in paragraphs if len(p) > 300]
if len(long_paras) > len(paragraphs) * 0.3:
self.issues.append({
"type": "内容质量",
"category": "段落过长",
"detail": f"超过30%的段落长度>300字(共{len(paragraphs)}段,{len(long_paras)}段过长),在手机上阅读体验差",
"suggestion": "将长段拆分为2-3个短段,每段不超过150-200字"
})
# 图片检测(各平台阈值不同)
imgs = re.findall(r'<img', text)
min_imgs = {"zhihu": 1, "wechat": 1, "xiaohongshu": 1}.get(platform, 1)
if not imgs:
self.issues.append({
"type": "内容质量",
"category": "缺少配图",
"detail": "正文中没有图片,建议插入配图提升阅读体验",
"suggestion": "在关键位置插入配图(数据截图、金句卡片、操作步骤图等)"
})
elif len(imgs) < min_imgs:
self.issues.append({
"type": "内容质量",
"category": "配图不足",
"detail": f"只有{len(imgs)}张图,建议至少{min_imgs}张配图提升阅读体验",
"suggestion": "在文章中间部分增加配图"
})
def _check_platform_engagement(self, text: str, platform: str):
"""检查平台特有的互动/传播要素"""
plain = re.sub(r'<[^>]+>', '', text)
if platform == "zhihu":
# 知乎需要讨论引导(评论区互动是核心算法权重)
if not any(kw in plain for kw in ('你觉得', '你怎么看', '欢迎在评论区', '说说你的', '欢迎讨论', '你怎么想')):
self.issues.append({
"type": "内容质量",
"category": "缺少互动引导",
"detail": "知乎文章建议在末尾加讨论引导(如「你觉得呢?欢迎在评论区聊聊」),提升互动率",
"suggestion": "在文章末尾添加一个问题或讨论话题,引导读者评论"
})
# 检查是否有数据引用(知乎读者看重可信度)
if not any(kw in plain for kw in ('据统计', '调研显示', '数据显示', '报告指出', '根据', '研究表明', '调查显示')):
self.issues.append({
"type": "内容质量",
"category": "缺少数据引用",
"detail": "知乎文章建议引用具体数据或报告来支撑观点,增强可信度",
"suggestion": "在关键观点处引用权威数据来源"
})
if platform == "xiaohongshu":
# 小红书需要收藏引导(收藏率是推荐算法核心指标)
if '收藏' not in plain:
self.issues.append({
"type": "内容质量",
"category": "缺少收藏引导",
"detail": "小红书笔记建议在末尾加收藏引导(如「觉得有用点个收藏」),提升收藏率",
"suggestion": "在末尾添加收藏引导语"
})
# 小红书段落需非常短
paragraphs = re.findall(r'<p>(.*?)</p>', text, re.DOTALL)
long_paras = [p for p in paragraphs if len(p) > 150]
if len(long_paras) > len(paragraphs) * 0.2:
self.issues.append({
"type": "内容质量",
"category": "段落过长",
"detail": f"小红书建议每段不超过80-100字,当前{len(long_paras)}/{len(paragraphs)}段超过150字",
"suggestion": "将长段拆分为1-2句的短段落,每段不超过100字"
})
# 小红书需要至少一些 emoji
if not re.search(r'[\U0001F300-\U0001F9FF\u2600-\u27BF]', plain):
self.issues.append({
"type": "内容质量",
"category": "缺少emoji",
"detail": "小红书笔记建议适当使用emoji来增加视觉吸引力",
"suggestion": "在标题、章节分隔或重点句前添加相关emoji"
})
def _check_inline_images(self, html: str):
"""检查图片是否以内联方式嵌入(data:image)"""
# 提取所有 img 标签的 src 属性值
srcs = re.findall(r'<img\b[^>]*src=[\'"]([^\'"]+)[\'"]', html, re.IGNORECASE)
for src in srcs:
if not src.startswith('data:image/'):
self.issues.append({
"type": "资源合规",
"category": "图片内联",
"detail": f"图片未内联: {src[:50]}... 需手动修复"
})
def _check_timeliness(self, text: str):
years = re.findall(r'(19\d{2}|20[0-4]\d)', text)
outdated = {y for y in years if int(y) < 2025}
if outdated:
self.issues.append({
"type": "平台规则",
"category": "时效性",
"detail": f"使用过时年份: {', '.join(sorted(outdated))},需更新为2025年及以后的数据",
"suggestion": "替换为最新数据,或使用'近期'等模糊表述"
})
def check_article(html_content: str, platform: str, topic_data: Dict = None, platform_config: Dict = None) -> Dict:
"""便捷函数:执行完整合规检查"""
checker = ComplianceChecker(platform_config=platform_config)
return checker.check_text(html_content, platform, topic_data)
if __name__ == "__main__":
# 测试
test_html = "<html><body><h1>测试</h1>内容涉及赌博网站</body></html>"
result = check_article(test_html, "zhihu")
print(json.dumps(result, ensure_ascii=False, indent=2))