#!/usr/bin/env python3 """ 内容合规审查模块 检查文章是否符合法律法规、平台规则、品牌规范 """ import re import json 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 Exception: 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'([^<]+)', text) or re.search(r']*>([^<]+)', 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'
([^<]+)
', text) or re.search(r'
([^<]+)
', 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'

(.*?)

', 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']+>', '', 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'

(.*?)

', 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']*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'\b(19\d{2}|20[0-4]\d)\b', 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 = "

测试

内容涉及赌博网站" result = check_article(test_html, "zhihu") print(json.dumps(result, ensure_ascii=False, indent=2))