#!/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, save_article 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": 3000, "style": "深度长文分析", "min_chars": 1500}, "wechat": {"max_chars": 1500, "style": "亲切口语化", "min_chars": 800}, "xiaohongshu": {"max_chars": 800, "style": "图文笔记，emoji+标签", "min_chars": 300}, } 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() 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: lines = text.split('\n') cleaned = [] in_code_fence = False for line in lines: if line.strip().startswith('```'): in_code_fence = not in_code_fence continue if in_code_fence: continue line = re.sub(r'^#{1,6}\s+', '', line) line = re.sub(r'^[\-\*\+]\s+', '', line) line = re.sub(r'^\d+[\.\)]\s+', '', line) line = re.sub(r'\*{1,3}([^*]+)\*{1,3}', r'\1', line) cleaned.append(line) return '\n'.join(cleaned).strip() @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) -> str: content = section.get('content', '').strip() # 大纲要点格式（>40% 行以 -/*/** 开头）应始终由 LLM 展开为连贯段落 if HAVE_LLM and self._is_bullet_only(content): logger.info(f"使用 LLM 扩写章节(要点→段落): {section['title']}") prompt = f"""你是一个资深作者，正在写一篇关于「{self.topic['title']}」的文章。请写「{section['title']}」这一节。 今天日期：{datetime.datetime.now().strftime('%Y年%m月%d日')}。 笔记要点： {content} 【输出要求】 输出3-6段纯粹、流畅的段落文字，每节内容根据平台需求控制在200-800字之间。 格式： - 禁止任何标题/列表/格式标记（#、-、*、1.、**等） - 每段3-5句，段间空行分隔 - 用「你」或「我们」视角，自然口语化 内容要求（让文章在各平台能被推荐）： - 开头直接切入痛点或反常识观点，抓住注意力 - 每个观点配具体案例或数据（用「据统计」「调研显示」等），不要空泛说理 - 有独特判断和立场，避免正确废话 - 回答「所以呢」——读者看完能带走什么 - 结尾有情绪感召力，让人想点赞/收藏/转发 直接输出段落正文，不要任何附加说明。""" try: expanded = call_llm(prompt, temperature=0.6) if expanded: cleaned = self._clean_markdown(expanded.strip()) if cleaned: return cleaned except Exception as e: logger.warning(f"LLM 扩写失败: {e}") # Fallback: 将 bullet points 展开为段落（过滤噪音行） 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_full_markdown(self) -> str: sections = self._parse_outline_sections() parts = [] for sec in sections: if sec['level'] == 1: if sec.get('content'): expanded = self._expand_section(sec) 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) 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) 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) if tags_text: tags = [t.strip('#') for t in tags_text.strip().split() if t.strip('#')] if tags: return " ".join(f'{t}' 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'{t}' 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字最佳 - 有情绪感召力：共鸣/好奇/焦虑/期待 - 参考近期10万+标题的语气节奏 - 直接输出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) 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 = "{{TITLE}}

{{TITLE}}

" html = template.replace("{{TITLE}}", title).replace("{{DATE}}", TODAY).replace("{{GEN_TIME}}", GEN_TIME) html_content = _md_parser(adapted) # WeChat: insert topic-relevant image at start of body if platform == "wechat": import base64 topic_title = self.topic.get('title', title) topic_field = self.topic.get('field', '') safe_title = topic_title.replace('&', '&').replace('<', '<').replace('>', '>').replace('"', '"').replace("'", ''') safe_field = topic_field.replace('&', '&').replace('<', '<').replace('>', '>') lines = [] chars_per_line = 24 for i in range(0, len(safe_title), chars_per_line): lines.append(safe_title[i:i+chars_per_line]) if not lines: lines = ['配图'] line_y = 220 - (len(lines) - 1) * 20 title_texts = ''.join(f'{l}' for i, l in enumerate(lines)) field_text = f'{safe_field}' if safe_field else '' img_svg = f''' {title_texts} {field_text} 宇之然 · 配图（可替换） ''' img_b64 = 'data:image/svg+xml;base64,' + base64.b64encode(img_svg.encode('utf-8')).decode('ascii') img_tag = f'

{safe_title}

\n' h1_end = html_content.find('') if h1_end != -1: html_content = html_content[:h1_end + 5] + '\n' + img_tag + html_content[h1_end + 5:] else: html_content = img_tag + html_content html = html.replace("", html_content) # 防御：清理可能在 LLM 输出中混入的 markdown 代码围栏和文件头 html = re.sub(r'^```+\w*\s*\n?', '', html) html = html.strip() tags_html = self._get_platform_tags(platform) if tags_html: html = html.replace("", tags_html) else: html = html.replace("", "") return html def save_html(self, html: str, platform: str) -> str: try: save_article(self.topic_id, platform, html) 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("开始撰写阶段") 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()