📝 docs(project): 添加开源社区标准文档与 CI 工作流
CI / Lint (ruff) (push) Has been cancelled
CI / Import Check (push) Has been cancelled

- 新增 GitHub Issue 模板(Bug 报告、功能请求)和 Pull Request 模板
- 新增 Code of Conduct(贡献者行为准则)和 Security Policy(安全政策)
- 新增 CI 工作流(GitHub Actions),包含 ruff 代码检查和导入验证
- 新增开发依赖文件 requirements-dev.txt

📦 build(ci): 配置 GitHub Actions 持续集成

- 在 push 到 main 分支和 pull request 时自动触发 CI
- 添加 lint 任务执行 ruff 代码风格检查
- 添加 import-check 任务验证核心服务模块导入

♻️ refactor(structure): 重构项目目录结构

- 将根目录的 6 个服务模块迁移至 services/ 包
- 更新所有相关文件的导入语句(main.py、ui/、services/)
- 根目录仅保留 main.py 作为唯一 Python 入口文件

🔧 chore(config): 调整配置和资源文件路径

- 将 config.json 移至 config/ 目录,更新相关引用
- 将个人头像图片移至 assets/faces/ 目录,更新 .gitignore
- 更新 Dockerfile 和 docker-compose.yml 中的配置路径

📝 docs(readme): 完善 README 文档

- 添加项目状态徽章(Python 版本、License、CI)
- 更新项目结构图反映实际目录布局
- 修正使用指南中的 Tab 名称和操作路径
- 替换 your-username 占位符为格式提示

🗑️ chore(cleanup): 清理冗余文件

- 删除旧版备份文件、测试脚本、临时记录和运行日志
- 删除散落的个人图片文件(已归档至 assets/faces/)
This commit is contained in:
2026-02-27 22:12:39 +08:00
parent b5deafa2cc
commit 2ba87c8f6e
58 changed files with 1119 additions and 468 deletions
+665
View File
@@ -0,0 +1,665 @@
"""
笔记数据分析 & 智能权重学习模块
定时抓取已发布笔记的互动数据,自动学习哪些内容受欢迎,生成加权主题池
"""
import json
import os
import re
import tempfile
import time
import logging
import math
from datetime import datetime, timedelta
from collections import defaultdict
logger = logging.getLogger(__name__)
ANALYTICS_FILE = "analytics_data.json"
WEIGHTS_FILE = "content_weights.json"
def _safe_int(val) -> int:
"""将 '1.2万' / '1234' / 1234 等格式转为整数"""
if isinstance(val, (int, float)):
return int(val)
if not val:
return 0
s = str(val).strip()
if "万" in s:
try:
return int(float(s.replace("万", "")) * 10000)
except ValueError:
return 0
try:
return int(float(s))
except ValueError:
return 0
class AnalyticsService:
"""笔记表现分析 & 权重学习引擎"""
def __init__(self, workspace_dir: str = "xhs_workspace"):
self.workspace_dir = workspace_dir
self.analytics_path = os.path.join(workspace_dir, ANALYTICS_FILE)
self.weights_path = os.path.join(workspace_dir, WEIGHTS_FILE)
self._analytics_data = self._load_json(self.analytics_path, {"notes": {}, "last_analysis": ""})
self._weights = self._load_json(self.weights_path, {
"topic_weights": {},
"style_weights": {},
"tag_weights": {},
"title_pattern_weights": {},
"time_weights": {},
"last_updated": "",
"analysis_history": [],
})
# ========== 持久化 ==========
@staticmethod
def _load_json(path: str, default: dict) -> dict:
if os.path.exists(path):
try:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except (json.JSONDecodeError, IOError) as e:
logger.warning("加载 %s 失败: %s,使用默认值", path, e)
return default.copy()
def _save_analytics(self):
os.makedirs(self.workspace_dir, exist_ok=True)
target = self.analytics_path
target_dir = os.path.dirname(os.path.abspath(target))
fd, tmp = tempfile.mkstemp(dir=target_dir, suffix=".tmp", prefix="analytics_")
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(self._analytics_data, f, ensure_ascii=False, indent=2)
os.replace(tmp, target)
except Exception:
try:
os.remove(tmp)
except OSError:
pass
raise
def _save_weights(self):
os.makedirs(self.workspace_dir, exist_ok=True)
target = self.weights_path
target_dir = os.path.dirname(os.path.abspath(target))
fd, tmp = tempfile.mkstemp(dir=target_dir, suffix=".tmp", prefix="weights_")
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(self._weights, f, ensure_ascii=False, indent=2)
os.replace(tmp, target)
except Exception:
try:
os.remove(tmp)
except OSError:
pass
raise
# ========== 数据采集 ==========
def collect_note_performance(self, mcp_client, user_id: str, xsec_token: str) -> dict:
"""
通过 MCP 获取我的所有笔记及其互动数据,存入 analytics_data.json
返回 {"total": N, "updated": M, "notes": [...]}
"""
logger.info("开始采集笔记表现数据 (user_id=%s)", user_id)
raw = mcp_client.get_user_profile(user_id, xsec_token)
text = ""
if isinstance(raw, dict):
# _call_tool 返回 {"success": True, "text": "...", "raw": <mcp原始响应>}
# 优先从 raw["raw"]["content"] 提取,兼容直接 content
inner_raw = raw.get("raw", {})
content_list = []
if isinstance(inner_raw, dict):
content_list = inner_raw.get("content", [])
if not content_list:
content_list = raw.get("content", [])
for item in content_list:
if isinstance(item, dict) and item.get("type") == "text":
text = item.get("text", "")
break
if not text:
text = raw.get("text", "")
# 解析 JSON
data = None
for attempt_fn in [
lambda t: json.loads(t),
lambda t: json.loads(re.search(r'```(?:json)?\s*\n([\s\S]+?)\n```', t).group(1)),
lambda t: json.loads(re.search(r'(\{[\s\S]*\})', t).group(1)),
]:
try:
data = attempt_fn(text)
if data:
break
except Exception:
continue
if not data:
return {"total": 0, "updated": 0, "error": "无法解析用户数据"}
feeds = data.get("feeds", [])
if not feeds:
return {"total": 0, "updated": 0, "error": "未找到笔记数据"}
notes_dict = self._analytics_data.get("notes", {})
updated = 0
note_summaries = []
for f in feeds:
nc = f.get("noteCard") or {}
# MCP 用户主页 feeds 中,笔记 ID 在 f["id"] 而非 nc["noteId"]
note_id = nc.get("noteId") or f.get("id", "") or f.get("noteId", "")
if not note_id:
logger.warning("跳过无 ID 的笔记条目: keys=%s", list(f.keys()))
continue
interact = nc.get("interactInfo") or {}
liked = _safe_int(interact.get("likedCount", 0))
# MCP 返回的用户主页笔记列表通常只有 likedCount
# 详情页才有评论数和收藏数,先用点赞数作为主指标
title = nc.get("displayTitle", "") or ""
note_type = nc.get("type", "normal") # normal / video
# 从本地备份的文案中提取主题、风格、标签
local_meta = self._find_local_meta(title)
note_data = {
"note_id": note_id,
"title": title,
"type": note_type,
"likes": liked,
"topic": local_meta.get("topic", ""),
"style": local_meta.get("style", ""),
"tags": local_meta.get("tags", []),
"sd_prompt": local_meta.get("sd_prompt", ""),
"collected_at": datetime.now().isoformat(),
}
# 更新或新增
old = notes_dict.get(note_id, {})
if old.get("likes", 0) != liked or not old:
updated += 1
notes_dict[note_id] = {**old, **note_data}
note_summaries.append(note_data)
self._analytics_data["notes"] = notes_dict
self._analytics_data["last_analysis"] = datetime.now().isoformat()
self._save_analytics()
logger.info("采集完成: 共 %d 篇笔记, 更新 %d 篇", len(feeds), updated)
return {"total": len(feeds), "updated": updated, "notes": note_summaries}
def collect_note_details(self, mcp_client, note_id: str, xsec_token: str):
"""获取单篇笔记的详细数据(点赞、评论数、收藏等)"""
try:
result = mcp_client.get_feed_detail(note_id, xsec_token, load_all_comments=False)
text = ""
if isinstance(result, dict):
# 兼容 _call_tool 包装格式
inner_raw = result.get("raw", {})
content_list = []
if isinstance(inner_raw, dict):
content_list = inner_raw.get("content", [])
if not content_list:
content_list = result.get("content", [])
for item in content_list:
if isinstance(item, dict) and item.get("type") == "text":
text = item.get("text", "")
break
if not text:
text = result.get("text", "")
if text:
data = None
try:
data = json.loads(text)
except Exception:
m = re.search(r'(\{[\s\S]*\})', text)
if m:
try:
data = json.loads(m.group(1))
except Exception:
pass
if data:
interact = data.get("interactInfo") or {}
comments = data.get("comments", [])
return {
"likes": _safe_int(interact.get("likedCount", 0)),
"comments_count": _safe_int(interact.get("commentCount", len(comments))),
"collects": _safe_int(interact.get("collectedCount", 0)),
"shares": _safe_int(interact.get("shareCount", 0)),
}
except Exception as e:
logger.warning("获取笔记 %s 详情失败: %s", note_id, e)
return None
def _find_local_meta(self, title: str) -> dict:
"""从本地 xhs_workspace 中查找匹配标题的备份文案,提取 topic/style/tags"""
result = {"topic": "", "style": "", "tags": [], "sd_prompt": ""}
if not title:
return result
# 搜索备份目录
try:
for dirname in os.listdir(self.workspace_dir):
dir_path = os.path.join(self.workspace_dir, dirname)
if not os.path.isdir(dir_path) or dirname.startswith("_"):
continue
txt_path = os.path.join(dir_path, "文案.txt")
if not os.path.exists(txt_path):
continue
try:
with open(txt_path, "r", encoding="utf-8") as f:
content = f.read()
# 检查标题是否匹配
if title[:10] in content or title in dirname:
# 提取元数据
for line in content.split("\n"):
if line.startswith("风格:"):
result["style"] = line.split(":", 1)[1].strip()
elif line.startswith("主题:"):
result["topic"] = line.split(":", 1)[1].strip()
elif line.startswith("标签:"):
tags_str = line.split(":", 1)[1].strip()
result["tags"] = [t.strip() for t in tags_str.split(",") if t.strip()]
elif line.startswith("SD Prompt:"):
result["sd_prompt"] = line.split(":", 1)[1].strip()
break
except Exception:
continue
except Exception:
pass
return result
# ========== 权重计算 ==========
def calculate_weights(self) -> dict:
"""
根据已采集的笔记表现数据,计算各维度权重
使用 互动得分 = likes * 1.0 + comments * 2.0 + collects * 1.5 加权
返回权重摘要
"""
notes = self._analytics_data.get("notes", {})
if not notes:
return {"error": "暂无笔记数据,请先采集"}
# 计算每篇笔记的综合得分
scored_notes = []
for nid, note in notes.items():
likes = note.get("likes", 0)
comments_count = note.get("comments_count", 0)
collects = note.get("collects", 0)
# 综合得分: 点赞权重 1.0, 评论权重 2.0(评论代表深度互动), 收藏权重 1.5
score = likes * 1.0 + comments_count * 2.0 + collects * 1.5
# 至少用点赞数保底
if score == 0:
score = likes
scored_notes.append({**note, "score": score, "note_id": nid})
if not scored_notes:
return {"error": "没有可分析的笔记"}
# 按得分排序
scored_notes.sort(key=lambda x: x["score"], reverse=True)
max_score = scored_notes[0]["score"] if scored_notes[0]["score"] > 0 else 1
# ---- 主题权重 ----
topic_scores = defaultdict(float)
topic_counts = defaultdict(int)
for note in scored_notes:
topic = note.get("topic", "").strip()
if topic:
topic_scores[topic] += note["score"]
topic_counts[topic] += 1
topic_weights = {}
for topic, total_score in topic_scores.items():
avg_score = total_score / topic_counts[topic]
# 归一化到 0-100
weight = min(100, int((avg_score / max_score) * 100)) if max_score > 0 else 50
# 多篇验证的加分
if topic_counts[topic] >= 3:
weight = min(100, weight + 10)
elif topic_counts[topic] >= 2:
weight = min(100, weight + 5)
topic_weights[topic] = {
"weight": weight,
"count": topic_counts[topic],
"avg_score": round(avg_score, 1),
"total_score": round(total_score, 1),
}
# ---- 风格权重 ----
style_scores = defaultdict(float)
style_counts = defaultdict(int)
for note in scored_notes:
style = note.get("style", "").strip()
if style:
style_scores[style] += note["score"]
style_counts[style] += 1
style_weights = {}
for style, total_score in style_scores.items():
avg = total_score / style_counts[style]
weight = min(100, int((avg / max_score) * 100)) if max_score > 0 else 50
style_weights[style] = {
"weight": weight,
"count": style_counts[style],
"avg_score": round(avg, 1),
}
# ---- 标签权重 ----
tag_scores = defaultdict(float)
tag_counts = defaultdict(int)
for note in scored_notes:
for tag in note.get("tags", []):
tag = tag.strip().lstrip("#")
if tag:
tag_scores[tag] += note["score"]
tag_counts[tag] += 1
tag_weights = {}
for tag, total_score in tag_scores.items():
avg = total_score / tag_counts[tag]
weight = min(100, int((avg / max_score) * 100)) if max_score > 0 else 50
tag_weights[tag] = {"weight": weight, "count": tag_counts[tag]}
# 排序后取 Top
tag_weights = dict(sorted(tag_weights.items(), key=lambda x: x[1]["weight"], reverse=True)[:30])
# ---- 标题模式权重 (提取 emoji/句式/长度特征) ----
title_patterns = defaultdict(list)
for note in scored_notes:
title = note.get("title", "")
if not title:
continue
# 检测标题特征
has_emoji = bool(re.search(r'[\U0001F600-\U0001F9FF\u2600-\u27BF]', title))
has_question = "?" in title or "?" in title
has_exclaim = "!" in title or "!" in title
has_ellipsis = "..." in title or "…" in title
length_bucket = "短(≤10)" if len(title) <= 10 else ("中(11-15)" if len(title) <= 15 else "长(16-20)")
for feature, val in [
("含emoji", has_emoji), ("疑问句式", has_question),
("感叹句式", has_exclaim), ("省略句式", has_ellipsis),
]:
if val:
title_patterns[feature].append(note["score"])
title_patterns[f"长度:{length_bucket}"].append(note["score"])
title_pattern_weights = {}
for pattern, scores in title_patterns.items():
avg = sum(scores) / len(scores) if scores else 0
title_pattern_weights[pattern] = {
"weight": min(100, int((avg / max_score) * 100)) if max_score > 0 else 50,
"count": len(scores),
"avg_score": round(avg, 1),
}
# ---- 发布时间权重 ----
time_scores = defaultdict(list)
for note in scored_notes:
collected = note.get("collected_at", "")
if collected:
try:
dt = datetime.fromisoformat(collected)
hour_bucket = f"{(dt.hour // 3) * 3:02d}-{(dt.hour // 3) * 3 + 3:02d}时"
time_scores[hour_bucket].append(note["score"])
except Exception:
pass
time_weights = {}
for bucket, scores in time_scores.items():
avg = sum(scores) / len(scores) if scores else 0
time_weights[bucket] = {
"weight": min(100, int((avg / max_score) * 100)) if max_score > 0 else 50,
"count": len(scores),
}
# ---- 保存权重 ----
self._weights.update({
"topic_weights": dict(sorted(topic_weights.items(), key=lambda x: x[1]["weight"], reverse=True)),
"style_weights": dict(sorted(style_weights.items(), key=lambda x: x[1]["weight"], reverse=True)),
"tag_weights": tag_weights,
"title_pattern_weights": title_pattern_weights,
"time_weights": time_weights,
"last_updated": datetime.now().isoformat(),
"total_notes_analyzed": len(scored_notes),
"top_note": {
"title": scored_notes[0].get("title", ""),
"score": scored_notes[0].get("score", 0),
"likes": scored_notes[0].get("likes", 0),
} if scored_notes else {},
})
# 追加分析历史
history = self._weights.get("analysis_history", [])
history.append({
"time": datetime.now().isoformat(),
"total_notes": len(scored_notes),
"avg_score": round(sum(n["score"] for n in scored_notes) / len(scored_notes), 1),
"top_topic": list(topic_weights.keys())[0] if topic_weights else "",
})
# 只保留最近 50 条
self._weights["analysis_history"] = history[-50:]
self._save_weights()
return {
"total_notes": len(scored_notes),
"top_topics": list(topic_weights.items())[:10],
"top_styles": list(style_weights.items())[:5],
"top_tags": list(tag_weights.items())[:10],
"title_patterns": title_pattern_weights,
"top_note": scored_notes[0] if scored_notes else None,
}
# ========== 加权主题选择 ==========
def get_weighted_topic(self, base_topics: list[str] = None) -> str:
"""
根据权重从主题池中加权随机选择一个主题
如果没有权重数据, 退回均匀随机
"""
import random
topic_weights = self._weights.get("topic_weights", {})
if not topic_weights:
# 无权重数据,从基础池中随机
return random.choice(base_topics) if base_topics else "日常分享"
# 合并: 已有权重的主题 + base_topics 中新的主题
all_topics = {}
for topic, info in topic_weights.items():
all_topics[topic] = info.get("weight", 50)
if base_topics:
for t in base_topics:
if t not in all_topics:
all_topics[t] = 30 # 新主题给一个基础权重
# 加权随机选择
topics = list(all_topics.keys())
weights = [max(1, all_topics[t]) for t in topics] # 确保权重 >= 1
chosen = random.choices(topics, weights=weights, k=1)[0]
logger.info("加权选题: %s (权重: %s)", chosen, all_topics.get(chosen, "?"))
return chosen
def get_weighted_style(self, base_styles: list[str] = None) -> str:
"""根据权重选择风格"""
import random
style_weights = self._weights.get("style_weights", {})
if not style_weights:
return random.choice(base_styles) if base_styles else "真实分享"
all_styles = {}
for style, info in style_weights.items():
all_styles[style] = info.get("weight", 50)
if base_styles:
for s in base_styles:
if s not in all_styles:
all_styles[s] = 30
styles = list(all_styles.keys())
weights = [max(1, all_styles[s]) for s in styles]
return random.choices(styles, weights=weights, k=1)[0]
def get_top_tags(self, n: int = 8) -> list[str]:
"""获取权重最高的 N 个标签"""
tag_weights = self._weights.get("tag_weights", {})
if not tag_weights:
return []
sorted_tags = sorted(tag_weights.items(), key=lambda x: x[1].get("weight", 0), reverse=True)
return [t[0] for t in sorted_tags[:n]]
def get_title_advice(self) -> str:
"""根据标题模式权重生成建议"""
patterns = self._weights.get("title_pattern_weights", {})
if not patterns:
return "暂无标题分析数据"
sorted_p = sorted(patterns.items(), key=lambda x: x[1].get("weight", 0), reverse=True)
advice_parts = []
for p_name, p_info in sorted_p[:5]:
advice_parts.append(f" • {p_name}: 权重 {p_info['weight']}分 (出现{p_info['count']}次)")
return "\n".join(advice_parts)
# ========== LLM 深度分析 ==========
def generate_llm_analysis_prompt(self) -> str:
"""生成给 LLM 分析笔记表现的 prompt 数据部分"""
notes = self._analytics_data.get("notes", {})
if not notes:
return ""
# 按点赞排序
sorted_notes = sorted(notes.values(), key=lambda x: x.get("likes", 0), reverse=True)
lines = []
for i, note in enumerate(sorted_notes[:20]):
lines.append(
f"#{i+1} 「{note.get('title', '无标题')}」\n"
f" 点赞: {note.get('likes', 0)} | 主题: {note.get('topic', '未知')} | "
f"风格: {note.get('style', '未知')}\n"
f" 标签: {', '.join(note.get('tags', []))}"
)
return "\n".join(lines)
# ========== 报告生成 ==========
def generate_report(self) -> str:
"""生成 Markdown 格式的分析报告"""
weights = self._weights
notes = self._analytics_data.get("notes", {})
if not notes:
return "## 📊 暂无分析数据\n\n请先点击「采集数据」获取笔记表现数据,再点击「计算权重」。"
total = len(notes)
last_updated = weights.get("last_updated", "未知")
# Top Note
top_note = weights.get("top_note", {})
top_note_str = f"**{top_note.get('title', '')}** (❤️ {top_note.get('likes', 0)})" if top_note else "暂无"
lines = [
f"## 📊 智能内容学习报告",
f"",
f"🕐 最后更新: {last_updated[:19] if last_updated else '从未'}",
f"📝 分析笔记数: **{total}** 篇",
f"🏆 最佳笔记: {top_note_str}",
"",
"---",
"",
]
# 主题权重
topic_w = weights.get("topic_weights", {})
if topic_w:
lines.append("### 🎯 主题权重排行")
lines.append("| 排名 | 主题 | 权重 | 笔记数 | 平均得分 |")
lines.append("|:---:|------|:---:|:---:|:---:|")
for idx, (topic, info) in enumerate(list(topic_w.items())[:10]):
bar = "█" * (info["weight"] // 10) + "░" * (10 - info["weight"] // 10)
lines.append(
f"| {idx+1} | {topic} | {bar} {info['weight']} | {info['count']} | {info['avg_score']} |"
)
lines.append("")
# 风格权重
style_w = weights.get("style_weights", {})
if style_w:
lines.append("### 🎨 风格权重排行")
for style, info in list(style_w.items())[:5]:
bar = "█" * (info["weight"] // 10) + "░" * (10 - info["weight"] // 10)
lines.append(f"- **{style}**: {bar} {info['weight']}分 ({info['count']}篇)")
lines.append("")
# 标签权重
tag_w = weights.get("tag_weights", {})
if tag_w:
lines.append("### 🏷️ 高权重标签 (Top 10)")
top_tags = list(tag_w.items())[:10]
tag_strs = [f"`#{t}` ({info['weight']})" for t, info in top_tags]
lines.append(" | ".join(tag_strs))
lines.append("")
# 标题模式
title_p = weights.get("title_pattern_weights", {})
if title_p:
lines.append("### ✏️ 标题模式分析")
sorted_p = sorted(title_p.items(), key=lambda x: x[1].get("weight", 0), reverse=True)
for p_name, p_info in sorted_p[:6]:
lines.append(f"- **{p_name}**: 权重 {p_info['weight']} (出现 {p_info['count']} 次)")
lines.append("")
# 建议
lines.append("---")
lines.append("### 💡 智能建议")
if topic_w:
top_3 = list(topic_w.keys())[:3]
lines.append(f"- 📌 **高权重主题**: 优先创作 → {', '.join(top_3)}")
if tag_w:
hot_tags = [f"#{t}" for t in list(tag_w.keys())[:5]]
lines.append(f"- 🏷️ **推荐标签**: {' '.join(hot_tags)}")
if title_p:
best_pattern = max(title_p.items(), key=lambda x: x[1].get("weight", 0))
lines.append(f"- ✏️ **标题建议**: 多用「{best_pattern[0]}」(权重{best_pattern[1]['weight']})")
lines.append("")
lines.append(f"> 💡 启用「智能加权发布」后,自动发布将按权重倾斜生成高表现内容")
return "\n".join(lines)
def get_weighted_topics_display(self) -> str:
"""获取加权后的主题列表(用于UI显示)"""
topic_w = self._weights.get("topic_weights", {})
if not topic_w:
return ""
# 按权重排序,返回逗号分隔
sorted_topics = sorted(topic_w.items(), key=lambda x: x[1].get("weight", 0), reverse=True)
return ", ".join([t[0] for t in sorted_topics[:15]])
@property
def has_weights(self) -> bool:
"""是否已有权重数据"""
return bool(self._weights.get("topic_weights"))
@property
def weights_summary(self) -> str:
"""一行权重摘要"""
tw = self._weights.get("topic_weights", {})
total = self._weights.get("total_notes_analyzed", 0)
if not tw:
return "暂无权重数据"
top = list(tw.keys())[:3]
return f"{total}篇笔记 | 热门: {', '.join(top)}"
+3 -2
View File
@@ -14,12 +14,12 @@ _STARTUP_REG_KEY = r"Software\Microsoft\Windows\CurrentVersion\Run"
def _get_startup_script_path() -> str:
"""获取启动脚本路径(.vbs 静默启动,不弹黑窗)"""
return os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "_autostart.vbs")
return os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "scripts", "_autostart.vbs")
def _get_startup_bat_path() -> str:
"""获取启动 bat 路径"""
return os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "_autostart.bat")
return os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "scripts", "_autostart.bat")
def _create_startup_scripts():
@@ -35,6 +35,7 @@ def _create_startup_scripts():
# 创建 bat
bat_path = _get_startup_bat_path()
os.makedirs(os.path.dirname(bat_path), exist_ok=True)
bat_content = f"""@echo off
cd /d "{app_dir}"
"{venv_python}" "{main_script}"
+268
View File
@@ -0,0 +1,268 @@
"""
配置管理模块
支持多配置项、默认值回退、自动保存
"""
import json
import os
import tempfile
import logging
logger = logging.getLogger(__name__)
# 敏感字段 keyring 存储常量
_KEYRING_PLACEHOLDER = "[keyring]"
_KEYRING_SERVICE = "autobot_xhs"
_PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
CONFIG_FILE = os.path.join(_PROJECT_ROOT, "config", "config.json")
OUTPUT_DIR = "xhs_workspace"
DEFAULT_CONFIG = {
"api_key": "",
"base_url": "https://api.openai.com/v1",
"sd_url": "http://127.0.0.1:7860",
"mcp_url": "http://localhost:18060/mcp",
"model": "gpt-3.5-turbo",
"persona": "性感福利主播,身材火辣衣着大胆,专注分享穿衣显身材和私房写真风穿搭",
"auto_reply_enabled": False,
"schedule_enabled": False,
"my_user_id": "",
"active_llm": "",
"llm_providers": [],
"use_smart_weights": True,
# SD 图片生成参数
"quality_mode": "标准 (约1分钟)",
"sd_steps": 20,
"sd_cfg_scale": 5.5,
"sd_negative_prompt": "",
# 自动运营调度参数
"sched_comment_on": True,
"sched_like_on": True,
"sched_fav_on": True,
"sched_reply_on": True,
"sched_publish_on": True,
"sched_c_min": 15,
"sched_c_max": 45,
"sched_l_min": 10,
"sched_l_max": 30,
"sched_like_count": 5,
"sched_fav_min": 12,
"sched_fav_max": 35,
"sched_fav_count": 3,
"sched_r_min": 20,
"sched_r_max": 60,
"sched_reply_max": 3,
"sched_p_min": 60,
"sched_p_max": 180,
"sched_start_hour": 8,
"sched_end_hour": 23,
"auto_like_count": 5,
"auto_fav_count": 3,
"auto_reply_max": 5,
# 智能学习参数
"learn_interval": 6,
# 内容排期参数
"queue_gen_count": 3,
}
class ConfigManager:
"""配置管理器 - 单例模式"""
_instance = None
_config = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
if self._config is None:
self._config = self._load()
def _load(self) -> dict:
"""从文件加载配置,缺失项用默认值填充"""
config = DEFAULT_CONFIG.copy()
if os.path.exists(CONFIG_FILE):
try:
with open(CONFIG_FILE, "r", encoding="utf-8") as f:
saved = json.load(f)
config.update(saved)
except (json.JSONDecodeError, IOError) as e:
logger.warning("配置文件读取失败,使用默认值: %s", e)
return config
def save(self):
"""原子写:临时文件 + os.replace,防止写中断导致数据损坏"""
config_dir = os.path.dirname(os.path.abspath(CONFIG_FILE)) or "."
try:
fd, tmp_path = tempfile.mkstemp(dir=config_dir, suffix=".tmp", prefix="config_")
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(self._config, f, indent=4, ensure_ascii=False)
os.replace(tmp_path, CONFIG_FILE)
except Exception:
try:
os.remove(tmp_path)
except OSError:
pass
raise
except IOError as e:
logger.error("配置保存失败: %s", e)
def get_secure(self, key: str, default: str = "") -> str:
"""读取敏感配置,优先级: 环境变量 AUTOBOT_<KEY> > keyring > config.json 明文(自动迁移)"""
env_val = os.environ.get(f"AUTOBOT_{key.upper()}")
if env_val:
return env_val
try:
import keyring
kr_val = keyring.get_password(_KEYRING_SERVICE, key)
if kr_val is not None:
return kr_val
except Exception as e:
logger.warning("keyring 读取失败,使用明文回退: %s", e)
current = self._config.get(key, default)
if current and current != _KEYRING_PLACEHOLDER:
# 有明文值,尝试自动迁移到 keyring
try:
import keyring
keyring.set_password(_KEYRING_SERVICE, key, current)
self._config[key] = _KEYRING_PLACEHOLDER
self.save()
logger.info("已将 '%s' 自动迁移至系统 keyring", key)
except Exception as e:
logger.warning("keyring 迁移失败,继续使用明文: %s", e)
return current
return default
def set_secure(self, key: str, value: str):
"""将敏感值存入 keyring(不可用时降级为明文),config.json 中写入占位符"""
try:
import keyring
keyring.set_password(_KEYRING_SERVICE, key, value)
self._config[key] = _KEYRING_PLACEHOLDER
self.save()
except Exception as e:
logger.warning("keyring 不可用,明文保存 '%s': %s", key, e)
self._config[key] = value
self.save()
def get(self, key: str, default=None):
"""获取配置项"""
return self._config.get(key, default)
def set(self, key: str, value):
"""设置配置项并自动保存"""
self._config[key] = value
self.save()
def update(self, data: dict):
"""批量更新配置"""
self._config.update(data)
self.save()
@property
def all(self) -> dict:
"""返回全部配置(副本)"""
return self._config.copy()
def ensure_workspace(self):
"""确保工作空间目录存在"""
os.makedirs(OUTPUT_DIR, exist_ok=True)
# ---------- 多 LLM 提供商管理 ----------
def get_llm_providers(self) -> list[dict]:
"""获取所有 LLM 提供商配置"""
providers = self._config.get("llm_providers", [])
# 兼容旧配置: 如果 providers 为空但有 api_key,自动迁移
if not providers and self._config.get("api_key"):
default_provider = {
"name": "默认",
"api_key": self._config["api_key"],
"base_url": self._config.get("base_url", "https://api.openai.com/v1"),
}
providers = [default_provider]
self._config["llm_providers"] = providers
self._config["active_llm"] = "默认"
self.save()
return providers
def get_llm_provider_names(self) -> list[str]:
"""获取所有提供商名称列表"""
return [p["name"] for p in self.get_llm_providers()]
def get_active_llm(self) -> dict | None:
"""获取当前激活的 LLM 提供商配置(自动解析 keyring 中的 api_key)"""
active_name = self._config.get("active_llm", "")
providers = self.get_llm_providers()
provider = None
for p in providers:
if p["name"] == active_name:
provider = p
break
if provider is None:
provider = providers[0] if providers else None
if provider is None:
return None
# 解析 api_key:占位符时从 keyring 读取真实值
api_key = provider.get("api_key", "")
if api_key == _KEYRING_PLACEHOLDER:
secure_key = f"provider_{provider['name']}_api_key"
api_key = self.get_secure(secure_key, "")
return {**provider, "api_key": api_key}
def add_llm_provider(self, name: str, api_key: str, base_url: str) -> str:
"""添加一个 LLM 提供商,返回状态消息"""
name = name.strip()
if not name:
return "❌ 名称不能为空"
if not api_key.strip():
return "❌ API Key 不能为空"
providers = self.get_llm_providers()
for p in providers:
if p["name"] == name:
return f"❌ 名称「{name}」已存在,请换一个"
# 安全存储 api_key
stored_key = api_key.strip()
secure_key = f"provider_{name}_api_key"
try:
import keyring
keyring.set_password(_KEYRING_SERVICE, secure_key, stored_key)
stored_key = _KEYRING_PLACEHOLDER
except Exception as e:
logger.warning("keyring 不可用,api_key 明文存储: %s", e)
providers.append({
"name": name,
"api_key": stored_key,
"base_url": (base_url or "https://api.openai.com/v1").strip().rstrip("/"),
})
self._config["llm_providers"] = providers
if not self._config.get("active_llm"):
self._config["active_llm"] = name
self.save()
return f"✅ 已添加「{name}」"
def remove_llm_provider(self, name: str) -> str:
"""删除一个 LLM 提供商"""
providers = self.get_llm_providers()
new_providers = [p for p in providers if p["name"] != name]
if len(new_providers) == len(providers):
return f"⚠️ 未找到「{name}」"
self._config["llm_providers"] = new_providers
if self._config.get("active_llm") == name:
self._config["active_llm"] = new_providers[0]["name"] if new_providers else ""
self.save()
return f"✅ 已删除「{name}」"
def set_active_llm(self, name: str):
"""切换当前激活的 LLM 提供商"""
self._config["active_llm"] = name
# 同步到兼容字段
p = self.get_active_llm()
if p:
self._config["api_key"] = p["api_key"]
self._config["base_url"] = p["base_url"]
self.save()
+4 -4
View File
@@ -8,10 +8,10 @@ import logging
import gradio as gr
from config_manager import ConfigManager
from llm_service import LLMService
from sd_service import SDService, get_model_profile_info
from mcp_client import get_mcp_client
from .config_manager import ConfigManager
from .llm_service import LLMService
from .sd_service import SDService, get_model_profile_info
from .mcp_client import get_mcp_client
logger = logging.getLogger("autobot")
cfg = ConfigManager()
+6 -6
View File
@@ -11,12 +11,12 @@ import logging
from PIL import Image
from config_manager import ConfigManager, OUTPUT_DIR
from llm_service import LLMService
from sd_service import SDService, get_sd_preset
from mcp_client import get_mcp_client
from services.connection import _get_llm_config
from services.persona import _resolve_persona
from .config_manager import ConfigManager, OUTPUT_DIR
from .llm_service import LLMService
from .sd_service import SDService, get_sd_preset
from .mcp_client import get_mcp_client
from .connection import _get_llm_config
from .persona import _resolve_persona
logger = logging.getLogger("autobot")
cfg = ConfigManager()
+4 -4
View File
@@ -6,10 +6,10 @@ import logging
import gradio as gr
from mcp_client import get_mcp_client
from llm_service import LLMService
from services.connection import _get_llm_config
from services.hotspot import _pick_from_cache, _set_cache, _get_cache
from .mcp_client import get_mcp_client
from .llm_service import LLMService
from .connection import _get_llm_config
from .hotspot import _pick_from_cache, _set_cache, _get_cache
logger = logging.getLogger("autobot")
+4 -4
View File
@@ -7,10 +7,10 @@ import logging
import gradio as gr
from llm_service import LLMService
from mcp_client import get_mcp_client
from services.connection import _get_llm_config
from services.persona import _resolve_persona
from .llm_service import LLMService
from .mcp_client import get_mcp_client
from .connection import _get_llm_config
from .persona import _resolve_persona
logger = logging.getLogger("autobot")
+963
View File
@@ -0,0 +1,963 @@
"""
LLM 服务模块
封装对 OpenAI 兼容 API 的调用,包含文案生成、热点分析、评论回复等 Prompt
"""
import requests
import json
import re
import random
import logging
logger = logging.getLogger(__name__)
# ================= Prompt 模板 =================
PROMPT_COPYWRITING = """
你是一个真实的小红书博主,正在用手机编辑一篇笔记。你不是内容专家,你只是一个想认真分享的普通人。
【你的写作状态】:
想象你刚体验完某件事(试了一个产品/去了一个地方/学到一个技巧),打开小红书想跟朋友们聊聊。你不会字字斟酌,就是把感受写出来。
【标题规则】(严格执行):
1. 长度限制:必须控制在 18 字以内(含Emoji),绝对不能超过 20 字!
2. 像你发朋友圈的语气,口语化、有情绪感。可以用疑问句、感叹句、省略句
3. 可以加1-2个emoji,但不要堆砌
4. 禁止广告法违禁词("第一" "最" "顶级"等)
5. 好的标题示例:"后悔没早买!这个真的绝了" "姐妹们被我找到了" "求求你们别再踩这个坑了"
6. 避免AI感标题:不要用"震惊!" "必看!" "干货"这种过于营销的开头
【正文规则——像说话一样写】:
1. 想象你在跟闺蜜/朋友面对面聊天,把她说的话打下来就对了
2. 正文控制在 400-600 字
3. 不要像写作文一样"首先、其次、最后",用碎碎念的方式自然展开
4. 可以有小情绪:吐槽、感叹、自嘲、开心炸裂都行
5. emoji不要每句话都有,穿插在情绪高点就好(一段文字2-4个emoji足够)
6. 真人笔记特征:
- 会有"话说" "对了" "哦对" 这种口语转折
- 会有"不是我说" "真的会谢" "笑不活了"这种网络表达
- 会有不完整的句子、省略号、波浪号
- 段落长短不一,有的段就一句话,有的段会稍长
7. 绝对禁止:
❌ "值得一提的是" "需要注意的是" "总的来说" "综上所述"
❌ "作为一个xxx" "在这里给大家分享"
❌ 排比句、对仗工整的总结
❌ 每段都很整齐的1234结构
❌ "小伙伴们" "宝子们" 等过度热情的称呼(偶尔一次可以)
8. 结尾加 5-8 个相关话题标签(#)
【!!极其重要 — 反AI检测规则!!】:
你的文案会被AI检测工具审查,必须通过以下防检测措施:
1. 句子长度必须参差不齐:有的句子就3-5个字,有的20多字,绝对不能每句都差不多长
2. 不要所有段落逻辑都很通顺——真人写东西会突然跑题、突然想起另一件事
3. 偶尔犯一点小"不规范":比如某句话没写完就换话题了、用了个不太准确的词
4. 标点符号不要太规范:真人用手机打字经常漏掉逗号、句号不加,偶尔连续用两个问号??
5. 用词不要太"正确":真人会说"巨好用" "绝绝子" "无敌了" 而不是"非常好用" "十分出色"
6. 避免完美的逻辑链条:不要每段都工工整整地推进论点,真人笔记是跳跃式的
7. 偶尔口语化到"学渣"程度:"就 很那个 你懂的" "属于是" "多少有点" "怎么说呢"
8. 绝对不要用"然而" "此外" "因此" "尽管" "虽然...但是..."这些书面连接词
【绘图 Prompt】:
{sd_prompt_guide}
返回 JSON 格式:
{{"title": "...", "content": "...", "sd_prompt": "...", "tags": ["标签1", "标签2", ...]}}
"""
PROMPT_PERFORMANCE_ANALYSIS = """
你是一个有实战经验的小红书运营数据分析师。下面是一个博主已发布的笔记数据,按互动量从高到低排列:
{note_data}
【权重学习分析任务】:
请深度分析这些笔记的互动数据,找出「什么样的内容最受欢迎」的规律。
请分析以下维度:
1. **高表现内容特征**:表现好的笔记有什么共同特征?主题、标题套路、风格、标签……越具体越好
2. **低表现内容反思**:表现差的笔记问题出在哪?是选题不行、标题没吸引力、还是其他原因?
3. **用户偏好画像**:从数据反推,关注这个账号的用户最喜欢什么样的内容?
4. **内容优化建议**:给出 5 个具体的下一步内容方向,每个都要说清楚为什么推荐
5. **标题优化建议**:总结 3 个高互动标题的写法模板,直接给出可套用的句式
6. **最佳实践标签**:推荐 10 个最有流量潜力的标签组合
注意:
- 用数据说话,不要空谈
- 建议要具体到可以直接执行的程度
- 不要说废话和套话
返回 JSON 格式:
{{"high_perform_features": "...", "low_perform_issues": "...", "user_preference": "...", "content_suggestions": [{{"topic": "...", "reason": "...", "priority": 1-5}}], "title_templates": ["模板1", "模板2", "模板3"], "recommended_tags": ["标签1", "标签2", ...]}}
"""
PROMPT_WEIGHTED_COPYWRITING = """
你是一个真实的小红书博主,正在用手机编辑一篇笔记。
【智能学习洞察——基于你过去笔记的数据分析】:
{weight_insights}
【创作要求】:
基于以上数据洞察,请创作一篇更容易获得高互动的笔记。要把数据分析的结论融入创作中,但写出来的内容要自然,不能看出是"为了数据而写"。
【标题规则】(严格执行):
1. 长度限制:必须控制在 18 字以内(含Emoji),绝对不能超过 20 字!
2. 参考高互动标题的模式:{title_advice}
3. 口语化,有情绪感,像发朋友圈
4. 禁止广告法违禁词
【正文规则——像说话一样写】:
1. 想象你在跟闺蜜/朋友面对面聊天
2. 正文控制在 400-600 字
3. 自然展开,不要分点罗列
4. 可以有小情绪:吐槽、感叹、自嘲、开心炸裂
5. emoji穿插在情绪高点,不要每句都有
6. 绝对禁止 AI 痕迹书面用语
【推荐标签】:优先使用这些高权重标签 → {hot_tags}
【绘图 Prompt】:
{sd_prompt_guide}
返回 JSON 格式:
{{"title": "...", "content": "...", "sd_prompt": "...", "tags": ["标签1", "标签2", ...]}}
"""
PROMPT_HOTSPOT_ANALYSIS = """
你是一个有实战经验的小红书运营人。下面是搜索到的热门笔记信息:
{feed_data}
你要像一个真正做过内容运营的人来分析这些数据,说话方式接地气一点,不要写得像论文。
请分析以下方面:
1. **热门选题方向**:提炼 3-5 个最火的细分选题。不要只写大方向如"美妆""穿搭",要具体到用户真正感兴趣的切入角度
2. **标题拆解**:这些高赞标题的套路是什么?比如用了什么句式、什么情绪钩子、什么悬念。直接举例说明,不要写得很理论
3. **内容结构**:爆款笔记是怎么组织内容的?是先说痛点再给方案?还是直接晒结果?用大白话说清楚
4. **推荐模仿方案**:给出 3 个具体的选题建议。要具体到"如果是我来写,我会这样切入"的程度,不要泛泛而谈
注意:
- 你的分析是给一个普通创作者看的,不是给甲方看报告
- 说人话,不要堆砌专业术语
- "内容矩阵""赛道""心智占领""用户画像"这种运营黑话能不用就不用
- 建议要可执行,不要写"结合自身特色打造差异化内容"这种废话
返回 JSON 格式:
{{"hot_topics": ["...", "..."], "title_patterns": ["...", "..."], "content_structure": "...", "suggestions": [{{"topic": "...", "reason": "..."}}]}}
"""
PROMPT_COMMENT_REPLY = """
你现在是一个真实的小红书博主,你的人设是:{persona}。
有粉丝在你的笔记下留言了,你需要像真人一样自然回复。
【核心模拟原则——极其重要】:
你必须模拟一个真实的、活生生的小红书用户在手机上随手打字的状态:
- 打字随意,不追求语法完美,偶尔可以有口语化的省略
- 真人不会每句话都加 emoji,有时一个都不加也很正常
- 绝对不要用"哈哈哈"开头(这是 AI 最常见的暴露特征)
- 不要用书面化的转折词("不过""然而""但是""确实"),用口语("哎""诶""嘿""是吧""对对对""哇塞")
- 回复可以很短,甚至只有几个字("谢谢宝!""对的呀""试试看~")
- 绝对不要用逗号和句号写出完整的完美句子,真人打字经常没有标点
- 不要用感叹号过多,真人回复感叹号一般不超过1个
- 不要用"呢""哦""呀"等语气词堆砌在一起
【回复策略】:
- 如果是夸奖/支持 → 简短感谢,可以适当撒娇或害羞("嘿嘿被你发现了""爱你""抱住")
- 如果是提问 → 直接回答,不要先说"这个问题很好"之类的废话
- 如果是质疑/负面 → 轻松带过或幽默化解,不要长篇大论解释
- 如果是分享经历 → 表示共鸣,简短回应
- 如果评论很简短(比如"好看""不错") → 你的回复也应该很简短
【禁止出现的 AI 特征】:
❌ "谢谢你的xxx" "感谢你的分享/支持/喜欢"
❌ "作为一个xxx" "我觉得xxx"
❌ 完整的主谓宾结构句子
❌ 连续使用3个以上emoji
❌ "哈哈哈"开头
❌ 每句话都有标点符号
❌ 排比句、对仗句
❌ "宝子""姐妹""家人们"等称呼(除非人设就是这种风格)
笔记标题:{post_title}
粉丝评论:{comment}
直接输出回复内容,不要有任何解释。回复长度请匹配对方评论的长度——对方说的短你也短,对方说的长你可以稍长一点。
"""
PROMPT_PROACTIVE_COMMENT = """
你必须完美伪装成一个真实的小红书用户。你的人设是:{persona}。
你正在刷小红书,看到了一篇笔记,你想留一条看起来像真人写的评论。
【核心!!——你是一个在手机上随手打字的普通人】:
把自己想象成一个躺在沙发上刷手机的人,看到一篇笔记后随手评论。
你打字很随意,不会字字斟酌,想到什么说什么。
【真人评论特征——必须遵循】:
1. 长度自然变化:有时候一句话("救命也太好看了吧"),有时候两三句,极少超过50字
2. 真人打字习惯:
- 经常省略主语("看完立马下单了" 而不是 "我看完后立马下单了")
- 会用缩写和网络用语("绝绝子""yyds""蹲一个""dd""awsl")
- 感叹用语如 "天哪""救命""啊啊啊""绝了" 而不是文绉绉的"真的很棒"
- 偶尔打错字也ok但不要刻意
3. emoji 使用规则:
- 50%的概率不加任何emoji
- 加的话最多1-2个,而且偏好 😭🫠❤️🥺😍 这类情绪化的
- 不要用 ✨💫🌟 这种博主式的装饰emoji
4. 绝对不要分点列举!真人评论从不分1234条说
5. 不要用完整标点,真人评论经常没逗号句号
【评论类型——随机选择一种自然风格】:
- 分享真实感受("这个颜色实物真的绝了 上次路过柜台试了一下就走不动了")
- 提一个具体问题("这个是什么色号呀""博主身高多少 我怕买了不合适")
- 表达种草("看完直接去搜了""钱包在哭泣")
- 补充相关经验("我之前买过xxx 感觉跟这个搭也蛮好看的")
- 简短共鸣("真的!""笑死""太真实了""懂了")
【绝对禁止——这些是AI评论的特征】:
❌ "写得真好" "内容很有价值" "干货满满" "收藏了"
❌ "博主太厉害了" "学到了" "受益匪浅" "非常实用"
❌ "我也觉得xxx" "我认为xxx" 这种过于理性客观的表达
❌ "首先...其次...最后..." 任何分点罗列
❌ 以"哈哈"开头
❌ 超过3个emoji
❌ 完整规范的标点使用
❌ 每句话都很完整很正式
❌ 同时出现"!"和emoji(选一个就够了)
❌ 把笔记标题的关键词重复一遍(比如笔记标题说"穿搭"你就评论"穿搭真好看")
【笔记信息】:
标题:{post_title}
正文摘要:{post_content}
【已有评论参考(避免重复)】:
{existing_comments}
请直接输出一条评论,不要有任何解释或前缀。记住:你是一个真人,不是AI。
"""
PROMPT_COPY_WITH_REFERENCE = """
你是一个真实的小红书博主,正在参考一些热门笔记来写一篇自己的原创内容。
你不是在写营销文案,你只是觉得这些笔记写得不错,想借鉴思路写一篇自己的体验分享。
【参考笔记】:
{reference_notes}
【创作主题】:{topic}
【风格要求】:{style}
【标题规则】:
1. 长度限制:必须控制在 18 字以内(含Emoji),绝对不能超过 20 字!
2. 学习参考笔记标题的情绪感和口语感,但内容完全原创
3. 写得像你发给朋友看的那种,不要像广告
【正文规则——写得像真人】:
1. 想象你是刚体验完然后打开小红书写笔记,把你的真实感受和过程写出来
2. 正文控制在 400-600 字
3. 真人写法:
- 开头可以直接说事,不需要"嗨大家好"之类的开场白
- 中间夹杂一些个人感受和小吐槽("一开始还在犹豫 结果用了之后真香")
- 不要面面俱到什么优点都说一遍,挑2-3个最有感触的重点说
- 可以适当说一两个小缺点,让内容更真实("唯一的缺点就是xxx 但瑕不掩瑜")
- 段落自然分割,有的段一两句,有的段稍长
4. emoji 穿插在情绪高点,不要每句都有,整篇 6-10 个足够
5. 绝对禁止:
❌ 排比句、对仗句("不仅...而且..." "既...又...")
❌ "值得一提" "需要注意" "总结一下" 等总结性书面用语
❌ 每个段落都很工整的1234结构
❌ 面面俱到地罗列所有优点
6. 结尾加 5-8 个话题标签(#)
【绘图 Prompt】:
{sd_prompt_guide}
返回 JSON 格式:
{{"title": "...", "content": "...", "sd_prompt": "...", "tags": ["标签1", "标签2", ...]}}
"""
class LLMService:
"""LLM API 服务封装"""
# 当主模型返回空内容时,依次尝试的备选模型列表
FALLBACK_MODELS = ["deepseek-v3", "gemini-2.5-flash", "deepseek-v3.1"]
def __init__(self, api_key: str, base_url: str, model: str = "gpt-3.5-turbo"):
self.api_key = api_key
self.base_url = base_url.rstrip("/")
self.model = model
@staticmethod
def get_sd_prompt_guide(sd_model_name: str = None, persona: str = None) -> str:
"""根据当前 SD 模型 + 人设 生成 LLM 使用的绘图 Prompt 指南(含反 AI 检测指导 + 人设视觉风格)"""
from sd_service import SD_MODEL_PROFILES, detect_model_profile, get_persona_sd_profile
key = detect_model_profile(sd_model_name) if sd_model_name else "juggernautXL"
profile = SD_MODEL_PROFILES.get(key, SD_MODEL_PROFILES["juggernautXL"])
arch = profile.get("arch", "sdxl")
display = profile.get("display_name", key)
desc = profile.get("description", "")
# 通用反 AI 检测 prompt 技巧
anti_detect_tips = (
"\n\n【重要 - 反AI检测】生成的图片需要尽量像真实手机拍摄:\n"
"- 在 prompt 中加入真实感关键词:shot on iPhone, casual snapshot, real life, candid photo\n"
"- 加入微小不完美:slight motion blur, natural background, everyday environment\n"
"- 避免过度完美的构图词:不要用 perfect composition, symmetrical, flawless 等\n"
"- 光线自然化:用 natural daylight, indoor ambient light, window light 而非 studio lighting\n"
"- 模拟手机拍照特征:phone camera, slightly overexposed, casual angle, not centered\n"
)
# 人设视觉风格指导 (追加于末尾)
persona_guide = ""
persona_sd = get_persona_sd_profile(persona) if persona else None
if persona_sd and persona_sd.get("llm_guide"):
persona_guide = persona_sd["llm_guide"]
# 赛博AI虚拟博主不需要反AI检测(大方承认是AI)
if "赛博AI虚拟博主" in (persona or ""):
anti_detect_tips = (
"\n\n【注意 - 本人设无需反AI检测】该博主大方承认是AI生成,不需要伪装真实感。\n"
"- 尽情使用 perfect, flawless, ultra detailed 等高质量词\n"
"- 追求视觉震撼和极致精致,不需要模拟手机拍照\n"
"- 可以使用 studio lighting, dramatic lighting, cinematic 等专业光效词\n"
)
if key == "majicmixRealistic":
base = (
f"生成 Stable Diffusion 英文提示词,当前使用模型: {display} ({desc})\n"
"该模型擅长东亚网红/朋友圈自拍风格,请按以下规则生成 sd_prompt:\n"
"- 人物要求(最重要!):必须是东亚面孔中国人,必须描述眼睛、肤色、表情、妆容\n"
"- 推荐使用 (权重:数值) 语法加强关键词,例如 (almond eyes:1.3), (glossy lips:1.2)\n"
"- 颜值核心词(必选2-3个):(bright sparkling eyes:1.2), (glossy lips:1.2), (rosy cheeks:1.1),\n"
" (soft smile:1.2), (dewy glowing skin:1.2), (long lashes:1.1), (charming gaze:1.1)\n"
"- 风格关键词:RAW photo, realistic, photorealistic, instagram aesthetic\n"
"- 氛围词:soft lighting, warm tone, phone camera feel, shallow depth of field\n"
"- ❌ 严禁使用:skin pores, natural imperfections, skin texture(会让皮肤变粗糙)\n"
"- 非常适合:自拍、穿搭展示、美妆效果、生活日常、闺蜜合照风格\n"
"- 示范 prompt(参考此质量和格式):\n"
" (best quality:1.4), RAW photo, (photorealistic:1.4), (asian girl:1.3), (almond eyes:1.3),\n"
" (bright sparkling eyes:1.2), (dewy glowing skin:1.2), (natural makeup:1.3), (glossy lips:1.2),\n"
" (soft smile:1.2), wearing cream linen dress, standing on sunny street, shallow depth of field,\n"
" warm tone, instagram aesthetic, phone camera feel\n"
"- 用英文逗号分隔"
)
elif key == "realisticVision":
base = (
f"生成 Stable Diffusion 英文提示词,当前使用模型: {display} ({desc})\n"
"该模型擅长写实摄影风格,请按以下规则生成 sd_prompt:\n"
"- 人物要求(最重要!):必须是东亚面孔中国人,必须描述眼睛、肤色、表情\n"
"- 推荐使用 (权重:数值) 语法,例如 (realistic:1.4), (almond eyes:1.2)\n"
"- 颜值核心词(必选2-3个):(bright clear eyes:1.2), (charming gaze:1.1), (glossy lips:1.1),\n"
" (dewy smooth skin:1.2), (soft smile:1.1), (long lashes:1.1), (rosy cheeks:1.1)\n"
"- 风格关键词:RAW photo, DSLR, street photography, film color grading\n"
"- 镜头感:shot on Sony A7, 85mm lens, f/1.8, depth of field, bokeh\n"
"- ❌ 严禁使用:skin pores, detailed skin texture, natural imperfections(会让皮肤变难看)\n"
"- 非常适合:街拍、旅行照、真实场景、有故事感的画面\n"
"- 示范 prompt(参考此质量和格式):\n"
" RAW photo, (best quality:1.4), (realistic:1.4), (photorealistic:1.4), (asian:1.2),\n"
" (almond eyes:1.2), (bright clear eyes:1.2), (dewy smooth skin:1.2), (natural makeup:1.2),\n"
" (soft smile:1.1), wearing light blue shirt, walking in old town alley,\n"
" shot on Sony A7, 85mm lens, f/1.8, golden hour, depth of field\n"
"- 用英文逗号分隔"
)
else: # juggernautXL (SDXL)
base = (
f"生成 Stable Diffusion 英文提示词,当前使用模型: {display} ({desc})\n"
"该模型为 SDXL 架构,擅长电影级大片质感,请按以下规则生成 sd_prompt:\n"
"- 人物要求(最重要!):必须是东亚面孔中国人,绝对禁止西方人特征\n"
"- 不要使用 (权重:数值) 括号语法,SDXL 模型直接用逗号分隔即可\n"
"- 颜值核心词(必选2-3个):bright sparkling eyes, glossy lips, rosy cheeks,\n"
" gentle smile, luminous dewy skin, long lashes, charming expression, defined brows\n"
"- 质量词:masterpiece, best quality, ultra detailed, 8k uhd, high resolution\n"
"- 风格:photorealistic, cinematic lighting, commercial photography\n"
"- 光影:golden hour, soft diffused light, volumetric lighting, soft catchlights in eyes\n"
"- 非常适合:商业摄影、时尚大片、复杂光影场景、杂志封面风格\n"
"- 示范 prompt(参考此质量和格式):\n"
" masterpiece, best quality, ultra detailed, photorealistic, cinematic lighting,\n"
" chinese beauty, east asian features, almond eyes, bright sparkling eyes, long lashes,\n"
" luminous dewy skin, natural makeup, glossy lips, gentle smile, rosy cheeks,\n"
" wearing white summer dress, standing in flower garden, golden hour, soft bokeh background,\n"
" commercial photography, fashion editorial style\n"
"- 用英文逗号分隔"
)
# Task 4.1-4.4: 中国审美人物描述规则
chinese_aesthetic_guide = (
"\n\n【人物描述规则 - 中国审美标准】\n"
"生成人物 prompt 时,请严格遵守以下规则:\n\n"
"❌ 禁止使用通用美丽词汇(太泛,效果弱):\n"
" - 禁止: beautiful, pretty, gorgeous, attractive, stunning\n"
" - 原因: 这类词在中文语境下容易偏向西方模型默认审美\n\n"
"✅ 三维人物描述法(眼睛 / 肤色 / 气质)——必须各选至少1个:\n"
" 眼睛维度: almond eyes, double eyelid, bright clear eyes, "
"gentle gaze, expressive eyes, long eyelashes\n"
" 肤色维度: porcelain skin, luminous fair skin, milky white skin, "
"translucent skin, dewy complexion, peach-tinted cheeks\n"
" 气质维度: elegant temperament, gentle demeanor, refined bearing, "
"graceful posture, youthful vitality, scholarly aura\n\n"
"✅ 必须加表情/神态词(让人物鲜活有魅力,这非常重要!):\n"
" 从以下选1-2个: soft smile, gentle smile, charming gaze, bright eyes, "
"sweet expression, confident look, warm smile, playful smile\n"
" 绝对禁止生成没有表情词的人物 prompt — 没有表情词会导致人物呆板!\n\n"
"✅ 必须加妆容词(哪怕是日常淡妆,也要显得精致上镜):\n"
" 从以下选1-2个: natural makeup, light makeup, glossy lips, "
"rosy cheeks, subtle blush, defined brows, dewy skin finish\n\n"
"📸 构图策略(人物 vs 场景):\n"
" - 人物特写 (>50%画面): 突出五官气质,加 face close-up, portrait, "
"shallow depth of field, bokeh background\n"
" - 全身/中景: 突出整体氛围,加 full body shot, environmental portrait, "
"natural setting, lifestyle photography\n"
" - 半身照 (最常用): half body shot, upper body, 搭配场景关键词\n\n"
"💡 专属光线词(能凸显东亚肤色最佳状态):\n"
" - 室内: soft diffused light, window side lighting, warm ambient light, "
"ring light (美妆专用)\n"
" - 室外: golden hour, overcast daylight (均匀柔和), dappled sunlight (树影)\n"
" - 通用提升: skin luminance, luminous glow, soft catchlights in eyes\n"
" - 避免: harsh noon sunlight, under-eye shadows, top-down lighting, flat lighting\n"
)
return base + chinese_aesthetic_guide + anti_detect_tips + persona_guide
def _chat(self, system_prompt: str, user_message: str,
json_mode: bool = True, temperature: float = 0.8) -> str:
"""底层聊天接口(含空返回检测、json_mode 回退、模型降级)"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
if json_mode:
user_message = user_message + "\n请以json格式返回。"
# 构建要尝试的模型列表:主模型 + 备选模型(去重)
models_to_try = [self.model] + [m for m in self.FALLBACK_MODELS if m != self.model]
last_error = None
for model_idx, current_model in enumerate(models_to_try):
payload = {
"model": current_model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
],
"temperature": temperature,
}
if json_mode:
payload["response_format"] = {"type": "json_object"}
try:
resp = requests.post(
f"{self.base_url}/chat/completions",
headers=headers, json=payload, timeout=90
)
resp.raise_for_status()
content = resp.json()["choices"][0]["message"]["content"]
# 检测空返回 — 如果启用了 json_mode 且返回为空,回退去掉 response_format 重试
if not content or not content.strip():
if json_mode:
logger.warning("[%s] LLM 返回空内容 (json_mode=True),关闭 json_mode 回退重试...", current_model)
payload.pop("response_format", None)
resp2 = requests.post(
f"{self.base_url}/chat/completions",
headers=headers, json=payload, timeout=90
)
resp2.raise_for_status()
content = resp2.json()["choices"][0]["message"]["content"]
if not content or not content.strip():
# 当前模型完全无法返回内容,尝试下一个模型
if model_idx < len(models_to_try) - 1:
next_model = models_to_try[model_idx + 1]
logger.warning("[%s] 返回空内容,自动降级到模型: %s", current_model, next_model)
continue
raise RuntimeError(f"所有模型均返回空内容(已尝试: {', '.join(models_to_try[:model_idx+1])})")
if model_idx > 0:
logger.info("模型降级成功: %s → %s", self.model, current_model)
return content
except requests.exceptions.HTTPError as e:
status = getattr(resp, 'status_code', 0)
body = getattr(resp, 'text', '')[:300]
# 某些模型/提供商不支持 response_format,自动回退重试
if json_mode and status in (400, 422, 500):
logger.warning("[%s] json_mode 请求失败 (HTTP %s),关闭 response_format 回退重试...", current_model, status)
payload.pop("response_format", None)
try:
resp2 = requests.post(
f"{self.base_url}/chat/completions",
headers=headers, json=payload, timeout=90
)
resp2.raise_for_status()
content = resp2.json()["choices"][0]["message"]["content"]
if content and content.strip():
if model_idx > 0:
logger.info("模型降级成功: %s → %s", self.model, current_model)
return content
except Exception:
pass
# 当前模型失败,尝试下一个
last_error = ConnectionError(f"LLM API 错误 ({status}): {body}")
if model_idx < len(models_to_try) - 1:
logger.warning("[%s] HTTP %s 失败,降级到: %s", current_model, status, models_to_try[model_idx + 1])
continue
raise last_error
except requests.exceptions.Timeout:
last_error = TimeoutError(f"[{current_model}] LLM 请求超时")
if model_idx < len(models_to_try) - 1:
logger.warning("[%s] 请求超时,降级到: %s", current_model, models_to_try[model_idx + 1])
continue
raise TimeoutError("LLM 请求超时,所有模型均超时,请检查网络")
except (ConnectionError, RuntimeError):
raise
except Exception as e:
last_error = RuntimeError(f"LLM 调用异常: {e}")
if model_idx < len(models_to_try) - 1:
logger.warning("[%s] 调用异常 (%s),降级到: %s", current_model, e, models_to_try[model_idx + 1])
continue
raise last_error
raise last_error or RuntimeError("LLM 调用失败: 未知错误")
def _parse_json(self, text: str) -> dict:
"""从 LLM 返回文本中解析 JSON(多重容错)"""
if not text or not text.strip():
raise ValueError("LLM 返回内容为空,无法解析 JSON")
raw = text.strip()
# 策略1: 去除 markdown 代码块
cleaned = re.sub(r"```(?:json)?\s*", "", raw)
cleaned = re.sub(r"```", "", cleaned).strip()
# 策略2: 直接解析
try:
return json.loads(cleaned)
except json.JSONDecodeError:
pass
# 策略3: 提取最外层的 { ... } 块
match = re.search(r'(\{[\s\S]*\})', cleaned)
if match:
try:
return json.loads(match.group(1))
except json.JSONDecodeError:
pass
# 策略4: 逐行查找 JSON 开始位置
for i, ch in enumerate(cleaned):
if ch == '{':
try:
return json.loads(cleaned[i:])
except json.JSONDecodeError:
pass
break
# 策略5: 尝试修复常见问题(尾部多余逗号、缺少闭合括号)
try:
# 去除尾部多余逗号
fixed = re.sub(r',\s*([}\]])', r'\1', cleaned)
return json.loads(fixed)
except json.JSONDecodeError:
pass
# 全部失败,打日志并抛出有用的错误信息
preview = raw[:500] if len(raw) > 500 else raw
logger.error("JSON 解析全部失败,LLM 原始返回: %s", preview)
raise ValueError(
f"LLM 返回内容无法解析为 JSON。\n"
f"返回内容前200字: {raw[:200]}\n\n"
f"💡 可能原因: 模型不支持 JSON 输出格式,建议更换模型重试"
)
# ---------- 业务方法 ----------
def get_models(self) -> list[str]:
"""获取可用模型列表"""
url = f"{self.base_url}/models"
headers = {"Authorization": f"Bearer {self.api_key}"}
try:
resp = requests.get(url, headers=headers, timeout=10)
resp.raise_for_status()
text = resp.text.strip()
if not text:
logger.warning("GET %s 返回空响应", url)
return []
data = resp.json()
return [item["id"] for item in data.get("data", [])]
except Exception as e:
logger.warning("获取模型列表失败 (%s): %s", url, e)
return []
def generate_copy(self, topic: str, style: str, sd_model_name: str = None, persona: str = None) -> dict:
"""生成小红书文案(含重试逻辑,自动适配SD模型,支持人设)"""
sd_guide = self.get_sd_prompt_guide(sd_model_name, persona=persona)
system_prompt = PROMPT_COPYWRITING.format(sd_prompt_guide=sd_guide)
user_msg = f"主题:{topic}\n风格:{style}"
if persona:
user_msg = f"【博主人设】:{persona}\n请以此人设的视角和风格创作。\n\n{user_msg}"
last_error = None
for attempt in range(2):
try:
# 第二次尝试不使用 json_mode(兼容不支持的模型)
use_json_mode = (attempt == 0)
content = self._chat(
system_prompt,
user_msg,
json_mode=use_json_mode,
temperature=0.92,
)
data = self._parse_json(content)
# 强制标题长度限制
title = data.get("title", "")
if len(title) > 20:
title = title[:20]
data["title"] = title
# 去 AI 化后处理
if "content" in data:
data["content"] = self._humanize_content(data["content"])
return data
except (json.JSONDecodeError, ValueError) as e:
last_error = e
if attempt == 0:
logger.warning("文案生成 JSON 解析失败 (尝试 %d/2): %s,将关闭 json_mode 重试", attempt + 1, e)
continue
else:
logger.error("文案生成 JSON 解析失败 (尝试 %d/2): %s", attempt + 1, e)
raise RuntimeError(f"文案生成失败: LLM 返回无法解析为 JSON,已重试 2 次。\n最后错误: {last_error}")
def generate_copy_with_reference(self, topic: str, style: str,
reference_notes: str, sd_model_name: str = None, persona: str = None) -> dict:
"""参考热门笔记生成文案(含重试逻辑,自动适配SD模型,支持人设)"""
sd_guide = self.get_sd_prompt_guide(sd_model_name, persona=persona)
prompt = PROMPT_COPY_WITH_REFERENCE.format(
reference_notes=reference_notes, topic=topic, style=style,
sd_prompt_guide=sd_guide,
)
user_msg = f"请创作关于「{topic}」的小红书笔记"
if persona:
user_msg = f"【博主人设】:{persona}\n请以此人设的视角和风格创作。\n\n{user_msg}"
last_error = None
for attempt in range(2):
try:
use_json_mode = (attempt == 0)
content = self._chat(
prompt, user_msg,
json_mode=use_json_mode, temperature=0.92,
)
data = self._parse_json(content)
title = data.get("title", "")
if len(title) > 20:
data["title"] = title[:20]
if "content" in data:
data["content"] = self._humanize_content(data["content"])
return data
except (json.JSONDecodeError, ValueError) as e:
last_error = e
if attempt == 0:
logger.warning("参考文案生成 JSON 解析失败 (尝试 %d/2): %s,将关闭 json_mode 重试", attempt + 1, e)
continue
else:
logger.error("参考文案生成 JSON 解析失败 (尝试 %d/2): %s", attempt + 1, e)
raise RuntimeError(f"参考文案生成失败: LLM 返回无法解析为 JSON,已重试 2 次。\n最后错误: {last_error}")
def analyze_hotspots(self, feed_data: str) -> dict:
"""分析热门内容趋势(含重试逻辑)"""
prompt = PROMPT_HOTSPOT_ANALYSIS.format(feed_data=feed_data)
last_error = None
for attempt in range(2):
try:
use_json_mode = (attempt == 0)
content = self._chat(prompt, "请分析以上热门笔记数据",
json_mode=use_json_mode)
return self._parse_json(content)
except (json.JSONDecodeError, ValueError) as e:
last_error = e
if attempt == 0:
logger.warning("热点分析 JSON 解析失败 (尝试 %d/2): %s,将关闭 json_mode 重试", attempt + 1, e)
continue
else:
logger.error("热点分析 JSON 解析失败 (尝试 %d/2): %s", attempt + 1, e)
raise RuntimeError(f"热点分析失败: LLM 返回无法解析为 JSON,已重试 2 次。\n最后错误: {last_error}")
@staticmethod
def _humanize_content(text: str) -> str:
"""后处理: 深度去除 AI 书面痕迹,模拟真人手机打字风格"""
t = text
# ========== 第一层: 替换过于书面化/AI化的表达 ==========
ai_phrases = {
"值得一提的是": "对了",
"需要注意的是": "不过要注意",
"总的来说": "反正",
"综上所述": "总之",
"总而言之": "总之",
"不仅如此": "而且",
"与此同时": "然后",
"除此之外": "还有",
"众所周知": "",
"毋庸置疑": "",
"不言而喻": "",
"在这里给大家分享": "来分享",
"在此分享给大家": "分享一下",
"接下来让我们": "",
"话不多说": "",
"废话不多说": "",
"下面我来": "",
"让我来": "",
"首先我要说": "先说",
"我认为": "我觉得",
"我相信": "我觉得",
"事实上": "其实",
"实际上": "其实",
"毫无疑问": "",
"不可否认": "",
"客观来说": "",
"坦白说": "",
"具体而言": "就是",
"简而言之": "就是说",
"换句话说": "就是",
"归根结底": "说白了",
"由此可见": "",
"正如我所说": "",
"正如前文所述": "",
"在我看来": "我觉得",
"从某种程度上说": "",
"在一定程度上": "",
"非常值得推荐": "真的可以试试",
"强烈推荐": "真心推荐",
"性价比极高": "性价比很高",
"给大家安利": "安利",
"为大家推荐": "推荐",
"希望对大家有所帮助": "",
"希望能帮到大家": "",
"以上就是": "",
"感谢阅读": "",
"感谢大家的阅读": "",
}
for old, new in ai_phrases.items():
t = t.replace(old, new)
# ========== 第二层: 去掉分点罗列感 ==========
t = re.sub(r'(?m)^首先[,,::\s]*', '', t)
t = re.sub(r'(?m)^其次[,,::\s]*', '', t)
t = re.sub(r'(?m)^最后[,,::\s]*', '', t)
t = re.sub(r'(?m)^再者[,,::\s]*', '', t)
t = re.sub(r'(?m)^另外[,,::\s]*', '', t)
# 去序号: "1. " "2、" "①" 等
t = re.sub(r'(?m)^[①②③④⑤⑥⑦⑧⑨⑩]\s*', '', t)
t = re.sub(r'(?m)^[1-9][.、))]\s*', '', t)
# ========== 第三层: 去掉AI常见的空洞开头 ==========
for prefix in ["嗨大家好!", "嗨,大家好!", "大家好,", "大家好!",
"哈喽大家好!", "Hello大家好!", "嗨~", "hey~",
"各位姐妹大家好!", "各位宝子们好!"]:
if t.startswith(prefix):
t = t[len(prefix):].strip()
# ========== 第四层: 标点符号真人化 ==========
# AI 特征: 每句话都有完整标点 → 真人经常不加标点或只用逗号
sentences = t.split('\n')
humanized_lines = []
for line in sentences:
if not line.strip():
humanized_lines.append(line)
continue
# 随机去掉句末句号 (真人经常不打句号)
if line.rstrip().endswith('。') and random.random() < 0.35:
line = line.rstrip()[:-1]
# 随机把部分逗号替换成空格或什么都不加 (模拟打字不加标点)
if random.random() < 0.15:
# 只替换一个逗号
comma_positions = [m.start() for m in re.finditer(r'[,,]', line)]
if comma_positions:
pos = random.choice(comma_positions)
line = line[:pos] + ' ' + line[pos+1:]
humanized_lines.append(line)
t = '\n'.join(humanized_lines)
# ========== 第五层: 随机添加真人口语化元素 ==========
# 在段落开头随机插入口语衔接词
oral_connectors = [
"对了 ", "哦对 ", "话说 ", "然后 ", "就是说 ", "emmm ", "嗯 ",
"说真的 ", "不是 ", "离谱的是 ", "我发现 ",
]
paragraphs = t.split('\n\n')
if len(paragraphs) > 2:
# 在中间段落随机加1-2个口语衔接词
inject_count = random.randint(1, min(2, len(paragraphs) - 2))
inject_indices = random.sample(range(1, len(paragraphs)), inject_count)
for idx in inject_indices:
if paragraphs[idx].strip() and not any(paragraphs[idx].strip().startswith(c.strip()) for c in oral_connectors):
connector = random.choice(oral_connectors)
paragraphs[idx] = connector + paragraphs[idx].lstrip()
t = '\n\n'.join(paragraphs)
# ========== 第六层: 句子长度打散 ==========
# AI 特征: 句子长度高度均匀 → 真人笔记长短参差不齐
# 随机把一些长句用换行打散
lines = t.split('\n')
final_lines = []
for line in lines:
# 超过60字的行, 随机在一个位置断句
if len(line) > 60 and random.random() < 0.3:
# 找到中间附近的标点位置断句
mid = len(line) // 2
best_pos = -1
for offset in range(0, mid):
for check_pos in [mid + offset, mid - offset]:
if 0 < check_pos < len(line) and line[check_pos] in ',。!?、,':
best_pos = check_pos
break
if best_pos > 0:
break
if best_pos > 0:
final_lines.append(line[:best_pos + 1])
final_lines.append(line[best_pos + 1:].lstrip())
continue
final_lines.append(line)
t = '\n'.join(final_lines)
# ========== 第七层: 随机注入微小不完美 ==========
# 真人打字偶尔有重复字、多余空格等
if random.random() < 0.2:
# 随机在某处加一个波浪号或省略号
insert_chars = ['~', '...', '~', '..']
lines = t.split('\n')
if lines:
target = random.randint(0, len(lines) - 1)
if lines[target].rstrip() and not lines[target].rstrip()[-1] in '~~.。!?!?':
lines[target] = lines[target].rstrip() + random.choice(insert_chars)
t = '\n'.join(lines)
# ========== 第八层: 清理 ==========
# 去掉连续3个以上的 emoji
t = re.sub(r'([\U0001F600-\U0001F9FF\u2600-\u27BF])\1{2,}', r'\1\1', t)
# 清理多余空行
t = re.sub(r'\n{3,}', '\n\n', t)
# 清理行首多余空格 (手机打字不会缩进)
t = re.sub(r'(?m)^[ \t]+', '', t)
return t.strip()
@staticmethod
def _humanize(text: str) -> str:
"""后处理: 深度去除 AI 评论/回复中的非人类痕迹"""
t = text.strip()
# 去掉前后引号包裹
if (t.startswith('"') and t.endswith('"')) or (t.startswith("'") and t.endswith("'")):
t = t[1:-1].strip()
# 去掉 AI 常见的前缀
for prefix in ["回复:", "回复:", "评论:", "评论:", "以下是", "好的,",
"当然,", "当然!", "谢谢你的", "感谢你的", "好的!",
"嗯,"]:
if t.startswith(prefix):
t = t[len(prefix):].strip()
# 去掉末尾多余的句号(真人评论很少用句号结尾)
if t.endswith("。"):
t = t[:-1]
# 去掉末尾的"哦" "呢" 堆叠 (AI 常见)
t = re.sub(r'[哦呢呀哈]{2,}$', lambda m: m.group()[0], t)
# 替换过于完整规范的标点为口语化
if random.random() < 0.25 and ',' in t:
# 随机去掉一个逗号
comma_pos = [m.start() for m in re.finditer(',', t)]
if comma_pos:
pos = random.choice(comma_pos)
t = t[:pos] + ' ' + t[pos+1:]
# 随机去掉末尾感叹号(真人不是每句都加!)
if t.endswith('!') and random.random() < 0.3:
t = t[:-1]
# 限制连续 emoji(最多2个)
t = re.sub(r'([\U0001F600-\U0001F9FF\u2600-\u27BF])\1{2,}', r'\1\1', t)
return t
def generate_reply(self, persona: str, post_title: str, comment: str) -> str:
"""AI 生成评论回复"""
prompt = PROMPT_COMMENT_REPLY.format(
persona=persona, post_title=post_title, comment=comment
)
raw = self._chat(prompt, "请生成回复", json_mode=False, temperature=0.95)
return self._humanize(raw)
def generate_proactive_comment(self, persona: str, post_title: str,
post_content: str, existing_comments: str = "") -> str:
"""AI 生成主动评论"""
prompt = PROMPT_PROACTIVE_COMMENT.format(
persona=persona, post_title=post_title,
post_content=post_content,
existing_comments=existing_comments or "暂无评论",
)
raw = self._chat(prompt, "请生成评论", json_mode=False, temperature=0.95)
return self._humanize(raw)
def analyze_note_performance(self, note_data: str) -> dict:
"""AI 深度分析笔记表现,生成内容策略建议"""
prompt = PROMPT_PERFORMANCE_ANALYSIS.format(note_data=note_data)
last_error = None
for attempt in range(2):
try:
use_json_mode = (attempt == 0)
content = self._chat(prompt, "请深度分析以上笔记数据,找出规律并给出优化建议",
json_mode=use_json_mode, temperature=0.7)
return self._parse_json(content)
except (json.JSONDecodeError, ValueError) as e:
last_error = e
if attempt == 0:
logger.warning("表现分析 JSON 解析失败 (尝试 %d/2): %s", attempt + 1, e)
continue
raise RuntimeError(f"笔记表现分析失败: {last_error}")
def generate_weighted_copy(self, topic: str, style: str,
weight_insights: str, title_advice: str,
hot_tags: str, sd_model_name: str = None, persona: str = None) -> dict:
"""基于权重学习生成高互动潜力的文案(自动适配SD模型,支持人设)"""
sd_guide = self.get_sd_prompt_guide(sd_model_name, persona=persona)
prompt = PROMPT_WEIGHTED_COPYWRITING.format(
weight_insights=weight_insights,
title_advice=title_advice,
hot_tags=hot_tags,
sd_prompt_guide=sd_guide,
)
user_msg = f"主题:{topic}\n风格:{style}\n请创作一篇基于数据洞察的高质量小红书笔记"
if persona:
user_msg = f"【博主人设】:{persona}\n请以此人设的视角和风格创作。\n\n{user_msg}"
last_error = None
for attempt in range(2):
try:
use_json_mode = (attempt == 0)
content = self._chat(
prompt,
user_msg,
json_mode=use_json_mode,
temperature=0.92,
)
data = self._parse_json(content)
title = data.get("title", "")
if len(title) > 20:
data["title"] = title[:20]
if "content" in data:
data["content"] = self._humanize_content(data["content"])
return data
except (json.JSONDecodeError, ValueError) as e:
last_error = e
if attempt == 0:
logger.warning("加权文案生成失败 (尝试 %d/2): %s", attempt + 1, e)
continue
raise RuntimeError(f"加权文案生成失败: {last_error}")
+481
View File
@@ -0,0 +1,481 @@
"""
小红书 MCP HTTP 客户端
封装对 xiaohongshu-mcp 服务 (http://localhost:18060/mcp) 的调用
"""
import requests
import json
import logging
import uuid
import base64
import re
import io
from PIL import Image
logger = logging.getLogger(__name__)
MCP_DEFAULT_URL = "http://localhost:18060/mcp"
MCP_TIMEOUT = 60 # 秒
# 全局客户端缓存 —— 同一 URL 复用同一实例,避免反复 initialize
_client_cache: dict[str, "MCPClient"] = {}
def get_mcp_client(base_url: str = MCP_DEFAULT_URL) -> "MCPClient":
"""获取 MCP 客户端(单例),同一 URL 复用同一实例"""
if base_url not in _client_cache:
_client_cache[base_url] = MCPClient(base_url)
client = _client_cache[base_url]
return client
class MCPClient:
"""小红书 MCP 服务的 HTTP 客户端封装"""
def __init__(self, base_url: str = MCP_DEFAULT_URL):
self.base_url = base_url
self.session = requests.Session()
self.session.headers.update({"Content-Type": "application/json"})
self._session_id = None
self._initialized = False
# ---------- 底层通信 ----------
def _call(self, method: str, params: dict = None, *,
is_notification: bool = False) -> dict:
"""发送 JSON-RPC 请求到 MCP 服务
Args:
is_notification: 若为 True 则不带 id(JSON-RPC 通知)
"""
payload = {
"jsonrpc": "2.0",
"method": method,
"params": params or {},
}
# JSON-RPC 通知不带 id
if not is_notification:
payload["id"] = str(uuid.uuid4())
headers = {}
if self._session_id:
headers["mcp-session-id"] = self._session_id
try:
resp = self.session.post(
self.base_url, json=payload, timeout=MCP_TIMEOUT, headers=headers
)
# 保存 session id
if "mcp-session-id" in resp.headers:
self._session_id = resp.headers["mcp-session-id"]
resp.raise_for_status()
# 通知不一定有响应体
if is_notification:
return {"status": "notified"}
data = resp.json()
if "error" in data:
logger.error("MCP error: %s", data["error"])
return {"error": data["error"]}
return data.get("result", data)
except requests.exceptions.ConnectionError:
logger.error("MCP 服务未启动或无法连接: %s", self.base_url)
return {"error": "MCP 服务未启动,请先启动 xiaohongshu-mcp"}
except requests.exceptions.Timeout:
logger.error("MCP 请求超时")
return {"error": "MCP 请求超时,请稍后重试"}
except Exception as e:
logger.error("MCP 调用异常: %s", e)
return {"error": str(e)}
def _ensure_initialized(self):
"""确保 MCP 连接已初始化"""
if not self._initialized:
result = self._call("initialize", {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "xhs-autobot", "version": "2.0.0"}
})
if "error" not in result:
# 发送 initialized 通知(JSON-RPC 通知不带 id)
self._call("notifications/initialized", {},
is_notification=True)
self._initialized = True
return result
return {"status": "already_initialized"}
def _reset(self):
"""重置初始化状态(下次调用会重新握手)"""
self._initialized = False
self._session_id = None
def _call_tool(self, tool_name: str, arguments: dict = None) -> dict:
"""调用 MCP 工具,400 错误时自动重试一次"""
self._ensure_initialized()
result = self._call("tools/call", {
"name": tool_name,
"arguments": arguments or {}
})
# 如果返回 400 相关错误,重置并重试一次
if isinstance(result, dict) and "error" in result:
err_msg = str(result["error"])
if "400" in err_msg or "Bad Request" in err_msg:
logger.warning("MCP 400 错误,重置会话后重试: %s", tool_name)
self._reset()
self._ensure_initialized()
result = self._call("tools/call", {
"name": tool_name,
"arguments": arguments or {}
})
# 提取文本和图片内容
if isinstance(result, dict) and "content" in result:
texts = []
images = []
for item in result["content"]:
if item.get("type") == "text":
texts.append(item["text"])
elif item.get("type") == "image":
# MCP 返回的 base64 图片
img_data = item.get("data", "")
if img_data:
images.append(img_data)
out = {"success": True, "text": "\n".join(texts), "raw": result}
if images:
out["images"] = images
return out
return result
# ---------- 登录 ----------
def get_login_qrcode(self) -> dict:
"""获取登录二维码,返回 {success, text, qr_image(PIL.Image)}"""
result = self._call_tool("get_login_qrcode")
if "error" in result:
return result
# 尝试解析 base64 图片
qr_image = None
if "images" in result and result["images"]:
try:
img_bytes = base64.b64decode(result["images"][0])
qr_image = Image.open(io.BytesIO(img_bytes))
except Exception as e:
logger.warning("二维码图片解析失败: %s", e)
result["qr_image"] = qr_image
return result
def check_login_status(self) -> dict:
"""检查小红书登录状态"""
return self._call_tool("check_login_status")
# ---------- 连接状态 ----------
def check_connection(self) -> tuple[bool, str]:
"""检查 MCP 服务是否可连接"""
result = self._call_tool("check_login_status")
if "error" in result:
return False, result["error"]
return True, result.get("text", "已连接")
# ---------- 搜索 ----------
def search_feeds(self, keyword: str, sort_by: str = "综合",
note_type: str = "不限", publish_time: str = "不限") -> dict:
"""搜索小红书内容"""
args = {
"keyword": keyword,
"filters": {
"sort_by": sort_by,
"note_type": note_type,
"publish_time": publish_time,
}
}
return self._call_tool("search_feeds", args)
# ---------- 推荐列表 ----------
def list_feeds(self) -> dict:
"""获取首页推荐列表"""
return self._call_tool("list_feeds")
# ---------- 笔记列表解析 ----------
@staticmethod
def _parse_feed_entries(text: str) -> list[dict]:
"""从 MCP 返回文本中解析笔记条目为结构化列表"""
entries = []
# 方式1: 尝试直接 JSON 解析
try:
data = json.loads(text)
feeds = []
if isinstance(data, dict) and "feeds" in data:
feeds = data["feeds"]
elif isinstance(data, list):
feeds = data
for feed in feeds:
note = feed.get("noteCard", {})
user = note.get("user", {})
interact = note.get("interactInfo", {})
entries.append({
"feed_id": feed.get("id", ""),
"xsec_token": feed.get("xsecToken", ""),
"title": note.get("displayTitle", "未知标题"),
"author": user.get("nickname", user.get("nickName", "")),
"user_id": user.get("userId", ""),
"likes": interact.get("likedCount", "0"),
"type": note.get("type", ""),
})
if entries:
return entries
except (json.JSONDecodeError, TypeError, AttributeError):
pass
# 方式2: 正则提取 —— 适配 MCP 的文本格式
# 匹配 feed_id (24位十六进制)
feed_ids = re.findall(r'(?:feed_id|id)["\s::]+([0-9a-f]{24})', text, re.I)
# 匹配 xsecToken
tokens = re.findall(r'(?:xsec_?[Tt]oken)["\s::]+([A-Za-z0-9+/=_-]{20,})', text, re.I)
# 匹配标题
titles = re.findall(r'(?:title|标题)["\s::]+(.+?)(?:\n|$)', text, re.I)
# 匹配 userId
user_ids = re.findall(r'(?:user_?[Ii]d|userId)["\s::]+([0-9a-f]{24})', text, re.I)
count = max(len(feed_ids), len(tokens))
for i in range(count):
entries.append({
"feed_id": feed_ids[i] if i < len(feed_ids) else "",
"xsec_token": tokens[i] if i < len(tokens) else "",
"title": titles[i].strip() if i < len(titles) else f"笔记 {i+1}",
"author": "",
"user_id": user_ids[i] if i < len(user_ids) else "",
"likes": "",
"type": "",
})
return entries
def list_feeds_parsed(self) -> list[dict]:
"""获取首页推荐并解析为结构化列表"""
result = self.list_feeds()
if "error" in result:
return []
return self._parse_feed_entries(result.get("text", ""))
def search_feeds_parsed(self, keyword: str, sort_by: str = "综合") -> list[dict]:
"""搜索笔记并解析为结构化列表"""
result = self.search_feeds(keyword, sort_by=sort_by)
if "error" in result:
return []
return self._parse_feed_entries(result.get("text", ""))
@staticmethod
def _extract_comment_obj(c: dict) -> dict:
"""从单个评论 JSON 对象提取结构化数据"""
user_info = c.get("userInfo") or c.get("user") or {}
return {
"comment_id": str(c.get("id", c.get("commentId", ""))),
"user_id": user_info.get("userId", user_info.get("user_id", "")),
"nickname": user_info.get("nickname", user_info.get("nickName", "未知")),
"content": c.get("content", ""),
"sub_comment_count": c.get("subCommentCount", 0),
}
@staticmethod
def _find_comment_list(data: dict) -> list:
"""在多种嵌套结构中定位评论列表"""
if not isinstance(data, dict):
return []
# 格式1: {"data": {"comments": {"list": [...]}}} —— 实际 MCP 返回
d = data.get("data", {})
if isinstance(d, dict):
cm = d.get("comments", {})
if isinstance(cm, dict) and "list" in cm:
return cm["list"]
if isinstance(cm, list):
return cm
# 格式2: {"comments": {"list": [...]}}
cm = data.get("comments", {})
if isinstance(cm, dict) and "list" in cm:
return cm["list"]
if isinstance(cm, list):
return cm
# 格式3: {"data": [{...}, ...]} (直接列表)
if isinstance(d, list):
return d
return []
@classmethod
def _parse_comments(cls, text: str) -> list[dict]:
"""从笔记详情文本中解析评论列表为结构化数据
返回: [{comment_id, user_id, nickname, content, sub_comment_count}, ...]
"""
comments = []
# 方式1: 尝试 JSON 解析(支持多种嵌套格式)
try:
data = json.loads(text)
raw_comments = []
if isinstance(data, list):
raw_comments = data
elif isinstance(data, dict):
raw_comments = cls._find_comment_list(data)
for c in raw_comments:
if isinstance(c, dict) and c.get("content"):
comments.append(cls._extract_comment_obj(c))
if comments:
return comments
except (json.JSONDecodeError, TypeError, AttributeError):
pass
# 方式2: 正则提取 —— 仅当 JSON 完全失败时使用
# 逐个评论块提取,避免跨评论字段错位
# 匹配 JSON 对象中相邻的 id + content + userInfo 组合
comment_blocks = re.finditer(
r'"id"\s*:\s*"([0-9a-fA-F]{20,26})"[^}]*?'
r'"content"\s*:\s*"([^"]{1,500})"[^}]*?'
r'"userInfo"\s*:\s*\{[^}]*?"userId"\s*:\s*"([0-9a-fA-F]{20,26})"'
r'[^}]*?"nickname"\s*:\s*"([^"]{1,30})"',
text, re.DOTALL
)
for m in comment_blocks:
comments.append({
"comment_id": m.group(1),
"user_id": m.group(3),
"nickname": m.group(4),
"content": m.group(2),
"sub_comment_count": 0,
})
return comments
# ---------- 帖子详情 ----------
def get_feed_detail(self, feed_id: str, xsec_token: str,
load_all_comments: bool = False) -> dict:
"""获取笔记详情"""
args = {
"feed_id": feed_id,
"xsec_token": xsec_token,
"load_all_comments": load_all_comments,
}
return self._call_tool("get_feed_detail", args)
def get_feed_comments(self, feed_id: str, xsec_token: str,
load_all: bool = True) -> list[dict]:
"""获取笔记评论列表(结构化)
直接返回解析好的评论列表,优先从 raw JSON 解析
"""
result = self.get_feed_detail(feed_id, xsec_token, load_all_comments=load_all)
if "error" in result:
return []
# 优先从 raw 结构中直接提取
raw = result.get("raw", {})
if raw and isinstance(raw, dict):
for item in raw.get("content", []):
if item.get("type") == "text":
try:
data = json.loads(item["text"])
comment_list = self._find_comment_list(data)
if comment_list:
return [self._extract_comment_obj(c)
for c in comment_list
if isinstance(c, dict) and c.get("content")]
except (json.JSONDecodeError, KeyError, TypeError):
pass
# 回退到 text 解析
text = result.get("text", "")
return self._parse_comments(text) if text else []
# ---------- 发布 ----------
def publish_content(self, title: str, content: str, images: list[str],
tags: list[str] = None, schedule_at: str = None) -> dict:
"""发布图文内容"""
args = {
"title": title,
"content": content,
"images": images,
}
if tags:
args["tags"] = tags
if schedule_at:
args["schedule_at"] = schedule_at
return self._call_tool("publish_content", args)
def publish_video(self, title: str, content: str, video_path: str,
tags: list[str] = None, schedule_at: str = None) -> dict:
"""发布视频内容"""
args = {
"title": title,
"content": content,
"video": video_path,
}
if tags:
args["tags"] = tags
if schedule_at:
args["schedule_at"] = schedule_at
return self._call_tool("publish_with_video", args)
# ---------- 评论 ----------
def post_comment(self, feed_id: str, xsec_token: str, comment: str) -> dict:
"""发表评论"""
return self._call_tool("post_comment_to_feed", {
"feed_id": feed_id,
"xsec_token": xsec_token,
"content": comment,
})
def reply_comment(self, feed_id: str, xsec_token: str,
comment_id: str, user_id: str, content: str) -> dict:
"""回复评论"""
return self._call_tool("reply_comment_in_feed", {
"feed_id": feed_id,
"xsec_token": xsec_token,
"comment_id": comment_id,
"user_id": user_id,
"content": content,
})
# ---------- 互动 ----------
def like_feed(self, feed_id: str, xsec_token: str, unlike: bool = False) -> dict:
"""点赞/取消点赞"""
return self._call_tool("like_feed", {
"feed_id": feed_id,
"xsec_token": xsec_token,
"unlike": unlike,
})
def favorite_feed(self, feed_id: str, xsec_token: str,
unfavorite: bool = False) -> dict:
"""收藏/取消收藏"""
return self._call_tool("favorite_feed", {
"feed_id": feed_id,
"xsec_token": xsec_token,
"unfavorite": unfavorite,
})
# ---------- 用户 ----------
def get_user_profile(self, user_id: str, xsec_token: str) -> dict:
"""获取用户主页信息"""
return self._call_tool("user_profile", {
"user_id": user_id,
"xsec_token": xsec_token,
})
# ---------- 登录管理 ----------
def delete_cookies(self) -> dict:
"""删除 cookies,重置登录状态"""
return self._call_tool("delete_cookies", {})
+1 -1
View File
@@ -4,7 +4,7 @@ services/persona.py
"""
import random
import logging
from config_manager import ConfigManager
from .config_manager import ConfigManager
logger = logging.getLogger("autobot")
cfg = ConfigManager()
+1 -1
View File
@@ -9,7 +9,7 @@ import logging
import matplotlib
import matplotlib.pyplot as plt
from mcp_client import get_mcp_client
from .mcp_client import get_mcp_client
_font_candidates = ["Microsoft YaHei", "SimHei", "PingFang SC", "WenQuanYi Micro Hei"]
for _fn in _font_candidates:
+629
View File
@@ -0,0 +1,629 @@
"""
发布队列模块
SQLite 持久化的内容排期 + 发布队列,支持草稿预审、定时发布、失败重试
"""
import sqlite3
import json
import os
import time
import logging
import threading
import base64
from datetime import datetime, timedelta
from typing import Optional
logger = logging.getLogger(__name__)
# 队列项状态
STATUS_DRAFT = "draft" # 草稿 — 待审核
STATUS_APPROVED = "approved" # 已审核 — 待排期或立即可发布
STATUS_SCHEDULED = "scheduled" # 已排期 — 定时发布
STATUS_PUBLISHING = "publishing" # 发布中
STATUS_PUBLISHED = "published" # 已发布
STATUS_FAILED = "failed" # 发布失败
STATUS_REJECTED = "rejected" # 已拒绝/丢弃
ALL_STATUSES = [STATUS_DRAFT, STATUS_APPROVED, STATUS_SCHEDULED,
STATUS_PUBLISHING, STATUS_PUBLISHED, STATUS_FAILED, STATUS_REJECTED]
STATUS_LABELS = {
STATUS_DRAFT: "📝 草稿",
STATUS_APPROVED: "✅ 待发布",
STATUS_SCHEDULED: "🕐 已排期",
STATUS_PUBLISHING: "🚀 发布中",
STATUS_PUBLISHED: "✅ 已发布",
STATUS_FAILED: "❌ 失败",
STATUS_REJECTED: "🚫 已拒绝",
}
MAX_RETRIES = 2
class PublishQueue:
"""发布队列管理器 (SQLite 持久化)"""
def __init__(self, workspace_dir: str):
self.db_path = os.path.join(workspace_dir, "publish_queue.db")
os.makedirs(workspace_dir, exist_ok=True)
self._init_db()
# 发布中状态恢复 (启动时把 publishing → failed)
self._recover_stale()
def _get_conn(self) -> sqlite3.Connection:
conn = sqlite3.connect(self.db_path, timeout=10)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
return conn
def _init_db(self):
"""初始化数据库表"""
conn = self._get_conn()
try:
conn.execute("""
CREATE TABLE IF NOT EXISTS queue (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
content TEXT NOT NULL DEFAULT '',
sd_prompt TEXT DEFAULT '',
tags TEXT DEFAULT '[]',
image_paths TEXT DEFAULT '[]',
backup_dir TEXT DEFAULT '',
status TEXT NOT NULL DEFAULT 'draft',
scheduled_time TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
published_at TEXT,
topic TEXT DEFAULT '',
style TEXT DEFAULT '',
persona TEXT DEFAULT '',
error_message TEXT DEFAULT '',
retry_count INTEGER DEFAULT 0,
publish_result TEXT DEFAULT ''
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_queue_status ON queue(status)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_queue_scheduled ON queue(scheduled_time)
""")
conn.commit()
finally:
conn.close()
def _recover_stale(self):
"""启动时将残留的 publishing 状态恢复为 failed"""
conn = self._get_conn()
try:
conn.execute(
"UPDATE queue SET status = ?, error_message = '程序重启,发布中断' "
"WHERE status = ?",
(STATUS_FAILED, STATUS_PUBLISHING),
)
conn.commit()
finally:
conn.close()
# ---------- CRUD ----------
def add(self, title: str, content: str, sd_prompt: str = "",
tags: list = None, image_paths: list = None,
backup_dir: str = "", topic: str = "", style: str = "",
persona: str = "", status: str = STATUS_DRAFT,
scheduled_time: str = None) -> int:
"""添加一个队列项,返回 ID"""
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
conn = self._get_conn()
try:
cur = conn.execute(
"""INSERT INTO queue (title, content, sd_prompt, tags, image_paths,
backup_dir, status, scheduled_time, created_at, updated_at,
topic, style, persona)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(title, content, sd_prompt,
json.dumps(tags or [], ensure_ascii=False),
json.dumps(image_paths or [], ensure_ascii=False),
backup_dir, status, scheduled_time, now, now,
topic, style, persona),
)
conn.commit()
item_id = cur.lastrowid
logger.info("📋 队列添加 #%d: %s [%s]", item_id, title[:20], status)
return item_id
finally:
conn.close()
def get(self, item_id: int) -> Optional[dict]:
"""获取单个队列项"""
conn = self._get_conn()
try:
row = conn.execute("SELECT * FROM queue WHERE id = ?", (item_id,)).fetchone()
return self._row_to_dict(row) if row else None
finally:
conn.close()
def update_status(self, item_id: int, status: str,
error_message: str = "", publish_result: str = ""):
"""更新状态"""
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
conn = self._get_conn()
try:
fields = "status = ?, updated_at = ?"
params = [status, now]
if error_message:
fields += ", error_message = ?"
params.append(error_message)
if publish_result:
fields += ", publish_result = ?"
params.append(publish_result)
if status == STATUS_PUBLISHED:
fields += ", published_at = ?"
params.append(now)
params.append(item_id)
conn.execute(f"UPDATE queue SET {fields} WHERE id = ?", params)
conn.commit()
finally:
conn.close()
def update_content(self, item_id: int, title: str = None, content: str = None,
sd_prompt: str = None, tags: list = None,
scheduled_time: str = None):
"""更新内容 (仅 draft/approved/scheduled/failed 状态可编辑)"""
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
conn = self._get_conn()
try:
row = conn.execute("SELECT status FROM queue WHERE id = ?", (item_id,)).fetchone()
if not row or row["status"] in (STATUS_PUBLISHING, STATUS_PUBLISHED):
return False
sets, params = ["updated_at = ?"], [now]
if title is not None:
sets.append("title = ?"); params.append(title)
if content is not None:
sets.append("content = ?"); params.append(content)
if sd_prompt is not None:
sets.append("sd_prompt = ?"); params.append(sd_prompt)
if tags is not None:
sets.append("tags = ?"); params.append(json.dumps(tags, ensure_ascii=False))
if scheduled_time is not None:
sets.append("scheduled_time = ?"); params.append(scheduled_time)
params.append(item_id)
conn.execute(f"UPDATE queue SET {', '.join(sets)} WHERE id = ?", params)
conn.commit()
return True
finally:
conn.close()
def delete(self, item_id: int) -> bool:
"""删除队列项 (仅非 publishing 状态可删)"""
conn = self._get_conn()
try:
row = conn.execute("SELECT status FROM queue WHERE id = ?", (item_id,)).fetchone()
if not row or row["status"] == STATUS_PUBLISHING:
return False
conn.execute("DELETE FROM queue WHERE id = ?", (item_id,))
conn.commit()
return True
finally:
conn.close()
def approve(self, item_id: int, scheduled_time: str = None) -> bool:
"""审核通过 → 进入待发布或排期"""
conn = self._get_conn()
try:
row = conn.execute("SELECT status FROM queue WHERE id = ?", (item_id,)).fetchone()
if not row or row["status"] not in (STATUS_DRAFT, STATUS_FAILED):
return False
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
new_status = STATUS_SCHEDULED if scheduled_time else STATUS_APPROVED
conn.execute(
"UPDATE queue SET status = ?, scheduled_time = ?, updated_at = ?, "
"error_message = '', retry_count = 0 WHERE id = ?",
(new_status, scheduled_time, now, item_id),
)
conn.commit()
return True
finally:
conn.close()
def reject(self, item_id: int) -> bool:
"""拒绝/丢弃"""
return self._set_status_if(item_id, STATUS_REJECTED,
allowed_from=[STATUS_DRAFT, STATUS_APPROVED, STATUS_SCHEDULED, STATUS_FAILED])
def retry(self, item_id: int) -> bool:
"""失败项重试"""
conn = self._get_conn()
try:
row = conn.execute("SELECT status, retry_count FROM queue WHERE id = ?", (item_id,)).fetchone()
if not row or row["status"] != STATUS_FAILED:
return False
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
conn.execute(
"UPDATE queue SET status = ?, updated_at = ?, error_message = '' WHERE id = ?",
(STATUS_APPROVED, now, item_id),
)
conn.commit()
return True
finally:
conn.close()
def _set_status_if(self, item_id: int, new_status: str, allowed_from: list) -> bool:
conn = self._get_conn()
try:
row = conn.execute("SELECT status FROM queue WHERE id = ?", (item_id,)).fetchone()
if not row or row["status"] not in allowed_from:
return False
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
conn.execute("UPDATE queue SET status = ?, updated_at = ? WHERE id = ?",
(new_status, now, item_id))
conn.commit()
return True
finally:
conn.close()
# ---------- 查询 ----------
def list_by_status(self, statuses: list = None, limit: int = 50) -> list[dict]:
"""按状态查询队列项"""
conn = self._get_conn()
try:
if statuses:
placeholders = ",".join("?" * len(statuses))
rows = conn.execute(
f"SELECT * FROM queue WHERE status IN ({placeholders}) "
"ORDER BY created_at DESC LIMIT ?",
statuses + [limit],
).fetchall()
else:
rows = conn.execute(
"SELECT * FROM queue ORDER BY created_at DESC LIMIT ?", (limit,)
).fetchall()
return [self._row_to_dict(r) for r in rows]
finally:
conn.close()
def get_pending_publish(self) -> list[dict]:
"""获取待发布项: approved 或 scheduled 且已到时间"""
conn = self._get_conn()
try:
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
rows = conn.execute(
"""SELECT * FROM queue
WHERE (status = ? OR (status = ? AND scheduled_time <= ?))
ORDER BY
CASE WHEN scheduled_time IS NOT NULL THEN scheduled_time
ELSE created_at END ASC
LIMIT 10""",
(STATUS_APPROVED, STATUS_SCHEDULED, now),
).fetchall()
return [self._row_to_dict(r) for r in rows]
finally:
conn.close()
def count_by_status(self) -> dict:
"""统计各状态数量"""
conn = self._get_conn()
try:
rows = conn.execute(
"SELECT status, COUNT(*) as cnt FROM queue GROUP BY status"
).fetchall()
return {r["status"]: r["cnt"] for r in rows}
finally:
conn.close()
def get_calendar_data(self, days: int = 30) -> list[dict]:
"""获取日历数据 (最近 N 天的发布/排期概览)"""
conn = self._get_conn()
try:
cutoff = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S")
rows = conn.execute(
"""SELECT id, title, status, scheduled_time, published_at, created_at
FROM queue
WHERE created_at >= ? OR scheduled_time >= ? OR published_at >= ?
ORDER BY COALESCE(scheduled_time, published_at, created_at) ASC""",
(cutoff, cutoff, cutoff),
).fetchall()
return [dict(r) for r in rows]
finally:
conn.close()
# ---------- 辅助 ----------
@staticmethod
def _row_to_dict(row: sqlite3.Row) -> dict:
"""Row → dict, 并解析 JSON 字段"""
d = dict(row)
for key in ("tags", "image_paths"):
if key in d and isinstance(d[key], str):
try:
d[key] = json.loads(d[key])
except json.JSONDecodeError:
d[key] = []
return d
def format_queue_table(self, statuses: list = None, limit: int = 30) -> str:
"""生成 Markdown 格式的队列表格"""
items = self.list_by_status(statuses, limit)
if not items:
return "📭 队列为空"
lines = ["| # | 状态 | 标题 | 主题 | 排期时间 | 创建时间 |",
"|---|------|------|------|----------|----------|"]
for item in items:
status_label = STATUS_LABELS.get(item["status"], item["status"])
sched = item.get("scheduled_time") or "—"
if sched != "—":
sched = sched[:16] # 去掉秒
created = item["created_at"][:16] if item.get("created_at") else "—"
title_short = (item.get("title") or "")[:18]
topic_short = (item.get("topic") or "")[:10]
lines.append(f"| {item['id']} | {status_label} | {title_short} | {topic_short} | {sched} | {created} |")
# 统计摘要
counts = self.count_by_status()
summary_parts = []
for s, label in STATUS_LABELS.items():
cnt = counts.get(s, 0)
if cnt > 0:
summary_parts.append(f"{label}: {cnt}")
summary = " · ".join(summary_parts) if summary_parts else "全部为空"
return f"**队列统计**: {summary}\n\n" + "\n".join(lines)
def format_calendar(self, days: int = 14) -> str:
"""生成简易日历视图 (Markdown)"""
data = self.get_calendar_data(days)
if not data:
return "📅 暂无排期数据"
# 按日期分组
by_date = {}
for item in data:
# 优先用排期时间,其次发布时间,最后创建时间
dt_str = item.get("scheduled_time") or item.get("published_at") or item["created_at"]
date_key = dt_str[:10] if dt_str else "未知"
by_date.setdefault(date_key, []).append(item)
lines = ["### 📅 内容日历 (近 %d 天)\n" % days]
today = datetime.now().strftime("%Y-%m-%d")
for date_key in sorted(by_date.keys()):
marker = " 📌 **今天**" if date_key == today else ""
lines.append(f"**{date_key}**{marker}")
for item in by_date[date_key]:
status_icon = STATUS_LABELS.get(item["status"], "❓")
time_part = ""
if item.get("scheduled_time"):
time_part = f" ⏰{item['scheduled_time'][11:16]}"
elif item.get("published_at"):
time_part = f" ✅{item['published_at'][11:16]}"
title_short = (item.get("title") or "无标题")[:20]
lines.append(f" - {status_icon} #{item['id']} {title_short}{time_part}")
lines.append("")
return "\n".join(lines)
def format_preview(self, item_id: int) -> str:
"""生成单个项目的详细预览 (Markdown)"""
item = self.get(item_id)
if not item:
return "❌ 未找到该队列项"
status_label = STATUS_LABELS.get(item["status"], item["status"])
tags = item.get("tags", [])
tags_str = " ".join(f"#{t}" for t in tags) if tags else "无标签"
images = item.get("image_paths", [])
img_count = len(images) if images else 0
lines = [
f"## {status_label} #{item['id']}",
f"### 📌 {item.get('title', '无标题')}",
"",
item.get("content", "无正文"),
"",
f"---",
f"**主题**: {item.get('topic', '—')} · **风格**: {item.get('style', '—')}",
f"**标签**: {tags_str}",
f"**图片**: {img_count} 张",
f"**人设**: {(item.get('persona') or '—')[:30]}",
]
if item.get("scheduled_time"):
lines.append(f"**排期**: {item['scheduled_time']}")
if item.get("error_message"):
lines.append(f"**错误**: ❌ {item['error_message']}")
if item.get("backup_dir"):
lines.append(f"**备份**: `{item['backup_dir']}`")
if item.get("sd_prompt"):
lines.append(f"**SD提示词**: {item['sd_prompt'][:100]}{'...' if len(item.get('sd_prompt', '')) > 100 else ''}")
lines.extend([
f"**创建**: {item.get('created_at', '—')}",
f"**更新**: {item.get('updated_at', '—')}",
])
if item.get("published_at"):
lines.append(f"**发布**: {item['published_at']}")
# 展示图片(如有)
if images:
lines.append("\n### 📷 生成的图片\n")
for i, img in enumerate(images):
# 检查图片是否存在
exists = os.path.exists(img)
status_icon = "✅" if exists else "❌"
# 转换为绝对路径
abs_path = os.path.abspath(img) if exists else img
lines.append(f"{i+1}. {status_icon} `{abs_path}`")
# 如果图片存在,将图片转换为 base64 编码嵌入
if exists:
try:
# 读取图片并转换为 base64
with open(abs_path, 'rb') as f:
img_data = f.read()
img_base64 = base64.b64encode(img_data).decode('utf-8')
# 根据文件扩展名确定 MIME 类型
ext = os.path.splitext(abs_path)[1].lower()
mime_types = {
'.jpg': 'image/jpeg',
'.jpeg': 'image/jpeg',
'.png': 'image/png',
'.gif': 'image/gif',
'.webp': 'image/webp',
'.bmp': 'image/bmp'
}
mime_type = mime_types.get(ext, 'image/jpeg')
# 使用 data URI 格式嵌入图片
data_uri = f"data:{mime_type};base64,{img_base64}"
lines.append(f" ![预览图{i+1}]({data_uri})")
lines.append(f" <small>图片大小: {len(img_data) / 1024:.1f} KB</small>")
except Exception as e:
lines.append(f" ⚠️ 无法读取图片: {e}")
else:
lines.append(f" ⚠️ 图片文件不存在")
lines.append("")
return "\n".join(lines)
class QueuePublisher:
"""后台队列发布处理器"""
def __init__(self, queue: PublishQueue):
self.queue = queue
self._running = threading.Event()
self._thread = None
self._publish_fn = None # 由外部注册的发布回调
self._log_fn = None # 日志回调
def set_publish_callback(self, fn):
"""注册发布回调: fn(item: dict) -> (success: bool, message: str)"""
self._publish_fn = fn
def set_log_callback(self, fn):
"""注册日志回调: fn(msg: str)"""
self._log_fn = fn
def _log(self, msg: str):
logger.info(msg)
if self._log_fn:
try:
self._log_fn(msg)
except Exception:
pass
def start(self, check_interval: int = 60):
"""启动后台队列处理"""
if self._running.is_set():
return
self._running.set()
self._thread = threading.Thread(
target=self._loop, args=(check_interval,), daemon=True
)
self._thread.start()
self._log("📋 发布队列处理器已启动")
def stop(self):
"""停止队列处理"""
self._running.clear()
self._log("📋 发布队列处理器已停止")
@property
def is_running(self) -> bool:
return self._running.is_set()
def _loop(self, interval: int):
while self._running.is_set():
try:
self._process_pending()
except Exception as e:
self._log(f"❌ 队列处理异常: {e}")
logger.error("队列处理异常: %s", e, exc_info=True)
# 等待,但可中断
for _ in range(interval):
if not self._running.is_set():
break
time.sleep(1)
def _process_pending(self):
"""处理所有待发布项"""
if not self._publish_fn:
return
pending = self.queue.get_pending_publish()
if not pending:
return
for item in pending:
if not self._running.is_set():
break
item_id = item["id"]
title = item.get("title", "")[:20]
self._log(f"📋 队列发布 #{item_id}: {title}")
# 标记为发布中
self.queue.update_status(item_id, STATUS_PUBLISHING)
try:
success, message = self._publish_fn(item)
if success:
self.queue.update_status(item_id, STATUS_PUBLISHED, publish_result=message)
self._log(f"✅ 队列发布成功 #{item_id}: {title}")
else:
retry_count = item.get("retry_count", 0) + 1
if retry_count <= MAX_RETRIES:
# 还有重试机会 → approved 状态等下一轮
self.queue.update_status(item_id, STATUS_APPROVED, error_message=f"第{retry_count}次失败: {message}")
conn = self.queue._get_conn()
try:
conn.execute("UPDATE queue SET retry_count = ? WHERE id = ?",
(retry_count, item_id))
conn.commit()
finally:
conn.close()
self._log(f"⚠️ #{item_id} 发布失败 (重试 {retry_count}/{MAX_RETRIES}): {message}")
else:
self.queue.update_status(item_id, STATUS_FAILED, error_message=message)
self._log(f"❌ #{item_id} 发布失败已达重试上限: {message}")
except Exception as e:
self.queue.update_status(item_id, STATUS_FAILED, error_message=str(e))
self._log(f"❌ #{item_id} 发布异常: {e}")
# 发布间隔 (模拟真人)
import random
wait = random.randint(5, 15)
self._log(f"⏳ 等待 {wait}s 后处理下一项...")
for _ in range(wait):
if not self._running.is_set():
break
time.sleep(1)
def publish_now(self, item_id: int) -> str:
"""立即发布指定项 (不经过后台循环)"""
if not self._publish_fn:
return "❌ 发布回调未注册"
item = self.queue.get(item_id)
if not item:
return "❌ 未找到队列项"
if item["status"] not in (STATUS_APPROVED, STATUS_SCHEDULED, STATUS_FAILED):
return f"❌ 当前状态 [{STATUS_LABELS.get(item['status'], item['status'])}] 不可发布"
self.queue.update_status(item_id, STATUS_PUBLISHING)
try:
success, message = self._publish_fn(item)
if success:
self.queue.update_status(item_id, STATUS_PUBLISHED, publish_result=message)
return f"✅ 发布成功: {message}"
else:
self.queue.update_status(item_id, STATUS_FAILED, error_message=message)
return f"❌ 发布失败: {message}"
except Exception as e:
self.queue.update_status(item_id, STATUS_FAILED, error_message=str(e))
return f"❌ 发布异常: {e}"
+7 -7
View File
@@ -6,17 +6,17 @@ import os
import time
import logging
from config_manager import ConfigManager, OUTPUT_DIR
from publish_queue import (
from .config_manager import ConfigManager, OUTPUT_DIR
from .publish_queue import (
PublishQueue, QueuePublisher,
STATUS_DRAFT, STATUS_APPROVED, STATUS_SCHEDULED, STATUS_PUBLISHING,
STATUS_PUBLISHED, STATUS_FAILED, STATUS_REJECTED, STATUS_LABELS,
)
from mcp_client import get_mcp_client
from services.connection import _get_llm_config
from services.persona import DEFAULT_TOPICS, DEFAULT_STYLES, _resolve_persona
from services.content import generate_copy, generate_images
from services.rate_limiter import _increment_stat, _clear_error_streak
from .mcp_client import get_mcp_client
from .connection import _get_llm_config
from .persona import DEFAULT_TOPICS, DEFAULT_STYLES, _resolve_persona
from .content import generate_copy, generate_images
from .rate_limiter import _increment_stat, _clear_error_streak
cfg = ConfigManager()
logger = logging.getLogger("autobot")
+11 -11
View File
@@ -12,23 +12,23 @@ from datetime import datetime
from PIL import Image
from config_manager import ConfigManager, OUTPUT_DIR
from llm_service import LLMService
from sd_service import SDService
from mcp_client import get_mcp_client
from services.rate_limiter import (
from .config_manager import ConfigManager, OUTPUT_DIR
from .llm_service import LLMService
from .sd_service import SDService
from .mcp_client import get_mcp_client
from .rate_limiter import (
_op_history, _daily_stats, DAILY_LIMITS,
_reset_daily_stats_if_needed,
_check_daily_limit, _increment_stat, _record_error,
_clear_error_streak, _is_in_cooldown, _is_in_operating_hours,
_get_stats_summary,
)
from services.persona import (
from .persona import (
DEFAULT_TOPICS, DEFAULT_STYLES, DEFAULT_COMMENT_KEYWORDS,
get_persona_keywords, _resolve_persona,
)
from services.connection import _get_llm_config
from analytics_service import AnalyticsService
from .connection import _get_llm_config
from .analytics_service import AnalyticsService
cfg = ConfigManager()
logger = logging.getLogger("autobot")
@@ -51,7 +51,7 @@ _auto_running = threading.Event()
_auto_thread: threading.Thread | None = None
_auto_log: list[str] = []
from services.rate_limiter import (
from .rate_limiter import (
_op_history, _daily_stats, DAILY_LIMITS,
_reset_daily_stats_if_needed,
_check_daily_limit, _increment_stat, _record_error,
@@ -59,7 +59,7 @@ from services.rate_limiter import (
_get_stats_summary,
)
from services.persona import (
from .persona import (
DEFAULT_PERSONAS, RANDOM_PERSONA_LABEL, PERSONA_POOL_MAP,
DEFAULT_TOPICS, DEFAULT_STYLES, DEFAULT_COMMENT_KEYWORDS,
_match_persona_pools, get_persona_topics, get_persona_keywords,
@@ -1068,7 +1068,7 @@ def stop_learn_scheduler():
# ==================================================
# Windows 开机自启管理
# ==================================================
from services.autostart import (
from .autostart import (
is_autostart_enabled, enable_autostart, disable_autostart, toggle_autostart,
)
File diff suppressed because it is too large Load Diff