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

fix: content quality, image format, task monitor, calendar data source, search UI & sort

Browse Source
This commit is contained in:
Yuzhiran Dev
2026-05-26 11:26:10 +08:00
parent b2d043b231
commit ac8644d752
36 changed files with 2294 additions and 873 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/collector.py
+10 -4
View File
@@ -23,6 +23,7 @@ import subprocess
# scripts/collector.py 位于 <project_root>/scripts/,因此向上2级即可
PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
sys.path.insert(0, str(PROJECT_ROOT / "platform" / "backend"))
# 配置路径
CONFIG_DIR = PROJECT_ROOT / "config"
@@ -39,6 +40,7 @@ logging.basicConfig(
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
DEFAULT_CHINA_PAINS = {
"循环消费": "以旧换新流程繁琐、二手商品信任缺失、租赁市场不规范",
"低碳出行": "新能源车充电设施不足、城市规划不支持骑行、通勤距离长",
@@ -364,8 +366,11 @@ class SustainabilityCollector:
content = title
search_text = (title + content).lower()
keywords = source_keywords if source_keywords else _load_rss_keywords()
if any(keyword.lower() in search_text for keyword in keywords):
if source_keywords:
keyword_match = any(keyword.lower() in search_text for keyword in source_keywords)
else:
keyword_match = True
if keyword_match:
articles.append({
'title': title,
'url': entry.get('link', ''),
@@ -395,7 +400,7 @@ class SustainabilityCollector:
if not query:
logger.warning(f"web_search源 {source.name} 未配置查询词")
return []
results = search(query, max_results=8, use_cache=False)
results = search(query, max_results=8)
articles = []
for r in results:
articles.append({
@@ -943,7 +948,8 @@ def main():
sys.exit(1)
except Exception as e:
logger.error(f"收集任务失败: {e}")
try: logger.error(f"收集任务失败: {e}")
except: pass
print(f"ERROR: {e}")
sys.exit(1)
scripts/compliance_checker.py
+145 -19
View File
@@ -35,6 +35,28 @@ PLATFORM_RULES = {
}
}
# AI 套话检测模式（一旦出现在正文中，说明写作痕迹明显）
AI_TELTALES = [
"说回到",
"一个真实的.*案例很能说明问题",
"这就是.*被.*后的样子",
"如果你也",
"值得注意的是",
"首先其次最后",
"综上所述",
"总的来说",
"说到这里",
"我们来总结一下",
"总而言之",
"我们不难发现",
"我们可以看出",
"从以上分析可以看出",
"无可否认",
"众所周知",
"毋庸置疑",
"不知大家有没有发现",
]
_cached_sensitive_words = None
_cached_platform_rules = None
@@ -139,15 +161,22 @@ class ComplianceChecker:
# 6. 最小字数检查
self._check_min_length(text, platform)
# 7. 结构完整性检查（必须包含关键章节）
self._check_required_sections(text)
# 7. 内容质量检查
self._check_ai_telltales(text)
self._check_pronoun_consistency(text, platform)
self._check_reading_experience(text, platform)
self._check_platform_engagement(text, platform)
self._check_inline_images(text)
self._check_timeliness(text)
hard_types = ('敏感词', '法律法规', '平台规则', '品牌规范', '资源合规')
hard_issues = [i for i in self.issues if i['type'] in hard_types]
return {
"passed": len(self.issues) == 0,
"passed": len(hard_issues) == 0,
"issues": self.issues,
"score": max(0, 100 - len(self.issues) * 10)
"score": max(0, 100 - sum(
10 if i['type'] in hard_types else 5 for i in self.issues
))
}
def _check_sensitive_words(self, text: str):
@@ -317,23 +346,120 @@ class ComplianceChecker:
"suggestion": "扩写内容至最低要求"
})
def _check_required_sections(self, text: str):
"""检查是否包含必要章节（如引言、核心观点、总结等）"""
required_headings = [
"引言", "核心观点", "受众痛点", "总结", "行动指南"
]
missing = []
for heading in required_headings:
# 检查 h2 或 h3 中是否出现 heading
if not re.search(r'<h[23][^>]*>.*' + re.escape(heading) + r'.*</h[23]>', text, re.IGNORECASE):
missing.append(heading)
if missing:
def _check_ai_telltales(self, text: str):
"""检查AI套话——正文中出现这些模式说明AI写作痕迹明显"""
plain = re.sub(r'<[^>]+>', '', text)
for pattern in AI_TELTALES:
if re.search(pattern, plain):
self.issues.append({
"type": "内容质量",
"category": "AI套话",
"detail": f"正文出现AI套话模式: 「{pattern}」",
"suggestion": "删除或替换为自然表达，不要让读者感觉是AI写的"
})
def _check_pronoun_consistency(self, text: str, platform: str):
"""检查人称一致性（尤其是微信文章）"""
if platform != "wechat":
return
plain = re.sub(r'<[^>]+>', '', text)
has_ni = '你' in plain
has_nimen = '你们' in plain
has_women = '我们' in plain
if has_nimen and has_ni:
self.issues.append({
"type": "结构完整",
"category": "章节缺失",
"detail": f"缺少必要章节：{', '.join(missing)}",
"suggestion": "补充缺失章节"
"type": "内容质量",
"category": "人称混用",
"detail": "微信文章中同时使用「你」和「你们」，建议统一为「你」",
"suggestion": "将所有「你们」替换为「你」，保持与读者的单数对话感"
})
if has_women and has_ni:
self.issues.append({
"type": "内容质量",
"category": "人称混用",
"detail": "微信文章中同时使用「我们」和「你」，建议统一视角",
"suggestion": "将「我们」替换为「你」或「我」，保持与读者对话而非说教"
})
def _check_reading_experience(self, text: str, platform: str = ""):
"""检查阅读体验（段落长度、配图）"""
paragraphs = re.findall(r'<p>(.*?)</p>', text, re.DOTALL)
long_paras = [p for p in paragraphs if len(p) > 300]
if len(long_paras) > len(paragraphs) * 0.3:
self.issues.append({
"type": "内容质量",
"category": "段落过长",
"detail": f"超过30%的段落长度>300字（共{len(paragraphs)}段，{len(long_paras)}段过长），在手机上阅读体验差",
"suggestion": "将长段拆分为2-3个短段，每段不超过150-200字"
})
# 图片检测（各平台阈值不同）
imgs = re.findall(r'<img', text)
min_imgs = {"zhihu": 1, "wechat": 1, "xiaohongshu": 1}.get(platform, 1)
if not imgs:
self.issues.append({
"type": "内容质量",
"category": "缺少配图",
"detail": "正文中没有图片，建议插入配图提升阅读体验",
"suggestion": "在关键位置插入配图（数据截图、金句卡片、操作步骤图等）"
})
elif len(imgs) < min_imgs:
self.issues.append({
"type": "内容质量",
"category": "配图不足",
"detail": f"只有{len(imgs)}张图，建议至少{min_imgs}张配图提升阅读体验",
"suggestion": "在文章中间部分增加配图"
})
def _check_platform_engagement(self, text: str, platform: str):
"""检查平台特有的互动/传播要素"""
plain = re.sub(r'<[^>]+>', '', text)
if platform == "zhihu":
# 知乎需要讨论引导（评论区互动是核心算法权重）
if not any(kw in plain for kw in ('你觉得', '你怎么看', '欢迎在评论区', '说说你的', '欢迎讨论', '你怎么想')):
self.issues.append({
"type": "内容质量",
"category": "缺少互动引导",
"detail": "知乎文章建议在末尾加讨论引导（如「你觉得呢？欢迎在评论区聊聊」），提升互动率",
"suggestion": "在文章末尾添加一个问题或讨论话题，引导读者评论"
})
# 检查是否有数据引用（知乎读者看重可信度）
if not any(kw in plain for kw in ('据统计', '调研显示', '数据显示', '报告指出', '根据', '研究表明', '调查显示')):
self.issues.append({
"type": "内容质量",
"category": "缺少数据引用",
"detail": "知乎文章建议引用具体数据或报告来支撑观点，增强可信度",
"suggestion": "在关键观点处引用权威数据来源"
})
if platform == "xiaohongshu":
# 小红书需要收藏引导（收藏率是推荐算法核心指标）
if '收藏' not in plain:
self.issues.append({
"type": "内容质量",
"category": "缺少收藏引导",
"detail": "小红书笔记建议在末尾加收藏引导（如「觉得有用点个收藏」），提升收藏率",
"suggestion": "在末尾添加收藏引导语"
})
# 小红书段落需非常短
paragraphs = re.findall(r'<p>(.*?)</p>', text, re.DOTALL)
long_paras = [p for p in paragraphs if len(p) > 150]
if len(long_paras) > len(paragraphs) * 0.2:
self.issues.append({
"type": "内容质量",
"category": "段落过长",
"detail": f"小红书建议每段不超过80-100字，当前{len(long_paras)}/{len(paragraphs)}段超过150字",
"suggestion": "将长段拆分为1-2句的短段落，每段不超过100字"
})
# 小红书需要至少一些 emoji
if not re.search(r'[\U0001F300-\U0001F9FF\u2600-\u27BF]', plain):
self.issues.append({
"type": "内容质量",
"category": "缺少emoji",
"detail": "小红书笔记建议适当使用emoji来增加视觉吸引力",
"suggestion": "在标题、章节分隔或重点句前添加相关emoji"
})
def _check_inline_images(self, html: str):
"""检查图片是否以内联方式嵌入（data:image）"""
# 提取所有 img 标签的 src 属性值
scripts/compliance_optimizer.py
+3 -3
View File
@@ -171,7 +171,7 @@ def polish_with_llm(html: str, platform: str, remaining_issues: Optional[List[Di
if remaining_issues:
issues_desc = "\n".join(
f"- [{i['type']}] {i.get('category','')}: {i.get('detail','')} (建议: {i.get('suggestion','')})"
f"- [{i['type']}] {i.get('category','')}: {i.get('detail','')}"
for i in remaining_issues
)
prompt = get_prompt("compliance_fix", issues_desc=issues_desc, html=html)
@@ -183,7 +183,7 @@ def polish_with_llm(html: str, platform: str, remaining_issues: Optional[List[Di
polished = strip_thinking_html(polished)
if '<h2' in polished or '<p>' in polished:
if len(polished) > len(html) * 0.3 and len(polished) > 100:
if not any(kw in polished[:100] for kw in ['保留', '建议', '可以', '应该', '推荐']):
if not any(kw in polished for kw in ['保留', '建议', '可以', '应该', '推荐', '改为', '替换为']):
tag = "针对性修复" if remaining_issues else "常规润色"
return polished, f"LLM {tag}"
logger.warning(f"LLM 优化输出异常（过短或含建议性文字），保留原文 (len={len(polished)})")
@@ -316,7 +316,7 @@ def main(topic_ids: List[str] = None):
for tid, scores in passed_scores.items():
avg_score = sum(scores) // len(scores)
update_topic_status(tid, 'ready', compliance_score=avg_score)
update_topic_status(tid, 'ready', compliance_score=avg_score, reviewed_at=datetime.datetime.now())
logger.info(f"选题 {tid} 状态 → ready（待发布）, 合规分={avg_score}")
report_file = DRAFTS_DIR / TODAY / "optimization_report.json"
scripts/content_cleaner.py
+7 -1
View File
@@ -110,6 +110,7 @@ def strip_thinking_html(html: str) -> str:
r'<p[^>]*>最后.*?</p>',
r'<p[^>]*>(总的来说|值得注意的是|换句话说|总而言之|简而言之|一言以蔽之|可以说|不难发现|由此可见|综上所述).*?</p>',
r'<div[^>]*>(总的来说|值得注意的是|换句话说|总而言之|简而言之|一言以蔽之|可以说|不难发现|由此可见|综上所述).*?</div>',
r'<p[^>]*>(开头钩子|核心观点|受众痛点|独特视角|差异化切入|内容形式).*?</p>',
]
for pat in patterns:
html = re.sub(pat, '', html, flags=re.IGNORECASE)
@@ -176,11 +177,16 @@ def clean_markdown_content(text: str) -> str:
return '\n'.join(cleaned).strip()
def clean_html_content(html: str) -> str:
"""清洗 HTML 输出：去 markdown 代码围栏头尾 + 去 AI 思考注释"""
"""清洗 HTML 输出：去 markdown 代码围栏 + AI 思考 + 图片占位 + 结构标签"""
html = re.sub(r'^```+\w*\s*\n?', '', html)
html = html.strip()
html = re.sub(r'\n?```+\s*$', '', html)
html = strip_thinking_html(html)
# 移除 LLM 插入的建议配图占位（<p><!-- 建议配图：... --></p>）
html = re.sub(r'<p>\s*<!--\s*建议配图.*?-->\s*</p>', '', html, flags=re.DOTALL)
# 移除可能变成可见文本的建议配图文字
html = re.sub(r'<!--\s*建议配图.*?-->', '', html, flags=re.DOTALL)
html = re.sub(r'建议配图：.*?(?=<|$)', '', html)
return html
def clean_full_pipeline(text: str, output_format: str = 'markdown') -> str:
scripts/db_helper.py
+3 -1
View File
@@ -71,7 +71,7 @@ def get_next_topic(priority: Optional[str] = None, db: Optional[Session] = None)
if close_db:
db.close()
def update_topic_status(topic_id: str, status: str, compliance_score: Optional[int] = None, db: Optional[Session] = None) -> bool:
def update_topic_status(topic_id: str, status: str, compliance_score: Optional[int] = None, db: Optional[Session] = None, reviewed_at: Optional[datetime] = None) -> bool:
close_db = False
if db is None:
db = SessionLocal()
@@ -84,6 +84,8 @@ def update_topic_status(topic_id: str, status: str, compliance_score: Optional[i
topic.updated_at = datetime.now()
if compliance_score is not None:
topic.compliance_score = compliance_score
if reviewed_at is not None:
topic.reviewed_at = reviewed_at
if status in ['ready', 'published']:
topic.generated_at = datetime.now()
db.commit()
scripts/image_generator.py
+171 -534
View File
@@ -1,547 +1,184 @@
#!/usr/bin/env python3
"""
文章配图自动生成器
基于PIL，根据文章标题、内容自动生成适合各平台的配图
多平台文章配图生成器
为知乎/公众号/小红书生成平台风格的 SVG 配图（base64 内联，无需外部资源）
"""
import base64, math, textwrap, re
from typing import List, Optional
import os
import sys
import json
import datetime
import logging
from pathlib import Path
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass
PLATFORM_STYLES = {
"zhihu": {
"primary": "#0084ff",
"primary_rgb": "0, 132, 255",
"accent": "#e8f4fd",
"gradient_start": "#e8f4fd",
"gradient_end": "#f5f9ff",
"card_bg": "#ffffff",
"title_color": "#1a1a1a",
"dim_color": "#c0c4cc",
},
"wechat": {
"primary": "#07c160",
"primary_rgb": "7, 193, 96",
"accent": "#f0faf4",
"gradient_start": "#f0faf4",
"gradient_end": "#e8f5ee",
"card_bg": "#ffffff",
"title_color": "#1a1a1a",
"dim_color": "#c0c4cc",
},
"xiaohongshu": {
"primary": "#ff2442",
"primary_rgb": "255, 36, 66",
"accent": "#fff5f5",
"gradient_start": "#fff5f5",
"gradient_end": "#fff0f0",
"card_bg": "#ffffff",
"title_color": "#262626",
"dim_color": "#bfbfbf",
},
}
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
# 确保项目根目录在路径中
PROJECT_ROOT = Path('/root/openclaw-workspace/projects/yu-zhi-ran')
sys.path.insert(0, str(PROJECT_ROOT))
sys.path.insert(0, str(PROJECT_ROOT / 'platform' / 'backend'))
from db_helper import get_topic_by_id
from app.models import Article
from app.database import SessionLocal
logger = logging.getLogger(__name__)
# 加载配置
CONFIG_DIR = PROJECT_ROOT / "config"
with open(CONFIG_DIR / "wecom_config.yaml", 'r', encoding='utf-8') as f:
wecom_config = yaml.safe_load(f)
@dataclass
class ImageSpec:
"""图片规格"""
platform: str
width: int
height: int
format: str = "PNG"
quality: int = 85
bg_color: Tuple[int, int, int] = (255, 255, 255) # 白色背景
accent_color: Tuple[int, int, int] = (76, 175, 80) # 品牌绿色 #4CAF50
text_color: Tuple[int, int, int] = (51, 51, 51) # 深灰色
class ImageGenerator:
"""图片生成器"""
def __init__(self, output_base: Path = None):
self.output_base = output_base or (PROJECT_ROOT / "automation" / "images" / "generated")
self.today = datetime.datetime.now().strftime("%Y-%m-%d")
self.output_dir = self.output_base / self.today
self.output_dir.mkdir(parents=True, exist_ok=True)
# 加载平台规格
self.platform_specs = {}
for platform, specs in wecom_config["image_specs"].items():
self.platform_specs[platform] = ImageSpec(
platform=platform,
width=specs["width"],
height=specs["height"],
format=specs["format"],
quality=specs["quality"]
)
# 字体路径
self.font_paths = self._find_chinese_fonts()
def _find_chinese_fonts(self) -> List[str]:
"""查找系统中可用的中文字体"""
font_paths = [
"/usr/share/fonts/truetype/wqy/wqy-microhei.ttc", # 文泉驿微米黑
"/usr/share/fonts/truetype/arphic/uming.ttc", # 文鼎PL中等
"/usr/share/fonts/truetype/liberation/LiberationSans-Regular.ttf",
"/System/Library/Fonts/PingFang.ttc", # macOS
"/System/Library/Fonts/STHeiti Medium.ttc", # macOS
"C:\\Windows\\Fonts\\msyh.ttc", # Windows
"C:\\Windows\\Fonts\\simsun.ttc"
]
available = [p for p in font_paths if os.path.exists(p)]
return available if available else [None] # 回退到默认字体
def _get_font(self, size: int, bold: bool = False) -> ImageFont.FreeTypeFont:
"""获取合适的中文字体"""
# 优先使用系统中文字体
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:
continue
return ImageFont.load_default()
def generate_cover_image(self, title: str, subtitle: str = "", platform: str = "zhihu") -> Path:
"""生成封面图"""
spec = self.platform_specs.get(platform, self.platform_specs["zhihu"])
# 创建图片
img = Image.new('RGB', (spec.width, spec.height), color=spec.bg_color)
draw = ImageDraw.Draw(img)
# 添加渐变背景
for y in range(spec.height):
# 从顶部到中间的渐变
ratio = y / (spec.height * 0.6)
r = int(255 * (1 - ratio) + 230 * ratio)
g = int(255 * (1 - ratio) + 240 * ratio)
b = int(255 * (1 - ratio) + 250 * ratio)
draw.line([(0, y), (spec.width, y)], fill=(r, g, b))
# 绘制品牌标识区域（底部条纹）
stripe_height = 20
stripe_y = spec.height - stripe_height - 30
draw.rectangle([0, stripe_y, spec.width, stripe_y + stripe_height], fill=spec.accent_color)
draw.text((20, stripe_y + 5), "宇之然", fill=(255, 255, 255), font=self._get_font(14))
# 标题排版
title_font = self._get_font(int(spec.height * 0.12), bold=True)
subtitle_font = self._get_font(int(spec.height * 0.06))
# 自动换行处理
max_width = spec.width * 0.9
title_lines = self._wrap_text(title, title_font, max_width)
subtitle_lines = self._wrap_text(subtitle, subtitle_font, max_width) if subtitle else []
# 计算总高度
line_spacing = 1.2
title_height = len(title_lines) * title_font.size * line_spacing
subtitle_height = len(subtitle_lines) * subtitle_font.size * line_spacing
total_text_height = title_height + subtitle_height + 20 # 间距
# 居中绘制
start_y = (spec.height - total_text_height) // 2
# 绘制标题
for i, line in enumerate(title_lines):
y = start_y + i * (title_font.size * line_spacing)
self._draw_centered_text(draw, line, y, spec.width, title_font, spec.text_color)
# 绘制副标题
if subtitle_lines:
subtitle_start_y = start_y + title_height + 10
for i, line in enumerate(subtitle_lines):
y = subtitle_start_y + i * (subtitle_font.size * line_spacing)
self._draw_centered_text(draw, line, y, spec.width, subtitle_font, (102, 102, 102))
# 保存图片
filename = f"cover_{platform}.{spec.format.lower()}"
output_path = self.output_dir / filename
img.save(output_path, quality=spec.quality)
return output_path
def generate_chart_image(self, chart_type: str, data: Dict, title: str, platform: str = "zhihu") -> Path:
"""生成数据图表"""
spec = self.platform_specs.get(platform, self.platform_specs["zhihu"])
img = Image.new('RGB', (spec.width, spec.height), color=(255, 255, 255))
draw = ImageDraw.Draw(img)
# 绘制标题
title_font = self._get_font(36, bold=True)
draw.text((50, 30), title, fill=spec.text_color, font=title_font)
# 根据图表类型绘制
if chart_type == "bar":
self._draw_bar_chart(draw, data, spec)
elif chart_type == "pie":
self._draw_pie_chart(draw, data, spec)
elif chart_type == "line":
self._draw_line_chart(draw, data, spec)
else:
# 默认显示文本
text_font = self._get_font(24)
draw.text((50, 150), f"图表类型: {chart_type}", fill=spec.text_color, font=text_font)
draw.text((50, 200), f"数据: {json.dumps(data, ensure_ascii=False)}", fill=spec.text_color, font=text_font)
# 水印
watermark_font = self._get_font(14)
draw.text((spec.width - 150, spec.height - 30), "数据来源: 宇之然", fill=(150, 150, 150), font=watermark_font)
filename = f"data_chart_{platform}.png"
output_path = self.output_dir / filename
img.save(output_path, quality=spec.quality)
return output_path
def _draw_bar_chart(self, draw: ImageDraw.Draw, data: Dict, spec: ImageSpec):
"""绘制柱状图"""
# 数据格式: {"label1": value1, "label2": value2, ...}
labels = list(data.keys())
values = list(data.values())
max_value = max(values) if values else 1
chart_area = {
"left": 100,
"top": 120,
"right": spec.width - 50,
"bottom": spec.height - 100
}
chart_width = chart_area["right"] - chart_area["left"]
chart_height = chart_area["bottom"] - chart_area["top"]
bar_width = chart_width // (len(values) * 2)
gap = bar_width
# 绘制坐标轴
draw.line([
(chart_area["left"], chart_area["top"]),
(chart_area["left"], chart_area["bottom"])
], fill=(0, 0, 0), width=2)
draw.line([
(chart_area["left"], chart_area["bottom"]),
(chart_area["right"], chart_area["bottom"])
], fill=(0, 0, 0), width=2)
# 绘制柱子
for i, (label, value) in enumerate(zip(labels, values)):
x = chart_area["left"] + i * (bar_width + gap) + gap // 2
bar_height = (value / max_value) * chart_height
y_bottom = chart_area["bottom"]
y_top = chart_area["bottom"] - bar_height
# 柱子（渐变色）
for y in range(int(y_top), int(y_bottom)):
ratio = (y - y_top) / bar_height if bar_height > 0 else 0
r = int(76 + (100-76) * ratio)
g = int(175 + (150-175) * ratio)
b = int(80 + (120-80) * ratio)
draw.line([(x, y), (x + bar_width, y)], fill=(r, g, b))
# 标签
label_font = self._get_font(18)
self._draw_centered_text(draw, label, y_bottom + 10, x + bar_width // 2, label_font, (80, 80, 80))
# 数值
value_font = self._get_font(20, bold=True)
self._draw_centered_text(draw, f"{value}", y_top - 10, x + bar_width // 2, value_font, spec.accent_color)
def _draw_pie_chart(self, draw: ImageDraw.Draw, data: Dict, spec: ImageSpec):
"""绘制饼图"""
# 简单实现：绘制圆形扇形
center_x, center_y = spec.width // 2, spec.height // 2
radius = min(spec.width, spec.height) // 3
total = sum(data.values()) if data else 1
angle_start = 0
# 颜色调色板
colors = [
(76, 175, 80), (33, 150, 83), (139, 195, 74),
(255, 193, 7), (255, 152, 0), (244, 67, 54)
]
for i, (label, value) in enumerate(data.items()):
angle_extent = (value / total) * 360
color = colors[i % len(colors)]
# 绘制扇形
draw.arc(
[center_x - radius, center_y - radius, center_x + radius, center_y + radius],
angle_start, angle_start + angle_extent,
fill=color, width=radius * 2
)
angle_start += angle_extent
# 画中心白圆形成饼图效果
inner_radius = radius * 0.5
draw.ellipse(
[center_x - inner_radius, center_y - inner_radius, center_x + inner_radius, center_y + inner_radius],
fill=(255, 255, 255)
)
# 绘制图例
legend_y = spec.height - 80
legend_x = 100
for i, (label, value) in enumerate(data.items()):
color = colors[i % len(colors)]
# 色块
draw.rectangle([legend_x, legend_y + i*25, legend_x+20, legend_y+20+i*25], fill=color)
# 标签
label_font = self._get_font(16)
draw.text((legend_x+30, legend_y+i*25), f"{label}: {value}", fill=(60, 60, 60), font=label_font)
def _draw_line_chart(self, draw: ImageDraw.Draw, data: Dict, spec: ImageSpec):
"""绘制折线图"""
# 简化版：显示文本描述
title_font = self._get_font(24)
draw.text((50, 100), "折线图 (数据趋势)", fill=spec.text_color, font=title_font)
items = list(data.items())
if not items:
draw.text((50, 150), "无可用数据", fill=(100, 100, 100), font=self._get_font(18))
return
# 列出数据
data_font = self._get_font(16)
y = 200
for label, value in items[:10]: # 限制显示数量
draw.text((50, y), f"{label}: {value}", fill=(80, 80, 80), font=data_font)
y += 25
def generate_concept_image(self, title: str, items: List[str], platform: str = "zhihu") -> Path:
"""生成概念示意图（用于行动清单等）"""
spec = self.platform_specs.get(platform, self.platform_specs["zhihu"])
img = Image.new('RGB', (spec.width, spec.height), color=(245, 245, 245))
draw = ImageDraw.Draw(img)
# 标题
title_font = self._get_font(42, bold=True)
self._draw_centered_text(draw, title, 60, spec.width, title_font, spec.text_color)
# 绘制项目列表（带复选框样式）
item_font = self._get_font(28)
start_y = 150
for i, item in enumerate(items[:8]): # 限制8个
y = start_y + i * 50
# 复选框
box_size = 30
box_x = (spec.width - 400) // 2
draw.rectangle([box_x, y, box_x + box_size, y + box_size], outline=spec.accent_color, width=3)
# 勾
check_font = self._get_font(24)
draw.text((box_x + 7, y + 2), "✓", fill=spec.accent_color, font=check_font)
# 文字
draw.text((box_x + box_size + 20, y + 5), item[:30], fill=(60, 60, 60), font=item_font)
filename = f"action_checklist_{platform}.png"
output_path = self.output_dir / filename
img.save(output_path, quality=spec.quality)
return output_path
def generate_equipment_list_image(self, items: List[Dict[str, str]], platform: str = "zhihu") -> Path:
"""生成装备清单图"""
spec = self.platform_specs.get(platform, self.platform_specs["zhihu"])
img = Image.new('RGB', (spec.width, spec.height), color=(255, 255, 255))
draw = ImageDraw.Draw(img)
# 标题
title = "装备清单"
title_font = self._get_font(38, bold=True)
draw.text((50, 40), title, fill=spec.text_color, font=title_font)
# 列头
headers = ["名称", "用途", "预算"]
header_font = self._get_font(24, bold=True)
col_width = spec.width // len(headers)
for i, header in enumerate(headers):
x = i * col_width + 20
draw.text((x, 100), header, fill=(100, 100, 100), font=header_font)
# 分隔线
draw.line([(50, 130), (spec.width-50, 130)], fill=(200, 200, 200), width=2)
# 绘制条目
item_font = self._get_font(20)
row_height = 40
y = 150
for item in items[:10]: # 最多10行
name = item.get("name", "")[:12]
purpose = item.get("purpose", "")[:10]
budget = item.get("budget", "")
draw.text((70, y), name, fill=(50, 50, 50), font=item_font)
draw.text((col_width + 70, y), purpose, fill=(50, 50, 50), font=item_font)
draw.text((2*col_width + 70, y), budget, fill=(50, 50, 50), font=item_font)
y += row_height
# 底部总预算
total_budget = sum([int(item.get("budget", "0").replace("元", "")) for item in items if item.get("budget", "").replace("元", "").isdigit()])
total_font = self._get_font(22, bold=True)
draw.text((50, spec.height - 50), f"总预算: {total_budget}元", fill=spec.accent_color, font=total_font)
filename = f"equipment_{platform}.png"
output_path = self.output_dir / filename
img.save(output_path, quality=spec.quality)
return output_path
def _wrap_text(self, text: str, font: ImageFont.FreeTypeFont, max_width: int) -> List[str]:
"""文本自动换行"""
words = list(text)
lines = []
current_line = ""
for char in words:
test_line = current_line + char
bbox = font.getbbox(test_line)
width = bbox[2] - bbox[0]
if width <= max_width:
current_line = test_line
else:
if current_line:
lines.append(current_line)
current_line = char
if current_line:
lines.append(current_line)
return lines if lines else [text]
def _draw_centered_text(self, draw: ImageDraw.Draw, text: str, y: int, center_x: int, font: ImageFont.FreeTypeFont, color: Tuple[int, int, int]):
"""绘制居中文本"""
bbox = font.getbbox(text)
text_width = bbox[2] - bbox[0]
x = center_x - text_width // 2
draw.text((x, y), text, fill=color, font=font)
def generate_all_placeholders(self, article_title: str, platform: str = "zhihu") -> Dict[str, Path]:
"""生成所有占位图片"""
files = {}
# 1. 封面图
files["cover"] = self.generate_cover_image(article_title, "宇之然 · 可持续生活指南", platform)
# 2. 数据图表示例
files["data_chart"] = self.generate_chart_image("bar", {"选项A": 45, "选项B": 32, "选项C": 23}, "数据对比", platform)
# 3. 概念图（行动清单）
files["action_checklist"] = self.generate_concept_image("立即行动清单", [
"第一步：记录现状，识别改进空间",
"第二步：尝试最小可行改变",
"第三步：评估效果，决定是否继续",
"第四步：建立习惯，持续改进"
], platform)
# 4. 装备清单图
files["equipment"] = self.generate_equipment_list_image([
{"name": "智能插座", "purpose": "定时控制", "budget": "50元"},
{"name": "土壤传感器", "purpose": "湿度监测", "budget": "80元"},
{"name": "自动灌溉", "purpose": "浇水", "budget": "120元"},
{"name": "LED补光灯", "purpose": "光照", "budget": "200元"}
], platform)
return files
def generate_for_topic(topic_id: str, platforms: List[str] = None) -> Dict[str, Dict[str, str]]:
"""为指定选题生成三平台配图，路径存入 articles 表"""
if platforms is None:
platforms = ["zhihu", "wechat", "xiaohongshu"]
topic = get_topic_by_id(topic_id)
if not topic:
raise ValueError(f"Topic {topic_id} not found")
title = topic.get("title", "无标题")
generator = ImageGenerator()
results = {}
for platform in platforms:
try:
files = generator.generate_all_placeholders(title, platform)
cover_path = str(files.get("cover", ""))
chart_path = str(files.get("data_chart", ""))
checklist_path = str(files.get("action_checklist", ""))
images = {
"cover": cover_path,
"chart": chart_path,
"checklist": checklist_path,
}
# 存入 DB
save_article_images(topic_id, platform, images)
results[platform] = images
logger.info(f" [{platform}] cover={Path(cover_path).name}" if cover_path else "")
except Exception as e:
logger.error(f" [{platform}] 生成失败: {e}")
results[platform] = {}
return results
FONT = "-apple-system, BlinkMacSystemFont, 'PingFang SC', 'Microsoft YaHei', 'Helvetica Neue', sans-serif"
def save_article_images(topic_id: str, platform: str, images: Dict[str, str]):
"""将图片路径写入 articles 表的 images 字段"""
db = SessionLocal()
def _wrap_chinese(text: str, chars_per_line: int = 14) -> List[str]:
"""将中文文本按字数折行，尽量在标点处断开"""
if not text:
return [""]
lines = []
remainder = text
while len(remainder) > chars_per_line:
chunk = remainder[:chars_per_line]
# 尝试在最后一个标点处断开
cut = max(chunk.rfind(c) + 1 for c in ("，", "、", "。", "！", "？", "：", "；", "）", " ", "—") if c in chunk[:-1])
if cut <= 0:
cut = chars_per_line
lines.append(remainder[:cut].strip())
remainder = remainder[cut:].strip()
if remainder:
lines.append(remainder)
return lines
def _to_base64(svg: str) -> str:
try:
from app.models import Article
article_id = f"{platform}_{topic_id}"
article = db.query(Article).filter(Article.id == article_id).first()
if article:
existing = article.images or {}
existing.update(images)
article.images = existing
else:
article = Article(
id=article_id,
topic_id=topic_id,
platform=platform,
file_path=f"db:{article_id}",
status="draft",
images=images,
)
db.add(article)
db.commit()
import cairosvg
png = cairosvg.svg2png(bytestring=svg.encode('utf-8'))
return 'data:image/png;base64,' + base64.b64encode(png).decode('ascii')
except Exception:
db.rollback()
raise
finally:
db.close()
return 'data:image/svg+xml;base64,' + base64.b64encode(svg.encode('utf-8')).decode('ascii')
def main():
import argparse
parser = argparse.ArgumentParser(description='文章配图生成器')
parser.add_argument('--topic-id', help='选题ID，指定则为选题生成配图')
args = parser.parse_args()
def _alt_attr(text: str) -> str:
return text.replace('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;').replace('"', '&quot;').replace("'", '&#39;')
if args.topic_id:
print(f"为选题 {args.topic_id} 生成配图...")
results = generate_for_topic(args.topic_id)
print(json.dumps({"topic_id": args.topic_id, "images": results}, ensure_ascii=False))
sys.exit(0)
"""测试主函数"""
generator = ImageGenerator()
# 测试生成图片
print(f"开始生成图片到: {generator.output_dir}")
# 生成所有类型的占位图
files = generator.generate_all_placeholders("上海阳台种菜一年：我收获的不仅是蔬菜", "zhihu")
print("\n生成的文件:")
for name, path in files.items():
print(f" - {name}: {path.name} ({path.stat().st_size // 1024}KB)")
print(f"\n✅ 图片生成完成，共 {len(files)} 张")
if __name__ == "__main__":
main()
def _img_tag(src: str, alt: str, width: int = 1080) -> str:
return f'<p><img src="{src}" alt="{_alt_attr(alt)}" style="width:100%;max-width:{width}px;border-radius:8px;"></p>\n'
def generate_lead(platform: str, title: str, field: str = "", brand: str = "宇之然") -> str:
"""生成文章头图（1080×600）"""
s = PLATFORM_STYLES.get(platform, PLATFORM_STYLES["zhihu"])
lines = _wrap_chinese(title, 16)
title_lines = ""
y_start = 160
for i, line in enumerate(lines[:3]):
title_lines += f'<text x="80" y="{y_start + i*60}" font-size="44" font-weight="bold" fill="{s["title_color"]}">{_alt_attr(line)}</text>\n'
field_badge = ""
if field:
field_badge = f'''
<rect x="80" y="{y_start + len(lines[:3]) * 60 + 20}" width="{len(field)*14 + 32}" height="34" rx="17" fill="{s["primary"]}" opacity="0.12"/>
<text x="96" y="{y_start + len(lines[:3]) * 60 + 43}" font-size="14" fill="{s["primary"]}" font-weight="500">{_alt_attr(field)}</text>'''
svg = f'''<svg xmlns="http://www.w3.org/2000/svg" width="1080" height="600" viewBox="0 0 1080 600" style="width:100%;max-width:1080px;border-radius:8px;">
<defs>
<linearGradient id="lead_bg" x1="0%" y1="0%" x2="100%" y2="100%"><stop offset="0%" stop-color="{s["gradient_start"]}"/><stop offset="100%" stop-color="{s["gradient_end"]}"/></linearGradient>
</defs>
<rect width="1080" height="600" fill="url(#lead_bg)"/>
<circle cx="120" cy="120" r="220" fill="{s["primary"]}" opacity="0.04"/>
<circle cx="960" cy="480" r="180" fill="{s["primary"]}" opacity="0.06"/>
<circle cx="540" cy="300" r="320" fill="{s["primary"]}" opacity="0.02"/>
<rect x="80" y="80" width="80" height="4" rx="2" fill="{s["primary"]}"/>
<g font-family="{FONT}">
{title_lines}
{field_badge}
<text x="80" y="540" font-size="14" fill="{s["dim_color"]}">{_alt_attr(brand)}</text>
</g>
</svg>'''
return _img_tag(_to_base64(svg), title)
def generate_section_card(platform: str, section_title: str, section_num: int) -> str:
"""生成章节分隔图（800×160）"""
s = PLATFORM_STYLES.get(platform, PLATFORM_STYLES["zhihu"])
num_text = f"{section_num:02d}"
svg = f'''<svg xmlns="http://www.w3.org/2000/svg" width="800" height="160" viewBox="0 0 800 160" style="width:100%;max-width:800px;border-radius:8px;">
<rect width="800" height="160" fill="{s["card_bg"]}" rx="10"/>
<rect x="0" y="0" width="5" height="160" fill="{s["primary"]}" rx="2.5"/>
<text x="36" y="72" font-size="56" font-weight="bold" fill="{s["primary"]}" opacity="0.12" font-family="{FONT}">{num_text}</text>
<text x="36" y="120" font-size="20" font-weight="bold" fill="{s["title_color"]}" font-family="{FONT}">{_alt_attr(section_title)}</text>
</svg>'''
return _img_tag(_to_base64(svg), section_title, 800)
def generate_quote_card(platform: str, quote: str, source: str = "") -> str:
"""生成金句卡片（800×280）"""
s = PLATFORM_STYLES.get(platform, PLATFORM_STYLES["zhihu"])
lines = _wrap_chinese(quote, 20)
quote_lines = ""
y_start = 110
for i, line in enumerate(lines[:4]):
quote_lines += f'<text x="80" y="{y_start + i*36}" font-size="18" fill="{s["title_color"]}" font-weight="500">{_alt_attr(line)}</text>\n'
source_line = ""
if source:
source_line = f'<text x="80" y="{y_start + min(len(lines), 4)*36 + 6}" font-size="13" fill="{s["dim_color"]}">{_alt_attr(f"— {source}")}</text>'
svg = f'''<svg xmlns="http://www.w3.org/2000/svg" width="800" height="280" viewBox="0 0 800 280" style="width:100%;max-width:800px;border-radius:8px;">
<rect width="800" height="280" fill="{s["accent"]}" rx="10"/>
<text x="40" y="80" font-size="56" fill="{s["primary"]}" opacity="0.2" font-family="Georgia, serif">"</text>
<g font-family="{FONT}">
{quote_lines}
{source_line}
</g>
</svg>'''
return _img_tag(_to_base64(svg), f"金句：{quote[:30]}", 800)
def generate_data_highlight(platform: str, number: str, label: str) -> str:
"""生成数据高亮卡片（800×220）"""
s = PLATFORM_STYLES.get(platform, PLATFORM_STYLES["zhihu"])
svg = f'''<svg xmlns="http://www.w3.org/2000/svg" width="800" height="220" viewBox="0 0 800 220" style="width:100%;max-width:800px;border-radius:8px;">
<rect width="800" height="220" fill="{s["primary"]}" rx="10"/>
<g font-family="{FONT}">
<text x="80" y="120" font-size="60" font-weight="bold" fill="#ffffff">{_alt_attr(number)}</text>
<text x="80" y="170" font-size="16" fill="#ffffff" opacity="0.85">{_alt_attr(label)}</text>
</g>
</svg>'''
return _img_tag(_to_base64(svg), label, 800)
def insert_lead_image(html: str, platform: str, title: str, field: str) -> str:
"""在 HTML 正文开头插入头图（仅必要环节）"""
lead = generate_lead(platform, title, field)
h1_end = html.find('</h1>')
if h1_end != -1:
html = html[:h1_end + 5] + '\n' + lead + html[h1_end + 5:]
else:
html = lead + html
return html
# 平台特定的配图密度
PLATFORM_IMAGE_COUNTS = {
"zhihu": {"lead": True, "sections": True, "quotes": 0, "data": 0, "density": "medium"},
"wechat": {"lead": True, "sections": True, "quotes": 0, "data": 0, "density": "medium"},
"xiaohongshu": {"lead": True, "sections": True, "quotes": 0, "data": 0, "density": "high"},
}
scripts/mcp_search_server.py
+300
View File
@@ -0,0 +1,300 @@
#!/usr/bin/env python3
"""
MCP Search Server — provides web search via opencode infrastructure.
Two search methods (automatic fallback):
1. npx opencode run (rate-limited but returns real web results)
2. opencode-go API + model training data (no rate limit, less fresh)
Usage:
python3 scripts/mcp_search_server.py # MCP server (stdio)
python3 scripts/mcp_search_server.py --query Q # one-shot search
python3 scripts/mcp_search_server.py --url U # one-shot webfetch
"""
import json, os, subprocess, sys, time
from pathlib import Path
from typing import Any, Dict, List, Optional
PROJECT_ROOT = Path(__file__).resolve().parent.parent
CACHE_FILE = PROJECT_ROOT / "automation" / "data" / "mcp_search_cache.json"
SESSION_FILE = PROJECT_ROOT / "automation" / "data" / "mcp_session.txt"
CACHE_TTL = 3600
SESSION_TITLE = "opencode搜索"
API_BASE = "https://opencode.ai/zen/go/v1"
API_KEY = os.environ.get("OPENCODE_API_KEY", "")
if not API_KEY:
try:
from dotenv import load_dotenv
env_path = PROJECT_ROOT / "platform" / "backend" / ".env"
load_dotenv(env_path)
API_KEY = os.environ.get("OPENCODE_API_KEY", "")
except Exception:
pass
# ── session (reuse same session for all MCP searches) ─────────────
def _load_session() -> Optional[str]:
if SESSION_FILE.exists():
try:
return SESSION_FILE.read_text().strip() or None
except Exception:
pass
return None
def _save_session_from_output(stdout: str):
for line in stdout.strip().split("\n"):
try:
ev = json.loads(line)
sid = ev.get("sessionID") or ev.get("part", {}).get("sessionID")
if sid:
SESSION_FILE.parent.mkdir(parents=True, exist_ok=True)
SESSION_FILE.write_text(sid)
return
except Exception:
pass
# ── cache ─────────────────────────────────────────────────────────
def _check_cache(query: str) -> Optional[List[Dict]]:
if not CACHE_FILE.exists():
return None
try:
data = json.loads(CACHE_FILE.read_text())
entry = data.get(query)
if entry and time.time() - entry.get("ts", 0) < CACHE_TTL:
return entry.get("results")
except Exception:
pass
return None
def _write_cache(query: str, results: List[Dict]):
CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
data = {}
if CACHE_FILE.exists():
try:
data = json.loads(CACHE_FILE.read_text())
except Exception:
pass
data[query] = {"ts": time.time(), "results": results}
keys = sorted(data.keys(), key=lambda k: data[k].get("ts", 0), reverse=True)[:200]
CACHE_FILE.write_text(json.dumps({k: data[k] for k in keys}, ensure_ascii=False))
# ── method 1: npx opencode run ────────────────────────────────────
def _search_via_opencode_cli(query: str, max_results: int) -> Optional[List[Dict]]:
"""Use npx opencode run to execute websearch tool (short timeout)."""
sid = _load_session()
args = ["npx", "opencode", "run", f"websearch {query}", "--format", "json", "--title", SESSION_TITLE]
if sid:
args.extend(["--session", sid, "--continue"])
try:
r = subprocess.run(
args, capture_output=True, text=True, timeout=15,
env={**os.environ, "OPENCODE_DISABLE_AUTOUPDATE": "1"}
)
except subprocess.TimeoutExpired:
return None
except Exception:
return None
if r.returncode != 0:
return None
# Save session ID for reuse
_save_session_from_output(r.stdout)
for line in r.stdout.strip().split("\n"):
try:
ev = json.loads(line)
if ev.get("type") == "tool_use":
part = ev.get("part", {})
state = part.get("state", {})
if part.get("tool") == "websearch" and state.get("status") == "completed":
data = json.loads(state["output"])
results = []
for item in (data.get("results") or [])[:max_results]:
url = (item.get("url") or "").strip()
title = (item.get("title") or "").strip()
excerpts = item.get("excerpts") or []
content = (excerpts[0] if excerpts else "")[:500]
if url and title:
results.append({"title": title, "url": url, "content": content, "source": "opencode_cli"})
return results
except Exception:
pass
return None
# ── method 2: opencode-go API + training data ─────────────────────
def _search_via_api(query: str, max_results: int) -> Optional[List[Dict]]:
"""Use opencode-go API to answer query from training data (no rate limit)."""
if not API_KEY:
return None
import requests
try:
resp = requests.post(
f"{API_BASE}/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
json={
"model": "deepseek-v4-flash",
"messages": [{"role": "user", "content": (
f"你现在是一个网络搜索工具。用户查询: {query[:100]}\n\n"
f"请根据你的训练数据，提供{max_results}条最相关的网页结果，包含标题、URL和摘要。"
f"以JSON格式输出: [{{\"title\":\"...\",\"url\":\"...\",\"content\":\"...\"}}]"
f"仅输出JSON数组，不要其他文字。如果URL不确定，用合理占位。"
)}],
"temperature": 0.3,
"max_tokens": 2000,
},
timeout=30
)
data = resp.json()
content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
# Extract JSON array
import re as _re
m = _re.search(r'\[.*?\]', content, _re.DOTALL)
if m:
items = json.loads(m.group())
if isinstance(items, list):
for item in items:
item["source"] = "opencode_api"
return items[:max_results]
except Exception:
pass
return None
# ── search ─────────────────────────────────────────────────────────
def web_search(query: str, max_results: int = 8) -> List[Dict]:
max_results = min(max_results, 10)
cached = _check_cache(query)
if cached:
return cached[:max_results]
results = _search_via_opencode_cli(query, max_results)
if results:
_write_cache(query, results)
return results
results = _search_via_api(query, max_results)
if results:
_write_cache(query, results)
return results
return []
def webfetch(url: str) -> Optional[str]:
sid = _load_session()
args = ["npx", "opencode", "run", f"webfetch {url}", "--format", "json", "--title", SESSION_TITLE]
if sid:
args.extend(["--session", sid, "--continue"])
try:
r = subprocess.run(
args, capture_output=True, text=True, timeout=60,
env={**os.environ, "OPENCODE_DISABLE_AUTOUPDATE": "1"}
)
if r.returncode == 0:
_save_session_from_output(r.stdout)
for line in r.stdout.strip().split("\n"):
try:
ev = json.loads(line)
if ev.get("type") == "tool_use":
p = ev.get("part", {})
s = p.get("state", {})
if p.get("tool") == "webfetch" and s.get("status") == "completed":
return s.get("output", "")[:10000]
except Exception:
pass
except Exception:
pass
return None
# ── MCP protocol (JSON-RPC 2.0 over stdio) ────────────────────────
def _read_msg() -> Optional[Dict]:
line = sys.stdin.readline()
if not line:
return None
try:
return json.loads(line)
except json.JSONDecodeError:
return None
def _send_msg(msg: Dict):
sys.stdout.write(json.dumps(msg, ensure_ascii=False) + "\n")
sys.stdout.flush()
def _send_error(req_id: Any, code: int, message: str):
_send_msg({"jsonrpc": "2.0", "id": req_id, "error": {"code": code, "message": message}})
def _send_result(req_id: Any, result: Any):
_send_msg({"jsonrpc": "2.0", "id": req_id, "result": result})
def serve():
sys.stdin.reconfigure(encoding="utf-8")
sys.stdout.reconfigure(encoding="utf-8")
while True:
msg = _read_msg()
if msg is None:
break
req_id = msg.get("id")
method = msg.get("method", "")
params = msg.get("params", {})
if method == "initialize":
_send_result(req_id, {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {"listChanged": False}},
"serverInfo": {"name": "opencode-search-mcp", "version": "1.0.0"}
})
elif method == "notifications/initialized":
pass
elif method == "tools/list":
_send_result(req_id, {"tools": [
{"name": "web_search", "description": "Search the web. Returns up to 10 results with title, url, content.", "inputSchema": {
"type": "object", "properties": {
"query": {"type": "string", "description": "Search query"},
"max_results": {"type": "number", "description": "Max results (1-10)", "default": 8}
}, "required": ["query"]
}},
{"name": "webfetch", "description": "Fetch and extract content from a URL.", "inputSchema": {
"type": "object", "properties": {"url": {"type": "string", "description": "URL to fetch"}},
"required": ["url"]
}}
]})
elif method == "tools/call":
name = params.get("name", "")
args = params.get("arguments", {})
try:
if name == "web_search":
results = web_search(args.get("query", ""), int(args.get("max_results", 8)))
_send_result(req_id, {"content": [{"type": "text", "text": json.dumps(results, ensure_ascii=False)}]})
elif name == "webfetch":
content = webfetch(args.get("url", ""))
_send_result(req_id, {"content": [{"type": "text", "text": content or "Failed to fetch URL"}]})
else:
_send_error(req_id, -32601, f"Unknown tool: {name}")
except Exception as e:
_send_error(req_id, -32603, str(e))
elif method == "shutdown":
_send_result(req_id, {})
break
else:
_send_error(req_id, -32601, f"Unknown method: {method}")
def main():
if "--query" in sys.argv:
idx = sys.argv.index("--query")
q = sys.argv[idx + 1] if idx + 1 < len(sys.argv) else ""
print(json.dumps(web_search(q), ensure_ascii=False, indent=2))
return
if "--url" in sys.argv:
idx = sys.argv.index("--url")
u = sys.argv[idx + 1] if idx + 1 < len(sys.argv) else ""
print(webfetch(u) or "Failed")
return
serve()
if __name__ == "__main__":
main()
scripts/opencode_search.py
+60 -21
View File
@@ -23,13 +23,57 @@ logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(level
logger = logging.getLogger(__name__)
SEARCH_CACHE_FILE = PROJECT_ROOT / "automation" / "data" / "search_cache.json"
SESSION_FILE = PROJECT_ROOT / "automation" / "data" / "opencode_session.txt"
def _get_or_create_session() -> Optional[str]:
"""获取或创建持久 session ID"""
if SESSION_FILE.exists():
try:
sid = SESSION_FILE.read_text().strip()
if sid:
result = subprocess.run(
["npx", "opencode", "run", "ping", "--session", sid, "--format", "json"],
capture_output=True, text=True, timeout=10,
cwd=str(PROJECT_ROOT),
env={**os.environ, "OPENCODE_DISABLE_AUTOUPDATE": "1"}
)
if result.returncode == 0:
return sid
except Exception:
pass
result = subprocess.run(
["npx", "opencode", "run", "init", "--format", "json"],
capture_output=True, text=True, timeout=30,
cwd=str(PROJECT_ROOT),
env={**os.environ, "OPENCODE_DISABLE_AUTOUPDATE": "1"}
)
for line in result.stdout.strip().split("\n"):
try:
event = json.loads(line)
sid = event.get("sessionID") or event.get("part", {}).get("sessionID")
if sid:
SESSION_FILE.write_text(sid)
return sid
except Exception:
pass
return None
_session_id = None
def _run_opencode(prompt: str, timeout: int = 60) -> Optional[str]:
"""调用 opencode run 执行任务，返回文本输出"""
global _session_id
if _session_id is None:
_session_id = _get_or_create_session()
args = ["npx", "opencode", "run", prompt, "--format", "json"]
if _session_id:
args.extend(["--session", _session_id, "--continue"])
try:
result = subprocess.run(
["npx", "opencode", "run", prompt, "--format", "json"],
args,
capture_output=True, text=True, timeout=timeout,
cwd=str(PROJECT_ROOT),
env={**os.environ, "OPENCODE_DISABLE_AUTOUPDATE": "1"}
@@ -66,28 +110,23 @@ def _run_opencode(prompt: str, timeout: int = 60) -> Optional[str]:
def search_via_opencode(query: str, max_results: int = 5) -> List[Dict]:
"""通过 opencode 联网搜索"""
prompt = f"""用 webfetch 搜索：{query}
只输出 JSON 数组 [{{"title":"标题","url":"链接","content":"摘要"}}]，最多 {max_results} 条，不要其他文字。"""
output = _run_opencode(prompt, timeout=90)
if not output:
return []
m = re.search(r'\[\s*\{.*\}\s*\]', output, re.DOTALL)
if not m:
logger.warning(f"未找到JSON数组: {output[:150]}")
return []
"""通过 MCP 搜索工具联网搜索（替代脆弱的 npx prompt 方式）"""
try:
results = json.loads(m.group())
if isinstance(results, list):
for r in results:
r["source"] = "opencode_webfetch"
logger.info(f"opencode 搜索 '{query[:20]}': {len(results)} 条")
return results[:max_results]
from search_utils import search
return search(query, max_results)
except Exception as e:
logger.warning(f"JSON解析失败: {e}")
return []
logger.warning("search_utils 不可用，回退子进程: %s", e)
result = subprocess.run(
[sys.executable, str(PROJECT_ROOT / "scripts" / "mcp_search_server.py"),
"--query", query],
capture_output=True, text=True, timeout=90,
)
if result.returncode == 0:
try:
return json.loads(result.stdout)[:max_results]
except Exception:
pass
return []
def refresh_cache():
scripts/prompt_loader.py
+125 -16
View File
@@ -1,3 +1,20 @@
#!/usr/bin/env python3
"""
Prompt 配置加载器
架构：DB 为主，代码仅作初始种子。
- _PROMPT_DEFAULTS: 仅用于首次自动写入 DB（种子数据），不作为运行时 fallback
- prompt_configs 表: 运行时唯一来源，修改 prompt 直接改 DB
- DB 不可用时: 退化到代码默认值（仅用于紧急回退，不是日常模式）
新增 prompt 流程：
1. 在 _PROMPT_DEFAULTS 添加定义
2. 重启后自动补入 DB（仅当该 key 不在 DB 中时）
3. 后续修改直接在 DB 操作，不再改代码
修改 prompt 流程：
直接 UPDATE prompt_configs SET content = '...' WHERE key = '...';
"""
import os, sys
from pathlib import Path
from typing import Dict, Any, Optional
@@ -6,6 +23,11 @@ PROJECT_ROOT = Path(__file__).parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
sys.path.insert(0, str(PROJECT_ROOT / 'platform' / 'backend'))
# ============================================================
# 种子数据（初始默认值）
# 修改 prompt 请直接在 DB 操作，不要改这里
# 仅在表里没有该 key 时用于首次填充
# ============================================================
_PROMPT_DEFAULTS = {
"topics_trends": {
"content": "你是中文互联网趋势分析师。请列出今天（{date}）中文互联网上最值得创作的10个话题。\n\n要求：\n1. 覆盖领域：{domains}\n2. 从真实用户角度出发\n3. 每个话题需包含：\n - \"domain\": 领域\n - \"topic\": 话题名称\n - \"reason\": 为什么现在讨论这个（1句话，有具体事件/数据支撑）\n - \"hot_keywords\": 3-5个搜索词（含1-2个长尾词）\n - \"platform\": 最适合分发的平台（知乎/小红书/微信/多平台）\n - \"seo_angle\": 从什么角度切入能获得搜索流量（1句话）\n - \"engagement\": 高/中/低\n\n输出 JSON 数组。只输出 JSON，不要其他文字。",
@@ -23,22 +45,22 @@ _PROMPT_DEFAULTS = {
"variables": ["gaps"],
},
"section_expansion": {
"content": "你是一个资深作者，正在写一篇关于「{topic_title}」的文章。请写「{section_title}」这一节。\n\n今天日期：{date}。\n\n笔记要点：\n{content}\n\n【输出要求】\n输出3-6段纯粹、流畅的段落文字，每节内容根据平台需求控制在200-800字之间。\n\n格式：\n- 禁止任何标题/列表/格式标记（#、-、*、1.、**等）\n- 每段3-5句，段间空行分隔\n- 用「你」或「我们」视角，自然口语化\n\n内容要求（让文章在各平台能被推荐）：\n- 开头直接切入痛点或反常识观点，抓住注意力\n- 每个观点配具体案例或数据（用「据统计」「调研显示」等），不要空泛说理\n- 有独特判断和立场，避免正确废话\n- 回答「所以呢」——读者看完能带走什么\n- 结尾有情绪感召力，让人想点赞/收藏/转发\n\n直接输出段落正文，不要任何附加说明。",
"content": "你是一个资深作者，正在写一篇关于「{topic_title}」的文章。请写「{section_title}」这一节。\n\n今天日期：{date}。\n\n笔记要点：\n{content}\n\n【输出要求】\n输出3-6段纯粹、流畅的段落文字，每节内容根据平台需求控制在200-800字之间。\n\n格式：\n- 禁止任何标题/列表/格式标记（#、-、*、1.、**等）\n- 每段3-5句，段间空行分隔\n- 用「你」或「我」视角，自然口语化。全文统一使用「你」称呼读者，不要混用「你们」\n- **禁止以「引言」「核心观点」「受众痛点」「总结」「开头钩子」「独特视角」这类结构标签开头**——直接从场景或痛点切入\n\n内容要求（让文章在各平台能被推荐）：\n- 开头直接切入场景或痛点，一句话抓住注意力，不要铺垫\n- 每个观点配具体案例或数据（用「据统计」「调研显示」等），不要空泛说理\n- 有独特判断和立场，避免正确废话\n- 回答「所以呢」——读者看完能带走什么\n- **避免AI套话**：不要出现「一个真实的XX案例很能说明问题」「说回到XX这件事」「这就是XX被XX后的样子」「如果你也XX」「值得注意的是」「首先其次最后」「综上所述」\n- 结尾有情绪感召力，让人想点赞/收藏/转发\n- 引用数据或案例时，在行内用（来源：报告/案例名称）标注\n\n直接输出段落正文，不要任何附加说明。\n\n【注意：如果你是写最后一节，在正文写完后加一行 --- 分隔，然后写 **参考资料**，每行一个来源：- 来源名称（简要说明）。前面几节不要加这个。】",
"temperature": 0.75, "max_tokens": 3000,
"variables": ["topic_title", "section_title", "date", "content"],
},
"title_optimize_zhihu": {
"content": "你是一个知乎内容专家。为以下文章起3个高点击率标题。\n\n标题：{title}\n核心观点：{core}\n受众痛点：{pain}\n领域：{field}\n\n要求：\n- 信息密度高，SEO关键词靠前\n- 偏好数字、对比、悬念、痛点类标题\n- 20字以内\n- 不要「如何...」开头\n- 有独特视角和差异化\n- 能引发讨论\n\n输出3个选项，格式：\n1. 标题A\n2. 标题B\n3. 标题C\n只输出标题，不要其他文字。",
"content": "你是一个知乎高赞标题专家。为以下文章起3个高收藏率标题。\n\n标题：{title}\n核心观点：{core}\n受众痛点：{pain}\n领域：{field}\n\n要求：\n- 信息密度高，包含搜索关键词\n- 优先使用数字、对比、悬念、痛点\n- 20字以内\n- 不要「如何...」开头\n- 有独特判断和立场，能引发讨论或反对\n- 制造「不点开就亏了」的紧迫感\n\n输出3个选项，格式：\n1. 标题A\n2. 标题B\n3. 标题C\n只输出标题，不要其他文字。",
"temperature": 0.8, "max_tokens": 1500,
"variables": ["title", "core", "pain", "field"],
},
"title_optimize_wechat": {
"content": "你是一个公众号资深作者。为以下文章起3个10万+潜力标题。\n\n标题：{title}\n核心观点：{core}\n\n要求：\n- 制造好奇心和话题感\n- 包含微信SEO关键词\n- 口语化，避免感叹号堆砌\n- 15-25字\n- 有情感共鸣或争议性\n\n输出3个选项，格式：\n1. 标题A\n2. 标题B\n3. 标题C\n只输出标题，不要其他文字。",
"content": "你是一个10万+公众号爆款标题专家。为以下文章起3个高打开率标题。\n\n文章主题：{title}\n核心观点：{core}\n\n要求：\n- 包含身份标签（如「打工人」「30岁后」「职场人」「普通上班族」）\n- 包含情绪钩子（焦虑/反常识/后悔/稀缺）\n- 包含微信SEO关键词（用户会搜的词）\n- 15-25字\n- 口语化，避免感叹号堆砌\n- 忌笼统——越具体越好\n\n格式模板参考：\n- 「身份+痛点+方案」：打工人学了一堆AI工具，为什么还在加班？\n- 「反常识+数据」：用了AI效率反而更低了？73%的上班族掉进了这个坑\n- 「结果+身份」：每天省出2小时后，我才发现自己以前有多傻\n\n输出3个选项，格式：\n1. 标题A\n2. 标题B\n3. 标题C\n只输出标题，不要其他文字。",
"temperature": 0.8, "max_tokens": 1500,
"variables": ["title", "core"],
},
"title_optimize_xhs": {
"content": "你是一个小红书爆款专家。为以下文章起3个热门标题。\n\n标题：{title}\n核心观点：{core}\n\n要求：\n- 20字以内\n- 爆款模式：数字+结果 / 痛点+方案 / 反常识\n- 包含小红书SEO关键词\n- 1个精确emoji\n- 有场景感、结果感、满足感\n- 不要「必看/收藏/码住」\n\n输出3个选项，格式：\n1. 标题A\n2. 标题B\n3. 标题C\n只输出标题，不要其他文字。",
"content": "你是一个小红书爆款笔记专家。为以下文章起3个高赞标题。\n\n标题：{title}\n核心观点：{core}\n\n要求：\n- 18字以内\n- 爆款公式：身份/场景+数字+结果，或痛点+方案+反差\n- 包含小红书SEO关键词\n- 1个精确emoji（不要用🔥💥❌❓这几个滥用的）\n- 有场景感、结果感、获得感\n- 忌笼统：不要「必看/收藏/码住/绝了」\n\n输出3个选项，格式：\n1. 标题A\n2. 标题B\n3. 标题C\n只输出标题，不要其他文字。",
"temperature": 0.8, "max_tokens": 1500,
"variables": ["title", "core"],
},
@@ -53,7 +75,7 @@ _PROMPT_DEFAULTS = {
"variables": ["date", "year", "title", "field", "core", "pain", "angle", "cases_summary"],
},
"compliance_fix": {
"content": "你是一个专业的内容合规优化助手。以下文章存在合规问题，请逐一修复并输出完整HTML。\n\n需修复的问题：\n{issues_desc}\n\n原文：\n{html}\n\n要求：\n- 只修复上述问题，不改变文章结构和核心内容\n- 保持<h2>, <h3>, <p>等标签结构不变\n- 修复后内容依然保持可读性和自然语感（不要因为合规变成生硬的表达）\n- 替换敏感词时选择意思相近的替代词，不删节重要信息",
"content": "你是一个专业的内容合规与质量优化助手。以下文章存在需要优化的地方，请逐一修复并输出完整HTML。\n\n需修复的问题：\n{issues_desc}\n\n原文：\n{html}\n\n要求：\n- 只修复上述问题，不改变文章结构和核心内容\n- 保持<h2>, <h3>, <p>等标签结构不变\n- 修复后内容依然保持可读性和自然语感\n- 替换敏感词时选择意思相近的替代词，不删节重要信息\n- 如果问题是AI套话，直接删除或改写那些词句\n- 如果问题是人称混用，统一为「你」\n- 如果问题是缺少配图，在关键位置插入 `<p></p>` 空段落占位，配图由后续流程统一处理",
"temperature": 0.3, "max_tokens": 8000,
"variables": ["issues_desc", "html"],
},
@@ -68,7 +90,7 @@ _PROMPT_DEFAULTS = {
"variables": ["n", "cat_names", "n2", "src_summary", "year"],
},
"tags_generation": {
"content": "为以下文章生成{platform}标签（3-5个）。\n\n标题：{title}\n领域：{field}\n核心观点：{core}\n\n要求：每个2-4字。直接输出标签，空格分隔。不要输出思考过程。",
"content": "为以下文章生成{platform}标签（5-8个）。\n\n标题：{title}\n领域：{field}\n核心观点：{core}\n\n要求：\n- 每个标签2-5字\n- 包含1-2个搜索流量词（用户在{platform}会搜的词）\n- 包含1-2个热门话题词\n- 标签之间要有层次：大领域→小话题→具体场景\n- 不要重复意思相近的标签\n\n直接输出标签，空格分隔。不要输出思考过程。",
"temperature": 0.3, "max_tokens": 500,
"variables": ["platform", "title", "field", "core"],
},
@@ -76,12 +98,45 @@ _PROMPT_DEFAULTS = {
_DB_CACHE: Dict[str, Dict[str, Any]] = {}
_CACHE_LOADED = False
_DB_AVAILABLE = False # True if DB was successfully loaded at least once
def _seed_missing_prompts():
"""将代码默认值中不存在的 prompt 自动补入 DB"""
try:
if os.getenv('USE_POSTGRES', 'true') == 'true':
from app.database import SessionLocal
from app.models import PromptConfig
db = SessionLocal()
try:
existing_keys = {p.key for p in db.query(PromptConfig).all()}
for key, cfg in _PROMPT_DEFAULTS.items():
if key not in existing_keys:
pc = PromptConfig(
key=key,
content=cfg['content'],
temperature=cfg.get('temperature'),
max_tokens=cfg.get('max_tokens'),
enabled=True,
module_id='default',
variables=cfg.get('variables', []),
)
db.add(pc)
print(f"[prompt_loader] auto-seeded prompt: {key}")
if any(key not in existing_keys for key in _PROMPT_DEFAULTS):
db.commit()
finally:
db.close()
except Exception:
pass
def _ensure_db_loaded():
global _DB_CACHE, _CACHE_LOADED
global _DB_CACHE, _CACHE_LOADED, _DB_AVAILABLE
if _CACHE_LOADED:
return
# 自动补缺：将代码默认值中不存在的 prompt 写入 DB
_seed_missing_prompts()
try:
if os.getenv('USE_POSTGRES', 'true') == 'true':
from app.database import SessionLocal
@@ -94,6 +149,7 @@ def _ensure_db_loaded():
"temperature": p.temperature,
"max_tokens": p.max_tokens,
}
_DB_AVAILABLE = True
finally:
db.close()
except Exception:
@@ -102,16 +158,30 @@ def _ensure_db_loaded():
def get_prompt(key: str, **kwargs) -> str:
"""获取 prompt 内容
优先级：DB → 代码默认值（仅 DB 不可用时）→ 空字符串
"""
_ensure_db_loaded()
cleaned = []
for k, v in kwargs.items():
cleaned.append((k, str(v)))
kwargs = dict(cleaned)
if key in _DB_CACHE:
content = _DB_CACHE[key]["content"]
elif key in _PROMPT_DEFAULTS:
for k, v in kwargs.items():
content = content.replace("{" + k + "}", str(v))
return content
# 仅 DB 不可用时回退代码默认值
if not _DB_AVAILABLE and key in _PROMPT_DEFAULTS:
content = _PROMPT_DEFAULTS[key]["content"]
else:
return ""
for k, v in kwargs.items():
content = content.replace("{" + k + "}", str(v))
return content
for k, v in kwargs.items():
content = content.replace("{" + k + "}", str(v))
return content
return ""
def get_prompt_params(key: str) -> Dict[str, Any]:
@@ -121,7 +191,7 @@ def get_prompt_params(key: str) -> Dict[str, Any]:
"temperature": _DB_CACHE[key].get("temperature"),
"max_tokens": _DB_CACHE[key].get("max_tokens"),
}
if key in _PROMPT_DEFAULTS:
if not _DB_AVAILABLE and key in _PROMPT_DEFAULTS:
return {
"temperature": _PROMPT_DEFAULTS[key].get("temperature"),
"max_tokens": _PROMPT_DEFAULTS[key].get("max_tokens"),
@@ -129,8 +199,47 @@ def get_prompt_params(key: str) -> Dict[str, Any]:
return {}
def seed_prompts(force: bool = False):
"""手动触发种子同步（用于 CLI 或 API）
force=True: 用代码默认值覆盖 DB
force=False: 仅补充 DB 中不存在的 key
"""
try:
from app.database import SessionLocal
from app.models import PromptConfig
db = SessionLocal()
try:
existing = {p.key: p for p in db.query(PromptConfig).all()}
for key, cfg in _PROMPT_DEFAULTS.items():
if force or key not in existing:
if key in existing:
p = existing[key]
p.content = cfg['content']
p.temperature = cfg.get('temperature')
p.max_tokens = cfg.get('max_tokens')
else:
p = PromptConfig(
key=key,
content=cfg['content'],
temperature=cfg.get('temperature'),
max_tokens=cfg.get('max_tokens'),
enabled=True,
module_id='default',
variables=cfg.get('variables', []),
)
db.add(p)
db.commit()
finally:
db.close()
return {"ok": True, "action": "force" if force else "seed"}
except Exception as e:
return {"ok": False, "error": str(e)}
def reload_prompts():
global _CACHE_LOADED, _DB_CACHE
global _CACHE_LOADED, _DB_CACHE, _DB_AVAILABLE
_CACHE_LOADED = False
_DB_CACHE = {}
_ensure_db_loaded()
_DB_AVAILABLE = False
_ensure_db_loaded()
scripts/search_utils.py
+290
View File
@@ -0,0 +1,290 @@
#!/usr/bin/env python3
"""
统一搜索工具：DB驱动多提供商自动降级
"""
import json, logging, os, sys, time
from pathlib import Path
from typing import List, Dict, Optional
logger = logging.getLogger(__name__)
PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
sys.path.insert(0, str(PROJECT_ROOT / "platform" / "backend"))
try:
from app.database import SessionLocal
from app.models import SearchProvider
HAVE_DB = True
except ImportError:
HAVE_DB = False
SEARCH_CACHE_FILE = PROJECT_ROOT / "automation" / "data" / "search_cache.json"
def _get_providers() -> List[Dict]:
"""从 DB 加载启用的搜索提供商，按优先级排序（自动跨日重置用量）"""
if not HAVE_DB:
return []
try:
import datetime as _dt
db = SessionLocal()
today = _dt.date.today()
rows = db.query(SearchProvider).filter(
SearchProvider.enabled == True
).order_by(SearchProvider.priority).all()
needs_commit = False
for r in rows:
if (r.usage_today or 0) > 0 and r.last_used_at:
last_date = r.last_used_at
if hasattr(last_date, 'date'):
last_date = last_date.date()
elif isinstance(last_date, _dt.datetime):
last_date = last_date.date()
if last_date < today:
r.usage_today = 0
needs_commit = True
if needs_commit:
db.commit()
db.close()
return [r.to_dict() if hasattr(r, 'to_dict') else {
"id": r.id, "name": r.name, "provider_type": r.provider_type,
"api_key": r.api_key or "", "api_url": r.api_url or "",
"priority": r.priority, "daily_limit": r.daily_limit,
"usage_today": getattr(r, 'usage_today', 0),
} for r in rows]
except Exception as e:
logger.warning("加载搜索提供商失败: %s", e)
return []
def reset_all_usage():
"""手动重置所有提供商当日用量（供 API/定时任务调用）"""
if not HAVE_DB:
return
try:
db = SessionLocal()
db.query(SearchProvider).update({SearchProvider.usage_today: 0})
db.commit()
db.close()
logger.info("所有搜索提供商用量已重置")
except Exception as e:
logger.warning("重置用量失败: %s", e)
def _increment_usage(provider_id: int):
"""增加提供商当日用量"""
if not HAVE_DB:
return
try:
db = SessionLocal()
p = db.query(SearchProvider).filter(SearchProvider.id == provider_id).first()
if p:
p.usage_today = (p.usage_today or 0) + 1
p.last_used_at = __import__('datetime').datetime.now(__import__('datetime').timezone.utc)
db.commit()
db.close()
except Exception:
pass
def _call_baidu(api_key: str, api_url: str, query: str, max_results: int) -> List[Dict]:
import requests
resp = requests.post(
api_url or "https://qianfan.baidubce.com/v2/ai_search/web_search",
headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
json={
"messages": [{"role": "user", "content": query}],
"search_source": "baidu_search_v2",
"resource_type_filter": [{"type": "web", "top_k": max_results}],
},
timeout=15
)
if resp.status_code != 200:
logger.warning("百度搜索返回 %s: %s", resp.status_code, resp.text[:100])
return []
data = resp.json()
results = data.get("results", []) or data.get("webPages", {}).get("value", [])
return [{
"title": r.get("title", "")[:120],
"url": r.get("url", "") or r.get("link", ""),
"content": r.get("snippet", "") or r.get("content", "") or r.get("summary", "")[:300],
"source": "baidu",
} for r in results[:max_results]]
def _call_qiniu(api_key: str, api_url: str, query: str, max_results: int) -> List[Dict]:
import requests
resp = requests.post(
api_url or "https://api.qnaigc.com/v1/search/web",
headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
json={"query": query, "max_results": max_results, "search_type": "web"},
timeout=15
)
if resp.status_code != 200:
logger.warning("七牛搜索返回 %s: %s", resp.status_code, resp.text[:100])
return []
data = resp.json()
items = data.get("results", data.get("data", []))
return [{
"title": r.get("title", "")[:120],
"url": r.get("url", ""),
"content": r.get("content", "") or r.get("snippet", "")[:300],
"source": "qiniu",
} for r in items[:max_results]]
def _call_tinyfish(api_key: str, api_url: str, query: str, max_results: int) -> List[Dict]:
import requests
resp = requests.get(
api_url or "https://api.search.tinyfish.ai",
params={"query": query, "location": "CN", "language": "zh"},
headers={"X-API-Key": api_key},
timeout=15
)
if resp.status_code != 200:
logger.warning("TinyFish搜索返回 %s: %s", resp.status_code, resp.text[:100])
return []
data = resp.json()
items = data.get("results", [])
return [{
"title": r.get("title", "")[:120],
"url": r.get("url", ""),
"content": r.get("snippet", "")[:300],
"source": "tinyfish",
} for r in items[:max_results]]
def _call_bing(api_key: str, api_url: str, query: str, max_results: int) -> List[Dict]:
import requests
resp = requests.get(
api_url or "https://api.bing.microsoft.com/v7.0/search",
params={"q": query, "count": max_results, "mkt": "zh-CN"},
headers={"Ocp-Apim-Subscription-Key": api_key},
timeout=15
)
if resp.status_code != 200:
logger.warning("Bing搜索返回 %s: %s", resp.status_code, resp.text[:100])
return []
data = resp.json()
items = data.get("webPages", {}).get("value", [])
return [{
"title": r.get("name", "")[:120],
"url": r.get("url", ""),
"content": r.get("snippet", "")[:300],
"source": "bing",
} for r in items[:max_results]]
def _call_mcp(api_key: str, api_url: str, query: str, max_results: int) -> List[Dict]:
"""Call the MCP search server directly (no API key needed)."""
import subprocess
try:
r = subprocess.run(
[sys.executable, str(PROJECT_ROOT / "scripts" / "mcp_search_server.py"),
"--query", query],
capture_output=True, text=True, timeout=90,
)
if r.returncode != 0:
logger.warning("MCP搜索子进程返回非零: %s", r.stderr[:100])
return []
results = json.loads(r.stdout)
if isinstance(results, list):
for res in results:
res["source"] = "opencode"
return results[:max_results]
except json.JSONDecodeError as e:
logger.warning("MCP搜索JSON解析失败: %s", e)
except subprocess.TimeoutExpired:
logger.warning("MCP搜索超时 (90s)")
except Exception as e:
logger.warning("MCP搜索失败: %s", e)
return []
_PROVIDER_CALLS = {
"baidu": _call_baidu,
"qiniu": _call_qiniu,
"tinyfish": _call_tinyfish,
"bing": _call_bing,
"mcp": _call_mcp,
}
def search(query: str, max_results: int = 5) -> List[Dict]:
"""统一搜索：DB提供商 → 本地缓存 → 空结果"""
providers = _get_providers()
for p in providers:
if (p.get("usage_today") or 0) >= (p.get("daily_limit") or 99999):
logger.info("提供商 %s 已达日限 %s，跳过", p.get("name"), p.get("daily_limit"))
continue
if not p.get("api_key") and p.get("provider_type") != "mcp":
logger.info("提供商 %s 未配置 API Key，跳过", p.get("name"))
continue
call_fn = _PROVIDER_CALLS.get(p.get("provider_type"))
if not call_fn:
continue
try:
results = call_fn(p["api_key"], p.get("api_url", ""), query, max_results)
if results:
_increment_usage(p["id"])
logger.info("搜索 '%s' 通过 %s 获得 %d 条结果", query[:20], p.get("name"), len(results))
return results
logger.warning("提供商 %s 返回空结果", p.get("name"))
except Exception as e:
logger.warning("提供商 %s 失败: %s", p.get("name"), e)
continue
logger.info("搜索 '%s' 无结果（所有提供商均不可用）", query[:20])
return []
def search_from_cache(query: str, max_results: int = 5) -> List[Dict]:
"""从本地缓存读取搜索结果"""
if not SEARCH_CACHE_FILE.exists():
return []
try:
cache = json.loads(SEARCH_CACHE_FILE.read_text(encoding="utf-8"))
meta = cache.get("_metadata", {})
updated = meta.get("updated_at", "")
if updated:
import datetime
age = (datetime.datetime.now() - datetime.datetime.fromisoformat(updated)).total_seconds()
if age > 129600:
logger.warning("搜索缓存过时(%dh)，跳过", int(age // 3600))
return []
results = cache.get(query, [])
return results[:max_results]
except Exception:
return []
def enrich_topic_research(topic: dict, max_results: int = 5) -> str:
"""对选题进行网络搜索，返回格式化的研究发现文本"""
title = topic.get('title', '')
field = topic.get('field', '')
queries = [title]
if field and field not in title:
queries.append(f"{field} {title[:40]}")
seen_urls = set()
results = []
for q in queries:
for r in search(q, max_results):
url = r.get('url', '')
if url and url not in seen_urls:
seen_urls.add(url)
results.append(r)
if not results:
return ""
lines = ["\n## 网络搜索参考", ""]
for r in results[:max_results]:
snippet = r.get('snippet', r.get('content', ''))
lines.append(f"- **{r.get('title', '无标题')}**")
lines.append(f" {snippet[:200]}")
if r.get('url'):
lines.append(f" [{r['url']}]")
lines.append("")
return "\n".join(lines)
scripts/web_search.py
+9 -1
View File
@@ -127,7 +127,15 @@ def save_to_cache(query: str, results: List[Dict]):
def search(query: str, max_results: int = 5) -> List[Dict]:
"""统一搜索接口：缓存 → API → 网页抓取"""
"""统一搜索接口：DB提供商 → 缓存 → API → 网页抓取"""
try:
from search_utils import search as db_search
results = db_search(query, max_results)
if results:
return results
except Exception:
pass
results = search_from_cache(query, max_results)
if results:
return results
scripts/writer.py
+75 -38
View File
@@ -14,6 +14,7 @@ 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
@@ -206,6 +207,15 @@ class Writer:
expanded = self._expand_section(sec)
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:
@@ -213,27 +223,71 @@ class Writer:
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 adapted.count('#') == 0:
adapted = f"# {self.topic['title']}\n\n{adapted}"
# 结尾加收藏引导（小红书算法权重：收藏率）
if '收藏' not in adapted:
adapted += "\n\n✨ 觉得有用的话点个收藏吧，下次需要的时候随时翻出来看~"
return adapted
if platform == "wechat":
result = []
for line in lines:
line = line.replace('我', '你')
# 人称统一：我们→我，你们→你
line = line.replace('我们', '我').replace('你们', '你').replace('我', '你')
if line.startswith('### '):
result.append(f"\n**{line[4:]}**\n")
elif line.startswith('## '):
@@ -338,12 +392,20 @@ class Writer:
if not line:
continue
line = re.sub(r'^\d+[.、)\s]+', '', line)
line = line.strip('*#- \t')
line = line.strip('*#- \t"\'"''"')
# 跳过思考/建议类输出（如"不如："、"或者："、"建议方案"等）
if re.match(r'^(不如|或者|建议|推荐|参考|方案[一二三]|第[一二三]种|以[下是]|标题[一二三]|选项)', line):
continue
if line:
titles.append(line)
if titles:
logger.info(f"标题优化 [{platform}]: {titles[0][:50]}...")
return titles[0]
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
@@ -361,43 +423,18 @@ class Writer:
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('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;').replace('"', '&quot;').replace("'", '&apos;')
safe_field = topic_field.replace('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;')
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'<text x="540" y="{line_y + i*55}" font-size="36" fill="#1a1a1a" font-weight="bold">{l}</text>' for i, l in enumerate(lines))
field_text = f'<text x="540" y="{line_y + len(lines)*55 + 30}" font-size="20" fill="#98a2b3">{safe_field}</text>' if safe_field else ''
img_svg = f'''<svg xmlns="http://www.w3.org/2000/svg" width="1080" height="600" viewBox="0 0 1080 600" style="width:100%;max-width:1080px;border-radius:8px;background:linear-gradient(135deg,#f0f4ff,#e8f0fe)">
<rect width="1080" height="600" fill="url(#bg)"/>
<defs><linearGradient id="bg" x1="0%" y1="0%" x2="100%" y2="100%"><stop offset="0%" style="stop-color:#f0f4ff"/><stop offset="100%" style="stop-color:#e8f0fe"/></linearGradient></defs>
<g transform="translate(540,300)" text-anchor="middle" font-family="-apple-system,BlinkMacSystemFont,Helvetica Neue,PingFang SC,Microsoft YaHei,sans-serif">
<rect x="-60" y="-100" width="120" height="4" rx="2" fill="#409eff"/>
{title_texts}
{field_text}
<text y="100" font-size="14" fill="#c0c4cc">宇之然 · 配图（可替换）</text>
</g></svg>'''
img_b64 = 'data:image/svg+xml;base64,' + base64.b64encode(img_svg.encode('utf-8')).decode('ascii')
img_tag = f'<p><img src="{img_b64}" alt="{safe_title}" style="width:100%;max-width:1080px;border-radius:8px;"></p>\n'
h1_end = html_content.find('</h1>')
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_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)