Ubuntu
1032 lines
39 KiB
Python
1032 lines
39 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
CMS 改写 → 公众号发布 一键流水线
|
||
|
||
两种模式:
|
||
1. 自动改写模式 (默认): python pipeline.py <article_id>
|
||
自己调 DeepSeek LLM 改写，然后发布
|
||
|
||
2. 传入改写内容模式: python pipeline.py <article_id> --content-file <markdown文件>
|
||
用预改写好的内容直接发布（志玲模式）
|
||
|
||
环境:
|
||
依赖 openai 库: pip install openai
|
||
API Key 从 ~/.nanobot/config.json 读取 DeepSeek 配置
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
import re
|
||
import json
|
||
import time
|
||
import random
|
||
import argparse
|
||
import yaml
|
||
from typing import Optional, Dict, Any
|
||
|
||
# 确保脚本在 rewrite/ 目录执行也能正常工作
|
||
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
PROJECT_ROOT = os.path.dirname(SCRIPT_DIR)
|
||
AUTO_CONFIG_PATH = os.path.join(SCRIPT_DIR, "auto_config.yaml")
|
||
BAOYU_DIR = "/home/ubuntu/.nanobot/workspace/skills/baoyu-post-to-wechat"
|
||
|
||
sys.path.insert(0, BAOYU_DIR)
|
||
sys.path.insert(0, SCRIPT_DIR)
|
||
|
||
from cms_ops import get_article_full, update_article_content, delete_article
|
||
from baoyu_optimized import WeChatPublisher
|
||
|
||
# ---------- 配置 ----------
|
||
|
||
MODEL = "deepseek-v4-flash"
|
||
NANOBOT_CONFIG = os.path.expanduser("~/.nanobot/config.json")
|
||
|
||
SYSTEM_PROMPT = """你是一个专业的医疗健康内容编辑。请将下面的文章改写为适合微信公众号发布的版本。
|
||
|
||
改写要求:
|
||
1. **第一行必须是标题**: 以 `# ` 开头。⚠️ **必须重新创作标题，严禁照搬原标题原句**。可以换表述、调语序、换措辞，但必须保持原文核心含义不变（不曲解、不偏离主旨）。标题中禁止使用任何emoji表情符号
|
||
2. **保持医疗准确性**: 不得改变任何医学事实、数据、诊断标准、用药建议
|
||
3. **风格口语化**: 用亲近自然的语气，像医生朋友在聊天。正文适当使用emoji（每段不超过1个），但标题禁止emoji
|
||
4. **保留结构**: 保留原文的章节标题和信息层次，但标题可以重新提炼得更吸引人
|
||
5. **精简优化**: 去除重复表述，合并相近内容，让文章更紧凑
|
||
6. **清理无关内容**: 去掉原文作者署名、转载来源、第三方推广链接（如"相关阅读"、"更多精彩内容"、"推荐阅读"、"参考文献"等板块）
|
||
7. **适合手机阅读**: 段落短小（2-3句一段），善用小标题和列表
|
||
8. **只输出改写后的 markdown**，不要加任何解释说明
|
||
|
||
输出格式:
|
||
- 第一行: `# 改写后的标题`
|
||
- 正文: 纯 markdown，可以包含 **粗体**、无序列表 -、## 二级标题、段间空行分隔"""
|
||
|
||
|
||
# ---------- 图片生成配置 ----------
|
||
|
||
IMAGE_GEN_ENABLED = True # 是否启用配图生成
|
||
IMAGE_GEN_COUNT = 1 # 目标生成图片数量
|
||
IMAGE_STORAGE_DIR = os.path.join(PROJECT_ROOT, "storage", "admin", "images")
|
||
BANNERS_DIR = os.path.join(PROJECT_ROOT, "storage", "admin", "images", "banners")
|
||
ADS_IMAGE = os.path.join(PROJECT_ROOT, "storage", "admin", "images", "ads", "0.jpg")
|
||
IMAGE_PROMPT_SYSTEM = """你是一个图片提示词专家。根据下面给出的文章片段，生成一个适合配图的简短中文描述（用于AI生图），要求：
|
||
|
||
1. 描述要简洁具体，20-40个字
|
||
2. 风格要求：温暖、专业、治愈的视觉风格
|
||
3. 内容要与上下文紧密相关
|
||
4. 只输出描述文字，不要加任何解释或前缀"""
|
||
|
||
# 混元生图 (TokenHub MAS API)
|
||
HUNYUAN_API_KEY = "sk-qgPNJwuOGji7I7ixqNbbkWaTIvF1MrYTRWQnEZ3ZRVU4u7y3"
|
||
HUNYUAN_API_BASE = "https://tokenhub.tencentmaas.com/v1/api/image"
|
||
HUNYUAN_MODEL = "hy-image-v3.0"
|
||
HUNYUAN_LITE_MODEL = "hy-image-lite"
|
||
HUNYUAN_LITE_ENDPOINT = "https://tokenhub.tencentmaas.com/v1/api/image/lite"
|
||
|
||
# Seedream（降级）
|
||
SEEDREAM_API_KEY = "83a29955-c633-449f-97f1-f49b62295cf6"
|
||
SEEDREAM_BASE_URL = "https://ark.cn-beijing.volces.com/api/v3"
|
||
SEEDREAM_MODEL = "doubao-seedream-4-5-251128"
|
||
|
||
# ---------- 工具函数 ----------
|
||
|
||
def load_deepseek_key() -> str:
|
||
"""从 nanobot 配置读取 DeepSeek API Key"""
|
||
try:
|
||
with open(NANOBOT_CONFIG) as f:
|
||
config = json.load(f)
|
||
key = config["providers"]["deepseek"]["apiKey"]
|
||
if not key:
|
||
raise ValueError("DeepSeek API key 为空")
|
||
return key
|
||
except Exception as e:
|
||
print(f"❌ 读取 DeepSeek API Key 失败: {e}", file=sys.stderr)
|
||
print(f" 配置文件: {NANOBOT_CONFIG}", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
|
||
def call_deepseek(markdown: str, api_key: str) -> str:
|
||
"""调用 DeepSeek API 改写文章"""
|
||
try:
|
||
from openai import OpenAI
|
||
except ImportError:
|
||
print("❌ 需要安装 openai: pip install openai", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
client = OpenAI(api_key=api_key, base_url="https://api.deepseek.com")
|
||
|
||
print("⏳ 调 DeepSeek 改写中...", file=sys.stderr)
|
||
response = client.chat.completions.create(
|
||
model=MODEL,
|
||
messages=[
|
||
{"role": "system", "content": SYSTEM_PROMPT},
|
||
{"role": "user", "content": markdown},
|
||
],
|
||
temperature=0.7,
|
||
max_tokens=4096,
|
||
)
|
||
result = response.choices[0].message.content.strip()
|
||
tokens = response.usage.total_tokens if response.usage else "?"
|
||
print(f" 消耗 {tokens} tokens", file=sys.stderr)
|
||
return result
|
||
|
||
|
||
def resolve_cover_path(cover_url: Optional[str]) -> Optional[str]:
|
||
"""将 CMS 封面 URL 转为本地绝对路径"""
|
||
if not cover_url:
|
||
return None
|
||
# cover_url 格式: /storage/admin/images/202605/xxx.jpg
|
||
path = os.path.join(PROJECT_ROOT, "..", cover_url.lstrip("/"))
|
||
# 某些情况下 cover_url 可能不带前导 /
|
||
alt_path = os.path.join(PROJECT_ROOT, "storage", "admin", "images",
|
||
*cover_url.split("/")[-2:])
|
||
for p in [path, alt_path]:
|
||
if os.path.exists(p):
|
||
return os.path.normpath(p)
|
||
print(f"⚠️ 封面图未找到: {cover_url}", file=sys.stderr)
|
||
return None
|
||
|
||
|
||
# ---------- 标题提取 ----------
|
||
|
||
def _extract_title(markdown: str) -> tuple[str, str]:
|
||
"""从 markdown 首行提取 `# 标题`，返回 (标题, 去除标题后的正文)"""
|
||
lines = markdown.split("\n")
|
||
for i, line in enumerate(lines):
|
||
stripped = line.strip()
|
||
if stripped.startswith("# ") and not stripped.startswith("## "):
|
||
title = stripped[2:].strip()
|
||
# 去掉标题行，保留后续内容
|
||
remaining = "\n".join(lines[i + 1:]).strip()
|
||
return title, remaining
|
||
# 没有找到标题，返回空标题 + 原文
|
||
return "", markdown
|
||
|
||
|
||
# ---------- emoji 清理 ----------
|
||
|
||
def _strip_emoji(text: str) -> str:
|
||
"""移除文本中的 emoji 表情符号，用于清理标题。
|
||
|
||
策略：先移除非 BMP 字符（覆盖绝大多数现代 emoji），
|
||
再清理少量 BMP 区 emoji 符号。
|
||
保留 ⚠️、①②、℃ 等实用符号不被误删。
|
||
"""
|
||
# 1. 移除非 BMP 字符（Supplementary Plane，覆盖 😀🚀🎯 等几乎全部多彩 emoji）
|
||
text = re.sub(r'[^\u0000-\uFFFF]+', '', text)
|
||
|
||
# 2. 移除 BMP 区的 emoji/装饰符号
|
||
emoji_pattern = re.compile(
|
||
"["
|
||
"\U00002702-\U000027B0" # Dingbats (✂-➰)
|
||
"\U00002194-\U00002199" # Arrows (↔↕↖↗↘↙)
|
||
"\U00002B05-\U00002B07" # More arrows (⬅⬆⬇)
|
||
"\U00002934-\U00002935" # Curved arrows (⤴⤵)
|
||
"\U000025AA-\U000025AB" # Small squares (▪▫)
|
||
"\U000025B6" # Play ▶
|
||
"\U000025C0" # Reverse ◀
|
||
"\U000025FB-\U000025FE" # Medium/white squares (◻◼◽◾)
|
||
"\U00002614-\U00002615" # Umbrella/Rain (☔☕)
|
||
"\U00002648-\U00002653" # Zodiac etc (♈-♓)
|
||
"\U000026AA-\U000026AB" # Circles (⚪⚫)
|
||
"\U000026BD-\U000026BE" # Sports (⚽⚾)
|
||
"\U000026C4-\U000026C5" # Snowman/No clouds (⛄⛅)
|
||
"\U000026CE" # Ophiuchus (⛎)
|
||
"\U000026D4" # No entry (⛔)
|
||
"\U000026EA" # Church (⛪)
|
||
"\U000026F2-\U000026F3" # Fountain/Flag (⛲⛳)
|
||
"\U000026F5" # Sailboat (⛵)
|
||
"\U000026FA" # Tent (⛺)
|
||
"\U000026FD" # Fuel pump (⛽)
|
||
"\U00002714" # Checkmark (✔)
|
||
"\U00002716" # Cross (✖)
|
||
"\U00002728" # Sparkles (✨)
|
||
"\U00002733-\U00002734" # Star-tip (✳✴)
|
||
"\U00002744-\U00002745" # Snowflake (❄❅)
|
||
"\U00002747" # Sparkle (❇)
|
||
"\U00002757" # Cross mark (❌)
|
||
"\U00002763-\U00002764" # Heart (❣❤)
|
||
"\U00002795-\U00002797" # Plus/Minus/Divide (➕➖➗)
|
||
"\U00002B1B-\U00002B1C" # Large squares (⬛⬜)
|
||
"\U00002B50" # Star ⭐
|
||
"\U00002B55" # Circle ⭕
|
||
"\U00003030" # Wavy dash 〰
|
||
"\U0000303D" # Part alternation mark 〽
|
||
"\U00003297" # Congrats ㊗
|
||
"\U00003299" # Secret ㊙
|
||
"\U0000FE00-\U0000FE0F" # Variation Selectors
|
||
"\U0000200D" # ZWJ
|
||
"]+", flags=re.UNICODE)
|
||
return emoji_pattern.sub('', text).strip()
|
||
|
||
|
||
# ---------- 封面图 ----------
|
||
|
||
def _random_banner() -> Optional[str]:
|
||
"""从 banners 目录随机选一张图片，返回绝对路径"""
|
||
if not os.path.isdir(BANNERS_DIR):
|
||
return None
|
||
images = [f for f in os.listdir(BANNERS_DIR)
|
||
if f.lower().endswith(('.jpg', '.jpeg', '.png'))]
|
||
if not images:
|
||
return None
|
||
chosen = random.choice(images)
|
||
path = os.path.join(BANNERS_DIR, chosen)
|
||
print(f" 🎲 随机封面: {chosen}", file=sys.stderr)
|
||
return path
|
||
|
||
|
||
# ---------- 内容清理 ----------
|
||
|
||
def _clean_content(markdown: str) -> str:
|
||
"""清理第三方链接、作者署名等无关内容"""
|
||
lines = markdown.split("\n")
|
||
cleaned = []
|
||
skip_until_section_end = False
|
||
|
||
for line in lines:
|
||
stripped = line.strip().lower()
|
||
|
||
# 检测需要跳过的板块标题
|
||
if re.match(r'^#{1,3}\s*(相关阅读|更多精彩|推荐阅读|延伸阅读|参考文献|参考资料|热门推荐|往期回顾|猜你喜欢|广告)', stripped):
|
||
skip_until_section_end = True
|
||
continue
|
||
|
||
# 板块结束（遇到新标题 或 空行不再累加）
|
||
if skip_until_section_end:
|
||
if stripped.startswith("#") and not re.match(r'^#{1,3}\s*(相关阅读|更多精彩|推荐阅读|延伸阅读|参考文献|参考资料|热门推荐|往期回顾|猜你喜欢|广告)', stripped):
|
||
skip_until_section_end = False
|
||
cleaned.append(line)
|
||
continue
|
||
|
||
# 跳过带来源/作者标记的行（行首）
|
||
if re.match(r'^(原文作者|作者[：:]|来源[：:]|转载自|文[／/]\s*|编辑[：:]|排版[：:]|审核[：:])', stripped):
|
||
continue
|
||
|
||
# 跳过含作者+格式的元数据行（如 "尿酸痛风 作者: 刘松 格式: Markdown"）
|
||
if re.search(r'作者[：:]\s*\S+', stripped) and re.search(r'格式[：:]', stripped):
|
||
continue
|
||
|
||
# 跳过独立的格式声明行
|
||
if re.match(r'^格式[:：]\s*(Markdown|HTML|TXT|TEXT)', stripped):
|
||
continue
|
||
|
||
# 跳过明显的第三方链接（纯 URL 行）
|
||
if re.match(r'^https?://', stripped) and 'mp.weixin' not in stripped and 'niaodaifu' not in stripped:
|
||
continue
|
||
|
||
cleaned.append(line)
|
||
|
||
return "\n".join(cleaned)
|
||
|
||
|
||
# ---------- 内容质量检查 ----------
|
||
|
||
# LLM 占位回复检测模式（当原文无文字时 DeepSeek 会返回这类消息）
|
||
_LLM_PLACEHOLDER_PATTERNS = [
|
||
r'原文内容(全部|仅|只).*(图片|链接|无法显示)',
|
||
r'没有可改写的文本',
|
||
r'请提供.*文字',
|
||
r'请.*提供.*文章.*内容',
|
||
r'一旦获得文字内容',
|
||
r'您好，您提供的.*(图片|链接)',
|
||
r'为了生成适合',
|
||
r'无法生成适合',
|
||
r'请确保文章包含文字',
|
||
]
|
||
|
||
_MIN_TEXT_LENGTH = 200 # 原文纯文本低于此值视为无实质内容
|
||
|
||
|
||
def _strip_images(markdown: str) -> str:
|
||
"""移除 markdown 图片语法，返回纯文本"""
|
||
return re.sub(r'!\[.*?\]\([^)]+\)', '', markdown)
|
||
|
||
|
||
def _check_original_content(content: str, article_id: int) -> int:
|
||
"""
|
||
检查原文是否有足够的文字内容。
|
||
返回纯文本字符数；低于 _MIN_TEXT_LENGTH 删除文章并抛出 ValueError。
|
||
"""
|
||
text = _strip_images(content)
|
||
# 也移除 markdown 链接、HTML 标签等
|
||
text = re.sub(r'\[([^\]]*)\]\([^)]+\)', r'\1', text) # 链接文字保留
|
||
text = re.sub(r'<[^>]+>', '', text) # HTML 标签
|
||
text = re.sub(r'\s+', ' ', text).strip()
|
||
text_len = len(text)
|
||
if text_len < _MIN_TEXT_LENGTH:
|
||
delete_article(article_id)
|
||
print(f" 🗑️ 已删除短文章 ID={article_id}（{text_len} 字符）", file=sys.stderr)
|
||
raise ValueError(
|
||
f"原文文字内容不足（{text_len} 字符 < {_MIN_TEXT_LENGTH}），"
|
||
f"可能全部为图片，已删除"
|
||
)
|
||
return text_len
|
||
|
||
|
||
def _check_llm_response(content: str, article_id: int) -> None:
|
||
"""
|
||
检查 LLM 返回是否是无意义的占位回复。
|
||
如果是占位回复则抛出 ValueError。
|
||
"""
|
||
for pattern in _LLM_PLACEHOLDER_PATTERNS:
|
||
if re.search(pattern, content):
|
||
delete_article(article_id)
|
||
print(f" 🗑️ 已删除占位回复文章 ID={article_id}", file=sys.stderr)
|
||
raise ValueError(
|
||
f"LLM 返回疑似占位回复（匹配模式: {pattern}），已删除"
|
||
)
|
||
# 额外检查：改写后内容过短（< 200 字）也视为无效
|
||
if len(content) < _MIN_TEXT_LENGTH:
|
||
delete_article(article_id)
|
||
print(f" 🗑️ 已删除改写后过短文章 ID={article_id}（{len(content)} 字符）", file=sys.stderr)
|
||
raise ValueError(
|
||
f"改写后内容过短（{len(content)} 字符 < {_MIN_TEXT_LENGTH}），已删除"
|
||
)
|
||
|
||
|
||
# ---------- 正文末尾广告图 ----------
|
||
|
||
def _append_ads_image(markdown: str) -> str:
|
||
"""在文章末尾添加固定广告图"""
|
||
if not os.path.exists(ADS_IMAGE):
|
||
print(f" ⚠️ 广告图不存在: {ADS_IMAGE}", file=sys.stderr)
|
||
return markdown
|
||
web_path = "/storage/admin/images/ads/0.jpg"
|
||
print(f" 📢 追加广告图: {web_path}", file=sys.stderr)
|
||
return markdown.rstrip() + f"\n\n\n"
|
||
|
||
|
||
# ---------- 图片生成 ----------
|
||
|
||
def _make_image_prompt(context_text: str, api_key: str) -> str:
|
||
"""用 LLM 从文章片段生成图片 prompt"""
|
||
try:
|
||
from openai import OpenAI
|
||
except ImportError:
|
||
print("⚠️ openai 未安装，使用默认图片 prompt", file=sys.stderr)
|
||
return "温暖的医疗健康主题插图，简洁专业的扁平设计风格"
|
||
|
||
client = OpenAI(api_key=api_key, base_url="https://api.deepseek.com")
|
||
try:
|
||
response = client.chat.completions.create(
|
||
model=MODEL,
|
||
messages=[
|
||
{"role": "system", "content": IMAGE_PROMPT_SYSTEM},
|
||
{"role": "user", "content": context_text},
|
||
],
|
||
temperature=0.3,
|
||
max_tokens=200,
|
||
)
|
||
prompt = response.choices[0].message.content.strip()
|
||
print(f" 🎨 图片 prompt: {prompt}", file=sys.stderr)
|
||
return prompt
|
||
except Exception as e:
|
||
print(f" ⚠️ 图片 prompt 生成失败: {e}", file=sys.stderr)
|
||
return "温暖的医疗健康主题插图，简洁专业的扁平设计风格"
|
||
|
||
|
||
def _hunyuan_lite_image(prompt: str, output_path: str,
|
||
resolution: str = "1024:1024") -> bool:
|
||
"""腾讯混元生图 hy-image-lite (轻量同步版)
|
||
|
||
POST 一次直接返回图片 URL，无需轮询。价格更低。
|
||
注意：Lite 固定输出 1024×1024，不支持自定义分辨率。
|
||
"""
|
||
import requests
|
||
|
||
if not HUNYUAN_API_KEY:
|
||
return False
|
||
|
||
try:
|
||
payload = {
|
||
"model": HUNYUAN_LITE_MODEL,
|
||
"prompt": prompt,
|
||
"rsp_img_type": "url",
|
||
}
|
||
|
||
headers = {
|
||
"Authorization": f"Bearer {HUNYUAN_API_KEY}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
|
||
resp = requests.post(HUNYUAN_LITE_ENDPOINT, json=payload,
|
||
headers=headers, timeout=60)
|
||
resp.raise_for_status()
|
||
result = resp.json()
|
||
|
||
data = result.get("data", [])
|
||
if not data:
|
||
print(f" ⚠️ 混元 Lite 无数据: {result}", file=sys.stderr)
|
||
return False
|
||
|
||
img_url = data[0].get("url", "")
|
||
if not img_url:
|
||
print(f" ⚠️ 混元 Lite 无 URL: {data[0]}", file=sys.stderr)
|
||
return False
|
||
|
||
r = requests.get(img_url, timeout=60)
|
||
r.raise_for_status()
|
||
img_data = r.content
|
||
|
||
os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)
|
||
with open(output_path, "wb") as f:
|
||
f.write(img_data)
|
||
print(f" ✅ 混元 Lite: {os.path.basename(output_path)} ({len(img_data)} bytes)", file=sys.stderr)
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f" ⚠️ 混元 Lite 异常: {e}", file=sys.stderr)
|
||
return False
|
||
|
||
|
||
def _generate_image(prompt: str, output_path: str,
|
||
resolution: str = "1024:1024") -> bool:
|
||
"""生成封面配图：优先混元 Lite → 混元 V3.0 → Seedream"""
|
||
# 1) 混元 Lite (轻量版，更便宜，同步返回)
|
||
success = _hunyuan_lite_image(prompt, output_path, resolution)
|
||
if success:
|
||
return True
|
||
# 2) 混元 V3.0 (标准版)
|
||
print(" ↪ 降级到混元 V3.0...", file=sys.stderr)
|
||
success = _hunyuan_image(prompt, output_path, resolution)
|
||
if success:
|
||
return True
|
||
# 3) Seedream
|
||
print(" ↪ 降级到 Seedream...", file=sys.stderr)
|
||
return _seedream_image(prompt, output_path, resolution)
|
||
|
||
|
||
def _hunyuan_image(prompt: str, output_path: str,
|
||
resolution: str = "1024:1024") -> bool:
|
||
"""腾讯混元生图 hy-image-v3.0 (TokenHub MAS API)
|
||
|
||
异步提交流程: submit → poll query → download
|
||
"""
|
||
import requests
|
||
|
||
if not HUNYUAN_API_KEY:
|
||
return False
|
||
|
||
try:
|
||
# Step 1: 提交生图任务
|
||
submit_url = f"{HUNYUAN_API_BASE}/submit"
|
||
submit_payload = {
|
||
"model": HUNYUAN_MODEL,
|
||
"prompt": prompt,
|
||
"logo_add": 0, # 不添加"AI生成"水印
|
||
}
|
||
# 可选参数：尺寸
|
||
parts = resolution.replace(":", "x").split("x")
|
||
if len(parts) == 2:
|
||
try:
|
||
w, h = int(parts[0]), int(parts[1])
|
||
submit_payload["width"] = w
|
||
submit_payload["height"] = h
|
||
except ValueError:
|
||
pass
|
||
|
||
headers = {
|
||
"Authorization": f"Bearer {HUNYUAN_API_KEY}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
|
||
resp = requests.post(submit_url, json=submit_payload, headers=headers, timeout=30)
|
||
resp.raise_for_status()
|
||
result = resp.json()
|
||
|
||
task_id = result.get("id")
|
||
if not task_id:
|
||
print(f" ⚠️ 混元提交失败: {result}", file=sys.stderr)
|
||
return False
|
||
|
||
print(f" 📤 混元任务已提交: {task_id}", file=sys.stderr)
|
||
|
||
# Step 2: 轮询查询结果
|
||
query_url = f"{HUNYUAN_API_BASE}/query"
|
||
query_payload = {"model": HUNYUAN_MODEL, "id": task_id}
|
||
|
||
for _ in range(30): # 最多等 150 秒
|
||
time.sleep(5)
|
||
qr = requests.post(query_url, json=query_payload, headers=headers, timeout=30)
|
||
qr.raise_for_status()
|
||
qresult = qr.json()
|
||
|
||
status = qresult.get("status", "")
|
||
if status == "completed":
|
||
# 提取图片 URL (Tencent MAS 返回 data 数组)
|
||
data = qresult.get("data", [])
|
||
if not data:
|
||
print(f" ⚠️ 混元无图片: {qresult}", file=sys.stderr)
|
||
return False
|
||
|
||
img_obj = data[0]
|
||
img_url = img_obj.get("url", "")
|
||
img_b64 = img_obj.get("b64_json", "")
|
||
|
||
if img_url:
|
||
r = requests.get(img_url, timeout=60)
|
||
r.raise_for_status()
|
||
img_data = r.content
|
||
elif img_b64:
|
||
import base64
|
||
img_data = base64.b64decode(img_b64)
|
||
else:
|
||
print(f" ⚠️ 混元无可用图片格式: {img_obj}", file=sys.stderr)
|
||
return False
|
||
|
||
os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)
|
||
with open(output_path, "wb") as f:
|
||
f.write(img_data)
|
||
print(f" ✅ 混元生图: {os.path.basename(output_path)} ({len(img_data)} bytes)", file=sys.stderr)
|
||
return True
|
||
|
||
elif status in ("failed", "cancelled"):
|
||
print(f" ⚠️ 混元任务失败: status={status}, {qresult}", file=sys.stderr)
|
||
return False
|
||
|
||
# 其他状态 (processing, queued, etc.) — 继续等待
|
||
print(f" ⏳ 混元: {status}", file=sys.stderr)
|
||
|
||
print(f" ⚠️ 混元超时 (150s)", file=sys.stderr)
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f" ⚠️ 混元异常: {e}", file=sys.stderr)
|
||
return False
|
||
|
||
|
||
def _seedream_image(prompt: str, output_path: str,
|
||
resolution: str = "1024:1024") -> bool:
|
||
"""Seedream (火山引擎 ARK)"""
|
||
import requests
|
||
|
||
# Seedream 要求 ≥ 3,686,400 像素，自动放大
|
||
parts = resolution.replace(":", "x").split("x")
|
||
if len(parts) == 2:
|
||
try:
|
||
w, h = int(parts[0]), int(parts[1])
|
||
if w * h < 3686400:
|
||
scale = (3686400 / (w * h)) ** 0.5
|
||
w2, h2 = (int(w * scale) // 16) * 16, (int(h * scale) // 16) * 16
|
||
resolution = f"{w2}x{h2}"
|
||
except ValueError:
|
||
pass
|
||
|
||
try:
|
||
url = f"{SEEDREAM_BASE_URL}/images/generations"
|
||
resp = requests.post(
|
||
url,
|
||
headers={
|
||
"Authorization": f"Bearer {SEEDREAM_API_KEY}",
|
||
"Content-Type": "application/json",
|
||
},
|
||
json={
|
||
"model": SEEDREAM_MODEL,
|
||
"prompt": prompt,
|
||
"size": resolution,
|
||
"n": 1,
|
||
},
|
||
timeout=120,
|
||
)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
image_url = data["data"][0].get("url")
|
||
if not image_url:
|
||
return False
|
||
|
||
r = requests.get(image_url, timeout=60)
|
||
r.raise_for_status()
|
||
|
||
os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)
|
||
with open(output_path, "wb") as f:
|
||
f.write(r.content)
|
||
print(f" ✅ Seedream: {os.path.basename(output_path)} ({len(r.content)} bytes)", file=sys.stderr)
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f" ⚠️ Seedream: {e}", file=sys.stderr)
|
||
return False
|
||
|
||
|
||
def _remove_watermark(image_path: str) -> None:
|
||
"""去除混元生图右下角'AI生成'水印：OpenCV修复 + 背景覆盖 + 官网logo叠加"""
|
||
try:
|
||
import cv2
|
||
import numpy as np
|
||
from PIL import Image, ImageDraw
|
||
|
||
LOGO_PATH = os.path.join(
|
||
PROJECT_ROOT,
|
||
"storage/admin/images/setting/20260506_100405_89971b62.png",
|
||
)
|
||
if not os.path.exists(LOGO_PATH):
|
||
return
|
||
|
||
img_cv = cv2.imread(image_path)
|
||
if img_cv is None:
|
||
return
|
||
h, w = img_cv.shape[:2]
|
||
|
||
# ---- 第1步：检测水印区域并修复 ----
|
||
# 右下角扫描: 30%宽 × 25%高（扩大扫描范围以捕获完整水印）
|
||
scan_w = int(w * 0.30)
|
||
scan_h = int(h * 0.25)
|
||
corner = img_cv[h - scan_h : h, w - scan_w : w]
|
||
|
||
# 转灰度，用大核高斯模糊做背景估计，diff 找出局部异常亮区
|
||
gray = cv2.cvtColor(corner, cv2.COLOR_BGR2GRAY)
|
||
blurred = cv2.GaussianBlur(gray, (51, 51), 0)
|
||
diff = cv2.subtract(gray, blurred)
|
||
# 水印文字比周围亮，自适应阈值
|
||
diff_max = diff.max()
|
||
threshold = max(8, int(diff_max * 0.35))
|
||
_, wm_mask = cv2.threshold(diff, threshold, 255, cv2.THRESH_BINARY)
|
||
|
||
# 膨胀连接文字笔画，扩大修复区域
|
||
kernel = np.ones((5, 5), np.uint8)
|
||
wm_mask = cv2.dilate(wm_mask, kernel, iterations=2)
|
||
|
||
# 补充右下角最小覆盖区（兜底：混元水印始终在右下角）
|
||
fixed_mask = np.zeros((scan_h, scan_w), dtype=np.uint8)
|
||
fixed_mask[int(scan_h * 0.4):, int(scan_w * 0.6):] = 255
|
||
wm_mask = cv2.bitwise_or(wm_mask, fixed_mask)
|
||
|
||
wm_pixels = wm_mask.sum() // 255
|
||
# 记录水印在扫描区中的边界框（全图坐标）
|
||
ys, xs = np.where(wm_mask > 0)
|
||
wm_left = w - scan_w + xs.min()
|
||
wm_top = h - scan_h + ys.min()
|
||
wm_right = w - scan_w + xs.max()
|
||
wm_bottom = h - scan_h + ys.max()
|
||
|
||
if wm_pixels > 200:
|
||
# 扩展到全图做修复（加大 inpaint 半径）
|
||
full_mask = np.zeros((h, w), dtype=np.uint8)
|
||
full_mask[h - scan_h : h, w - scan_w : w] = wm_mask
|
||
img_cv = cv2.inpaint(img_cv, full_mask, 5, cv2.INPAINT_TELEA)
|
||
print(f" 🎨 水印修复完成 ({wm_pixels} 像素)", file=sys.stderr)
|
||
else:
|
||
# 未检测到水印时，使用默认右下角区域
|
||
wm_left = int(w * 0.70)
|
||
wm_top = int(h * 0.75)
|
||
wm_right = w
|
||
wm_bottom = h
|
||
print(f" ℹ️ 未检测到明确水印，使用默认右下角覆盖", file=sys.stderr)
|
||
|
||
cv2.imwrite(image_path, img_cv, [cv2.IMWRITE_JPEG_QUALITY, 95])
|
||
|
||
# ---- 第2步：背景覆盖 + 叠加官网logo ----
|
||
img_pil = Image.open(image_path).convert("RGBA")
|
||
logo = Image.open(LOGO_PATH).convert("RGBA")
|
||
|
||
# 计算覆盖区域：从水印左上角延伸到图片右下角，加 padding
|
||
padding = int(min(w, h) * 0.015)
|
||
patch_left = max(0, wm_left - padding)
|
||
patch_top = max(0, wm_top - padding)
|
||
patch_right = w # 延伸到右边缘
|
||
patch_bottom = h # 延伸到底边缘
|
||
patch_w = patch_right - patch_left
|
||
patch_h = patch_bottom - patch_top
|
||
|
||
# 创建白色圆角背景（高不透明度确保完全覆盖水印）
|
||
patch = Image.new("RGBA", (patch_w, patch_h), (0, 0, 0, 0))
|
||
draw = ImageDraw.Draw(patch)
|
||
radius = max(10, int(min(patch_w, patch_h) * 0.04))
|
||
draw.rounded_rectangle(
|
||
[(0, 0), (patch_w - 1, patch_h - 1)],
|
||
radius=radius,
|
||
fill=(255, 255, 255, 245),
|
||
)
|
||
img_pil.paste(patch, (patch_left, patch_top), patch)
|
||
|
||
# 官网logo大小：按 patch 短边的 75% 等比缩放
|
||
logo_h = max(int(min(patch_w, patch_h) * 0.75), 100)
|
||
logo_w = int(logo_h * logo.size[0] / logo.size[1])
|
||
logo_r = logo.resize((logo_w, logo_h), Image.LANCZOS)
|
||
|
||
# Logo 居中放置在覆盖区域
|
||
logo_x = patch_left + (patch_w - logo_w) // 2
|
||
logo_y = patch_top + (patch_h - logo_h) // 2
|
||
img_pil.paste(logo_r, (logo_x, logo_y), logo_r)
|
||
|
||
img_pil = img_pil.convert("RGB")
|
||
img_pil.save(image_path, quality=95)
|
||
print(
|
||
f" 🏷️ 官网logo叠加完成 (覆盖{scan_w}x{scan_h}→{patch_w}x{patch_h}, "
|
||
f"logo:{logo_w}x{logo_h})",
|
||
file=sys.stderr,
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"去水印失败 ({image_path}): {e}")
|
||
print(f" ⚠️ 去水印失败: {e}", file=sys.stderr)
|
||
|
||
|
||
def _find_image_spots(markdown: str, count: int = 2) -> list:
|
||
"""在 markdown 中找适合插入图片的位置，返回 [(插入点索引, 上下文), ...]"""
|
||
lines = markdown.split("\n")
|
||
candidates = []
|
||
|
||
# 找段落间的空行位置作为候选
|
||
for i in range(1, len(lines) - 1):
|
||
# 空行 + 下一行有内容 → 段落边界
|
||
if lines[i].strip() == "" and i + 1 < len(lines) and lines[i + 1].strip():
|
||
# 确保不在标题前插入（标题通常以 ## 开头）
|
||
if not lines[i + 1].strip().startswith("#"):
|
||
# 取前后各 3 行作为上下文
|
||
ctx_start = max(0, i - 3)
|
||
ctx_end = min(len(lines), i + 4)
|
||
context = "\n".join(lines[ctx_start:ctx_end])
|
||
candidates.append((i + 1, context)) # 在空行后插入
|
||
|
||
# 均匀选取 count 个点
|
||
if not candidates:
|
||
return []
|
||
|
||
if len(candidates) <= count:
|
||
return candidates
|
||
|
||
# 均匀分布：从中间选取
|
||
step = len(candidates) // count
|
||
selected = [candidates[step * j] for j in range(count)]
|
||
return selected
|
||
|
||
|
||
def _insert_images(markdown: str, api_key: str) -> tuple[str, list[str]]:
|
||
"""分析 markdown、生成配图并插入，返回 (新 markdown, 生成的图片 web_path 列表)"""
|
||
if not IMAGE_GEN_ENABLED:
|
||
return markdown, []
|
||
|
||
spots = _find_image_spots(markdown, IMAGE_GEN_COUNT)
|
||
if not spots:
|
||
print(" ℹ️ 未找到合适的图片插入位置", file=sys.stderr)
|
||
return markdown, []
|
||
|
||
print(f" 🖼️ 计划插入 {len(spots)} 张配图", file=sys.stderr)
|
||
|
||
lines = markdown.split("\n")
|
||
generated: list[tuple[int, str]] = [] # (line_index, image_path)
|
||
|
||
for idx, (line_idx, context) in enumerate(spots):
|
||
# 生成图片 prompt
|
||
img_prompt = _make_image_prompt(context, api_key)
|
||
|
||
# 输出路径：storage/admin/images/YYYYMM/
|
||
month_dir = time.strftime("%Y%m")
|
||
filename = f"ai_gen_{int(time.time())}_{idx}.jpg"
|
||
output_dir = os.path.join(IMAGE_STORAGE_DIR, month_dir)
|
||
output_path = os.path.join(output_dir, filename)
|
||
|
||
ok = _generate_image(img_prompt, output_path)
|
||
if ok:
|
||
# logo_add=0 已消除水印，无需后处理覆盖
|
||
# _remove_watermark(output_path)
|
||
# 相对路径用于 markdown
|
||
web_path = f"/storage/admin/images/{month_dir}/{filename}"
|
||
generated.append((line_idx, web_path))
|
||
else:
|
||
print(f" ❌ 图片 {idx+1} 生成失败，跳过", file=sys.stderr)
|
||
|
||
if not generated:
|
||
return markdown, []
|
||
|
||
# 从后往前插入（避免索引偏移）
|
||
for line_idx, web_path in sorted(generated, key=lambda x: x[0], reverse=True):
|
||
img_md = f"\n\n"
|
||
lines.insert(line_idx, img_md)
|
||
|
||
# 提取生成的图片路径列表（按插入顺序）
|
||
generated_paths = [web_path for _, web_path in sorted(generated, key=lambda x: x[0])]
|
||
return "\n".join(lines), generated_paths
|
||
|
||
|
||
# ---------- 微信图片上传 ----------
|
||
|
||
def _upload_html_images(html: str, publisher) -> str:
|
||
"""找出 HTML 中以 /storage/ 开头的本地图片，上传到微信并替换为 CDN 地址"""
|
||
import re
|
||
from html import unescape
|
||
|
||
# 匹配 <img src="..." ...> 标签，提取 src 和整个标签
|
||
img_pattern = re.compile(r'<img[^>]+src=["\']([^"\']+)["\'][^>]*>', re.IGNORECASE)
|
||
|
||
storage_base = PROJECT_ROOT
|
||
|
||
def upload_and_replace(match):
|
||
full_tag = match.group(0)
|
||
src = match.group(1)
|
||
|
||
# 只处理以 /storage/ 开头的本地路径
|
||
if not src.startswith("/storage/"):
|
||
return full_tag
|
||
|
||
local_path = os.path.normpath(os.path.join(storage_base, src.lstrip("/")))
|
||
if not os.path.exists(local_path):
|
||
print(f" ⚠️ 图片不存在: {local_path}", file=sys.stderr)
|
||
return full_tag
|
||
|
||
try:
|
||
resp = publisher.upload_image(local_path, upload_type="body")
|
||
wechat_url = resp.get("url", "")
|
||
if wechat_url:
|
||
print(f" ✅ 已上传到微信: {os.path.basename(src)}", file=sys.stderr)
|
||
# 替换整个 img 标签，只改 src
|
||
return full_tag.replace(src, wechat_url)
|
||
else:
|
||
print(f" ❌ 微信未返回URL: {os.path.basename(src)}", file=sys.stderr)
|
||
return full_tag
|
||
except Exception as e:
|
||
print(f" ❌ 上传失败 {os.path.basename(src)}: {e}", file=sys.stderr)
|
||
return full_tag
|
||
|
||
result = img_pattern.sub(upload_and_replace, html)
|
||
return result
|
||
|
||
|
||
# ---------- 核心流程 ----------
|
||
|
||
def run_pipeline(
|
||
article_id: int,
|
||
rewritten_content: Optional[str] = None,
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
一键改写→发布流程。
|
||
|
||
Args:
|
||
article_id: CMS 文章 ID
|
||
rewritten_content: 预改写好的 markdown。None 则自动调 LLM
|
||
|
||
Returns:
|
||
{"article_id": ..., "title": ..., "media_id": ..., "db_updated": bool}
|
||
"""
|
||
# 1. 查文章
|
||
article = get_article_full(article_id)
|
||
if not article:
|
||
raise ValueError(f"文章 ID={article_id} 不存在")
|
||
|
||
title = article["title"]
|
||
author = "尿大夫" # 改写后统一使用尿大夫署名
|
||
digest = article.get("summary") or ""
|
||
original = article["content"]
|
||
|
||
print(f"📄 文章: ID={article_id} | {title}", file=sys.stderr)
|
||
print(f" 原文字数: {len(original)} chars", file=sys.stderr)
|
||
|
||
# 2. 改写（两种模式）
|
||
api_key = None # will be set in auto mode
|
||
if rewritten_content:
|
||
print(" 模式: 手动改写（跳过 LLM）", file=sys.stderr)
|
||
new_content = rewritten_content
|
||
else:
|
||
# 自动改写前先检查原文内容质量
|
||
text_len = _check_original_content(original, article_id)
|
||
print(f" 原文有效文字: {text_len} chars", file=sys.stderr)
|
||
print(" 模式: 自动改写（DeepSeek）", file=sys.stderr)
|
||
api_key = load_deepseek_key()
|
||
new_content = call_deepseek(original, api_key)
|
||
# 检查 LLM 返回是否有效
|
||
_check_llm_response(new_content, article_id)
|
||
|
||
print(f" 改写后字数: {len(new_content)} chars", file=sys.stderr)
|
||
|
||
# 2.1 提取改写后的标题（# 标题）+ 去emoji
|
||
rewritten_title, new_content = _extract_title(new_content)
|
||
if rewritten_title:
|
||
print(f" ✏️ 改写标题: {rewritten_title}", file=sys.stderr)
|
||
title = _strip_emoji(rewritten_title)
|
||
if title != rewritten_title:
|
||
print(f" 🧹 去emoji后: {title}", file=sys.stderr)
|
||
else:
|
||
print(f" ℹ️ 未检测到改写标题，保留原标题", file=sys.stderr)
|
||
# 兜底：原标题也去emoji
|
||
title = _strip_emoji(title)
|
||
|
||
# 2.2 清理第三方链接/作者署名
|
||
new_content = _clean_content(new_content)
|
||
print(f" 清理后字数: {len(new_content)} chars", file=sys.stderr)
|
||
|
||
# 2.3 生成配图并插入
|
||
print("🎨 生成配图...", file=sys.stderr)
|
||
if rewritten_content:
|
||
new_content, gen_image_paths = _insert_images(new_content, load_deepseek_key())
|
||
else:
|
||
new_content, gen_image_paths = _insert_images(new_content, api_key)
|
||
|
||
# 2.4 末尾添加广告图
|
||
new_content = _append_ads_image(new_content)
|
||
|
||
# 3. 转 HTML
|
||
publisher = WeChatPublisher()
|
||
html_content = publisher.markdown_to_html(new_content)
|
||
|
||
# 3.5 上传 HTML 中的本地图片到微信（将 /storage/... 转为微信 CDN 地址）
|
||
html_content = _upload_html_images(html_content, publisher)
|
||
|
||
# 4. 封面图：优先用 AI 生成的配图（已去水印+叠加logo），再降级到 banners/原封面
|
||
cover_path = None
|
||
cover_web_path = None
|
||
if gen_image_paths:
|
||
# 解析第一张AI生成图的本地路径
|
||
first_web = gen_image_paths[0]
|
||
cover_path = os.path.normpath(os.path.join(PROJECT_ROOT, first_web.lstrip("/")))
|
||
cover_web_path = first_web
|
||
print(f" 🎨 AI配图作为封面: {first_web}", file=sys.stderr)
|
||
if not cover_path or not os.path.exists(cover_path):
|
||
cover_web_path = None # 本地不存在，不更新DB
|
||
cover_path = _random_banner()
|
||
if not cover_path:
|
||
print(" ℹ️ banners 目录无图，尝试用文章封面", file=sys.stderr)
|
||
cover_url = article.get("cover_image") or ""
|
||
cover_path = resolve_cover_path(cover_url)
|
||
|
||
# 5. 创建公众号草稿
|
||
print("📝 创建公众号草稿...", file=sys.stderr)
|
||
media_id = publisher.create_draft(
|
||
title=title,
|
||
content=html_content,
|
||
author=author,
|
||
digest=digest,
|
||
cover_image_path=cover_path,
|
||
)
|
||
print(f" media_id: {media_id}", file=sys.stderr)
|
||
|
||
# 5.5 根据配置决定是否自动发布
|
||
publish_mode = "publish" # 默认值
|
||
if os.path.exists(AUTO_CONFIG_PATH):
|
||
try:
|
||
with open(AUTO_CONFIG_PATH, 'r', encoding='utf-8') as f:
|
||
auto_cfg = yaml.safe_load(f)
|
||
publish_mode = auto_cfg.get("publish_mode", "publish")
|
||
except Exception:
|
||
pass
|
||
|
||
if publish_mode == "publish":
|
||
print("📢 自动发布草稿...", file=sys.stderr)
|
||
try:
|
||
publisher.publish_draft(media_id)
|
||
print(" ✅ 已提交发布", file=sys.stderr)
|
||
except Exception as e:
|
||
print(f" ⚠️ 发布失败: {e}", file=sys.stderr)
|
||
else:
|
||
print(" 📋 草稿模式: 跳过自动发布，请手动确认", file=sys.stderr)
|
||
|
||
# 6. 更新数据库（含标题、作者署名、封面图）
|
||
ok = update_article_content(
|
||
article_id, new_content,
|
||
status="published",
|
||
new_title=title,
|
||
new_author=author,
|
||
new_cover_image=cover_web_path,
|
||
)
|
||
print(f"{'✅' if ok else '❌'} 数据库更新: {'成功' if ok else '失败'}", file=sys.stderr)
|
||
|
||
return {
|
||
"article_id": article_id,
|
||
"title": title,
|
||
"media_id": media_id,
|
||
"db_updated": ok,
|
||
}
|
||
|
||
|
||
# ---------- CLI ----------
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(
|
||
description="CMS 草稿改写 → 公众号发布 一键流水线"
|
||
)
|
||
parser.add_argument("article_id", type=int, help="CMS 文章 ID")
|
||
parser.add_argument(
|
||
"--content-file", "-f",
|
||
help="改写好的 markdown 文件路径。不传则自动调 DeepSeek 改写"
|
||
)
|
||
parser.add_argument(
|
||
"--dry-run", action="store_true",
|
||
help="只改写不发布，输出改写结果到 stdout"
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
# 读取改写内容
|
||
rewritten = None
|
||
if args.content_file:
|
||
with open(args.content_file) as f:
|
||
rewritten = f.read()
|
||
|
||
# 干跑模式：只改写输出
|
||
if args.dry_run:
|
||
article = get_article_full(args.article_id)
|
||
if not article:
|
||
print(f"❌ 文章 ID={args.article_id} 不存在", file=sys.stderr)
|
||
sys.exit(1)
|
||
api_key = load_deepseek_key()
|
||
result = call_deepseek(article["content"], api_key)
|
||
print(result)
|
||
return
|
||
|
||
# 正式执行
|
||
try:
|
||
result = run_pipeline(args.article_id, rewritten)
|
||
print(json.dumps(result, ensure_ascii=False, indent=2), flush=True)
|
||
except Exception as e:
|
||
print(f"❌ 流水线失败: {e}", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|