最后运行{{ mod.last_run || '从未' }}
+ 最后运行{{ mod.last_run || '从未' }}✅成功❌失败⏳运行中
下次运行{{ mod.next_run || '—' }}
- 累计运行{{ mod.total_runs }} 次 {{ mod.success_runs }} 成功 {{ mod.failed_runs }} 失败
+ 累计运行{{ mod.total_runs }} 次 {{ mod.success_runs }} 成功 {{ mod.failed_runs }} 失败 {{ mod.running }} 运行中
+ 🔍 {{ showSearch ? '收起搜索' : '展开搜索' }}
+ 筛选条件已生效，共 {{ filteredTopics.length }} 条
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ 查询
+ 重置
+
+
+
+
+
@@ -100,6 +147,7 @@
]*>(开头钩子|核心观点|受众痛点|独特视角|差异化切入|内容形式).*?',
]
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 插入的建议配图占位（）
+ html = re.sub(r'{_alt_attr(line)} \n'
+
+ field_badge = ""
+ if field:
+ field_badge = f'''
+ {_alt_attr(field)} '''
+
+ svg = f''''''
+ 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''''''
+ 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'{_alt_attr(line)} \n'
+
+ source_line = ""
+ if source:
+ source_line = f'{_alt_attr(f"— {source}")} '
+
+ svg = f''''''
+ 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''''''
+ 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('')
+ 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"},
+}
diff --git a/scripts/mcp_search_server.py b/scripts/mcp_search_server.py
new file mode 100644
index 0000000..024ed91
--- /dev/null
+++ b/scripts/mcp_search_server.py
@@ -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()
diff --git a/scripts/opencode_search.py b/scripts/opencode_search.py
index 8969708..aa43c86 100644
--- a/scripts/opencode_search.py
+++ b/scripts/opencode_search.py
@@ -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():
diff --git a/scripts/prompt_loader.py b/scripts/prompt_loader.py
index c36f786..2608041 100644
--- a/scripts/prompt_loader.py
+++ b/scripts/prompt_loader.py
@@ -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- 保持,
,
{l} ' for i, l in enumerate(lines))
- field_text = f'{safe_field} ' if safe_field else ''
- img_svg = f''''''
- img_b64 = 'data:image/svg+xml;base64,' + base64.b64encode(img_svg.encode('utf-8')).decode('ascii')
- img_tag = f'
@@ -177,6 +225,7 @@
|
状态：{{ getStatusLabel(previewTopic.status) }}
创作：{{ formatDate(previewTopic.generated_at) }}
+ 审查：{{ formatDate(previewTopic.reviewed_at) }}
发布：{{ formatDate(previewTopic.published_at) }}
@@ -204,16 +253,66 @@ const TopicsApp = {
publishPlatforms: { zhihu: true, wechat: true, xiaohongshu: true },
publishing: false,
savingContent: false,
- currentPage: 1, pageSize: 10
+ currentPage: 1, pageSize: 10,
+ sortField: 'created_at', sortOrder: 'ascending',
+ showSearch: false,
+ searchForm: { id: '', title: '', field: '', createdStart: null, createdEnd: null, generatedStart: null, generatedEnd: null, reviewedStart: null, reviewedEnd: null, publishedStart: null, publishedEnd: null }
}
},
computed: {
+ searchActive() {
+ const f = this.searchForm;
+ return !!(f.id || f.title || f.field || f.createdStart || f.createdEnd || f.generatedStart || f.generatedEnd || f.reviewedStart || f.reviewedEnd || f.publishedStart || f.publishedEnd);
+ },
filteredTopics() {
- if (this.filterStatus === 'today') return this.todayTopics;
- if (!this.filterStatus) return this.topics;
- const map = { 'pending': ['pending','待处理'], 'review': ['review','待审查'], 'ready': ['ready','待发布'], 'published': ['published','已发布'] };
- const allowed = map[this.filterStatus] || [this.filterStatus];
- return this.topics.filter(t => allowed.includes(t.status));
+ let list;
+ if (this.filterStatus === 'today') list = this.todayTopics;
+ else if (!this.filterStatus) list = this.topics;
+ else {
+ const map = { 'pending': ['pending','待处理'], 'review': ['review','待审查'], 'ready': ['ready','待发布'], 'published': ['published','已发布'] };
+ const allowed = map[this.filterStatus] || [this.filterStatus];
+ list = this.topics.filter(t => allowed.includes(t.status));
+ }
+ // 搜索过滤
+ const sf = this.searchForm;
+ if (sf.id || sf.title || sf.field || sf.createdStart || sf.createdEnd || sf.generatedStart || sf.generatedEnd || sf.reviewedStart || sf.reviewedEnd || sf.publishedStart || sf.publishedEnd) {
+ list = list.filter(t => {
+ if (sf.id && !t.id.toLowerCase().includes(sf.id.toLowerCase())) return false;
+ if (sf.title && !t.title.toLowerCase().includes(sf.title.toLowerCase())) return false;
+ if (sf.field && !(t.field || '').toLowerCase().includes(sf.field.toLowerCase())) return false;
+ const inRange = (val, start, end) => {
+ if (!val) return !start && !end;
+ const d = val.slice(0, 10);
+ if (start && d < start) return false;
+ if (end && d > end) return false;
+ return true;
+ };
+ return inRange(t.created_at, sf.createdStart, sf.createdEnd)
+ && inRange(t.generated_at, sf.generatedStart, sf.generatedEnd)
+ && inRange(t.reviewed_at, sf.reviewedStart, sf.reviewedEnd)
+ && inRange(t.published_at, sf.publishedStart, sf.publishedEnd);
+ });
+ }
+ if (this.sortField) {
+ const field = this.sortField;
+ const isDateField = ['created_at','generated_at','reviewed_at','published_at','updated_at'].includes(field);
+ list = [...list].sort((a, b) => {
+ const va = a[field], vb = b[field];
+ if (!va && !vb) return 0;
+ if (!va) return 1;
+ if (!vb) return -1;
+ let cmp;
+ if (isDateField) {
+ const ta = new Date(va.replace(' ', 'T')).getTime();
+ const tb = new Date(vb.replace(' ', 'T')).getTime();
+ cmp = ta - tb;
+ } else {
+ cmp = String(va).localeCompare(String(vb), 'zh', { numeric: true });
+ }
+ return this.sortOrder === 'descending' ? -cmp : cmp;
+ });
+ }
+ return list;
},
paginatedTopics() {
const start = (this.currentPage - 1) * this.pageSize;
@@ -356,13 +455,42 @@ const TopicsApp = {
const title = titleEl ? titleEl.textContent.trim() : (this.previewTopic?.title || '');
const body = doc.body;
if (body) {
- body.querySelectorAll('script, style, nav, .header, .tags, footer, .interaction, svg, img, button').forEach(el => el.remove());
+ body.querySelectorAll('script, style, nav, .header, .tags, footer, .interaction, button, .ad, aside, .comment').forEach(el => el.remove());
+ }
+
+ if (platform === 'xiaohongshu') {
+ const lines = [];
+ body.querySelectorAll('h1,h2,h3,h4,p,li').forEach(el => {
+ const text = el.textContent.trim();
+ if (!text || text.length < 2) return;
+ const tag = el.tagName.toLowerCase();
+ const prefix = tag.startsWith('h') ? '\n### ' : '- ';
+ const clean = text.replace(/\n/g, ' ').replace(/^[\s#]+|[\s#]+$/g, '');
+ if (clean) lines.push(prefix + clean);
+ });
+ const md = `**${title}**\n\n${lines.join('\n')}`;
+ navigator.clipboard.writeText(md).then(() => this.$message.success('✅ 已复制 Markdown（小红书格式）')).catch(() => this.$message.error('❌ 复制失败'));
+ } else {
+ const container = doc.createElement('div');
+ if (titleEl) {
+ const h1 = doc.createElement('h1');
+ h1.textContent = title;
+ container.appendChild(h1);
+ }
+ body.querySelectorAll('h2,h3,h4,p,li,blockquote,img,pre,code,table,hr').forEach(el => {
+ const clone = el.cloneNode(true);
+ container.appendChild(clone);
+ });
+ const cleanHtml = container.innerHTML;
+ const blob = new Blob([cleanHtml], { type: 'text/html' });
+ const richText = new Blob([cleanHtml], { type: 'text/plain' });
+ const item = new ClipboardItem({ 'text/html': blob, 'text/plain': richText });
+ navigator.clipboard.write([item]).then(() => {
+ this.$message.success(`✅ 已复制（含格式和配图），Ctrl+V 粘贴到${platform === 'zhihu' ? '知乎' : '微信公众号'}`);
+ }).catch(() => {
+ navigator.clipboard.writeText(cleanHtml).then(() => this.$message.success('✅ 已复制 HTML')).catch(() => this.$message.error('❌ 复制失败'));
+ });
}
- const contentEls = body ? Array.from(body.querySelectorAll('p, h1, h2, h3, h4, li')) : [];
- const text = contentEls.map(el => el.textContent.trim()).filter(t => t && t.length > 1).join('\n\n');
- navigator.clipboard.writeText(`标题：${title}\n\n内容：\n${text}`)
- .then(() => this.$message.success(`已复制内容，请前往${platform}粘贴发布`))
- .catch(() => this.$message.error('复制失败，请手动复制'));
},
async createTopic(topic) {
if (this.isStatus(topic, 'published')) { this.$message.info('已发布选题不可创作'); return; }
@@ -451,6 +579,11 @@ const TopicsApp = {
} catch (e) { this.$message.error('保存失败: ' + e.message); }
finally { this.savingContent = false; }
},
+ doSearch() { this.currentPage = 1; },
+ resetSearch() {
+ this.searchForm = { id: '', title: '', field: '', createdStart: null, createdEnd: null, generatedStart: null, generatedEnd: null, reviewedStart: null, reviewedEnd: null, publishedStart: null, publishedEnd: null };
+ this.currentPage = 1;
+ },
handleLogout() { localStorage.removeItem('authToken'); localStorage.removeItem('userRole'); localStorage.removeItem('currentUser'); window.location.href = '/login.html'; },
redirectToPage(page) { window.location.href = page.startsWith('/') ? page : '/' + page; },
getStatusLabel(status) { return { 'pending': '待处理', 'review': '待审查', 'ready': '待发布', 'published': '已发布' }[status] || status; },
@@ -471,6 +604,11 @@ const TopicsApp = {
table.toggleRowSelection(row, !allChecked);
}
},
+ handleSortChange({ prop, order }) {
+ this.sortField = prop || 'created_at';
+ this.sortOrder = order || 'ascending';
+ this.currentPage = 1;
+ },
formatDate(dateStr) {
if (!dateStr) return '-';
try { return new Date(dateStr.replace(' ', 'T')).toLocaleString('zh-CN', { year: 'numeric', month: '2-digit', day: '2-digit', hour: '2-digit', minute: '2-digit' }); }
@@ -482,8 +620,11 @@ const TopicsApp = {
mounted() {
const token = localStorage.getItem('authToken');
if (!token) { window.location.href = '/login.html'; return; }
- const urlFilter = new URLSearchParams(window.location.search).get('filter');
+ const params = new URLSearchParams(window.location.search);
+ const urlFilter = params.get('filter');
if (urlFilter) this.filterStatus = urlFilter;
+ const topicId = params.get('topic_id');
+ if (topicId) { this.searchForm.id = topicId; this.showSearch = true; }
fetch('/api/auth/me', { headers: { 'Authorization': 'Bearer ' + token } })
.then(r => r.ok ? r.json() : Promise.reject())
.then(data => { this.currentUser = data.user; this.isAdmin = data.user.role === 'admin'; this.isLoggedIn = true; this.fetchTopics(); this.fetchTodayCount(); })
diff --git a/scripts/collector.py b/scripts/collector.py
index 7be6122..c692243 100644
--- a/scripts/collector.py
+++ b/scripts/collector.py
@@ -23,6 +23,7 @@ import subprocess
# scripts/collector.py 位于 /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)
diff --git a/scripts/compliance_checker.py b/scripts/compliance_checker.py
index a7d8c91..ad501e9 100644
--- a/scripts/compliance_checker.py
+++ b/scripts/compliance_checker.py
@@ -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']*>.*' + re.escape(heading) + r'.*', 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']+>', '', 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'' 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"
diff --git a/scripts/content_cleaner.py b/scripts/content_cleaner.py
index c2bbece..80ecd71 100644
--- a/scripts/content_cleaner.py
+++ b/scripts/content_cleaner.py
@@ -110,6 +110,7 @@ def strip_thinking_html(html: str) -> str:
r']*>最后.*?',
r']*>(总的来说|值得注意的是|换句话说|总而言之|简而言之|一言以蔽之|可以说|不难发现|由此可见|综上所述).*?',
r']*>(总的来说|值得注意的是|换句话说|总而言之|简而言之|一言以蔽之|可以说|不难发现|由此可见|综上所述).*?
',
+ r'(.*?)
', 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'(.*?)
', 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 属性值 diff --git a/scripts/compliance_optimizer.py b/scripts/compliance_optimizer.py index ac31bdb..782f4e6 100644 --- a/scripts/compliance_optimizer.py +++ b/scripts/compliance_optimizer.py @@ -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 '\s*\s*
', '', html, flags=re.DOTALL) + # 移除可能变成可见文本的建议配图文字 + html = re.sub(r'', '', html, flags=re.DOTALL) + html = re.sub(r'建议配图：.*?(?=<|$)', '', html) return html def clean_full_pipeline(text: str, output_format: str = 'markdown') -> str: diff --git a/scripts/db_helper.py b/scripts/db_helper.py index 3e3513b..9d15c57 100644 --- a/scripts/db_helper.py +++ b/scripts/db_helper.py @@ -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() diff --git a/scripts/image_generator.py b/scripts/image_generator.py index 26562b8..43cfffe 100644 --- a/scripts/image_generator.py +++ b/scripts/image_generator.py @@ -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('&', '&').replace('<', '<').replace('>', '>').replace('"', '"').replace("'", ''') - 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() \ No newline at end of file +def _img_tag(src: str, alt: str, width: int = 1080) -> str: + return f', ,
等标签结构不变\n- 修复后内容依然保持可读性和自然语感（不要因为合规变成生硬的表达）\n- 替换敏感词时选择意思相近的替代词，不删节重要信息", + "content": "你是一个专业的内容合规与质量优化助手。以下文章存在需要优化的地方，请逐一修复并输出完整HTML。\n\n需修复的问题：\n{issues_desc}\n\n原文：\n{html}\n\n要求：\n- 只修复上述问题，不改变文章结构和核心内容\n- 保持
, ,
等标签结构不变\n- 修复后内容依然保持可读性和自然语感\n- 替换敏感词时选择意思相近的替代词，不删节重要信息\n- 如果问题是AI套话，直接删除或改写那些词句\n- 如果问题是人称混用，统一为「你」\n- 如果问题是缺少配图，在关键位置插入 `
` 空段落占位，配图由后续流程统一处理", "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() \ No newline at end of file + _DB_AVAILABLE = False + _ensure_db_loaded() diff --git a/scripts/search_utils.py b/scripts/search_utils.py new file mode 100644 index 0000000..9ee9435 --- /dev/null +++ b/scripts/search_utils.py @@ -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) diff --git a/scripts/web_search.py b/scripts/web_search.py index 42f71d9..1ed8df1 100644 --- a/scripts/web_search.py +++ b/scripts/web_search.py @@ -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 diff --git a/scripts/writer.py b/scripts/writer.py index 0d26703..e68aedf 100644 --- a/scripts/writer.py +++ b/scripts/writer.py @@ -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('&', '&').replace('<', '<').replace('>', '>').replace('"', '"').replace("'", ''') - safe_field = topic_field.replace('&', '&').replace('<', '<').replace('>', '>') - lines = [] - chars_per_line = 24 - for i in range(0, len(safe_title), chars_per_line): - lines.append(safe_title[i:i+chars_per_line]) - if not lines: - lines = ['配图'] - line_y = 220 - (len(lines) - 1) * 20 - title_texts = ''.join(f'