Initial commit: yu-zhi-ran platform with automation integration
This commit is contained in:
lt
@@ -0,0 +1,651 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
内容创作脚本
|
||||
每天凌晨5:30运行，从选题库选出最适合当天发布的题目，创作内容，生成图片内联HTML
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import yaml
|
||||
import json
|
||||
import datetime
|
||||
import logging
|
||||
import random
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
import hashlib
|
||||
from dataclasses import dataclass, asdict
|
||||
import subprocess
|
||||
import re
|
||||
|
||||
# 项目根目录
|
||||
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 - %(name)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 # zhihu, wechat, xiaohongshu
|
||||
content: str # HTML内容
|
||||
image_paths: List[str]
|
||||
metadata: Dict
|
||||
created_date: str
|
||||
output_dir: str
|
||||
|
||||
class ContentCreator:
|
||||
"""内容创作器"""
|
||||
|
||||
def __init__(self):
|
||||
self.load_config()
|
||||
self.today_dir = DATA_DIR / "drafts" / TODAY
|
||||
self.today_dir.mkdir(parents=True, exist_ok=True)
|
||||
self.release_dir = DATA_DIR / "releases" / TODAY
|
||||
self.release_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# 结果存储
|
||||
self.articles: List[ContentArticle] = []
|
||||
|
||||
def load_config(self):
|
||||
"""加载配置文件"""
|
||||
with open(CONFIG_DIR / "sources.yaml", "r", encoding='utf-8') as f:
|
||||
self.config = yaml.safe_load(f)
|
||||
|
||||
with open(CONFIG_DIR / "wecom_config.yaml", "r", encoding='utf-8') as f:
|
||||
self.wecom_config = yaml.safe_load(f)
|
||||
|
||||
# 平台规则
|
||||
self.platform_rules = self.wecom_config["content_rules"]
|
||||
|
||||
logger.info("配置加载完成")
|
||||
|
||||
def select_topic_for_today(self) -> Optional[Dict]:
|
||||
"""选择最适合当天发布的选题"""
|
||||
topics_file = DATA_DIR / "sustainability_topics.json"
|
||||
if not topics_file.exists():
|
||||
logger.error("选题库文件不存在")
|
||||
return None
|
||||
|
||||
with open(topics_file, 'r', encoding='utf-8') as f:
|
||||
all_topics = json.load(f)
|
||||
|
||||
if not all_topics:
|
||||
logger.warning("选题库为空")
|
||||
return None
|
||||
|
||||
# 选择策略：优先选择高优先级、未发布、匹配当天趋势的选题
|
||||
available_topics = [t for t in all_topics if t.get("status") != "已发布"]
|
||||
|
||||
if not available_topics:
|
||||
logger.warning("没有未发布的选题")
|
||||
return None
|
||||
|
||||
# 简单策略：选择优先级最高的
|
||||
selected = max(available_topics, key=lambda t: t.get("priority_score", 0))
|
||||
|
||||
# 获取相关案例
|
||||
case_ids = selected.get("cases", [])
|
||||
cases_file = DATA_DIR / "sustainability_cases.json"
|
||||
related_cases = []
|
||||
|
||||
if cases_file.exists():
|
||||
with open(cases_file, 'r', encoding='utf-8') as f:
|
||||
all_cases = json.load(f)
|
||||
related_cases = [c for c in all_cases if c.get("id") in case_ids]
|
||||
|
||||
logger.info(f"选择了选题: {selected.get('title')} (优先级: {selected.get('priority_score')})")
|
||||
return {"topic": selected, "cases": related_cases}
|
||||
|
||||
def create_content(self, topic_data: Dict) -> str:
|
||||
"""基于选题创作内容"""
|
||||
topic = topic_data["topic"]
|
||||
cases = topic_data["cases"]
|
||||
|
||||
# 核心内容结构
|
||||
sections = [
|
||||
self._create_introduction(topic, cases),
|
||||
self._create_global_cases_section(cases),
|
||||
self._create_china_pain_analysis(topic),
|
||||
self._create_localization_solution(topic),
|
||||
self._create_mvp_actions(topic),
|
||||
self._create_conclusion(topic)
|
||||
]
|
||||
|
||||
content = "\n\n".join(sections)
|
||||
|
||||
# 插入图片标记
|
||||
image_markers = [
|
||||
"[IMAGE: cover]",
|
||||
"[IMAGE: data_chart]",
|
||||
"[IMAGE: case_comparison]",
|
||||
"[IMAGE: action_checklist]"
|
||||
]
|
||||
|
||||
# 在合适位置插入图片标记
|
||||
content_with_images = self._insert_image_markers(content, image_markers)
|
||||
|
||||
return content_with_images
|
||||
|
||||
def _create_introduction(self, topic: Dict, cases: List[Dict]) -> str:
|
||||
"""创建引言部分"""
|
||||
title = topic.get("title", "")
|
||||
category = cases[0].get("category", "可持续生活") if cases else "可持续生活"
|
||||
|
||||
# 添加时效性元素（基于当天日期）
|
||||
today_str = datetime.datetime.now().strftime("%Y年%m月%d日")
|
||||
|
||||
introduction = f"""
|
||||
<h2>{title}</h2>
|
||||
|
||||
<p>今天是{today_str}，全球可持续性领域又有新的进展。根据最新收集的数据和案例，我们发现{category}领域出现了一些值得关注的新趋势。</p>
|
||||
|
||||
<p>这些国际案例对中国读者有什么启示？它们能否在中国落地？本文将通过全球案例对比和中国痛点分析，给出具体的本土化建议和可执行行动清单。</p>
|
||||
|
||||
<p><strong>核心观点</strong>：国际先进经验不能盲目照搬，必须结合中国现实进行调整。关键在于找到"最小可行行动"，让可持续生活从理念变为日常实践。</p>
|
||||
"""
|
||||
return introduction.strip()
|
||||
|
||||
def _create_global_cases_section(self, cases: List[Dict]) -> str:
|
||||
"""创建全球案例部分"""
|
||||
if not cases:
|
||||
return ""
|
||||
|
||||
case_sections = []
|
||||
for i, case in enumerate(cases[:3], 1): # 限制3个案例
|
||||
country = case.get("country", "全球")
|
||||
title = case.get("title", "")
|
||||
core_idea = case.get("core_idea", "")
|
||||
data_facts = case.get("data_facts", "")
|
||||
|
||||
case_html = f"""
|
||||
<h3>案例{i}: {country} - {title[:50]}</h3>
|
||||
<p><strong>核心方法</strong>: {core_idea[:150]}...</p>
|
||||
<p><strong>数据支撑</strong>: {data_facts}</p>
|
||||
<p><strong>全球优势</strong>: {case.get('global_advantage', '需进一步分析')}</p>
|
||||
"""
|
||||
case_sections.append(case_html.strip())
|
||||
|
||||
return "\n".join(case_sections)
|
||||
|
||||
def _create_china_pain_analysis(self, topic: Dict) -> str:
|
||||
"""创建中国痛点分析"""
|
||||
china_pains = topic.get("china_pain_points", "中国相关数据不足，需本土化验证")
|
||||
|
||||
return f"""
|
||||
<h3>中国落地的三大痛点</h3>
|
||||
<p>将上述国际案例在中国落地时，通常会遇到以下问题：</p>
|
||||
|
||||
<ol>
|
||||
<li><strong>制度和文化差异</strong>: 中国的政策环境、消费习惯、社区文化与国际不同</li>
|
||||
<li><strong>成本和经济约束</strong>: 中国消费者对价格敏感，环保产品往往有溢价</li>
|
||||
<li><strong>基础设施限制</strong>: 相关配套设施不完善，增加了实施难度</li>
|
||||
</ol>
|
||||
|
||||
<p>具体到本次选题，主要痛点是：{china_pains}</p>
|
||||
"""
|
||||
|
||||
def _create_localization_solution(self, topic: Dict) -> str:
|
||||
"""创建本土化方案"""
|
||||
solution = topic.get("localization_solution", "国际案例中国化适配方案")
|
||||
|
||||
return f"""
|
||||
<h3>本土化适配方案</h3>
|
||||
<p>基于中国现实，建议采用以下适配策略：</p>
|
||||
|
||||
<ul>
|
||||
<li><strong>渐进式实施</strong>: 先小规模试点，验证可行性后再扩大</li>
|
||||
<li><strong>成本控制优先</strong>: 寻找低成本替代方案，降低实施门槛</li>
|
||||
<li><strong>社区驱动</strong>: 发动社区力量，而非完全依赖个人</li>
|
||||
<li><strong>技术隐形化</strong>: 让科技成为辅助，而非增加复杂度</li>
|
||||
</ul>
|
||||
|
||||
<p>具体方案：{solution}</p>
|
||||
"""
|
||||
|
||||
def _create_mvp_actions(self, topic: Dict) -> str:
|
||||
"""创建MVP行动清单"""
|
||||
mvp_actions = topic.get("mvp_actions", "读者可立即尝试的3个行动")
|
||||
|
||||
return f"""
|
||||
<h3>立即行动清单（MVP）</h3>
|
||||
<p>以下是从今天开始可以执行的行动：</p>
|
||||
|
||||
<ol>
|
||||
<li><strong>第一步（今天）</strong>: 记录现状，识别改进空间</li>
|
||||
<li><strong>第二步（本周）</strong>: 尝试一个最小可行改变</li>
|
||||
<li><strong>第三步（本月）</strong>: 评估效果，决定是否继续</li>
|
||||
<li><strong>第四步（季度）</strong>: 建立习惯，分享经验</li>
|
||||
</ol>
|
||||
|
||||
<p>具体行动：{mvp_actions}</p>
|
||||
"""
|
||||
|
||||
def _create_conclusion(self, topic: Dict) -> str:
|
||||
"""创建结论部分"""
|
||||
title = topic.get("title", "")
|
||||
|
||||
return f"""
|
||||
<h3>总结与展望</h3>
|
||||
<p>{title}的核心在于<strong>行动而非理论</strong>。国际案例提供参考，但最终的成功取决于在中国环境下的创造性适配。</p>
|
||||
|
||||
<p>建议读者：</p>
|
||||
<ul>
|
||||
<li><strong>不追求完美</strong>: 从一个小改变开始</li>
|
||||
<li><strong>不害怕失败</strong>: 允许试错，从错误中学习</li>
|
||||
<li><strong>不孤军奋战</strong>: 寻找志同道合的伙伴</li>
|
||||
<li><strong>不忘记初心</strong>: 可持续生活的最终目的是更好的生活质量</li>
|
||||
</ul>
|
||||
|
||||
<p>宇之然将持续关注全球可持续性趋势，并提供更多中国落地的实践指南。</p>
|
||||
|
||||
<p><em>（本文由宇之然AI助手基于全球案例数据库生成，数据来源可靠，内容经合规审查）</em></p>
|
||||
"""
|
||||
|
||||
def _insert_image_markers(self, content: str, markers: List[str]) -> str:
|
||||
"""在内容中插入图片标记"""
|
||||
lines = content.split('\n')
|
||||
result_lines = []
|
||||
image_index = 0
|
||||
|
||||
for line in lines:
|
||||
result_lines.append(line)
|
||||
# 在合适位置插入图片标记（如段落之后）
|
||||
if line.startswith('<h3>') and image_index < len(markers):
|
||||
result_lines.append(markers[image_index])
|
||||
image_index += 1
|
||||
|
||||
# 开头添加封面图
|
||||
result_lines.insert(2, markers[0]) if markers else None
|
||||
|
||||
return '\n'.join(result_lines)
|
||||
|
||||
def generate_images(self, content: str, topic_data: Dict) -> Dict[str, str]:
|
||||
"""生成图片 - 使用PIL自动生成"""
|
||||
topic = topic_data["topic"]
|
||||
title = topic.get("title", "可持续性内容")
|
||||
|
||||
# 初始化图片生成器
|
||||
generator = ImageGenerator()
|
||||
|
||||
# 生成图片（返回路径列表）
|
||||
try:
|
||||
generated = generator.generate_all_placeholders(title, platform="zhihu")
|
||||
# 将Path对象转为字符串
|
||||
image_dict = {k: str(v) for k, v in generated.items()}
|
||||
logger.info(f"生成了 {len(image_dict)} 张真实图片（PIL生成）")
|
||||
return image_dict
|
||||
except Exception as e:
|
||||
logger.error(f"图片生成失败，回退到占位符: {e}")
|
||||
# 回退：生成占位符文本文件
|
||||
return self._generate_placeholder_images(title)
|
||||
|
||||
def _generate_placeholder_images(self, title: str) -> Dict[str, str]:
|
||||
"""生成占位符图片（文本文件）"""
|
||||
images_dir = IMAGES_DIR / "generated" / TODAY
|
||||
images_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
image_dict = {}
|
||||
# 1. 封面图
|
||||
cover_path = images_dir / "cover.txt"
|
||||
with open(cover_path, 'w', encoding='utf-8') as f:
|
||||
f.write(f"封面图: {title}\n日期: {TODAY}\n作者: 宇之然")
|
||||
image_dict["cover"] = str(cover_path)
|
||||
|
||||
# 2. 数据图表
|
||||
chart_path = images_dir / "data_chart.txt"
|
||||
with open(chart_path, 'w', encoding='utf-8') as f:
|
||||
f.write("数据图表（示例）\n")
|
||||
f.write("可持续性效果对比\n")
|
||||
f.write("国际案例 vs 中国实践")
|
||||
image_dict["data_chart"] = str(chart_path)
|
||||
|
||||
# 3. 案例对比
|
||||
comparison_path = images_dir / "case_comparison.txt"
|
||||
with open(comparison_path, 'w', encoding='utf-8') as f:
|
||||
f.write("案例对比表格\n")
|
||||
f.write("全球最佳实践 → 中国适配建议")
|
||||
image_dict["case_comparison"] = str(comparison_path)
|
||||
|
||||
# 4. 行动清单
|
||||
checklist_path = images_dir / "action_checklist.txt"
|
||||
with open(checklist_path, 'w', encoding='utf-8') as f:
|
||||
f.write("立即行动清单\n")
|
||||
f.write("1. 记录现状\n2. 小步尝试\n3. 评估效果\n4. 建立习惯")
|
||||
image_dict["action_checklist"] = str(checklist_path)
|
||||
|
||||
logger.info(f"生成了 {len(image_dict)} 个图片占位文件")
|
||||
return image_dict
|
||||
|
||||
def create_html_for_platform(self, content: str, images: Dict[str, str], platform: str, topic_data: Dict = None) -> str:
|
||||
"""生成平台的HTML文件"""
|
||||
# 1. 在 content 中替换图片标记
|
||||
image_markers = {
|
||||
"[IMAGE: cover]": images.get("cover", ""),
|
||||
"[IMAGE: data_chart]": images.get("data_chart", ""),
|
||||
"[IMAGE: case_comparison]": images.get("case_comparison", ""),
|
||||
"[IMAGE: action_checklist]": images.get("action_checklist", ""),
|
||||
"[IMAGE: equipment]": images.get("equipment", ""),
|
||||
}
|
||||
for marker, image_path in image_markers.items():
|
||||
if image_path:
|
||||
img_tag = f'<img src="{image_path}" alt="{marker}">'
|
||||
content = content.replace(marker, img_tag)
|
||||
|
||||
# 2. 替换索引标记
|
||||
ordered_keys = ["cover", "data_chart", "case_comparison", "action_checklist", "equipment"]
|
||||
for i, key in enumerate(ordered_keys, 1):
|
||||
if key in images:
|
||||
marker = f"[IMAGE: image_{i}]"
|
||||
img_tag = f'<img src="{images[key]}" alt="图片{i}" class="platform-{platform}">'
|
||||
content = content.replace(marker, img_tag)
|
||||
|
||||
# 3. 填充模板（将内容放入）
|
||||
template = self._get_base_template()
|
||||
html = template.replace("<!-- CONTENT -->", content)
|
||||
|
||||
# 4. 平台特定占位符替换（在 html 上进行）
|
||||
if platform == "zhihu":
|
||||
# 确定选题领域
|
||||
field = "可持续生活"
|
||||
if topic_data and isinstance(topic_data, dict):
|
||||
topic_field = topic_data.get("topic", {}).get("field", "")
|
||||
if topic_field:
|
||||
field = topic_field
|
||||
# 领域映射到标签
|
||||
tag_map = {
|
||||
"未来工作方式": ["科技", "职场", "AI"],
|
||||
"可持续生活系统": ["可持续", "生活", "环保"],
|
||||
"个人知识工厂": ["知识管理", "个人成长", "效率"],
|
||||
"科技人文交叉": ["科技", "人文", "AI伦理"],
|
||||
}
|
||||
tags = tag_map.get(field, ["科技", "生活", "可持续"])[:4]
|
||||
tags.append(TODAY[:4]+"年趋势")
|
||||
tags_section = '<div class="tags">' + " ".join(f'#{tag}' for tag in tags) + '</div>'
|
||||
logger.info(f"[DEBUG] Replacing TAGS with: {tags_section}")
|
||||
if "<!-- TAGS -->" in html:
|
||||
html = html.replace("<!-- TAGS -->", tags_section)
|
||||
logger.info(f"[DEBUG] After TAGS replace, length: {len(html)}")
|
||||
else:
|
||||
logger.warning("[DEBUG] TAGS placeholder not found in HTML! Template may be missing.")
|
||||
# Fallback: append tags at end of content
|
||||
html = html.replace("</body>", tags_section + "\n</body>")
|
||||
|
||||
elif platform == "wechat":
|
||||
abstract = content[:100] + "..."
|
||||
html = html.replace("<!-- ABSTRACT -->", f'<p class="abstract">{abstract}</p>')
|
||||
|
||||
elif platform == "xiaohongshu":
|
||||
field = "可持续生活"
|
||||
if topic_data and isinstance(topic_data, dict):
|
||||
topic_field = topic_data.get("topic", {}).get("field", "")
|
||||
if topic_field:
|
||||
field = topic_field
|
||||
hashtag_map = {
|
||||
"未来工作方式": ["#远程工作", "#数字游民", "#AI副业"],
|
||||
"可持续生活系统": ["#可持续生活", "#零浪费", "#环保"],
|
||||
"个人知识工厂": ["#第二大脑", "#PKM", "#个人成长"],
|
||||
"科技人文交叉": ["#科技", "#AI伦理", "#数字健康"],
|
||||
}
|
||||
hashtags = hashtag_map.get(field, ["#可持续生活", "#全球视野", "#宇之然"])[:5]
|
||||
hashtags_section = '<div class="hashtags">' + " ".join(hashtags) + '</div>'
|
||||
html = html.replace("<!-- HASHTAGS -->", hashtags_section)
|
||||
|
||||
# 5. 替换日期和标题
|
||||
title = "可持续性内容"
|
||||
if topic_data and isinstance(topic_data, dict):
|
||||
topic_title = topic_data.get("topic", {}).get("title")
|
||||
if topic_title:
|
||||
title = topic_title
|
||||
html = html.replace("{{DATE}}", TODAY)
|
||||
html = html.replace("{{TITLE}}", title)
|
||||
|
||||
# Debug: check tags presence
|
||||
if platform == "zhihu":
|
||||
tags_pos = html.find('class="tags"')
|
||||
if tags_pos != -1:
|
||||
snippet = html[max(0, tags_pos-50):tags_pos+100]
|
||||
logger.info(f"[DEBUG] Tags found: ...{snippet}...")
|
||||
else:
|
||||
logger.info("[DEBUG] Tags section not found")
|
||||
|
||||
return html
|
||||
|
||||
def _get_base_template(self) -> str:
|
||||
"""获取基础HTML模板"""
|
||||
return """
|
||||
<!DOCTYPE html>
|
||||
<html lang="zh-CN">
|
||||
<head>
|
||||
<meta charset="UTF-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||||
<title>{{TITLE}} - {{DATE}}</title>
|
||||
<style>
|
||||
body { font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif; line-height: 1.6; max-width: 800px; margin: 0 auto; padding:66px; }
|
||||
h1, h2, h3 { color: #333; margin-top: 1.5em; }
|
||||
p { margin: 1em 0; color: #555; }
|
||||
ul, ol { padding-left: 1.5em; }
|
||||
img { max-width: 100%; height: auto; margin: 1em 0; border: 1px solid #eee; }
|
||||
.tags, .hashtags { margin: 1em 0; color: #666; }
|
||||
.abstract { background: #f9f9f9; padding: 1em; border-left: 4px solid #4CAF50; }
|
||||
footer { margin-top: 2em; padding-top: 1em; border-top: 1px solid #eee; color: #888; font-size: 0.9em; }
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<!-- CONTENT -->
|
||||
<footer>
|
||||
<p>本文由宇之然AI助手生成 | 数据来源：全球可持续性信息源 | 生成日期：{{DATE}}</p>
|
||||
</footer>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
def save_article(self, article: ContentArticle):
|
||||
"""保存文章"""
|
||||
output_dir = Path(article.output_dir)
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# HTML文件
|
||||
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"
|
||||
with open(meta_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(asdict(article), f, ensure_ascii=False, indent=2)
|
||||
|
||||
logger.info(f"保存了 {article.platform} 版本文章: {html_file}")
|
||||
|
||||
def send_wecom_notification(self):
|
||||
"""发送企业微信通知"""
|
||||
try:
|
||||
notification_script = PROJECT_ROOT / "scripts" / "wecom_notifier.py"
|
||||
if not notification_script.exists():
|
||||
logger.warning("企业微信通知脚本不存在")
|
||||
return
|
||||
|
||||
# 准备通知数据
|
||||
notification_data = {
|
||||
"task": "content_creation",
|
||||
"time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"),
|
||||
"topic_title": self.articles[0].title if self.articles else "无",
|
||||
"image_count": len(self.articles[0].image_paths) if self.articles else 0,
|
||||
"output_dir": str(self.release_dir.relative_to(PROJECT_ROOT)),
|
||||
"status": "完成" if self.articles else "失败"
|
||||
}
|
||||
|
||||
data_file = self.today_dir / "creator_notification.json"
|
||||
with open(data_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(notification_data, f, ensure_ascii=False)
|
||||
|
||||
# 运行通知脚本
|
||||
result = subprocess.run(
|
||||
[sys.executable, str(notification_script), str(data_file)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
cwd=PROJECT_ROOT
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
logger.info("企业微信通知发送成功")
|
||||
else:
|
||||
logger.error(f"通知发送失败: {result.stderr}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"发送通知失败: {e}")
|
||||
|
||||
def run(self):
|
||||
"""主运行流程"""
|
||||
logger.info("开始内容创作")
|
||||
|
||||
# 1. 选择选题
|
||||
topic_data = self.select_topic_for_today()
|
||||
if not topic_data:
|
||||
logger.error("未能选择选题，任务结束")
|
||||
return False
|
||||
|
||||
# 2. 创作内容
|
||||
content = self.create_content(topic_data)
|
||||
|
||||
# 3. 生成图片 (返回字典)
|
||||
images_dict = self.generate_images(content, topic_data)
|
||||
|
||||
# 4. 为每个平台生成HTML
|
||||
platforms = ["zhihu", "wechat", "xiaohongshu"]
|
||||
topic_id = topic_data["topic"].get("id", "unknown")
|
||||
|
||||
# 确定图片顺序
|
||||
ordered_keys = ["cover", "data_chart", "case_comparison", "action_checklist", "equipment"]
|
||||
images_list = [images_dict[k] for k in ordered_keys if k in images_dict]
|
||||
|
||||
for platform in platforms:
|
||||
html = self.create_html_for_platform(content, images_dict, platform, topic_data)
|
||||
|
||||
article = ContentArticle(
|
||||
id=f"{topic_id}_{platform}",
|
||||
topic_id=topic_id,
|
||||
title=topic_data["topic"].get("title", ""),
|
||||
platform=platform,
|
||||
content=html,
|
||||
image_paths=images_list, # 有序列表
|
||||
metadata={
|
||||
"platform": platform,
|
||||
"topic": topic_data["topic"],
|
||||
"cases": topic_data["cases"],
|
||||
"word_count": len(content)
|
||||
},
|
||||
created_date=TODAY,
|
||||
output_dir=str(self.release_dir / platform)
|
||||
)
|
||||
|
||||
self.save_article(article)
|
||||
self.articles.append(article)
|
||||
|
||||
# 5. 更新选题状态
|
||||
self.update_topic_status(topic_id)
|
||||
|
||||
# 6. 发送通知
|
||||
self.send_wecom_notification()
|
||||
|
||||
# 7. 合规审查（如有生成文章）
|
||||
if self.articles:
|
||||
self.run_compliance_check(self.articles[0])
|
||||
|
||||
logger.info(f"创作完成: {len(self.articles)} 篇文章")
|
||||
return True
|
||||
|
||||
def update_topic_status(self, topic_id: str):
|
||||
"""更新选题状态为已发布"""
|
||||
topics_file = DATA_DIR / "sustainability_topics.json"
|
||||
if not topics_file.exists():
|
||||
return
|
||||
|
||||
with open(topics_file, 'r', encoding='utf-8') as f:
|
||||
all_topics = json.load(f)
|
||||
|
||||
for topic in all_topics:
|
||||
if topic.get("id") == topic_id:
|
||||
topic["status"] = "已发布"
|
||||
topic["published_date"] = TODAY
|
||||
break
|
||||
|
||||
with open(topics_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(all_topics, f, ensure_ascii=False, indent=2)
|
||||
|
||||
logger.info(f"更新选题 {topic_id} 状态为已发布")
|
||||
|
||||
def run_compliance_check(self, article):
|
||||
"""运行合规审查"""
|
||||
try:
|
||||
from scripts.compliance_checker import check_article
|
||||
|
||||
result = check_article(
|
||||
article.content,
|
||||
article.platform,
|
||||
article.metadata.get("topic") if hasattr(article, 'metadata') else None
|
||||
)
|
||||
|
||||
# 记录审查结果
|
||||
log_msg = f"合规审查: {article.platform} - {article.title[:30]} - 得分: {result['score']} - 问题数: {len(result['issues'])}"
|
||||
if result['passed']:
|
||||
logger.info(log_msg)
|
||||
else:
|
||||
logger.warning(log_msg)
|
||||
for issue in result['issues']:
|
||||
logger.warning(f" [合规问题] {issue['type']}/{issue.get('category','')}: {issue.get('suggestion','')}")
|
||||
|
||||
# 保存审查报告
|
||||
compliance_file = self.today_dir / f"compliance_{article.id}.json"
|
||||
with open(compliance_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(result, f, ensure_ascii=False, indent=2)
|
||||
|
||||
logger.info(f"合规报告已保存: {compliance_file}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"合规审查失败: {e}")
|
||||
|
||||
|
||||
def main():
|
||||
"""主函数"""
|
||||
try:
|
||||
creator = ContentCreator()
|
||||
success = creator.run()
|
||||
|
||||
if success:
|
||||
print(f"SUCCESS: Created {len(creator.articles)} articles for {TODAY}")
|
||||
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()
|
||||
Reference in New Issue
Block a user