Yuzhiran Dev
401 lines
15 KiB
Python
401 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
import json, datetime, logging, sys, re
|
||
from pathlib import Path
|
||
from typing import Dict, List
|
||
|
||
PROJECT_ROOT = Path(__file__).parent.parent
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
sys.path.insert(0, str(PROJECT_ROOT / 'platform' / 'backend'))
|
||
|
||
from db_helper import get_topic_by_id, update_topic_status
|
||
try:
|
||
from app.core.nvidia_client import call_llm
|
||
HAVE_LLM = True
|
||
except ImportError:
|
||
HAVE_LLM = False
|
||
|
||
import mistune
|
||
|
||
DATA_DIR = PROJECT_ROOT / "automation" / "data"
|
||
OUTLINE_DIR = DATA_DIR / "outlines"
|
||
RELEASE_DIR = DATA_DIR / "releases"
|
||
TEMPLATES_DIR = PROJECT_ROOT / "automation" / "templates"
|
||
LOGS_DIR = PROJECT_ROOT / "automation" / "logs"
|
||
TODAY = datetime.datetime.now().strftime("%Y-%m-%d")
|
||
GEN_TIME = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s - %(levelname)s - %(message)s',
|
||
handlers=[
|
||
logging.FileHandler(LOGS_DIR / f"writer_{TODAY}.log"),
|
||
logging.StreamHandler()
|
||
]
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_md_parser = mistune.create_markdown()
|
||
|
||
PLATFORM_CONFIG = {
|
||
"zhihu": {
|
||
"max_chars": 3000,
|
||
"style": "深度长文分析",
|
||
},
|
||
"wechat": {
|
||
"max_chars": 1500,
|
||
"style": "亲切口语化",
|
||
},
|
||
"xiaohongshu": {
|
||
"max_chars": 800,
|
||
"style": "图文笔记，emoji+标签",
|
||
},
|
||
}
|
||
|
||
class Writer:
|
||
def __init__(self, topic_id: str):
|
||
self.topic_id = topic_id
|
||
self.topic = self._load_topic()
|
||
outline_file = OUTLINE_DIR / TODAY / f"{topic_id}_outline.md"
|
||
if not outline_file.exists():
|
||
raise FileNotFoundError(f"Outline not found: {outline_file}")
|
||
self.outline_content = outline_file.read_text(encoding='utf-8')
|
||
self.release_dir = RELEASE_DIR / TODAY
|
||
self.release_dir.mkdir(parents=True, exist_ok=True)
|
||
research_file = DATA_DIR / "research" / TODAY / f"{topic_id}_research.md"
|
||
self.research_notes = research_file.read_text(encoding='utf-8') if research_file.exists() else ""
|
||
|
||
def _load_topic(self) -> Dict:
|
||
topic = get_topic_by_id(self.topic_id)
|
||
if not topic:
|
||
raise ValueError(f"Topic {self.topic_id} not found")
|
||
return topic
|
||
|
||
def _clean_title(self, title: str) -> str:
|
||
title = re.sub(r'[（(]约\s*\d+字[）)]', '', title)
|
||
title = re.sub(r'[（(]MVP[）)]', '', title)
|
||
title = re.sub(r'[（(][^）)]*?[）)]', '', title)
|
||
return title.strip()
|
||
|
||
def _parse_outline_sections(self) -> List[Dict]:
|
||
sections = []
|
||
current = None
|
||
for line in self.outline_content.splitlines():
|
||
if line.startswith("# "):
|
||
if current:
|
||
sections.append(current)
|
||
current = {"level": 1, "title": line[2:].strip(), "content": ""}
|
||
elif line.startswith("## "):
|
||
if current:
|
||
sections.append(current)
|
||
current = {"level": 2, "title": line[3:].strip(), "content": ""}
|
||
elif line.startswith("### "):
|
||
if current:
|
||
sections.append(current)
|
||
current = {"level": 3, "title": line[4:].strip(), "content": ""}
|
||
else:
|
||
if current and line.strip():
|
||
current['content'] = current.get('content', '') + line + "\n"
|
||
if current:
|
||
sections.append(current)
|
||
return sections
|
||
|
||
def _expand_section(self, section: Dict) -> str:
|
||
content = section.get('content', '').strip()
|
||
if len(content) > 200:
|
||
return content
|
||
if HAVE_LLM and len(content) < 150:
|
||
logger.info(f"使用 LLM 扩写章节: {section['title']}")
|
||
prompt = f"""你是一个真人写作者，正在写一篇关于「{self.topic['title']}」的文章。现在写「{section['title']}」这一节。
|
||
|
||
笔记要点：
|
||
{content}
|
||
|
||
要求（逐条对照）：
|
||
### 价值
|
||
- 回答读者一个具体问题或解决一个困惑
|
||
- 每个论点配真实案例或数据，不写空话
|
||
- **所有数据必须使用2025-2026年最新数据**，禁用过时数据
|
||
- 结束时读者要有「学到了」的感觉
|
||
|
||
### 真人感
|
||
- 用「你」或「我们」视角，不要用「我」
|
||
- 像人在自然说话，不是AI组装文字
|
||
- 避免「首先」「其次」「总的来说」「综上所述」「值得注意的是」
|
||
- 段落短，2-4句一段，节奏有变化
|
||
- 适当用反问或口语化表达
|
||
|
||
### SEO
|
||
- 自然融入1-2个目标搜索词，不生硬堆砌
|
||
- 第一句包含核心关键词
|
||
|
||
### 长度
|
||
- 200-400字，写到点子上就停
|
||
|
||
直接输出段落正文。"""
|
||
try:
|
||
expanded = call_llm(prompt, temperature=0.6, max_tokens=1500)
|
||
if expanded and len(expanded.strip()) > len(content):
|
||
return expanded.strip()
|
||
except Exception as e:
|
||
logger.warning(f"LLM 扩写失败: {e}")
|
||
|
||
# Fallback: 将 bullet points 展开为段落
|
||
lines = [l.strip() for l in content.split('\n') if l.strip()]
|
||
if lines:
|
||
sentences = []
|
||
for line in lines:
|
||
text = line.lstrip('- *').strip()
|
||
if text:
|
||
for prefix in ['- ', '* ', '1. ', '2. ', '3. ', '4. ', '5. ']:
|
||
if line.startswith(prefix):
|
||
text = line[len(prefix):]
|
||
break
|
||
if text[-1] not in '。！？；':
|
||
text += '。'
|
||
sentences.append(text)
|
||
if sentences:
|
||
return ' '.join(sentences)
|
||
return content
|
||
|
||
def generate_full_markdown(self) -> str:
|
||
sections = self._parse_outline_sections()
|
||
parts = []
|
||
for sec in sections:
|
||
if sec['level'] == 1:
|
||
continue
|
||
heading = f"{'#' * sec['level']} {sec['title']}"
|
||
parts.append(heading)
|
||
if sec.get('content'):
|
||
expanded = self._expand_section(sec)
|
||
parts.append(expanded + "\n")
|
||
full_md = "\n".join(parts).strip()
|
||
return full_md
|
||
|
||
def _adapt_for_platform(self, markdown: str, platform: str) -> str:
|
||
cfg = PLATFORM_CONFIG[platform]
|
||
max_c = cfg['max_chars']
|
||
lines = markdown.split('\n')
|
||
|
||
if platform == "xiaohongshu":
|
||
result = []
|
||
char_count = 0
|
||
for line in lines:
|
||
if char_count >= max_c:
|
||
break
|
||
if line.startswith('## '):
|
||
line = f"## ✨ {line[3:]}"
|
||
elif line.startswith('### '):
|
||
line = f"### 💡 {line[4:]}"
|
||
result.append(line)
|
||
char_count += len(line)
|
||
adapted = '\n'.join(result)
|
||
if adapted.count('#') == 0:
|
||
adapted = f"# {self.topic['title']}\n\n{adapted}"
|
||
return adapted
|
||
|
||
if platform == "wechat":
|
||
result = []
|
||
for line in lines:
|
||
line = line.replace('我', '你')
|
||
if line.startswith('### '):
|
||
result.append(f"\n**{line[4:]}**\n")
|
||
elif line.startswith('## '):
|
||
result.append(f"\n**{line[3:]}**\n")
|
||
elif line.strip() and len(line) > 80:
|
||
sentences = [s.strip() for s in line.replace('。', '。\n').split('\n') if s.strip()]
|
||
for s in sentences:
|
||
if s:
|
||
result.append(s)
|
||
else:
|
||
result.append(line)
|
||
adapted = '\n'.join(result)
|
||
if len(adapted) > max_c:
|
||
adapted = adapted[:max_c]
|
||
last = max(adapted.rfind('。'), adapted.rfind('\n'), adapted.rfind('！'))
|
||
if last > max_c // 2:
|
||
adapted = adapted[:last + 1]
|
||
return adapted
|
||
|
||
return markdown
|
||
|
||
def _get_platform_tags(self, platform: str) -> str:
|
||
field = self.topic.get('field', '')
|
||
title = self.topic.get('title', '')
|
||
core = self.topic.get('core_concept', '')
|
||
|
||
tag_prompts = {
|
||
"zhihu": f"为以下文章生成知乎标签（3-5个）。标题：{title} 领域：{field} 核心观点：{core} 每个2-4字。直接输出标签，空格分隔。不要输出思考过程。",
|
||
"wechat": f"为以下文章生成公众号标签（3-5个）。标题：{title} 领域：{field} 核心观点：{core} 每个2-4字。直接输出标签，空格分隔。不要输出思考过程。",
|
||
"xiaohongshu": f"为以下文章生成小红书标签（3-5个）。标题：{title} 领域：{field} 核心观点：{core} 每个2-4字。直接输出标签，空格分隔。不要输出思考过程。",
|
||
}
|
||
|
||
if HAVE_LLM:
|
||
prompt = tag_prompts.get(platform, f"根据文章信息生成适合{platform}的标签。标题：{title} 领域：{field} 核心观点：{core} 直接输出标签，空格分隔。")
|
||
try:
|
||
tags_text = call_llm(prompt, temperature=0.2, max_tokens=500)
|
||
if tags_text:
|
||
tags = [t.strip('#') for t in tags_text.strip().split() if t.strip('#')]
|
||
if tags:
|
||
return " ".join(f'<span class="tag">{t}</span>' for t in tags[:5])
|
||
except Exception:
|
||
pass
|
||
|
||
tags = []
|
||
if field:
|
||
import re
|
||
parts = re.split(r'[/、与和及]', field)
|
||
for p in parts:
|
||
p = p.strip()
|
||
if len(p) >= 2:
|
||
tags.append(p)
|
||
if len(parts) == 1 and len(parts[0]) > 4:
|
||
for i in range(0, len(parts[0]), 2):
|
||
chunk = parts[0][i:i+2]
|
||
if len(chunk) == 2:
|
||
tags.append(chunk)
|
||
tags.pop(0)
|
||
|
||
platform_extra = {"zhihu": ["职场"], "xiaohongshu": ["生活"]}
|
||
for t in platform_extra.get(platform, []):
|
||
if t not in tags:
|
||
tags.append(t)
|
||
|
||
if not tags:
|
||
tags = ["科技"]
|
||
|
||
seen = set()
|
||
return " ".join(f'<span class="tag">{t}</span>' for t in tags if t not in seen and not seen.add(t))
|
||
|
||
def _optimize_title(self, platform: str) -> str:
|
||
original = self.topic['title']
|
||
if not HAVE_LLM:
|
||
return original
|
||
|
||
title_templates = {
|
||
"zhihu": f"""你是一个知乎用户，在给自己的深度回答起高点击率标题。
|
||
|
||
原文标题：{original}
|
||
领域：{self.topic.get('field', '')}
|
||
|
||
要求：
|
||
- 有信息量：一看就知道能解决什么问题
|
||
- 含知乎搜索关键词（SEO）
|
||
- 带数字或对比最好（「3个方法」「从…到…」）
|
||
- 20字以内
|
||
- 参考知乎真实高赞标题，不要套路句式
|
||
- 避免「如何…」废句式、「XXX指南/手册/全攻略」
|
||
- 直接输出3个标题选项，每行一个，不要输出思考过程
|
||
|
||
生成 3 个选项，每行一个。""",
|
||
|
||
"wechat": f"""你是一个公众号作者，在给可能10万+的文章起标题。
|
||
|
||
原文标题：{original}
|
||
领域：{self.topic.get('field', '')}
|
||
|
||
要求：
|
||
- 制造好奇心和点击欲，让人觉得不点开会错过
|
||
- 包含微信搜索关键词（微信SEO）
|
||
- 口语化，不要书面腔
|
||
- 不要感叹号堆砌，不要「重磅/震惊/紧急」
|
||
- 字数15-25字最佳
|
||
- 直接输出3个标题选项，每行一个，不要输出思考过程
|
||
|
||
生成 3 个选项，每行一个。""",
|
||
|
||
"xiaohongshu": f"""你是一个小红书用户，在给笔记起能上热门推荐的标题。
|
||
|
||
原文标题：{original}
|
||
领域：{self.topic.get('field', '')}
|
||
|
||
要求：
|
||
- 20字以内
|
||
- 采用爆款模式：数字+结果/痛点+方案/反常识观点
|
||
- 包含小红书搜索关键词（SEO）
|
||
- 带1个emoji点缀
|
||
- 有场景感/结果感
|
||
- 不要「必看/收藏/码住」
|
||
- 像真实用户写的，不是运营写的
|
||
- 直接输出3个标题选项，每行一个，不要输出思考过程
|
||
|
||
生成 3 个选项，每行一个。""",
|
||
}
|
||
|
||
prompt = title_templates.get(platform, f"给以下文章改个吸引人的{platform}标题：{original}")
|
||
try:
|
||
resp = call_llm(prompt, temperature=0.7, max_tokens=500)
|
||
titles = []
|
||
for line in resp.strip().split('\n'):
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
line = re.sub(r'^\d+[.、)\s]+', '', line)
|
||
line = line.strip('*#- \t')
|
||
if line:
|
||
titles.append(line)
|
||
if titles:
|
||
logger.info(f"标题优化 [{platform}]: {titles[0][:50]}...")
|
||
return titles[0]
|
||
except Exception as e:
|
||
logger.warning(f"标题优化失败: {e}")
|
||
return original
|
||
|
||
def generate_platform_html(self, markdown: str, platform: str) -> str:
|
||
title = self._optimize_title(platform)
|
||
adapted = self._adapt_for_platform(markdown, platform)
|
||
tpl_path = TEMPLATES_DIR / f"{platform}.html"
|
||
if tpl_path.exists():
|
||
template = tpl_path.read_text(encoding='utf-8')
|
||
else:
|
||
template = "<!DOCTYPE html><html><head><meta charset='UTF-8'><title>{{TITLE}}</title><meta name='viewport' content='width=device-width'><style>body{max-width:800px;margin:0 auto;padding:20px;font-family:-apple-system,sans-serif;line-height:1.8}</style></head><body><h1>{{TITLE}}</h1><!-- CONTENT --></body></html>"
|
||
|
||
html = template.replace("{{TITLE}}", title).replace("{{DATE}}", TODAY).replace("{{GEN_TIME}}", GEN_TIME)
|
||
html_content = _md_parser(adapted)
|
||
html = html.replace("<!-- CONTENT -->", html_content)
|
||
|
||
tags_html = self._get_platform_tags(platform)
|
||
if tags_html:
|
||
html = html.replace("<!-- TAGS -->", tags_html)
|
||
else:
|
||
html = html.replace("<!-- TAGS -->", "")
|
||
|
||
return html
|
||
|
||
def save_html(self, html: str, platform: str) -> Path:
|
||
out_dir = self.release_dir / platform
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
filename = f"{platform}_{self.topic_id}.html"
|
||
out_path = out_dir / filename
|
||
out_path.write_text(html, encoding='utf-8')
|
||
logger.info(f"HTML 生成: {out_path}")
|
||
return out_path
|
||
|
||
def mark_draft(self):
|
||
update_topic_status(self.topic_id, 'review')
|
||
logger.info(f"选题 {self.topic_id} 状态已更新为待审查（数据库）")
|
||
|
||
def run(self):
|
||
logger.info("开始撰写阶段")
|
||
markdown = self.generate_full_markdown()
|
||
results = {}
|
||
for platform in ["zhihu", "wechat", "xiaohongshu"]:
|
||
html = self.generate_platform_html(markdown, platform)
|
||
results[platform] = str(self.save_html(html, platform))
|
||
self.mark_draft()
|
||
logger.info(f"撰写完成，状态已更新为待审查")
|
||
return {"ok": True, "files": results}
|
||
|
||
def main():
|
||
import argparse
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument('--topic-id', required=True, help='选题ID')
|
||
args = parser.parse_args()
|
||
|
||
w = Writer(args.topic_id)
|
||
result = w.run()
|
||
print(json.dumps(result, ensure_ascii=False))
|
||
sys.exit(0 if result['ok'] else 1)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|