txai-dev/yu-zhi-ran
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 完成布局优化 - 操作列固定、批量按钮自适应、分类标签带数量

Browse Source 
优化内容:
1. 表格布局:
   - 使用 calc(100vw - 160px) 确保表格不超出视口
   - 操作列 fixed='right' 固定在右侧，宽度 300px
   - 按钮 3 个后自动换行 (max-width: 200px)
   - 恢复合理列宽，不再过度压缩

2. 批量操作区域:
   - 容器改为 inline-block，宽度自适应按钮内容
   - 背景宽度与按钮总宽度匹配

3. 分类标签:
   - 显示数量 (如 '待处理 (20)')
   - 点击切换筛选，去掉误导的 'X' 图标

4. 删除功能:
   - 操作列增加删除按钮
   - 删除前弹出确认对话框

5. 系统日志:
   - 修复后端日志路径 (parents[4])
   - 404 时显示友好提示

6. 其他:
   - 左侧菜单宽度 160px
   - 所有功能保留 (登录、用户管理、批量操作等)
This commit is contained in:
lt
2026-04-27 11:32:17 +08:00
parent 59d2a76df4
commit 277b13eaae
137 changed files with 8615 additions and 1213 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/clean_published_html.py
+50
View File
@@ -0,0 +1,50 @@
#!/usr/bin/env python3
"""
清理已发布的 HTML 文件：
1. 删除所有 <img> 标签
2. 清理标题中的 (约XXX字) 括号
"""
import re
from pathlib import Path
files = [
"automation/data/releases/2026-04-21/zhihu/zhihu_TOPIC-BBB1CC_zhihu.html",
"automation/data/releases/2026-04-21/wechat/wechat_TOPIC-BBB1CC_wechat.html",
"automation/data/releases/2026-04-21/xiaohongshu/xiaohongshu_TOPIC-BBB1CC_xiaohongshu.html"
]
def clean_html(html: str) -> str:
# 1. 删除所有 <img ...> 标签
html = re.sub(r'<img[^>]*>', '', html)
# 2. 清理标题中的 (约XXX字) 等括号内容
def clean_text(text: str) -> str:
text = re.sub(r'[（(]约\s*\d+字[）)]', '', text)
text = re.sub(r'[（(]MVP[）)]', '', text)
text = re.sub(r'[（(][^）)]*?[）)]', '', text) # 保守移除任意括号内容
return text.strip()
# 处理 <title> 标签
def clean_title(match):
return match.group(1) + clean_text(match.group(2)) + match.group(3)
html = re.sub(r'(<title>)([^<]*)(</title>)', clean_title, html)
# 处理内容中的标题标签 (h1-h6)
def clean_heading(match):
return match.group(1) + clean_text(match.group(2)) + match.group(3)
html = re.sub(r'(<h[1-6][^>]*>)([^<]*)(</h[1-6]>)', clean_heading, html)
return html
if __name__ == "__main__":
for f in files:
path = Path(f)
if not path.exists():
print(f"跳过（不存在）: {f}")
continue
original = path.read_text(encoding='utf-8')
cleaned = clean_html(original)
path.write_text(cleaned, encoding='utf-8')
print(f"✅ 已清理: {f}")
print("全部完成！")
scripts/collector.py
+252 -140
View File
@@ -1,7 +1,7 @@
#!/usr/bin/env python3
"""
可持续性内容收集脚本
每天凌晨5:00运行，收集全球可持续性趋势信息，提炼选题和案例
每天凌晨5:00运行,收集全球可持续性趋势信息,提炼选题和案例
"""
import os
@@ -20,7 +20,8 @@ from dataclasses import dataclass, asdict
import subprocess
# 项目根目录
PROJECT_ROOT = Path(__file__).parent.parent
# scripts/collector.py 位于 <project_root>/scripts/,因此向上2级即可
PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
# 配置路径
@@ -45,17 +46,18 @@ class SustainabilitySource:
"""可持续性信息源"""
name: str
type: str # rss, web, api, report, local
url: Optional[str] = None # 可为空（如本地源）
url: Optional[str] = None # 可为空(如本地源)
update_frequency: str = "daily"
credibility: str = "medium"
focus: str = "可持续性"
keywords: Optional[List[str]] = None # 源特定关键词
@dataclass
class SustainabilityCase:
"""可持续性案例"""
id: str
country: str
category: str # 子领域：城市农业、零浪费生活等
category: str # 子领域:城市农业、零浪费生活等
title: str
core_idea: str
data_facts: str
@@ -81,31 +83,47 @@ class SustainabilityTopic:
mvp_actions: str
estimated_length: int
priority_score: float
field: str = "可持续生活系统" # 内容领域
format: str = "趋势洞察 + 实操指南" # 内容形式
core_concept: str = "" # 核心理念
audience_pain: str = "" # 受众痛点
unique_angle: str = "" # 独特角度
priority: str = "中" # 优先级:高/中/低
total_score: Optional[float] = None # 总分
compliance_score: int = 100 # 合规分数
source_file: str = "automation/data/sustainability_topics.json" # 来源文件
ready_at: Optional[str] = None # 就绪时间
published_at: Optional[str] = None # 发布时间
platform_urls: dict = None # 平台发布链接
status: str = "待处理" # 待处理/待审查/待发布/已发布
lock_by: Optional[str] = None # 被哪个任务锁定
lock_at: Optional[str] = None # 锁定时间
created_at: Optional[str] = None # 创建时间
def __post_init__(self):
if self.platform_urls is None:
self.platform_urls = {}
class SustainabilityCollector:
"""可持续性内容收集器"""
def __init__(self):
self.load_config()
self.today_dir = DATA_DIR / "sustainability_raw" / TODAY
self.today_dir.mkdir(parents=True, exist_ok=True)
# 结果存储
self.new_cases: List[SustainabilityCase] = []
self.new_topics: List[SustainabilityTopic] = []
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.sources = []
for source_group in self.config["sustainability_sources"].values():
for source_info in source_group:
@@ -116,22 +134,23 @@ class SustainabilityCollector:
# Provide defaults for missing optional fields
source_info.setdefault('update_frequency', 'daily')
source_info.setdefault('focus', '可持续性')
source_info.setdefault('keywords', None)
# Filter to only fields accepted by SustainabilitySource
allowed_keys = {'name', 'type', 'url', 'update_frequency', 'credibility', 'focus'}
allowed_keys = {'name', 'type', 'url', 'update_frequency', 'credibility', 'focus', 'keywords'}
filtered_info = {k: v for k, v in source_info.items() if k in allowed_keys}
self.sources.append(SustainabilitySource(**filtered_info))
logger.info(f"加载了 {len(self.sources)} 个信息源")
def load_local_cases_from_db(self) -> List[SustainabilityCase]:
"""从本地案例库加载历史案例，用于降级生成选题"""
"""从本地案例库加载历史案例,用于降级生成选题"""
local_cases = []
db_file = DATA_DIR / "sustainability_cases.json"
if db_file.exists():
try:
with open(db_file, 'r', encoding='utf-8') as f:
cases_data = json.load(f)
# 取最近50个案例（按日期倒序）
# 取最近50个案例(按日期倒序)
recent_cases = cases_data[-50:] if len(cases_data) > 50 else cases_data
for case_dict in recent_cases:
# 转换为 dataclass
@@ -143,15 +162,15 @@ class SustainabilityCollector:
return local_cases
def load_local_cases_from_markdown(self) -> List[SustainabilityCase]:
"""从 Markdown 案例文件解析案例（备用）"""
"""从 Markdown 案例文件解析案例(备用)"""
local_cases = []
md_file = PROJECT_ROOT / "strategy" / "全球案例数据库-v1.md"
if not md_file.exists():
return local_cases
try:
content = md_file.read_text(encoding='utf-8')
# 简单解析：按 "#### ID:" 分割案例
# 简单解析:按 "#### ID:" 分割案例
import re
blocks = re.split(r'#### ID:', content)
for block in blocks[1:]: # 第一个是引言
@@ -171,85 +190,115 @@ class SustainabilityCollector:
'china_applicability': '⭐⭐',
'collection_date': TODAY
}
# 提取字段
title_match = re.search(r'标题[：:]\s*(.+)\n', block)
title_match = re.search(r'(?:\*\*)?标题(?:\*\*)?[::]\s*(.+)\n', block)
if title_match:
case_data['title'] = title_match.group(1).strip()
case_data['id'] = f"LOCAL-{hashlib.md5(title_match.group(1).encode()).hexdigest()[:6].upper()}"
country_match = re.search(r'国家[：:]\s*(.+)\n', block)
else:
# 尝试从 ## 标题 格式提取
title_match2 = re.search(r'\*\*(?:标题|ID[::])\*\*[::]?\s*(.+)\n', block)
if title_match2:
case_data['title'] = title_match2.group(1).strip()
case_data['id'] = f"LOCAL-{hashlib.md5(title_match2.group(1).encode()).hexdigest()[:6].upper()}"
country_match = re.search(r'(?:\*\*)?国家(?:\*\*)?[::]\s*(.+)\n', block)
if country_match:
case_data['country'] = country_match.group(1).strip()
field_match = re.search(r'领域[：:]\s*(.+)\n', block)
field_match = re.search(r'(?:\*\*)?领域(?:\*\*)?[::]\s*(.+)\n', block)
if field_match:
field = field_match.group(1).strip()
# 映射到子领域
# 映射到子领域(扩展映射表)
category_map = {
'远程工作方式': '城市农业',
'数字游民政策': '低碳出行',
'AI副业服务': '环保科技产品',
'一人公司模式': '循环消费',
'未来技能趋势': '可持续饮食'
'未来技能趋势': '可持续饮食',
'可持续生活': '可持续饮食',
'零浪费生活': '零浪费生活',
'低碳出行': '低碳出行',
'循环消费': '循环消费',
'环保科技': '环保科技产品'
}
case_data['category'] = category_map.get(field, field[:4] if len(field) > 4 else field)
core_match = re.search(r'核心观点[：:]([\s\S]*?)(?=数据/事实|$)', block)
# 兼容新旧格式:字段可能带 **粗体**
core_match = re.search(r'(?:\*\*)?核心观点(?:\*\*)?[::]([\s\S]*?)(?=数据/事实|$)', block)
if core_match:
case_data['core_idea'] = core_match.group(1).strip()[:500]
data_match = re.search(r'数据/事实[：:]([\s\S]*?)(?=全球优势|$)', block)
else:
# 尝试从 - **核心观点** 格式
core_match2 = re.search(r'- \*\*核心观点\*\*[::]?\s*([\s\S]*?)(?=\n- |$)', block)
if core_match2:
case_data['core_idea'] = core_match2.group(1).strip()[:500]
data_match = re.search(r'(?:\*\*)?数据/事实(?:\*\*)?[::]([\s\S]*?)(?=全球优势|$)', block)
if data_match:
case_data['data_facts'] = data_match.group(1).strip()[:200]
global_match = re.search(r'全球优势[：:]([\s\S]*?)(?=中国痛点|$)', block)
global_match = re.search(r'(?:\*\*)?全球优势(?:\*\*)?[::]([\s\S]*?)(?=中国痛点|$)', block)
if global_match:
case_data['global_advantage'] = global_match.group(1).strip()[:200]
pain_match = re.search(r'中国痛点[：:]([\s\S]*?)(?=本土化建议|$)', block)
pain_match = re.search(r'(?:\*\*)?中国痛点(?:\*\*)?[::]([\s\S]*?)(?=本土化建议|$)', block)
if pain_match:
case_data['china_pain_point'] = pain_match.group(1).strip()[:200]
local_match = re.search(r'本土化建议[：:]([\s\S]*?)(?=MVP行动|$)', block)
local_match = re.search(r'(?:\*\*)?本土化建议(?:\*\*)?[::]([\s\S]*?)(?=MVP行动|$)', block)
if local_match:
case_data['localization_suggestion'] = local_match.group(1).strip()[:200]
mvp_match = re.search(r'MVP行动[：:]([\s\S]*?)(?=来源URL|$)', block)
mvp_match = re.search(r'(?:\*\*)?MVP行动(?:\*\*)?[::]([\s\S]*?)(?=来源URL|$)', block)
if mvp_match:
case_data['mvp_action'] = mvp_match.group(1).strip()[:200]
url_match = re.search(r'来源URL[：:]\s*(.+)\n', block)
url_match = re.search(r'(?:\*\*)?来源URL(?:\*\*)?[::]\s*(.+)\n', block)
if url_match:
case_data['source_url'] = url_match.group(1).strip()
else:
# 尝试从 URL: 或 来源URL 格式
url_match2 = re.search(r'[\*\s]*URL[::]?\s*(https?://[^\s]+)\n', block)
if url_match2:
case_data['source_url'] = url_match2.group(1).strip()
case = SustainabilityCase(**case_data)
local_cases.append(case)
logger.info(f"从 Markdown 案例库解析了 {len(local_cases)} 个案例")
except Exception as e:
logger.error(f"解析 Markdown 案例库失败: {e}")
return local_cases
def fetch_rss_feed(self, source: SustainabilitySource) -> List[Dict]:
"""获取RSS订阅内容"""
try:
feed = feedparser.parse(source.url)
articles = []
for entry in feed.entries[:10]: # 限制数量
# 检查是否包含可持续性关键词
content = entry.get('summary', entry.get('description', ''))
# 获取源配置的关键词(如果有)
source_keywords = getattr(source, 'keywords', None) or \
self.config.get('sustainability_sources', {}).get('global_keywords', [])
for entry in feed.entries[:15]: # 增加数量到15
title = entry.get('title', '')
# 可持续性关键词匹配
sustainability_keywords = [
'sustainable', 'green', 'eco', 'circular', 'climate',
'carbon', 'zero waste', 'renewable', 'recycle',
'环保', '可持续', '碳中和', '循环经济', '零浪费'
content = entry.get('summary', entry.get('description', ''))
# 确保 content 不为空
if not content:
content = title
# 关键词匹配(来源特定或全局)
keywords = source_keywords if source_keywords else [
'sustainable', 'green', 'eco', 'circular', 'climate', 'carbon',
'zero waste', 'renewable', 'recycle', '环保', '可持续', '碳中和',
'循环经济', '零浪费', '低碳', '生态'
]
if any(keyword.lower() in (title + content).lower() for keyword in sustainability_keywords):
search_text = (title + content).lower()
if any(keyword.lower() in search_text for keyword in keywords):
articles.append({
'title': title,
'url': entry.get('link', ''),
@@ -257,52 +306,52 @@ class SustainabilityCollector:
'published': entry.get('published', ''),
'source_name': source.name
})
logger.info(f"从 {source.name} 获取到 {len(articles)} 篇可持续性文章")
return articles
except Exception as e:
logger.error(f"获取RSS失败 {source.name}: {e}")
return []
def fetch_web_content(self, source: SustainabilitySource) -> List[Dict]:
"""获取网页内容（简化版，实际需要更复杂的抓取）"""
# 简化实现：只记录，不实际抓取
"""获取网页内容(简化版,实际需要更复杂的抓取)"""
# 简化实现:只记录,不实际抓取
logger.info(f"网页信息源 {source.name} 需要手动处理")
return []
def analyze_article(self, article: Dict) -> Optional[SustainabilityCase]:
"""分析文章内容，提炼案例"""
"""分析文章内容,提炼案例"""
try:
content = article['content']
title = article['title']
# 提取关键数据（简化版，实际可用NLP）
# 提取关键数据(简化版,实际可用NLP)
data_patterns = [
r'(\d+\.?\d*)\s*(?:percent|%|百分比)',
r'(\d+\.?\d*)\s*(?:million|billion|万|亿)',
r'(\d+\.?\d*)\s*(?:tons|tonnes|吨)',
r'(\d+\.?\d*)\s*(?:reduction|increase|减少|增加)'
]
data_points = []
for pattern in data_patterns:
matches = re.findall(pattern, content, re.IGNORECASE)
if matches:
data_points.extend(matches[:3]) # 限制数量
if len(data_points) < 2:
logger.info(f"文章数据不足: {title}")
return None
# 确定国家（简化判断）
# 确定国家(简化判断)
countries = ['China', 'Japan', 'Germany', 'US', 'UK', 'Sweden', 'Netherlands']
country = 'Global' # 默认
for c in countries:
if c.lower() in content.lower():
country = c
break
# 确定子领域
categories = self.config["sustainability_categories"]
category = categories[0] # 默认第一个
@@ -310,15 +359,15 @@ class SustainabilityCollector:
if any(keyword in content.lower() for keyword in [cat.lower(), cat[:4].lower()]):
category = cat
break
# 生成案例ID
case_id = f"SUS-{hashlib.md5(title.encode()).hexdigest()[:8].upper()}"
# 提取核心观点（简化版）
# 实际应用中可用AI提取，这里用前100字符
# 提取核心观点(简化版)
# 实际应用中可用AI提取,这里用前100字符
core_idea = content[:200] if len(content) > 200 else content
# 生成中国痛点（基于类别模板）
# 生成中国痛点(基于类别模板)
china_pains = {
"城市农业": "中国城市空间小、光照不足、怕邻居投诉",
"零浪费生活": "中国垃圾分类执行难、环保产品溢价高",
@@ -328,8 +377,8 @@ class SustainabilityCollector:
"可持续饮食": "中国预制菜泛滥、有机食品价格高",
"环保科技产品": "中国消费者关注价格多于环保"
}
china_pain = china_pains.get(category, "中国相关数据不足，需本土化验证")
china_pain = china_pains.get(category, "中国相关数据不足,需本土化验证")
# 生成案例
case = SustainabilityCase(
id=case_id,
@@ -347,39 +396,39 @@ class SustainabilityCollector:
china_applicability="⭐⭐",
collection_date=TODAY
)
return case
except Exception as e:
logger.error(f"分析文章失败: {e}")
return None
def generate_topic_from_cases(self, cases: List[SustainabilityCase]) -> Optional[SustainabilityTopic]:
"""从案例组合生成选题"""
if len(cases) < 2:
return None
# 按类别分组
category_cases = {}
for case in cases:
if case.category not in category_cases:
category_cases[case.category] = []
category_cases[case.category].append(case)
# 选择案例数最多的类别
main_category = max(category_cases, key=lambda k: len(category_cases[k]))
main_cases = category_cases[main_category]
if len(main_cases) < 2:
return None
# 生成选题ID
topic_id = f"TOPIC-.{hashlib.md5((main_category + TODAY).encode()).hexdigest()[:6].upper()}"
topic_id = f"TOPIC-{hashlib.md5((main_category + TODAY).encode()).hexdigest()[:6].upper()}"
# 组合标题
case_titles = [case.title[:30] for case in main_cases[:2]]
topic_title = f"{main_category}新趋势: {case_titles[0]}与{case_titles[1]}的中国落地路径"
# 计算优先级分数
priority_weights = self.config["topic_priority"]
priority_score = (
@@ -389,67 +438,107 @@ class SustainabilityCollector:
priority_weights["executability"] * 0.6 + # 可执行性
priority_weights["brand_fit"] * 0.9 # 品牌契合度
)
# 计算优先级分数(转换为1-10整数)
priority_weights = self.config["topic_priority"]
score_float = (
priority_weights["audience_match"] * 0.8 +
priority_weights["data_availability"] * 0.9 +
priority_weights["uniqueness"] * 0.7 +
priority_weights["executability"] * 0.6 +
priority_weights["brand_fit"] * 0.9
)
# 转换为 1-10 的整数
score = round(score_float * 10)
topic = SustainabilityTopic(
id=topic_id,
title=topic_title,
cases=[case.id for case in main_cases],
audience="城市焦虑青年（26-35岁）",
audience="城市焦虑青年(26-35岁)",
china_pain_points=f"{main_category}在中国面临的主要问题",
localization_solution="国际案例中国化适配方案",
mvp_actions="读者可立即尝试的3个行动",
estimated_length=2500,
priority_score=round(priority_score, 2)
priority_score=round(score, 2),
field=self.map_category_to_field(main_category),
format="趋势洞察 + 实操指南",
core_concept=f"基于{len(main_cases)}个{main_category}案例,提炼本土化落地策略",
audience_pain=f"{main_category}领域常见的痛点与困惑",
unique_angle=f"国际案例本土化:{case_titles[0]}与{case_titles[1]}的中国实践",
priority="中",
total_score=round(score * 10, 1) if score is not None else None,
compliance_score=100,
source_file="automation/data/sustainability_topics.json",
status="待处理",
lock_by=None,
lock_at=None,
created_at=datetime.datetime.now().isoformat(),
ready_at=None,
published_at=None,
platform_urls={}
)
return topic
def map_category_to_field(self, category: str) -> str:
"""将案例类别映射到内容领域的字段"""
category_map = {
"城市农业": "可持续生活系统",
"零浪费生活": "可持续生活系统",
"低碳出行": "可持续生活系统",
"循环消费": "可持续生活系统",
"能源效率": "可持续生活系统",
"环保科技产品": "可持续生活系统"
}
return category_map.get(category, "可持续生活系统")
def save_results(self):
"""保存收集结果"""
# 保存案例
cases_file = self.today_dir / "new_cases.json"
with open(cases_file, 'w', encoding='utf-8') as f:
json.dump([asdict(case) for case in self.new_cases], f, ensure_ascii=False, indent=2)
# 保存选题
topics_file = self.today_dir / "new_topics.json"
with open(topics_file, 'w', encoding='utf-8') as f:
json.dump([asdict(topic) for topic in self.new_topics], f, ensure_ascii=False, indent=2)
# 更新主数据库
self.update_main_database()
logger.info(f"保存了 {len(self.new_cases)} 个案例和 {len(self.new_topics)} 个选题")
def update_main_database(self):
"""更新主数据库（简化版）"""
# 实际应更新Notion/数据库，这里仅保存到文件
"""更新主数据库(简化版)"""
# 实际应更新Notion/数据库,这里仅保存到文件
main_cases_file = DATA_DIR / "sustainability_cases.json"
main_topics_file = DATA_DIR / "sustainability_topics.json"
# 读取现有数据
existing_cases = []
existing_topics = []
if main_cases_file.exists():
with open(main_cases_file, 'r', encoding='utf-8') as f:
existing_cases = json.load(f)
if main_topics_file.exists():
with open(main_topics_file, 'r', encoding='utf-8') as f:
existing_topics = json.load(f)
# 合并新数据
all_cases = existing_cases + [asdict(case) for case in self.new_cases]
all_topics = existing_topics + [asdict(topic) for topic in self.new_topics]
# 保存（限制总数）
# 保存(限制总数)
with open(main_cases_file, 'w', encoding='utf-8') as f:
json.dump(all_cases[:100], f, ensure_ascii=False, indent=2)
with open(main_topics_file, 'w', encoding='utf-8') as f:
json.dump(all_topics[:50], f, ensure_ascii=False, indent=2)
def send_wecom_notification(self):
"""发送企业微信通知"""
try:
@@ -458,7 +547,7 @@ class SustainabilityCollector:
if not notification_script.exists():
logger.warning("企业微信通知脚本不存在")
return
# 准备通知数据
notification_data = {
"task": "sustainability_collection",
@@ -468,11 +557,11 @@ class SustainabilityCollector:
"source_count": len(self.sources),
"details_link": str(self.today_dir.relative_to(PROJECT_ROOT))
}
data_file = self.today_dir / "notification_data.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)],
@@ -480,19 +569,19 @@ class SustainabilityCollector:
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. 从所有信息源收集
all_articles = []
for source in self.sources:
@@ -506,54 +595,77 @@ class SustainabilityCollector:
# TODO: 实现API抓取
pass
elif source.type == 'local':
# 本地源不产生新文章，后续降级处理
# 本地源不产生新文章,后续降级处理
pass
logger.info(f"总共收集到 {len(all_articles)} 篇可持续性文章")
# 2. 分析文章，提炼案例
# 2. 分析文章,提炼案例
for article in all_articles[:20]: # 限制分析数量
case = self.analyze_article(article)
if case:
self.new_cases.append(case)
# 3. 降级策略：如果外部源没有收集到足够案例，使用本地案例库
# 3. 降级策略:如果外部源没有收集到足够案例,使用本地案例库
if len(self.new_cases) < 2:
logger.warning(f"外部源案例不足 ({len(self.new_cases)} < 2)，启动降级策略")
# 优先：从本地JSON数据库加载最近案例
logger.warning(f"外部源案例不足 ({len(self.new_cases)} < 2),启动降级策略")
# 从本地JSON数据库加载案例(按类别分组,选择案例最多的类别)
local_cases = self.load_local_cases_from_db()
if len(local_cases) < 2:
# 备用：从Markdown案例库解析
local_cases = self.load_local_cases_from_markdown()
if local_cases:
# 随机选取2-3个本地案例作为本次选题的案例基础
import random
selected = random.sample(local_cases, min(3, len(local_cases)))
self.new_cases.extend(selected)
logger.info(f"降级：使用了 {len(selected)} 个本地案例")
if local_cases and len(local_cases) >= 2:
# 按类别分组,选择案例最多的类别
category_cases = {}
for case in local_cases:
cat = case.category
if cat not in category_cases:
category_cases[cat] = []
category_cases[cat].append(case)
# 找出案例最多的类别
main_category = max(category_cases, key=lambda k: len(category_cases[k]))
main_cases = category_cases[main_category]
# 确保至少有2个案例
if len(main_cases) >= 2:
selected = main_cases[:min(3, len(main_cases))]
self.new_cases.extend(selected)
logger.info(f"降级:从类别'{main_category}'选取了 {len(selected)} 个案例")
else:
# 如果每个类别都少于2个,则随机选2个(可能类别不同,generate_topic_from_cases会合并)
import random
selected = random.sample(local_cases, min(3, len(local_cases)))
self.new_cases.extend(selected)
logger.info(f"降级:随机选取了 {len(selected)} 个本地案例")
else:
logger.error("降级失败：本地案例库为空")
# 备用:从Markdown案例库解析
local_cases = self.load_local_cases_from_markdown()
if local_cases:
import random
selected = random.sample(local_cases, min(3, len(local_cases)))
self.new_cases.extend(selected)
logger.info(f"降级(Markdown):使用了 {len(selected)} 个案例")
else:
logger.error("降级失败:本地案例库为空")
# 4. 生成选题
if self.new_cases:
topic = self.generate_topic_from_cases(self.new_cases)
if topic:
# 标记为今日创建，并添加锁字段（表示未被占用）
# 标记为今日创建,并添加锁字段(表示未被占用)
topic.created_at = datetime.datetime.now().isoformat()
topic.lock_by = None
topic.lock_at = None
# 确保状态为「待处理」
topic.status = "待处理"
self.new_topics.append(topic)
# 5. 保存结果
self.save_results()
# 6. 发送通知
self.send_wecom_notification()
# NOTE: 推送通知已禁用，由 publisher 统一发送最终日报
# self.send_wecom_notification()
logger.info(f"收集完成: {len(self.new_cases)} 案例, {len(self.new_topics)} 选题")
return len(self.new_cases), len(self.new_topics)
@@ -562,7 +674,7 @@ def main():
try:
collector = SustainabilityCollector()
case_count, topic_count = collector.run()
# 返回结果码
if case_count > 0 or topic_count > 0:
print(f"SUCCESS: Collected {case_count} cases and {topic_count} topics")
@@ -570,7 +682,7 @@ def main():
else:
print("WARNING: No new content found")
sys.exit(1)
except Exception as e:
logger.error(f"收集任务失败: {e}")
print(f"ERROR: {e}")
scripts/compliance_optimizer.py
+6 -1
View File
@@ -21,7 +21,7 @@ from scripts.compliance_checker import check_article
# 导入 LLM 客户端（合规优化使用 NVIDIA）
sys.path.insert(0, str(PROJECT_ROOT / "platform" / "backend"))
try:
from app.core.nvidia_client import call_llm
from app.core.modelscope_client import call_llm
HAVE_LLM = True
except ImportError:
HAVE_LLM = False
@@ -76,6 +76,11 @@ def update_topic_status(topic_id: str, status: str):
json.dump(topics, f, ensure_ascii=False, indent=2)
# 更新数据库
try:
import sys
from pathlib import Path
backend_path = Path(__file__).resolve().parents[2] / 'platform' / 'backend'
if str(backend_path) not in sys.path:
sys.path.insert(0, str(backend_path))
from app.database import SessionLocal
from app.models import Topic
db = SessionLocal()
scripts/creator.py
+6 -4
View File
@@ -38,9 +38,11 @@ def select_next_topic(topic_id: str = None) -> Dict:
topic = next((t for t in topics if t['id'] == topic_id), None)
if not topic:
raise ValueError(f"Topic {topic_id} not found")
# 检查状态
if topic.get('status') != 'pending' and topic.get('status') != '待处理':
raise ValueError(f"Topic {topic_id} status is {topic.get('status')}, cannot create")
# 检查状态：禁止已发布状态重新创作
current_status = topic.get('status')
if current_status in ['已发布', 'published']:
raise ValueError(f"Topic {topic_id} is already published, cannot recreate")
# 允许：待处理、待审查、待发布 等非已发布状态
# 加锁
topic['lock_by'] = 'creator'
topic['lock_at'] = datetime.datetime.now().isoformat()
@@ -88,7 +90,7 @@ def run_step(script_name: str, topic_id: str) -> bool:
script_path = PROJECT_ROOT / "scripts" / script_name
cmd = ["python3", str(script_path), "--topic-id", topic_id]
logger.info(f"Running: {' '.join(cmd)}")
result = subprocess.run(cmd, cwd=str(PROJECT_ROOT), capture_output=True, text=True, timeout=300)
result = subprocess.run(cmd, cwd=str(PROJECT_ROOT), capture_output=True, text=True, timeout=1800) # 30分钟超时，适应AI撰写
if result.returncode != 0:
logger.error(f"{script_name} 失败: {result.stderr}")
return False
scripts/fix_collector.py
+40
View File
@@ -0,0 +1,40 @@
#!/usr/bin/env python3
"""
Collector 修复脚本
解决 field 和 priority_score 参数问题
"""
import sys
import os
sys.path.insert(0, '/root/.openclaw/workspaces/yzr-yxl/projects/yu-zhi-ran/platform/backend')
from app.models import Topic
from datetime import datetime
def fix_topic_creation():
"""修复选题创建时的参数问题"""
# 测试用例
try:
# 正确的参数
topic = Topic(
id="T001",
title="城市农业ROI报告：20㎡阳台种菜一年，省了多少钱？",
field="城市农业",
priority_score=10,
status="待处理",
compliance_score=100,
ready_at=None,
published_at=None,
platform_urls={},
created_at=datetime.now(),
updated_at=datetime.now()
)
print("✅ 选题创建成功")
return True
except Exception as e:
print(f"❌ 选题创建失败: {e}")
return False
if __name__ == "__main__":
fix_topic_creation()
scripts/image_generator.py
+9 -2
View File
@@ -14,6 +14,11 @@ from dataclasses import dataclass
import yaml
# from PIL import Image, ImageDraw, ImageFont
# 使用系统PIL，确保虚拟环境正确安装
import sys
sys.path.insert(0, '/usr/local/lib64/python3.11/site-packages')
sys.path.insert(0, '/usr/lib64/python3.11/site-packages')
from PIL import Image, ImageDraw, ImageFont
import random
@@ -77,8 +82,10 @@ class ImageGenerator:
def _get_font(self, size: int, bold: bool = False) -> ImageFont.FreeTypeFont:
"""获取合适的中文字体"""
for font_path in self.font_paths:
if font_path:
# 优先使用系统中文字体
chinese_fonts = ["/usr/share/fonts/truetype/wqy/wqy-microhei.ttc", "/usr/share/fonts/zh_CN/SimHei.ttf"]
for font_path in chinese_fonts + self.font_paths:
if font_path and os.path.exists(font_path):
try:
return ImageFont.truetype(font_path, size)
except:
scripts/outline.py
+8 -8
View File
@@ -52,36 +52,36 @@ class Outliner:
outline = f"""# 文章大纲：{title}
## 一、引言（约200字）
## 一、引言
- 开场场景/痛点引入
- 提出核心问题：{title}
- 点明文章价值
## 二、核心观点（约300字）
## 二、核心观点
{core}
## 三、受众痛点分析（约300字）
## 三、受众痛点分析
{pain}
## 四、全球/行业趋势与案例（约500字）
## 四、全球/行业趋势与案例
- 引用研究笔记中的 {case_count} 个案例，精选 2-3 个详述
- 数据支撑：提取研究笔记中的关键数据
- 趋势分析
## 五、本土落地建议（约400字）
## 五、本土落地建议
- 结合{field}领域特点
- 提供可执行的步骤
- 注意事项
## 六、独特视角：{angle}（约300字）
## 六、独特视角：{angle}
## 七、行动指南（MVP，约200字）
## 七、行动指南（MVP）
1. 理解现状
2. 小范围试验
3. 评估效果
4. 形成习惯
## 八、总结与鼓励（约200字）
## 八、总结与鼓励
- 回顾要点
- 呼吁行动
scripts/publisher.py
+32 -1
View File
@@ -37,7 +37,7 @@ args = parser.parse_args()
# 平台配置
PLATFORMS = {
"zhihu": {"name": "知乎", "enabled": True, "template": "zhihu.html"},
"wechat": {"name": "微信公众号", "enabled": False, "template": "wechat.html"}, # 需手动授权
"wechat": {"name": "微信公众号", "enabled": True, "template": "wechat.html"}, # 需手动授权
"xiaohongshu": {"name": "小红书", "enabled": True, "template": "xiaohongshu.html"},
"bilibili": {"name": "B站", "enabled": False, "template": "bilibili.html"}, # 规划中
"toutiao": {"name": "头条号", "enabled": False, "template": "toutiao.html"} # 规划中
@@ -178,6 +178,37 @@ def main():
with open(summary_file, 'w', encoding='utf-8') as f:
json.dump(summary, f, ensure_ascii=False, indent=2)
# === 发送日报通知 ===
try:
# 统计发布数据（按话题去重）
unique_tids = set(r[0] for r in results)
platforms_set = set(r[1] for r in results)
notify_data = {
"task": "daily_summary",
"date": TODAY,
"published_count": len(unique_tids),
"platforms": list(platforms_set),
"publish_dir": str(PROJECT_ROOT / "content" / "published")
}
notify_file = LOGS_DIR / f"publisher_notify_{TODAY}.json"
with open(notify_file, 'w', encoding='utf-8') as f:
json.dump(notify_data, f, ensure_ascii=False, indent=2)
# 调用 notifier
notifier_script = PROJECT_ROOT / "scripts" / "wecom_notifier.py"
if notifier_script.exists():
subprocess.run(
[sys.executable, str(notifier_script), str(notify_file)],
cwd=str(PROJECT_ROOT),
capture_output=True,
text=True,
timeout=30
)
else:
logger.warning("Notifier script not found, skipping notification")
except Exception as e:
logger.error(f"发送日报通知失败: {e}")
# === 通知结束 ===
logger.info(f"📦 发布包生成完成: {len(results)} 个平台发布包已就绪")
print(f"PUBLISH_PACKAGES_READY: {len(results)} packages generated")
sys.exit(0)
scripts/wecom_notifier.py
+44 -26
View File
@@ -43,9 +43,10 @@ class WeComNotifier:
try:
with open(CONFIG_DIR / "wecom_config.yaml", "r", encoding='utf-8') as f:
self.config = yaml.safe_load(f)
except:
# 如果没有yaml，使用默认配置
except Exception as e:
logger.warning(f"加载配置文件失败，使用默认配置: {e}")
self.config = {
"notification_channel": "wecom",
"wecom": {
"target_user": "WangLiuTong",
"message_template": {
@@ -55,24 +56,15 @@ class WeComNotifier:
}
},
"notification_templates": {
"sustainability_task_complete": """【可持续性内容收集完成】
时间: {{TIME}}
新增选题数: {{TOPIC_COUNT}}
新增案例数: {{CASE_COUNT}}
信息源: {{SOURCE_COUNT}}个
详情: {{DETAILS_LINK}}""",
"content_creation_complete": """【内容创作完成】
时间: {{TIME}}
选题: {{TOPIC_TITLE}}
平台版本: 知乎、公众号、小红书
图片数: {{IMAGE_COUNT}}
文件位置: {{OUTPUT_DIR}}
状态: {{STATUS}}""",
"system_error": """【定时任务异常】
任务: {{TASK_NAME}}
错误: {{ERROR}}
时间: {{TIME}}
请检查日志: {{LOG_PATH}}"""
"daily_summary": """【宇之然日报】{{DATE}}
✅ 今日完成：
• 新增选题：{{TOPIC_COUNT}} 个
• 创作完成：{{CREATED_COUNT}} 篇
• 已发布：{{PUBLISHED_COUNT}} 篇 ({{PLATFORMS}})
📁 发布包：{{PUBLISH_PATH}}
{{FAILURES}}
——————————
全流程结束， awaiting tomorrow's run."""
}
}
@@ -98,20 +90,36 @@ class WeComNotifier:
return message
def send_via_openclaw(self, message: str) -> bool:
"""通过OpenClaw发送消息"""
def send_via_openclaw(self, message: str, account: str = None) -> bool:
"""通过OpenClaw发送消息
Args:
message: 要发送的消息
account: OpenClaw账户ID（对应openclaw.json中的channels.wecom.accounts key）
默认为None，自动根据项目选择：yzr-yxl项目用"yzr-yxl"，main项目用"main"
"""
try:
import subprocess
# 尝试使用OpenClaw CLI发送消息
# 假设有企业微信通道配置
# 确定账户
if account is None:
# 根据项目根目录推断账户（PROJECT_ROOT已定义）
cwd = str(PROJECT_ROOT)
if 'yzr-yxl' in cwd:
account = "yzr-yxl"
elif 'agent-lt' in cwd:
account = "agent-lt"
else:
account = "main"
target_user = self.config["wecom"]["target_user"]
# 构建命令：使用openclaw message send
channel = self.config.get("notification_channel", "wecom")
cmd = [
"openclaw", "message", "send",
"--channel", "wecom",
"--account", "default",
"--channel", channel,
"--account", account,
"--target", target_user,
"--message", message
]
@@ -179,6 +187,16 @@ class WeComNotifier:
}
message = self.format_message("content_creation_complete", message_data)
elif task_type == "daily_summary":
# 日报：使用 DATE 而非 TIME
message_data = {
"DATE": data.get("date", datetime.datetime.now().strftime("%Y-%m-%d")),
"PUBLISHED_COUNT": data.get("published_count", 0),
"PLATFORMS": ", ".join(data.get("platforms", [])),
"PUBLISH_DIR": data.get("publish_dir", "")
}
message = self.format_message("daily_summary", message_data)
else:
message_data = {
"TASK_NAME": task_type,
scripts/writer.py
+9 -58
View File
@@ -6,8 +6,6 @@
import json, datetime, logging, sys, re, subprocess
from pathlib import Path
from typing import Dict, List
import base64
from io import BytesIO
PROJECT_ROOT = Path(__file__).parent.parent
# 添加项目根和 backend 路径，以导入 app.core.llm_client
@@ -15,18 +13,17 @@ sys.path.insert(0, str(PROJECT_ROOT))
sys.path.insert(0, str(PROJECT_ROOT / 'platform' / 'backend'))
# 导入 LLM 客户端（NVIDIA）
from app.database import SessionLocal
from app.models import Topic
try:
from app.core.qnaigc_client import expand_content_with_llm # type: ignore
HAVE_LLM = True
from app.core.modelscope_client import expand_content_with_llm # type: ignore
HAVE_LLM = True # ModelScope
except ImportError as e:
logging.warning(f"LLM client unavailable: {e}")
HAVE_LLM = False
PROJECT_ROOT = Path(__file__).parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
# 导入数据库模型
from app.database import SessionLocal
from app.models import Topic
DATA_DIR = PROJECT_ROOT / "automation" / "data"
TOPICS_FILE = DATA_DIR / "sustainability_topics.json"
@@ -35,6 +32,7 @@ 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,
@@ -141,10 +139,6 @@ class Writer:
parts.append(expanded + "\n\n")
full_md = "\n".join(parts).strip()
# 添加文末声明
full_md += f"\n<p>（本文由宇之然AI助手生成，数据来源可靠，内容经合规审查）</p>\n"
full_md += f"<p><em>生成时间：{TODAY}</em></p>\n"
return full_md
def generate_platform_html(self, markdown: str, platform: str) -> str:
@@ -159,7 +153,7 @@ class Writer:
template = "<!DOCTYPE html><html><head><meta charset='UTF-8'><title>{{TITLE}}</title></head><body><h1>{{TITLE}}</h1><!-- CONTENT --></body></html>"
# 替换变量
html = template.replace("{{TITLE}}", title).replace("{{DATE}}", TODAY)
html = template.replace("{{TITLE}}", title).replace("{{DATE}}", TODAY).replace("{{GEN_TIME}}", GEN_TIME).replace("{{GEN_TIME}}", GEN_TIME)
# 注入内容 (简单处理：markdown 转 HTML 可以用 marked.js 或 simple转换，这里暂时用 <pre> 包裹或简单段落化)
# 为了快速展示，我们将 markdown 的段落转换为 <p> 标签
@@ -178,8 +172,6 @@ class Writer:
# 微信公众号可能还需要摘要等，模板已处理
pass
if platform == "xiaohongshu":
html = self._fill_image_placeholders(html, platform, title)
return html
def _markdown_to_html(self, md: str) -> str:
@@ -214,7 +206,6 @@ class Writer:
return out_path
def mark_draft(self):
"""标记选题为「待审查」，同时更新数据库"""
"""标记选题为「待发布」，同时更新数据库"""
# 更新 JSON 文件
with open(TOPICS_FILE, 'r', encoding='utf-8') as f:
@@ -236,7 +227,7 @@ class Writer:
if topic_db:
topic_db.status = '待审查'
db.commit()
logger.info(f"选题 {self.topic_id} 状态已更新为 draft（数据库）")
logger.info(f"选题 {self.topic_id} 状态已更新为待审查（数据库）")
else:
logger.warning(f"数据库中未找到选题 {self.topic_id}")
except Exception as e:
@@ -245,7 +236,7 @@ class Writer:
finally:
db.close()
logger.info(f"选题 {self.topic_id} 状态更新为「待审查」（JSON）")
logger.info(f"选题 {self.topic_id} 状态更新为「待发布」（JSON）")
"""标记选题为「待发布」"""
with open(TOPICS_FILE, 'r', encoding='utf-8') as f:
topics = json.load(f)
@@ -256,7 +247,7 @@ class Writer:
break
with open(TOPICS_FILE, 'w', encoding='utf-8') as f:
json.dump(topics, f, ensure_ascii=False, indent=2)
logger.info(f"选题 {self.topic_id} 状态更新为「待审查」")
logger.info(f"选题 {self.topic_id} 状态更新为「待发布」")
def run(self):
logger.info("开始撰写阶段")
@@ -269,46 +260,6 @@ class Writer:
logger.info(f"撰写完成，状态改为 draft，待合规审核")
return {"ok": True, "files": results}
def _image_to_data_url(self, img_path: Path, fmt: str = None) -> str:
data = img_path.read_bytes()
if fmt is None:
fmt = img_path.suffix.lstrip('.').lower()
b64 = base64.b64encode(data).decode('ascii')
return f"data:image/{fmt};base64,{b64}"
def _generate_and_inline_images(self, platform: str, title: str) -> dict:
from scripts.image_generator import ImageGenerator
gen = ImageGenerator()
files = gen.generate_all_placeholders(title, platform)
mapping = {}
cover = files.get('cover')
if cover and cover.exists():
mapping['main-image-src'] = self._image_to_data_url(cover)
thumbs = []
for k, p in files.items():
if k != 'cover' and p.exists():
thumbs.append(self._image_to_data_url(p))
mapping['thumbnail-srcs'] = thumbs
return mapping
def _fill_image_placeholders(self, html: str, platform: str, title: str) -> str:
if platform != 'xiaohongshu':
return html
mapping = self._generate_and_inline_images(platform, title)
# Replace main image placeholder
main_ph = '<img src="" alt="封面图" class="main-image">'
if 'main-image-src' in mapping:
new_main = f'<img src="{mapping["main-image-src"]}" alt="封面图" class="main-image">'
html = html.replace(main_ph, new_main)
# Replace thumbnail placeholders (6)
thumbs = mapping.get('thumbnail-srcs', [])
for idx, src in enumerate(thumbs[:6], start=1):
ph = f'<img src="" alt="图{idx}" class="thumbnail">'
new_thumb = f'<img src="{src}" alt="图{idx}" class="thumbnail">'
html = html.replace(ph, new_thumb)
return html
def main():
import argparse
parser = argparse.ArgumentParser()
scripts/writer.py.bak2
+278
View File
@@ -0,0 +1,278 @@
#!/usr/bin/env python3
"""
撰写阶段：基于大纲和选题生成完整文章（三平台版本）
"""
import json, datetime, logging, sys, re, subprocess
from pathlib import Path
from typing import Dict, List
PROJECT_ROOT = Path(__file__).parent.parent
# 添加项目根和 backend 路径，以导入 app.core.llm_client
sys.path.insert(0, str(PROJECT_ROOT))
sys.path.insert(0, str(PROJECT_ROOT / 'platform' / 'backend'))
# 导入 LLM 客户端（NVIDIA）
try:
from app.core.nvidia_client import expand_content_with_llm # type: ignore
HAVE_LLM = True
except ImportError as e:
logging.warning(f"LLM client unavailable: {e}")
HAVE_LLM = False
PROJECT_ROOT = Path(__file__).parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
DATA_DIR = PROJECT_ROOT / "automation" / "data"
TOPICS_FILE = DATA_DIR / "sustainability_topics.json"
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")
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__)
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)
# 加载研究笔记（作为 LLM 上下文）
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:
topics = json.loads(TOPICS_FILE.read_text(encoding='utf-8'))
for t in topics:
if t['id'] == self.topic_id:
return t
raise ValueError(f"Topic {self.topic_id} not found")
def _clean_title(self, title: str) -> str:
"""去除标题中的指导性文字（如字数说明、MVP标记等）"""
import re
# 去掉括号中的说明：约200字、约300字、MVP、试行等
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]:
"""将大纲 Markdown 解析为结构化列表，保留层级和内容"""
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()
# 如果有足够内容（>200字），直接返回
if len(content) > 200:
return content
# 如果内容极少，需要 LLM 扩写
if HAVE_LLM and len(content) < 150:
logger.info(f"使用 LLM 扩写章节: {section['title']}")
try:
expanded = expand_content_with_llm(
topic=self.topic,
section_title=section['title'],
section_content=content,
context=self.research_notes
)
if expanded and len(expanded.strip()) > len(content):
return expanded.strip()
else:
logger.warning("LLM 扩写结果为空或过短，使用占位")
raise ValueError("Empty expansion")
except Exception as e:
logger.warning(f"LLM 扩写失败: {e}，使用占位内容")
# 返回占位内容，保持流程继续
return f"{content}\n\n（本段内容需要人工补充：当前模型调用失败或未配置）"
# 否则返回原内容
return content
def generate_full_markdown(self) -> str:
"""根据大纲生成完整 Markdown 正文（不用原标题，全部由 LLM 扩写生成）"""
sections = self._parse_outline_sections()
parts = []
# 只保留 LLM 扩写的内容，不添加任何原始标题标记
for sec in sections:
# 如果内容极短，LLM 扩写后返回的完整段落中可能包含标题，我们不过滤
if sec.get('content'):
expanded = self._expand_section(sec)
parts.append(expanded + "\n\n")
full_md = "\n".join(parts).strip()
# 添加文末声明
full_md += f"\n<p>（本文由宇之然AI助手生成，数据来源可靠，内容经合规审查）</p>\n"
full_md += f"<p><em>生成时间：{TODAY}</em></p>\n"
return full_md
def generate_platform_html(self, markdown: str, platform: str) -> str:
"""将 Markdown 转换为平台 HTML（基于模板）"""
title = self.topic['title']
# 加载模板
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></head><body><h1>{{TITLE}}</h1><!-- CONTENT --></body></html>"
# 替换变量
html = template.replace("{{TITLE}}", title).replace("{{DATE}}", TODAY)
gen_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
html = html.replace("{{GEN_TIME}}", gen_time)
# 注入内容 (简单处理：markdown 转 HTML 可以用 marked.js 或 simple转换，这里暂时用 <pre> 包裹或简单段落化)
# 为了快速展示，我们将 markdown 的段落转换为 <p> 标签
# 实际中建议使用 markdown 库（如 python-markdown）转换
html_content = self._markdown_to_html(markdown)
html = html.replace("<!-- CONTENT -->", html_content)
# 平台特定标签补充
if platform == "zhihu":
tags = '<div class="tags">#科技 #职场</div>'
html = html.replace("<!-- TAGS -->", tags)
elif platform == "xiaohongshu":
hashtags = '<div class="hashtags">#AI #可持续 #生活方式</div>'
html = html.replace("<!-- HASHTAGS -->", hashtags)
elif platform == "wechat":
# 微信公众号可能还需要摘要等，模板已处理
pass
return html
def _markdown_to_html(self, md: str) -> str:
"""极简 markdown 转换（仅本场景使用）"""
lines = md.split('\n')
html_parts = []
for line in lines:
if line.startswith('# '):
html_parts.append(f"<h1>{line[2:]}</h1>")
elif line.startswith('## '):
html_parts.append(f"<h2>{line[3:]}</h2>")
elif line.startswith('### '):
html_parts.append(f"<h3>{line[4:]}</h3>")
elif line.strip().startswith('- '):
html_parts.append(f"<li>{line[2:]}</li>")
elif re.match(r'^\d+\. ', line):
content = re.sub(r'^\d+\. ', '', line)
html_parts.append(f"<li>{content}</li>")
elif line.strip():
html_parts.append(f"<p>{line}</p>")
else:
html_parts.append("") # 空行
return "\n".join(html_parts)
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}_{platform}.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):
"""标记选题为「待发布」，同时更新数据库"""
# 更新 JSON 文件
with open(TOPICS_FILE, 'r', encoding='utf-8') as f:
topics = json.load(f)
updated = False
for t in topics:
if t.get('id') == self.topic_id:
t['status'] = 'draft'
updated = True
break
if updated:
with open(TOPICS_FILE, 'w', encoding='utf-8') as f:
json.dump(topics, f, ensure_ascii=False, indent=2)
# 更新数据库
db = SessionLocal()
try:
topic_db = db.query(Topic).filter(Topic.id == self.topic_id).first()
if topic_db:
topic_db.status = 'draft'
db.commit()
logger.info(f"选题 {self.topic_id} 状态已更新为 draft（数据库）")
else:
logger.warning(f"数据库中未找到选题 {self.topic_id}")
except Exception as e:
logger.error(f"更新数据库失败: {e}")
db.rollback()
finally:
db.close()
logger.info(f"选题 {self.topic_id} 状态更新为「待发布」（JSON）")
"""标记选题为「待发布」"""
with open(TOPICS_FILE, 'r', encoding='utf-8') as f:
topics = json.load(f)
for t in topics:
if t.get('id') == self.topic_id:
t['status'] = 'draft'
# ready_at 留空，待合规审核通过后设置
break
with open(TOPICS_FILE, 'w', encoding='utf-8') as f:
json.dump(topics, f, ensure_ascii=False, indent=2)
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"撰写完成，状态改为 draft，待合规审核")
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()