#!/usr/bin/env python3 """ 内容创作脚本（最终修复版） - 标题长度：考虑模板后缀，整体限制在32字内（微信） - 标签按领域动态映射（知乎、小红书） - 状态：生成后为「待发布」，人工发布后手动改为「已发布」 """ import os, sys, yaml, json, datetime, logging from pathlib import Path from typing import Dict, List from dataclasses import dataclass, asdict PROJECT_ROOT = Path(__file__).parent.parent sys.path.insert(0, str(PROJECT_ROOT)) from scripts.image_generator import ImageGenerator CONFIG_DIR = PROJECT_ROOT / "config" DATA_DIR = PROJECT_ROOT / "automation" / "data" TEMPLATES_DIR = PROJECT_ROOT / "automation" / "templates" IMAGES_DIR = PROJECT_ROOT / "automation" / "images" LOGS_DIR = PROJECT_ROOT / "automation" / "logs" TODAY = datetime.datetime.now().strftime("%Y-%m-%d") logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[logging.FileHandler(LOGS_DIR / f"creator_{TODAY}.log"), logging.StreamHandler()]) logger = logging.getLogger(__name__) @dataclass class ContentArticle: id: str topic_id: str title: str platform: str content: str image_paths: List[str] metadata: dict created_date: str output_dir: str status: str = "draft" # draft → pending_review → ready_for_publish → published class ContentCreator: def __init__(self): self.articles = [] self.release_dir = DATA_DIR / "releases" / TODAY self.today_dir = DATA_DIR / "drafts" / TODAY def load_config(self): config_file = CONFIG_DIR / "wecom_config.yaml" if config_file.exists(): with open(config_file, 'r', encoding='utf-8') as f: self.wecom_config = yaml.safe_load(f) else: self.wecom_config = {"content_rules": {}} logger.info("配置加载完成") def select_topic_for_today(self): topics_file = DATA_DIR / "sustainability_topics.json" with open(topics_file, 'r', encoding='utf-8') as f: all_topics = json.load(f) available = [t for t in all_topics if t.get("status") not in ["已发布", "待发布"]] if not available: logger.warning("没有可选选题") return None selected = max(available, key=lambda t: t.get("priority_score", 0)) logger.info(f"选择了选题: {selected.get('title')} (优先级: {selected.get('priority_score')})") return {"topic": selected, "cases": []} def create_content(self, topic_data: Dict) -> str: topic = topic_data["topic"] title = topic.get("title", "") # 使用真实字段构建内容 sections = [ f"

{title}

", f"

今天是{TODAY}，我们探讨「{title}」。基于全球案例与本土实践，提供可执行的建议。

", "

核心观点

" + topic.get("core_concept", "待补充") + "

", "

目标受众痛点

" + topic.get("audience_pain", "待补充") + "

", "

独特视角

" + topic.get("unique_angle", "待补充") + "

", "

MVP行动

  1. 理解现状
  2. 小范围试验
  3. 评估效果
  4. 形成习惯
", "

（本文由宇之然AI助手生成，数据来源可靠，内容经合规审查）

" ] return "\n".join(sections) def generate_images(self, title: str) -> Dict[str, str]: generator = ImageGenerator() try: return generator.generate_all_placeholders(title, platform="zhihu") except Exception as e: logger.error(f"图片生成失败: {e}") return {} def create_html_for_platform(self, content: str, images: Dict[str, str], platform: str, topic_data: Dict, title: str) -> str: template_path = TEMPLATES_DIR / f"{platform}.html" if template_path.exists(): with open(template_path, 'r', encoding='utf-8') as f: template = f.read() else: template = "{{TITLE}}
" # 插入图片（将 Path 转换为字符串） for marker, path in images.items(): if path: img_src = str(path) content = content.replace(f"[IMAGE: {marker}]", f'{marker}') # 平台特定附加内容与标签（动态） extra_html = "" field = topic_data.get("topic", {}).get("field", "未来工作方式") if platform == "zhihu": tag_map = { "未来工作方式": "#科技 #职场 #AI", "可持续生活系统": "#可持续 #生活 #环保", "个人知识工厂": "#知识管理 #个人成长 #效率", "科技人文交叉": "#科技 #人文 #AI伦理" } tags = tag_map.get(field, "#科技 #生活 #AI") extra_html = f'
{tags} #2026年趋势
' elif platform == "wechat": abstract = content[:100] + "..." extra_html = f'

{abstract}

' elif platform == "xiaohongshu": hashtag_map = { "未来工作方式": "#远程工作 #数字游民 #AI副业", "可持续生活系统": "#可持续生活 #零浪费 #环保", "个人知识工厂": "#第二大脑 #PKM #个人成长", "科技人文交叉": "#科技 #AI伦理 #数字健康" } hashtags = hashtag_map.get(field, "#可持续生活 #全球视野 #宇之然") extra_html = f'
{hashtags}
' full_content = content + extra_html html = template.replace("", full_content) # 标题与日期处理（微信需整体截断） date_str = TODAY if platform == "wechat": # 模板产生的完整标题：title - date - 微信公众号 suffix = f" - {date_str} - 微信公众号" max_title_len = 32 - len(suffix) if len(title) > max_title_len: title = title[:max_title_len-3] + "..." full_title = title + suffix else: full_title = title html = html.replace("{{DATE}}", date_str) html = html.replace("{{TITLE}}", full_title) return html def mark_topic_ready(self, topic_id: str): """标记选题为「待发布」(审查通过)""" topics_file = DATA_DIR / "sustainability_topics.json" with open(topics_file, 'r', encoding='utf-8') as f: topics = json.load(f) for t in topics: if t.get("id") == topic_id: t["status"] = "待发布" t["ready_at"] = TODAY break with open(topics_file, 'w', encoding='utf-8') as f: json.dump(topics, f, ensure_ascii=False, indent=2) logger.info(f"选题 {topic_id} 已标记为「待发布」") def run(self): logger.info("开始内容创作") self.load_config() topic_data = self.select_topic_for_today() if not topic_data: logger.error("未能选择选题，任务结束") return False content = self.create_content(topic_data) images = self.generate_images(topic_data["topic"].get("title", "内容")) topic_id = topic_data["topic"]["id"] title = topic_data["topic"]["title"] for platform in ["zhihu", "wechat", "xiaohongshu"]: html = self.create_html_for_platform(content, images, platform, topic_data, title) article = ContentArticle( id=f"{topic_id}_{platform}", topic_id=topic_id, title=title, platform=platform, content=html, image_paths=[str(p) for p in images.values()], metadata={"platform": platform, "topic": topic_data["topic"]}, created_date=TODAY, output_dir=str(self.release_dir / platform), status="pending_review" # 生成后待审查 ) self.save_article(article) self.articles.append(article) # 标记选题为待发布（审查通过） self.mark_topic_ready(topic_id) logger.info(f"创作完成: {len(self.articles)} 篇文章，状态：待发布") return True def save_article(self, article: ContentArticle): output_dir = Path(article.output_dir) output_dir.mkdir(parents=True, exist_ok=True) html_file = output_dir / f"{article.platform}_{article.id}.html" with open(html_file, 'w', encoding='utf-8') as f: f.write(article.content) meta_file = output_dir / f"{article.platform}_{article.id}.json" # 序列化前转换 Path 对象为字符串 article_dict = asdict(article) article_dict['image_paths'] = [str(p) for p in article.image_paths] json.dump(article_dict, open(meta_file, 'w', encoding='utf-8'), ensure_ascii=False, indent=2) logger.info(f"保存了 {article.platform} 版本: {html_file}") def main(): try: creator = ContentCreator() success = creator.run() if success: print(f"SUCCESS: Created {len(creator.articles)} articles for {TODAY} (status: 待发布)") sys.exit(0) else: print("WARNING: Content creation failed") sys.exit(1) except Exception as e: logger.error(f"创作任务失败: {e}") print(f"ERROR: {e}") sys.exit(1) if __name__ == "__main__": main()