Files
yu-zhi-ran/scripts/compliance_checker.py
T
lt 31d6306e3b feat: 数据源统一与前端预览修复
=== 后端核心 ===
- db_helper: 统一数据库访问抽象层
- system.py API:
  * 参数绑定修复: 使用 Body(embed=True) 接收 JSON
  * 添加请求日志记录
- sync.py: 仅导出 DB→JSON(备份)

=== 合规与流水线 ===
- compliance_checker: 标签检测优化(仅检查容器,避免正文误判)
- 所有脚本(creator/collector/writer/outline/research等)统一使用数据库

=== 前端改版 ===
- topics.html:
  * 创作/优化 API 路径修正
  * 预览弹窗重设计:多平台并行加载、富文本显示、单复制按钮
  * 状态中文映射(getStatusLabel)
  * 认证检查
- 所有 HTML 静态资源路径修复(移除 /static 前缀)

=== 数据一致性 ===
- 数据库状态统一为英文(pending/review/ready/published)
- 前端显示中文化映射

已测试 A03 流水线完整通过。
2026-05-07 11:25:42 +08:00

294 lines
12 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"导流"]
}
}
class ComplianceChecker:
"""合规审查器"""
def __init__(self):
self.issues = []
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_required_sections(text)
self._check_inline_images(text)
self._check_timeliness(text)
return {
"passed": len(self.issues) == 0,
"issues": self.issues,
"score": max(0, 100 - len(self.issues) * 10)
}
def _check_sensitive_words(self, text: str):
"""检查敏感词"""
for category, words in SENSITIVE_WORDS.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 = PLATFORM_RULES.get(platform, {})
# 标题长度(从HTML中提取)
title_match = re.search(r'<title>([^<]+)</title>', text) or re.search(r'<h1[^>]*>([^<]+)</h1>', text)
if title_match and rules.get("max_title_len"):
title_len = len(title_match.group(1))
if title_len > rules["max_title_len"]:
self.issues.append({
"type": "平台规则",
"category": "标题长度",
"detail": f"标题{title_len}字,超过{platform}限制{rules['max_title_len']}",
"suggestion": "缩短标题"
})
# 禁止的模式匹配
for pattern in rules.get("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 = rules.get("allowed_tags", [])
if allowed:
for tag in tags:
if tag not in allowed:
self.issues.append({
"type": "平台规则",
"category": "标签合规",
"tag": tag,
"suggestion": f"使用平台允许的标签,如{', '.join(allowed[: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标签)"""
# 简单去除HTML标签
plain = re.sub(r'<[^>]+>', '', text)
word_count = len(plain.strip())
min_words = PLATFORM_RULES.get(platform, {}).get("min_word_count", 1000)
if word_count < min_words:
self.issues.append({
"type": "内容完整度",
"category": "字数不足",
"detail": f"当前{word_count}字,低于平台要求{min_words}",
"suggestion": "扩写内容至最低要求"
})
def _check_required_sections(self, text: str):
"""检查是否包含必要章节(如引言、核心观点、总结等)"""
required_headings = [
"引言", "核心观点", "受众痛点", "总结", "行动指南"
]
missing = []
for heading in required_headings:
# 检查 h2 或 h3 中是否出现 heading
if not re.search(r'<h[23][^>]*>.*' + re.escape(heading) + r'.*</h[23]>', text, re.IGNORECASE):
missing.append(heading)
if missing:
self.issues.append({
"type": "结构完整",
"category": "章节缺失",
"detail": f"缺少必要章节:{', '.join(missing)}",
"suggestion": "补充缺失章节"
})
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) -> Dict:
"""便捷函数:执行完整合规检查"""
checker = ComplianceChecker()
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))