diff --git a/scripts/compliance_checker.py b/scripts/compliance_checker.py
index ad501e9..b8a57b6 100644
--- a/scripts/compliance_checker.py
+++ b/scripts/compliance_checker.py
@@ -55,10 +55,29 @@ AI_TELTALES = [
"众所周知",
"毋庸置疑",
"不知大家有没有发现",
+ "不可否认",
+ "毫无疑义",
+ "从某种意义上",
+ "从某种程度上",
+ "在一定程度上",
+ "换而言之",
+ "换言之",
+ "从本质",
+ "归根结底",
+ "说到底",
+ "这为我们提供了",
+ "为我们提供了宝贵的",
+ "引发了我们",
+ "不得不让人思考",
+ "引人深思",
+ "毫无悬念",
+ "毫无意外",
+ "毫无争议",
]
_cached_sensitive_words = None
_cached_platform_rules = None
+_cached_ai_telltales = None
def _load_sensitive_words():
global _cached_sensitive_words
@@ -121,6 +140,34 @@ def _load_platform_rules():
_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:
"""合规审查器"""
@@ -349,7 +396,8 @@ class ComplianceChecker:
def _check_ai_telltales(self, text: str):
"""检查AI套话——正文中出现这些模式说明AI写作痕迹明显"""
plain = re.sub(r'<[^>]+>', '', text)
- for pattern in AI_TELTALES:
+ patterns = _load_ai_telltales()
+ for pattern in patterns:
if re.search(pattern, plain):
self.issues.append({
"type": "内容质量",
@@ -359,26 +407,16 @@ class ComplianceChecker:
})
def _check_pronoun_consistency(self, text: str, platform: str):
- """检查人称一致性（尤其是微信文章）"""
- if platform != "wechat":
- return
+ """检查人称一致性"""
plain = re.sub(r'<[^>]+>', '', text)
has_ni = '你' in plain
has_nimen = '你们' in plain
- has_women = '我们' in plain
if has_nimen and has_ni:
self.issues.append({
"type": "内容质量",
"category": "人称混用",
- "detail": "微信文章中同时使用「你」和「你们」，建议统一为「你」",
- "suggestion": "将所有「你们」替换为「你」，保持与读者的单数对话感"
- })
- if has_women and has_ni:
- self.issues.append({
- "type": "内容质量",
- "category": "人称混用",
- "detail": "微信文章中同时使用「我们」和「你」，建议统一视角",
- "suggestion": "将「我们」替换为「你」或「我」，保持与读者对话而非说教"
+ "detail": "同时使用「你」和「你们」，建议统一为「你」",
+ "suggestion": "将「你们」替换为「你」，保持对话感"
})
def _check_reading_experience(self, text: str, platform: str = ""):
diff --git a/scripts/db_helper.py b/scripts/db_helper.py
index ffc4b1a..1cd3dcd 100644
--- a/scripts/db_helper.py
+++ b/scripts/db_helper.py
@@ -89,8 +89,14 @@ def update_topic_status(topic_id: str, status: str, compliance_score: Optional[i
topic.compliance_score = compliance_score
if reviewed_at is not None:
topic.reviewed_at = reviewed_at
- if status in ['ready', 'published']:
+ if status == 'review' and topic.generated_at is None:
topic.generated_at = datetime.now()
+ topic.reviewed_at = topic.reviewed_at or datetime.now()
+ if status == 'ready':
+ topic.ready_at = datetime.now().date()
+ topic.reviewed_at = topic.reviewed_at or datetime.now()
+ if status == 'published':
+ topic.published_at = datetime.now().date()
db.commit()
return True
finally:
@@ -243,7 +249,6 @@ def save_article(topic_id: str, platform: str, html_content: str, db: Optional[S
now = datetime.now()
if existing:
existing.html_content = html_content
- existing.compliance_score = existing.compliance_score
else:
article = Article(
id=article_id,
diff --git a/scripts/prompt_loader.py b/scripts/prompt_loader.py
index 6eb4985..852c90d 100644
--- a/scripts/prompt_loader.py
+++ b/scripts/prompt_loader.py
@@ -44,23 +44,33 @@ _PROMPT_DEFAULTS = {
"temperature": 0.6, "max_tokens": 2000,
"variables": ["gaps"],
},
- "section_expansion": {
- "content": "你是一个资深作者，正在写一篇关于「{topic_title}」的文章。请写「{section_title}」这一节。\n\n今天日期：{date}。\n\n笔记要点：\n{content}\n\n输出要求（200-800字，3-6段）：\n- 每段3-5句，段间空行分隔\n- 用「你」或「我」视角，自然口语化\n- 从具体场景或痛点切入，禁止以「引言」「核心观点」「总结」「开头钩子」这类标签开头\n- 每个论点配具体案例或数据，不要空谈道理\n- 回答读者「所以呢」——这对他有什么用\n\n格式：\n- 纯段落文字，不要标题/列表/标记（#、-、*、1.等）\n- 不要出现「首先其次最后」「总的来说」「值得注意的是」「综上所述」等AI套话\n- 来源URL请放在文末括号内",
- "temperature": 0.75, "max_tokens": 3000,
+ "section_expansion_zhihu": {
+ "content": "写「{topic_title}」文章中「{section_title}」这一节。\n\n今天：{date}。\n笔记要点：\n{content}\n\n写400-800字，3-6段。每段围绕一个论点展开：先摆事实/数据，再分析原因，最后说清这跟读者有什么关系。\n\n注意：\n- 每个论点必须配具体数据或案例（至少一个），不要空谈。引用研究、报告或行业事件\n- 用「你」称呼读者，像是在知乎回答里跟人讨论\n- 从具体问题切入，不要以「引言/总结/核心观点」这类标签开头\n- 回答「所以呢」——这条信息对读者有什么用\n- 避免「首先其次最后」「总的来说」「综上所述」「值得注意的是」这些表达\n- 来源放在文末括号内\n- 纯段落，不要列表/标题/序号标记",
+ "temperature": 0.78, "max_tokens": 4000,
+ "variables": ["topic_title", "section_title", "date", "content"],
+ },
+ "section_expansion_wechat": {
+ "content": "写「{topic_title}」文章中「{section_title}」这一节。\n\n今天：{date}。\n笔记要点：\n{content}\n\n写300-500字，用「我」的口吻叙述——像在跟朋友聊天，不是写文章。\n\n注意：\n- 从个人经历或感受切入，不要讲道理\n- 每个观点背后要有一个具体故事或观察\n- 段与段之间有情绪的起承转合，不是逻辑推理\n- 读起来像一个人在说话，而不是一篇文章在论述\n- 避免「首先其次最后」「总的来说」「综上所述」「值得注意的是」这些套路\n- 纯段落，不要标题/列表/序号标记",
+ "temperature": 0.8, "max_tokens": 3000,
+ "variables": ["topic_title", "section_title", "date", "content"],
+ },
+ "section_expansion_xiaohongshu": {
+ "content": "写「{topic_title}」笔记中「{section_title}」这一节。\n\n今天：{date}。\n笔记要点：\n{content}\n\n写100-200字，1-3小段。每段1-2句，直接给干货。\n\n注意：\n- 直接给结论或方法，不要铺垫\n- 可以用emoji（1个/段，不要堆砌）\n- 像博主在分享亲测有效的经验\n- 不要讲道理，只讲「是什么」和「怎么做」\n- 用「你」或直接省略主语\n- 避免「首先其次最后」「总的来说」「综上所述」「值得注意的是」这些套路\n- 纯段落，不要列表/标题/序号标记",
+ "temperature": 0.75, "max_tokens": 2000,
"variables": ["topic_title", "section_title", "date", "content"],
},
"title_optimize_zhihu": {
- "content": "你是一个知乎高赞标题专家。为以下文章起3个标题。\n\n文章主题：{title}\n核心观点：{core}\n受众痛点：{pain}\n领域：{field}\n\n要求：\n- 信息密度高，包含搜索关键词\n- 优先使用数字、对比、悬念、痛点\n- 20字以内\n- 不要「如何...」开头\n- 有独特判断和立场，能引发讨论\n- 制造「不点开就亏了」的紧迫感\n\n直接输出3个标题，每行一个，不要序号和说明。",
+ "content": "你是知乎老用户，经常上热榜。给这篇文章起3个标题候选：\n\n文章主题：{title}\n核心观点：{core}\n受众痛点：{pain}\n领域：{field}\n\n感觉像在跟人推荐一篇值得辩论的文章。信息要密、20字内、有明确立场、让人想点进来说两句。不要「如何」开头。\n\n每行一个，不要编号。",
"temperature": 0.8, "max_tokens": 1500,
"variables": ["title", "core", "pain", "field"],
},
"title_optimize_wechat": {
- "content": "你是一个公众号标题专家。为以下文章起3个标题。\n\n文章主题：{title}\n核心观点：{core}\n\n要求：\n- 包含身份标签（如「打工人」「30岁后」「普通上班族」）\n- 包含情绪钩子（焦虑/反常识/后悔/稀缺）\n- 包含微信SEO关键词\n- 15-25字，口语化\n- 忌笼统，越具体越好\n- 不要感叹号堆砌\n\n参考公式：\n- 「身份+痛点+方案」：打工人学了一堆AI工具，为什么还在加班？\n- 「反常识+数据」：用了AI效率反而更低了？73%的人掉进了这个坑\n- 「结果+身份」：每天省出2小时后，我才发现自己以前有多傻\n\n直接输出3个标题，每行一个，不要序号和说明。",
+ "content": "你是公众号编辑。这篇文章面向{field}领域的读者，帮起3个标题：\n\n文章主题：{title}\n核心观点：{core}\n\n要有身份代入感（打工人/30岁+/普通上班族）、有情绪钩子（焦虑/反常识/后悔）、含微信搜索关键词。15-25字，口语化，越具体越好。参考这种风格：「用了AI效率反而更低了？73%的人掉进了这个坑」\n\n每行一个，不要编号。",
"temperature": 0.8, "max_tokens": 1500,
- "variables": ["title", "core"],
+ "variables": ["title", "core", "field"],
},
"title_optimize_xhs": {
- "content": "你是一个小红书标题专家。为以下文章起3个标题。\n\n标题：{title}\n核心观点：{core}\n\n要求：\n- 18字以内\n- 爆款公式：身份/场景+数字+结果，或痛点+方案+反差\n- 包含小红书SEO关键词\n- 1个精确emoji（不要用🔥💥❌❓这几个滥用的）\n- 有场景感、结果感\n- 忌笼统：不要「必看/收藏/码住/绝了」\n\n直接输出3个标题，每行一个，不要序号和说明。",
+ "content": "你是小红书博主。给这篇笔记起3个标题：\n\n笔记主题：{title}\n核心：{core}\n\n每条18字以内，有具体场景和结果感。可以加1个精确emoji（不要🔥💥❌❓）。参考：「用了两年AI工具，推荐这5个就够了」这种风格。\n\n每行一个，不要编号。",
"temperature": 0.8, "max_tokens": 1500,
"variables": ["title", "core"],
},
diff --git a/scripts/writer.py b/scripts/writer.py
index 846b325..e4e09dd 100644
--- a/scripts/writer.py
+++ b/scripts/writer.py
@@ -43,9 +43,9 @@ logger = logging.getLogger(__name__)
_md_parser = mistune.create_markdown()
_FALLBACK_PLATFORM_CONFIG = {
- "zhihu": {"max_chars": 3000, "style": "深度长文分析", "min_chars": 1500},
- "wechat": {"max_chars": 1500, "style": "亲切口语化", "min_chars": 800},
- "xiaohongshu": {"max_chars": 800, "style": "图文笔记，emoji+标签", "min_chars": 300},
+ "zhihu": {"max_chars": 8000, "style": "深度长文分析", "min_chars": 3000},
+ "wechat": {"max_chars": 4000, "style": "个人叙事对话感", "min_chars": 2000},
+ "xiaohongshu": {"max_chars": 1000, "style": "图文笔记，精炼实用", "min_chars": 400},
}
def _load_platform_config() -> dict:
@@ -144,28 +144,30 @@ class Writer:
bullet_count = sum(1 for l in lines if l.startswith(('- ', '* ', '**', '+ ')))
return bullet_count / len(lines) > 0.4
- def _expand_section(self, section: Dict) -> str:
+ def _expand_section(self, section: Dict, platform: str = "zhihu") -> str:
content = section.get('content', '').strip()
- # 大纲要点格式（>40% 行以 -/*/** 开头）或内容过短（<100字）应由 LLM 展开为连贯段落
- if HAVE_LLM and (self._is_bullet_only(content) or (content and len(content) < 100)):
- logger.info(f"使用 LLM 扩写章节(要点→段落): {section['title']}")
- prompt = get_prompt("section_expansion",
+ platform_prompt_key = f"section_expansion_{platform}"
+ if platform not in ("zhihu", "wechat", "xiaohongshu"):
+ platform_prompt_key = "section_expansion_zhihu"
+
+ if HAVE_LLM and content:
+ logger.info(f"LLM 扩写 [{platform}]: {section['title']} ({len(content)} chars)")
+ prompt = get_prompt(platform_prompt_key,
topic_title=self.topic['title'],
section_title=section['title'],
date=datetime.datetime.now().strftime('%Y年%m月%d日'),
content=content,
)
try:
- params = get_prompt_params("section_expansion")
+ params = get_prompt_params(platform_prompt_key) or {"temperature": 0.75, "max_tokens": 3000}
expanded = call_llm(prompt, temperature=params.get("temperature", 0.75), max_tokens=params.get("max_tokens", 3000))
if expanded:
cleaned = self._clean_markdown(expanded.strip())
if cleaned:
return cleaned
except Exception as e:
- logger.warning(f"LLM 扩写失败: {e}")
+ logger.warning(f"LLM 扩写失败 [{platform}]: {e}")
- # Fallback: 将 bullet points 展开为段落（过滤噪音行）
lines = [l.strip() for l in content.split('\n') if not self._is_outline_noise(l)]
if lines:
sentences = []
@@ -185,17 +187,16 @@ class Writer:
return self._clean_markdown(result)
return ''
- def generate_full_markdown(self) -> str:
+ def generate_platform_markdown(self, platform: str = "zhihu") -> str:
sections = self._parse_outline_sections()
parts = []
for sec in sections:
if sec['level'] == 1:
if sec.get('content'):
- expanded = self._expand_section(sec)
+ expanded = self._expand_section(sec, platform)
if expanded:
parts.append(expanded + "\n")
continue
- # 跳过大纲结构噪音节点
title_stripped = sec['title'].strip()
if title_stripped in ('文章大纲', '大纲', '文章结构', '结构'):
continue
@@ -204,7 +205,7 @@ class Writer:
heading = f"{'#' * sec['level']} {sec['title']}"
parts.append(heading)
if sec.get('content'):
- expanded = self._expand_section(sec)
+ expanded = self._expand_section(sec, platform)
parts.append(expanded + "\n")
full_md = "\n".join(parts).strip()
# 收集所有引用来源，统一添加到文末
@@ -286,8 +287,6 @@ class Writer:
if platform == "wechat":
result = []
for line in lines:
- # 人称统一：我们→我，你们→你
- line = line.replace('我们', '我').replace('你们', '你').replace('我', '你')
if line.startswith('### '):
result.append(f"\n**{line[4:]}**\n")
elif line.startswith('## '):
@@ -370,6 +369,7 @@ class Writer:
prompt = get_prompt("title_optimize_wechat",
title=original,
core=self.topic.get('core_concept', ''),
+ field=self.topic.get('field', ''),
)
elif platform == "xiaohongshu":
prompt = get_prompt("title_optimize_xhs",
@@ -459,10 +459,10 @@ class Writer:
logger.info(f"选题 {self.topic_id} 状态已更新为待审查（数据库）")
def run(self):
- logger.info("开始撰写阶段")
- markdown = self.generate_full_markdown()
+ logger.info("开始撰写阶段（三平台独立展开）")
results = {}
for platform in ["zhihu", "wechat", "xiaohongshu"]:
+ markdown = self.generate_platform_markdown(platform)
html = self.generate_platform_html(markdown, platform)
results[platform] = str(self.save_html(html, platform))
self.mark_draft()