yuzhiran
d11d7f4980
db_helper.py: save_article now calculates and persists word_count generator.py: run_creator_blocking sets word_count for HTML-imported articles writer.py: fix title regex stripping content-leading numbers (35岁后→岁后) trends.py: fix Baidu hot_score str/int type comparison crash database.py: add missing content_tasks.org_id ALTER TABLE migration schemas.py + topics.py: topic list API returns article_count + articles[] previews topics.html: table view and card view show article badges with word counts, clickable to open preview Ultraworked with Sisyphus Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
653 lines
27 KiB
Python
653 lines
27 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
平台适配文章撰写
|
||
根据大纲和平台配置（字数/格式/配图要求），为知乎/公众号/小红书各平台生成适配内容
|
||
"""
|
||
import json, datetime, logging, sys, re
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional
|
||
|
||
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, save_article
|
||
from content_cleaner import strip_thinking, strip_ai_preface, clean_markdown_content, clean_html_content
|
||
from prompt_loader import get_prompt, get_prompt_params
|
||
from image_generator import insert_lead_image
|
||
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"
|
||
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()
|
||
|
||
_FALLBACK_PLATFORM_CONFIG = {
|
||
"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:
|
||
try:
|
||
from app.database import SessionLocal
|
||
from app.models import PlatformConfig
|
||
db = SessionLocal()
|
||
configs = db.query(PlatformConfig).all()
|
||
db.close()
|
||
result = {}
|
||
for c in configs:
|
||
result[c.platform] = {
|
||
"max_chars": c.max_words or _FALLBACK_PLATFORM_CONFIG.get(c.platform, {}).get("max_chars", 3000),
|
||
"style": c.default_format or _FALLBACK_PLATFORM_CONFIG.get(c.platform, {}).get("style", "深度内容"),
|
||
"min_chars": c.min_words or _FALLBACK_PLATFORM_CONFIG.get(c.platform, {}).get("min_chars", 300),
|
||
}
|
||
return result
|
||
except Exception:
|
||
pass
|
||
return dict(_FALLBACK_PLATFORM_CONFIG)
|
||
|
||
PLATFORM_CONFIG = _load_platform_config()
|
||
|
||
PLATFORM_NAMES = {
|
||
"zhihu": "知乎专栏",
|
||
"wechat": "微信公众号",
|
||
"xiaohongshu": "小红书",
|
||
}
|
||
|
||
def _extract_description(content: str, max_len: int = 200) -> str:
|
||
"""从 markdown 正文提取第一段有意义的文字作为 description"""
|
||
text = re.sub(r'^#\s+.*$', '', content, flags=re.MULTILINE)
|
||
text = re.sub(r'[#*>`~\[\]()\n]', ' ', text)
|
||
paragraphs = [p.strip() for p in text.split('\n\n') if p.strip()]
|
||
for p in paragraphs:
|
||
p = re.sub(r'\s+', ' ', p).strip()
|
||
if len(p) >= 15 and not p.startswith('http'):
|
||
return p[:max_len]
|
||
return content.replace('\n', ' ')[:max_len]
|
||
|
||
|
||
def _extract_tags_list(tags_html: str) -> list:
|
||
"""从 HTML tags 块提取纯标签列表"""
|
||
return re.findall(r'<span class="tag">([^<]+)</span>', tags_html)
|
||
|
||
|
||
def _detect_content_type(content: str) -> str:
|
||
"""检测内容类型: article / listicle / howto / faq / review"""
|
||
if not content:
|
||
return "article"
|
||
c = content.lower()
|
||
howto_score = 0
|
||
for p in ['步骤', '第一步', '第二步', '首先', '然后', '最后', 'step 1', 'step 2']:
|
||
if p in c:
|
||
howto_score += 1
|
||
faq_score = 0
|
||
for p in ['q：', 'a：', 'q:', 'a:', '问：', '答：', '什么', '如何', '怎么', '为什么']:
|
||
if p in c:
|
||
faq_score += 1
|
||
listicle_score = c.count('\n- ') + c.count('\n* ') + c.count('\n1. ')
|
||
if faq_score >= 4:
|
||
return "faq"
|
||
if howto_score >= 2:
|
||
return "howto"
|
||
if listicle_score >= 3:
|
||
return "listicle"
|
||
return "article"
|
||
|
||
|
||
def _extract_faq_pairs(content: str, max_pairs: int = 5) -> list:
|
||
"""从 markdown 中提取 FAQ 问答对"""
|
||
pairs = []
|
||
lines = content.split('\n')
|
||
current_q = None
|
||
for line in lines:
|
||
q_match = re.match(r'^(?:[Qq]|问)[：:]\s*(.+)$', line)
|
||
if q_match:
|
||
current_q = q_match.group(1).strip()
|
||
continue
|
||
a_match = re.match(r'^(?:[Aa]|答)[：:]\s*(.+)$', line)
|
||
if a_match and current_q:
|
||
pairs.append({"question": current_q, "answer": a_match.group(1).strip()[:200]})
|
||
current_q = None
|
||
if len(pairs) >= max_pairs:
|
||
break
|
||
return pairs
|
||
|
||
|
||
def _extract_howto_steps(content: str, max_steps: int = 8) -> list:
|
||
"""从 markdown 中提取 HowTo 步骤"""
|
||
steps = []
|
||
for m in re.finditer(r'(?:步骤|Step)\s*(\d+)[：:.\s)]+(.+)', content, re.IGNORECASE):
|
||
steps.append({"name": m.group(2).strip()[:100], "position": int(m.group(1))})
|
||
if not steps:
|
||
for i, m in enumerate(re.finditer(r'第一步[：:]\s*(.+?)\n|第二步[：:]\s*(.+?)\n|第三步[：:]\s*(.+?)\n', content)):
|
||
step_text = next(g for g in m.groups() if g)
|
||
steps.append({"name": step_text.strip()[:100], "position": i + 1})
|
||
return steps[:max_steps]
|
||
|
||
|
||
def inject_geo_metadata(html: str, title: str, content: str, platform: str, tags_html: str = "") -> str:
|
||
"""向 HTML <head> 注入 SEO/GEO 结构化元数据"""
|
||
description = _extract_description(content)
|
||
tags_list = _extract_tags_list(tags_html)
|
||
platform_name = PLATFORM_NAMES.get(platform, platform)
|
||
today = datetime.datetime.now().strftime("%Y-%m-%d")
|
||
content_type = _detect_content_type(content)
|
||
|
||
json_ld = {
|
||
"@context": "https://schema.org",
|
||
"@type": "Article",
|
||
"headline": title,
|
||
"description": description,
|
||
"datePublished": today,
|
||
"dateModified": today,
|
||
"author": {
|
||
"@type": "Organization",
|
||
"name": "宇之然",
|
||
"url": "https://yu-zhi-ran.com"
|
||
},
|
||
"publisher": {
|
||
"@type": "Organization",
|
||
"name": "宇之然",
|
||
"url": "https://yu-zhi-ran.com"
|
||
},
|
||
"mainEntityOfPage": {
|
||
"@type": "WebPage",
|
||
"@id": f"https://yu-zhi-ran.com/article/{platform}"
|
||
},
|
||
}
|
||
if tags_list:
|
||
json_ld["keywords"] = ", ".join(tags_list[:8])
|
||
|
||
if content_type == "faq":
|
||
faq_pairs = _extract_faq_pairs(content)
|
||
if faq_pairs:
|
||
json_ld["@type"] = "FAQPage"
|
||
json_ld["mainEntity"] = [
|
||
{"@type": "Question", "name": p["question"],
|
||
"acceptedAnswer": {"@type": "Answer", "text": p["answer"]}}
|
||
for p in faq_pairs
|
||
]
|
||
elif content_type == "howto":
|
||
howto_steps = _extract_howto_steps(content)
|
||
if howto_steps:
|
||
json_ld["@type"] = "HowTo"
|
||
json_ld["step"] = [
|
||
{"@type": "HowToStep", "position": s["position"],
|
||
"name": s["name"], "itemListElement": [
|
||
{"@type": "HowToDirection", "text": s["name"]}
|
||
]}
|
||
for s in howto_steps
|
||
]
|
||
elif content_type == "listicle":
|
||
json_ld["@type"] = "ItemList"
|
||
items = re.findall(r'^[-*\d]+[.、)\s]+(.+)$', content, re.MULTILINE)
|
||
json_ld["itemListElement"] = [
|
||
{"@type": "ListItem", "position": i + 1, "name": item.strip()[:100]}
|
||
for i, item in enumerate(items[:10])
|
||
]
|
||
|
||
json_ld_str = json.dumps(json_ld, ensure_ascii=False)
|
||
|
||
meta_tags = f"""
|
||
<meta name="description" content="{description}">
|
||
<meta name="keywords" content="{', '.join(tags_list[:8]) if tags_list else ''}">
|
||
<meta property="og:type" content="article">
|
||
<meta property="og:title" content="{title}">
|
||
<meta property="og:description" content="{description[:150]}">
|
||
<meta property="og:site_name" content="宇之然 | {platform_name}">
|
||
<meta property="article:published_time" content="{today}">
|
||
<meta property="article:author" content="宇之然">
|
||
<script type="application/ld+json">
|
||
{json_ld_str}
|
||
</script>"""
|
||
|
||
# 注入到 </head> 之前
|
||
html = html.replace("</head>", meta_tags + "\n</head>")
|
||
|
||
# 为 wechat + xiaohongshu 追加 Weibo/Wechat 兼容 meta
|
||
if platform in ("wechat", "xiaohongshu"):
|
||
html = html.replace("</head>", """
|
||
<meta property="og:image" content="https://yu-zhi-ran.com/og-image.png">
|
||
<meta name="weibo:webpage:source" content="宇之然">
|
||
</head>""")
|
||
|
||
return html
|
||
|
||
|
||
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')
|
||
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": "", "section_type": "normal"}
|
||
elif line.startswith("## "):
|
||
if current:
|
||
sections.append(current)
|
||
title = line[3:].strip()
|
||
stype = "noise" if title in ("文章大纲", "大纲", "文章结构", "结构") else "normal"
|
||
current = {"level": 2, "title": title, "content": "", "section_type": stype}
|
||
elif line.startswith("### "):
|
||
if current:
|
||
sections.append(current)
|
||
current = {"level": 3, "title": line[4:].strip(), "content": "", "section_type": "normal"}
|
||
else:
|
||
if current and line.strip():
|
||
current['content'] = current.get('content', '') + line + "\n"
|
||
if current:
|
||
sections.append(current)
|
||
return sections
|
||
|
||
@staticmethod
|
||
def _clean_markdown(text: str) -> str:
|
||
return clean_markdown_content(text)
|
||
|
||
@staticmethod
|
||
def _is_outline_noise(line: str) -> bool:
|
||
stripped = line.strip()
|
||
if not stripped:
|
||
return True
|
||
if stripped.startswith('---'):
|
||
return True
|
||
if '大纲生成时间' in stripped:
|
||
return True
|
||
if stripped.startswith('*大纲'):
|
||
return True
|
||
if re.match(r'^\*{0,3}【[^】]*】\*{0,3}\s*$', stripped):
|
||
return True
|
||
return False
|
||
|
||
def _is_bullet_only(self, text: str) -> bool:
|
||
"""检查内容是否主要是要点列表（大纲格式），需要 LLM 展开"""
|
||
lines = [l.strip() for l in text.split('\n') if l.strip()]
|
||
if not lines:
|
||
return False
|
||
bullet_count = sum(1 for l in lines if l.startswith(('- ', '* ', '**', '+ ')))
|
||
return bullet_count / len(lines) > 0.4
|
||
|
||
def _expand_section(self, section: Dict, platform: str = "zhihu") -> str:
|
||
content = section.get('content', '').strip()
|
||
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(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 扩写失败 [{platform}]: {e}")
|
||
|
||
lines = [l.strip() for l in content.split('\n') if not self._is_outline_noise(l)]
|
||
if lines:
|
||
sentences = []
|
||
for line in lines:
|
||
text = line
|
||
for prefix in ['- ', '* ', '1. ', '2. ', '3. ', '4. ', '5. ']:
|
||
if line.startswith(prefix):
|
||
text = line[len(prefix):]
|
||
break
|
||
text = text.strip()
|
||
if text:
|
||
if text[-1] not in '。！？；':
|
||
text += '。'
|
||
sentences.append(text)
|
||
if sentences:
|
||
result = ' '.join(sentences)
|
||
return self._clean_markdown(result)
|
||
return ''
|
||
|
||
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, platform)
|
||
if expanded:
|
||
parts.append(expanded + "\n")
|
||
continue
|
||
title_stripped = sec['title'].strip()
|
||
if title_stripped in ('文章大纲', '大纲', '文章结构', '结构'):
|
||
continue
|
||
if sec.get('section_type') == 'noise':
|
||
continue
|
||
heading = f"{'#' * sec['level']} {sec['title']}"
|
||
parts.append(heading)
|
||
if sec.get('content'):
|
||
expanded = self._expand_section(sec, platform)
|
||
parts.append(expanded + "\n")
|
||
full_md = "\n".join(parts).strip()
|
||
# 收集所有引用来源，统一添加到文末
|
||
refs = set()
|
||
for m in re.finditer(r'（来源：([^）]+)）', full_md):
|
||
refs.add(m.group(1).strip())
|
||
if refs:
|
||
# 去掉已有参考资料区，重新生成统一的
|
||
full_md = re.sub(r'\n---\n\*\*参考资料\*\*[\s\S]*$', '', full_md).strip()
|
||
ref_lines = "\n".join(f"- {r}" for r in sorted(refs))
|
||
full_md += f"\n\n---\n\n**参考资料**\n{ref_lines}"
|
||
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 == "zhihu":
|
||
result = []
|
||
in_list = False
|
||
for line in lines:
|
||
stripped = line.strip()
|
||
# 数据类行 → 引用格式（知乎文章中引用数据能增强可信度）
|
||
if any(stripped.startswith(p) for p in ('据统计', '调研显示', '数据显示', '报告指出', '根据', '数据显示')):
|
||
line = f"> {line}"
|
||
# 列表保持原样（知乎支持 markdown 列表）
|
||
if stripped.startswith('- ') or stripped.startswith('* '):
|
||
if not in_list:
|
||
result.append('')
|
||
in_list = True
|
||
else:
|
||
in_list = False
|
||
result.append(line)
|
||
adapted = '\n'.join(result)
|
||
# 末尾加讨论引导（知乎算法权重：互动率）
|
||
if not any(kw in adapted for kw in ('你觉得', '你怎么看', '欢迎在评论区', '说说你的')):
|
||
adapted += "\n\n---\n\n你觉得这个观点有道理吗？你在工作中有没有类似的经验？欢迎在评论区聊聊。"
|
||
return adapted
|
||
|
||
if platform == "xiaohongshu":
|
||
result = []
|
||
char_count = 0
|
||
last_was_heading = False
|
||
for line in lines:
|
||
if char_count >= max_c:
|
||
break
|
||
stripped = line.strip()
|
||
if line.startswith('## '):
|
||
if not last_was_heading and result:
|
||
result.append('')
|
||
char_count += 1
|
||
line = f"## ✨ {line[3:]}"
|
||
last_was_heading = True
|
||
elif line.startswith('### '):
|
||
if not last_was_heading and result:
|
||
result.append('')
|
||
char_count += 1
|
||
line = f"### 💡 {line[4:]}"
|
||
last_was_heading = True
|
||
else:
|
||
last_was_heading = False
|
||
# 超长段落后拆行 + 每段前加点缀
|
||
if stripped and len(stripped) > 60:
|
||
sentences = [s.strip() for s in stripped.replace('。', '。\n').split('\n') if s.strip()]
|
||
for s in sentences:
|
||
if s and char_count < max_c:
|
||
result.append(s)
|
||
char_count += len(s)
|
||
continue
|
||
result.append(line)
|
||
char_count += len(line)
|
||
adapted = '\n'.join(result)
|
||
# 结尾加收藏引导（小红书算法权重：收藏率）
|
||
if '收藏' not in adapted:
|
||
adapted += "\n\n✨ 觉得有用的话点个收藏吧，下次需要的时候随时翻出来看~"
|
||
return adapted
|
||
|
||
if platform == "wechat":
|
||
result = []
|
||
for line in lines:
|
||
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)
|
||
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": get_prompt("tags_generation", platform="知乎", title=title, field=field, core=core),
|
||
"wechat": get_prompt("tags_generation", platform="公众号", title=title, field=field, core=core),
|
||
"xiaohongshu": get_prompt("tags_generation", platform="小红书", title=title, field=field, core=core),
|
||
}
|
||
|
||
if HAVE_LLM:
|
||
prompt = tag_prompts.get(platform, get_prompt("tags_generation", platform=platform, title=title, field=field, core=core))
|
||
try:
|
||
params = get_prompt_params("tags_generation")
|
||
tags_text = call_llm(prompt, temperature=params.get("temperature", 0.3), max_tokens=params.get("max_tokens", 500))
|
||
tags_text = strip_thinking(tags_text)
|
||
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
|
||
|
||
if platform == "zhihu":
|
||
prompt = get_prompt("title_optimize_zhihu",
|
||
title=original,
|
||
core=self.topic.get('core_concept', ''),
|
||
pain=self.topic.get('audience_pain', ''),
|
||
field=self.topic.get('field', ''),
|
||
)
|
||
elif platform == "wechat":
|
||
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",
|
||
title=original,
|
||
core=self.topic.get('core_concept', ''),
|
||
)
|
||
else:
|
||
prompt = f"给以下文章改个吸引人的{platform}标题：{original}"
|
||
|
||
try:
|
||
if platform in ("zhihu", "wechat", "xiaohongshu"):
|
||
params = get_prompt_params(f"title_optimize_{platform}")
|
||
resp = call_llm(prompt, temperature=params.get("temperature", 0.8), max_tokens=params.get("max_tokens", 1500))
|
||
else:
|
||
resp = call_llm(prompt, temperature=0.7)
|
||
resp = strip_thinking(resp)
|
||
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 re.match(r'^(不如|或者|建议|推荐|参考|方案[一二三]|第[一二三]种|以[下是]|标题[一二三]|选项)', line):
|
||
continue
|
||
if line:
|
||
titles.append(line)
|
||
if titles:
|
||
best = titles[0][:80]
|
||
# 如果优化后标题与原文毫无关联或过短，回退原题
|
||
if len(best) < 4 or (len(set(best) & set(original)) < 2 and len(original) > 4):
|
||
logger.warning(f"标题优化结果异常「{best}」，回退原文")
|
||
return original
|
||
logger.info(f"标题优化 [{platform}]: {best}")
|
||
return best
|
||
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>"
|
||
|
||
adapted = strip_ai_preface(adapted)
|
||
html = template.replace("{{TITLE}}", title).replace("{{DATE}}", TODAY).replace("{{GEN_TIME}}", GEN_TIME)
|
||
html_content = _md_parser(adapted)
|
||
|
||
# 仅插入头图（每个平台一篇一张，不过度）
|
||
html_content = insert_lead_image(
|
||
html_content, platform,
|
||
title=self.topic.get('title', title),
|
||
field=self.topic.get('field', ''),
|
||
)
|
||
|
||
html = html.replace("<!-- CONTENT -->", html_content)
|
||
|
||
# 防御：清理可能在 LLM 输出中混入的 markdown 代码围栏和文件头
|
||
html = re.sub(r'^```+\w*\s*\n?', '', html)
|
||
html = re.sub(r'\n?```+\s*$', '', html)
|
||
html = html.strip()
|
||
|
||
tags_html = self._get_platform_tags(platform)
|
||
if tags_html:
|
||
html = html.replace("<!-- TAGS -->", tags_html)
|
||
else:
|
||
html = html.replace("<!-- TAGS -->", "")
|
||
|
||
html = inject_geo_metadata(html, title, adapted, platform, tags_html)
|
||
return html
|
||
|
||
def save_html(self, html: str, platform: str, *, title: str = "", content: str = "") -> str:
|
||
try:
|
||
save_article(self.topic_id, platform, html, title=title, content=content)
|
||
logger.info(f"文章写入数据库: {platform}_{self.topic_id}")
|
||
return f"db:{platform}_{self.topic_id}"
|
||
except Exception as e:
|
||
logger.warning(f"数据库保存失败: {e}")
|
||
return ""
|
||
|
||
def mark_draft(self):
|
||
update_topic_status(self.topic_id, 'review')
|
||
logger.info(f"选题 {self.topic_id} 状态已更新为待审查（数据库）")
|
||
|
||
def run(self):
|
||
logger.info("开始撰写阶段（三平台独立展开）")
|
||
results = {}
|
||
for platform in ["zhihu", "wechat", "xiaohongshu"]:
|
||
markdown = self.generate_platform_markdown(platform)
|
||
title = self._optimize_title(platform)
|
||
html = self.generate_platform_html(markdown, platform)
|
||
results[platform] = str(self.save_html(html, platform, title=title, content=markdown))
|
||
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()
|