commit 88faca150d7b914f3c7f64795d89a080c33ceb32
Author: zhoujie <929834232@qq.com>
Date: Sun Feb 8 14:21:50 2026 +0800
✨ feat(project): 初始化小红书AI爆文工坊V2.0项目
- 新增项目配置文件(.gitignore, config.json)和核心文档(Todo.md, mcp.md)
- 实现配置管理模块(config_manager.py)，支持单例模式和自动保存
- 实现LLM服务模块(llm_service.py)，包含文案生成、热点分析、评论回复等Prompt模板
- 实现SD服务模块(sd_service.py)，封装Stable Diffusion WebUI API调用
- 实现MCP客户端模块(mcp_client.py)，封装小红书MCP服务HTTP调用
- 实现主程序(main.py)，构建Gradio界面，包含内容创作、热点探测、评论管家、账号登录、数据看板五大功能模块
- 保留V1版本备份(main_v1_backup.py)供参考
- 添加项目依赖文件(requirements.txt)
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..4287568
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,3 @@
+xhs_workspace
+__pycache__
+*.log
\ No newline at end of file
diff --git a/Todo.md b/Todo.md
new file mode 100644
index 0000000..ab1c0a1
--- /dev/null
+++ b/Todo.md
@@ -0,0 +1 @@
+目前的脚本已经实现了从 “灵感 -> 文案 -> 绘图 -> 发布” 的核心闭环，作为一个个人辅助工具（MVP，最小可行性产品）已经非常出色了。但是，如果要作为一个专业的运营工具或者满足商业化需求，目前的版本还存在明显的短板。我将从 内容质量、运营闭环、账号安全、功能深度 四个维度为你进行全面分析，并给出升级建议。📊 当前功能评分表维度当前得分评价核心流程⭐⭐⭐⭐⭐流程跑通，无需在多个软件间切换，效率极大提升。内容质量⭐⭐⭐LLM 文案通用性强但个性不足；SD 绘图仅支持基础生图，缺乏精细控制。运营功能⭐⭐仅支持“发”，缺乏“看”（数据分析）和“回”（评论互动）。多媒体能力⭐⭐仅支持图片，不支持视频（尽管 MCP 支持）。稳定性⭐⭐⭐依赖本地环境和 Cookie 有效期，缺乏异常重试和账号管理。🔍 深度差距分析与改进建议1. 视觉能力的局限性 (痛点：图片不可控)目前使用的是基础的 txt2img（文生图）。问题：很难控制人物姿势、保持角色一致性（比如同一个博主IP）、或者在特定背景中植入产品。缺口：ControlNet 支持：无法指定姿势（Openpose）或线稿上色（Canny）。LoRA 切换：无法快速切换画风（如：二次元 vs 真实感 vs 胶片风）。Img2Img：无法基于参考图进行修改。💡 改进建议：在 UI 中增加 ControlNet 参数接口，或者增加“风格预设”下拉框（后台自动切换 LoRA）。2. 缺乏“选题与热点”辅助 (痛点：不知道写什么)目前主要依赖用户自己输入“主题”。问题：如果用户不知道最近什么火，写的文章可能没人看。缺口：MCP 搜索能力未利用：xiaohongshu-mcp 有 search_feeds 功能，但脚本里没用。💡 改进建议：增加 “热点探测” tab。逻辑：用户输入关键词 -> 调用 MCP 搜索 -> LLM 分析热门笔记的标题和结构 -> 生成“爆款模仿”方案。3. 缺失视频支持 (痛点：视频流量更大)小红书目前对视频流量扶持很大。问题：目前的 UI 和逻辑只支持图片。虽然 MCP 支持 publish_with_video，但你没接。💡 改进建议：UI 增加“上传视频”或“AI 生成视频”入口。接入 Runway / Luma API 或本地 AnimateDiff 生成几秒的动态视频。4. 运营互动的缺失 (痛点：发完不管)运营小红书，“养号”和“回复” 与发帖一样重要。问题：目前是“射后不理”。缺口：评论管理：无法自动回复评论，无法引导私域。数据反馈：发出去的笔记有多少阅读？脚本里看不到。💡 改进建议：增加 “评论管家” 模块：定期调用 MCP 获取新评论 -> LLM 生成回复 -> 调用 MCP 回复。增加 “数据看板”：调用 user_profile 展示昨日点赞涨粉数。5. 账号矩阵与安全性 (痛点：单点风险)问题：目前是单账号模式。缺口：多账号切换：如果我有 5 个号，需要反复手动扫码或替换 Cookie 文件。定时发布：只能“立即发布”。真正运营需要设定在晚高峰（18:00-21:00）自动发。💡 改进建议：引入简单的 SQLite 数据库或 JSON 文件管理多组 Cookie。引入 APScheduler 库，实现“存入草稿箱，特定时间自动调用 MCP 发布”。🛠️ 下一步升级路线图 (Roadmap)如果你想把这个脚本升级为V2.0 专业版，建议按以下顺序添加功能：第一阶段：补全 MCP 能力 (低成本，高回报)接入搜索功能：在写文案前，先让 AI 看 5 篇同类热门笔记。接入数据面板：在侧边栏显示当前账号粉丝数、获赞数。第二阶段：增强视觉 (提升内容力)SD 进阶：支持上传参考图 (img2img)。本地图库：有时候不想用 AI 图，想混排自己拍的照片，增加“本地上传”按钮。第三阶段：自动化运营 (解放双手)自动回复机器人：根据设定的人设（知性姐姐/毒舌博主）自动回评论。定时任务：设置一个队列，让它自己跑。
\ No newline at end of file
diff --git a/config.json b/config.json
new file mode 100644
index 0000000..52ab15c
--- /dev/null
+++ b/config.json
@@ -0,0 +1,10 @@
+{
+ "api_key": "sk-d212b926f51f4f0f9297629cd2ab77b4",
+ "base_url": "https://api.deepseek.com/v1",
+ "sd_url": "http://127.0.0.1:7860",
+ "mcp_url": "http://localhost:18060/mcp",
+ "model": "deepseek-reasoner",
+ "persona": "温柔知性的时尚博主",
+ "auto_reply_enabled": false,
+ "schedule_enabled": false
+}
\ No newline at end of file
diff --git a/config_manager.py b/config_manager.py
new file mode 100644
index 0000000..fdab8f9
--- /dev/null
+++ b/config_manager.py
@@ -0,0 +1,82 @@
+"""
+配置管理模块
+支持多配置项、默认值回退、自动保存
+"""
+import json
+import os
+import logging
+
+logger = logging.getLogger(__name__)
+
+CONFIG_FILE = "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,
+}
+
+
+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):
+ """保存配置到文件"""
+ try:
+ with open(CONFIG_FILE, "w", encoding="utf-8") as f:
+ json.dump(self._config, f, indent=4, ensure_ascii=False)
+ except IOError as e:
+ logger.error("配置保存失败: %s", e)
+
+ 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)
diff --git a/llm_service.py b/llm_service.py
new file mode 100644
index 0000000..eb47973
--- /dev/null
+++ b/llm_service.py
@@ -0,0 +1,223 @@
+"""
+LLM 服务模块
+封装对 OpenAI 兼容 API 的调用，包含文案生成、热点分析、评论回复等 Prompt
+"""
+import requests
+import json
+import re
+import logging
+
+logger = logging.getLogger(__name__)
+
+# ================= Prompt 模板 =================
+
+PROMPT_COPYWRITING = """
+你是一个小红书爆款内容专家。请根据用户主题生成内容。
+
+【标题规则】(严格执行)：
+1. 长度限制：必须控制在 18 字以内（含Emoji），绝对不能超过 20 字！
+2. 格式要求：Emoji + 爆点关键词 + 核心痛点。
+3. 禁忌：禁止使用"第一"、"最"、"顶级"等绝对化广告法违禁词。
+4. 风格：二极管标题（震惊/后悔/必看/避雷/哭了），具有强烈的点击欲望。
+
+【正文规则】：
+1. 口语化，多用Emoji，分段清晰，不堆砌长句。
+2. 正文控制在 600 字以内（小红书限制 1000 字）。
+3. 结尾必须有 5 个以上相关话题标签(#)。
+
+【绘图 Prompt】：
+生成对应的 Stable Diffusion 英文提示词，强调：masterpiece, best quality, 8k, soft lighting, ins style。
+
+返回 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}。
+有人在你的笔记下评论了，请你用符合人设的口吻回复。
+
+【规则】：
+1. 回复简洁，控制在 50 字以内
+2. 语气亲切自然，像和朋友聊天
+3. 适当加入 1-2 个 Emoji
+4. 如果是质疑，礼貌回应；如果是夸奖，真诚感谢
+
+笔记标题：{post_title}
+用户评论：{comment}
+
+直接返回回复内容，不需要 JSON 格式。
+"""
+
+PROMPT_PROACTIVE_COMMENT = """
+你是一个小红书活跃用户，人设为：{persona}。
+你正在浏览一篇笔记，想要留下一条真诚、有价值的评论，以提升互动和曝光。
+
+【笔记信息】：
+标题：{post_title}
+正文摘要：{post_content}
+
+【已有评论参考（可能为空）】：
+{existing_comments}
+
+【评论规则】：
+1. 评论简洁自然，控制在 30-80 字，不要像机器人
+2. 体现你对笔记内容的真实感受或个人经验
+3. 可以提问、分享类似经历、或表达共鸣
+4. 适当加入 1-2 个 Emoji，不要过多
+5. 不要重复已有评论的观点，找新角度
+6. 不要生硬带货或自我推广
+7. 语气因内容而异：教程类→请教/补充；种草类→分享体验；生活类→表达共鸣
+
+直接返回评论内容，不需要 JSON 格式。
+"""
+
+PROMPT_COPY_WITH_REFERENCE = """
+你是一个小红书爆款内容专家。参考以下热门笔记的风格和结构，创作全新原创内容。
+
+【参考笔记】：
+{reference_notes}
+
+【创作主题】：{topic}
+【风格要求】：{style}
+
+【标题规则】：
+1. 长度限制：必须控制在 18 字以内（含Emoji），绝对不能超过 20 字！
+2. 借鉴参考笔记的标题套路但内容必须原创。
+
+【正文规则】：
+1. 口语化，多用Emoji，分段清晰。
+2. 正文控制在 600 字以内。
+3. 结尾有 5 个以上话题标签(#)。
+
+【绘图 Prompt】：
+生成 Stable Diffusion 英文提示词。
+
+返回 JSON 格式：
+{{"title": "...", "content": "...", "sd_prompt": "...", "tags": ["标签1", "标签2", ...]}}
+"""
+
+
+class LLMService:
+ """LLM API 服务封装"""
+
+ 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
+
+ def _chat(self, system_prompt: str, user_message: str,
+ json_mode: bool = True, temperature: float = 0.8) -> str:
+ """底层聊天接口"""
+ headers = {
+ "Authorization": f"Bearer {self.api_key}",
+ "Content-Type": "application/json",
+ }
+ payload = {
+ "model": self.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"]
+ return content
+ except requests.exceptions.Timeout:
+ raise TimeoutError("LLM 请求超时，请检查网络或换一个模型")
+ except requests.exceptions.HTTPError as e:
+ raise ConnectionError(f"LLM API 错误 ({resp.status_code}): {resp.text[:200]}")
+ except Exception as e:
+ raise RuntimeError(f"LLM 调用异常: {e}")
+
+ def _parse_json(self, text: str) -> dict:
+ """从 LLM 返回文本中解析 JSON"""
+ cleaned = re.sub(r"```json\s*|```", "", text).strip()
+ return json.loads(cleaned)
+
+ # ---------- 业务方法 ----------
+
+ def get_models(self) -> list[str]:
+ """获取可用模型列表"""
+ url = f"{self.base_url}/models"
+ headers = {"Authorization": f"Bearer {self.api_key}"}
+ resp = requests.get(url, headers=headers, timeout=10)
+ resp.raise_for_status()
+ data = resp.json()
+ return [item["id"] for item in data.get("data", [])]
+
+ def generate_copy(self, topic: str, style: str) -> dict:
+ """生成小红书文案"""
+ content = self._chat(
+ PROMPT_COPYWRITING,
+ f"主题：{topic}\n风格：{style}"
+ )
+ data = self._parse_json(content)
+
+ # 强制标题长度限制
+ title = data.get("title", "")
+ if len(title) > 20:
+ title = title[:20]
+ data["title"] = title
+
+ return data
+
+ def generate_copy_with_reference(self, topic: str, style: str,
+ reference_notes: str) -> dict:
+ """参考热门笔记生成文案"""
+ prompt = PROMPT_COPY_WITH_REFERENCE.format(
+ reference_notes=reference_notes, topic=topic, style=style
+ )
+ content = self._chat(prompt, f"请创作关于「{topic}」的小红书笔记")
+ data = self._parse_json(content)
+
+ title = data.get("title", "")
+ if len(title) > 20:
+ data["title"] = title[:20]
+ return data
+
+ def analyze_hotspots(self, feed_data: str) -> dict:
+ """分析热门内容趋势"""
+ prompt = PROMPT_HOTSPOT_ANALYSIS.format(feed_data=feed_data)
+ content = self._chat(prompt, "请分析以上热门笔记数据")
+ return self._parse_json(content)
+
+ 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
+ )
+ return self._chat(prompt, "请生成回复", json_mode=False, temperature=0.9).strip()
+
+ 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 "暂无评论",
+ )
+ return self._chat(prompt, "请生成评论", json_mode=False, temperature=0.9).strip()
diff --git a/main.py b/main.py
new file mode 100644
index 0000000..ad70240
--- /dev/null
+++ b/main.py
@@ -0,0 +1,1131 @@
+"""
+小红书 AI 爆文生产工坊 V2.0
+全自动工作台：灵感 -> 文案 -> 绘图 -> 发布 -> 运营
+"""
+import gradio as gr
+import os
+import re
+import time
+import logging
+import platform
+import subprocess
+from PIL import Image
+
+from config_manager import ConfigManager, OUTPUT_DIR
+from llm_service import LLMService
+from sd_service import SDService, DEFAULT_NEGATIVE
+from mcp_client import MCPClient
+
+# ================= 日志配置 =================
+
+logging.basicConfig(
+ level=logging.INFO,
+ format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+ handlers=[
+ logging.StreamHandler(),
+ logging.FileHandler("autobot.log", encoding="utf-8"),
+ ],
+)
+logger = logging.getLogger("autobot")
+
+# 强制不走代理连接本地服务
+os.environ["NO_PROXY"] = "127.0.0.1,localhost"
+
+# ================= 全局服务初始化 =================
+
+cfg = ConfigManager()
+cfg.ensure_workspace()
+
+mcp = MCPClient(cfg.get("mcp_url", "http://localhost:18060/mcp"))
+
+# ==================================================
+# Tab 1: 内容创作
+# ==================================================
+
+
+def connect_llm(api_key, base_url):
+ """连接 LLM 并获取模型列表"""
+ if not api_key or not base_url:
+ return gr.update(choices=[]), "⚠️ 请先填写 API Key 和 Base URL"
+ try:
+ svc = LLMService(api_key, base_url)
+ models = svc.get_models()
+ cfg.update({"api_key": api_key, "base_url": base_url})
+ return (
+ gr.update(choices=models, value=models[0] if models else None),
+ f"✅ 已连接，加载 {len(models)} 个模型",
+ )
+ except Exception as e:
+ logger.error("LLM 连接失败: %s", e)
+ return gr.update(), f"❌ 连接失败: {e}"
+
+
+def connect_sd(sd_url):
+ """连接 SD 并获取模型列表"""
+ try:
+ svc = SDService(sd_url)
+ ok, msg = svc.check_connection()
+ if ok:
+ models = svc.get_models()
+ cfg.set("sd_url", sd_url)
+ return gr.update(choices=models, value=models[0] if models else None), f"✅ {msg}"
+ return gr.update(choices=[]), f"❌ {msg}"
+ except Exception as e:
+ logger.error("SD 连接失败: %s", e)
+ return gr.update(choices=[]), f"❌ SD 连接失败: {e}"
+
+
+def check_mcp_status(mcp_url):
+ """检查 MCP 连接状态"""
+ try:
+ client = MCPClient(mcp_url)
+ ok, msg = client.check_connection()
+ if ok:
+ cfg.set("mcp_url", mcp_url)
+ return f"✅ MCP 服务正常 - {msg}"
+ return f"❌ {msg}"
+ except Exception as e:
+ return f"❌ MCP 连接失败: {e}"
+
+
+# ==================================================
+# 小红书账号登录
+# ==================================================
+
+
+def get_login_qrcode(mcp_url):
+ """获取小红书登录二维码"""
+ try:
+ client = MCPClient(mcp_url)
+ result = client.get_login_qrcode()
+ if "error" in result:
+ return None, f"❌ 获取二维码失败: {result['error']}"
+ qr_image = result.get("qr_image")
+ msg = result.get("text", "")
+ if qr_image:
+ return qr_image, f"✅ 二维码已生成，请用小红书 App 扫码\n{msg}"
+ return None, f"⚠️ 未获取到二维码图片，MCP 返回:\n{msg}"
+ except Exception as e:
+ logger.error("获取登录二维码失败: %s", e)
+ return None, f"❌ 获取二维码失败: {e}"
+
+
+def check_login(mcp_url):
+ """检查小红书登录状态"""
+ try:
+ client = MCPClient(mcp_url)
+ result = client.check_login_status()
+ if "error" in result:
+ return f"❌ {result['error']}"
+ text = result.get("text", "")
+ if "未登录" in text:
+ return f"🔴 {text}"
+ return f"🟢 {text}"
+ except Exception as e:
+ return f"❌ 检查登录状态失败: {e}"
+
+
+def generate_copy(api_key, base_url, model, topic, style):
+ """生成文案"""
+ if not api_key:
+ return "", "", "", "", "❌ 缺少 API Key"
+ try:
+ svc = LLMService(api_key, base_url, model)
+ data = svc.generate_copy(topic, style)
+ cfg.set("model", model)
+ tags = data.get("tags", [])
+ return (
+ data.get("title", ""),
+ data.get("content", ""),
+ data.get("sd_prompt", ""),
+ ", ".join(tags) if tags else "",
+ "✅ 文案生成完毕",
+ )
+ except Exception as e:
+ logger.error("文案生成失败: %s", e)
+ return "", "", "", "", f"❌ 生成失败: {e}"
+
+
+def generate_images(sd_url, prompt, neg_prompt, model, steps, cfg_scale):
+ """生成图片"""
+ if not model:
+ return None, [], "❌ 未选择 SD 模型"
+ try:
+ svc = SDService(sd_url)
+ images = svc.txt2img(
+ prompt=prompt,
+ negative_prompt=neg_prompt,
+ model=model,
+ steps=int(steps),
+ cfg_scale=float(cfg_scale),
+ )
+ return images, images, f"✅ 生成 {len(images)} 张图片"
+ except Exception as e:
+ logger.error("图片生成失败: %s", e)
+ return None, [], f"❌ 绘图失败: {e}"
+
+
+def one_click_export(title, content, images):
+ """导出文案和图片到本地"""
+ if not title:
+ return "❌ 无法导出：没有标题"
+
+ safe_title = re.sub(r'[\\/*?:"<>|]', "", title)[:20]
+ folder_name = f"{int(time.time())}_{safe_title}"
+ folder_path = os.path.join(OUTPUT_DIR, folder_name)
+ os.makedirs(folder_path, exist_ok=True)
+
+ with open(os.path.join(folder_path, "文案.txt"), "w", encoding="utf-8") as f:
+ f.write(f"{title}\n\n{content}")
+
+ saved_paths = []
+ if images:
+ for idx, img in enumerate(images):
+ path = os.path.join(folder_path, f"图{idx+1}.png")
+ if isinstance(img, Image.Image):
+ img.save(path)
+ saved_paths.append(os.path.abspath(path))
+
+ # 尝试打开文件夹
+ try:
+ abs_path = os.path.abspath(folder_path)
+ if platform.system() == "Windows":
+ os.startfile(abs_path)
+ elif platform.system() == "Darwin":
+ subprocess.call(["open", abs_path])
+ else:
+ subprocess.call(["xdg-open", abs_path])
+ except Exception:
+ pass
+
+ return f"✅ 已导出至: {folder_path} ({len(saved_paths)} 张图片)"
+
+
+def publish_to_xhs(title, content, tags_str, images, local_images, mcp_url, schedule_time):
+ """通过 MCP 发布到小红书"""
+ if not title:
+ return "❌ 缺少标题"
+
+ client = MCPClient(mcp_url)
+
+ # 收集图片路径
+ image_paths = []
+
+ # 先保存 AI 生成的图片到临时目录
+ if images:
+ temp_dir = os.path.join(OUTPUT_DIR, "_temp_publish")
+ os.makedirs(temp_dir, exist_ok=True)
+ for idx, img in enumerate(images):
+ if isinstance(img, Image.Image):
+ path = os.path.abspath(os.path.join(temp_dir, f"ai_{idx}.png"))
+ img.save(path)
+ image_paths.append(path)
+
+ # 添加本地上传的图片
+ if local_images:
+ for img_file in local_images:
+ # Gradio File 组件返回的是 NamedString 或 tempfile path
+ img_path = img_file.name if hasattr(img_file, 'name') else str(img_file)
+ if os.path.exists(img_path):
+ image_paths.append(os.path.abspath(img_path))
+
+ if not image_paths:
+ return "❌ 至少需要 1 张图片才能发布"
+
+ # 解析标签
+ tags = [t.strip().lstrip("#") for t in tags_str.split(",") if t.strip()] if tags_str else None
+
+ # 定时发布
+ schedule = schedule_time if schedule_time and schedule_time.strip() else None
+
+ try:
+ result = client.publish_content(
+ title=title,
+ content=content,
+ images=image_paths,
+ tags=tags,
+ schedule_at=schedule,
+ )
+ if "error" in result:
+ return f"❌ 发布失败: {result['error']}"
+ return f"✅ 发布成功！\n{result.get('text', '')}"
+ except Exception as e:
+ logger.error("发布失败: %s", e)
+ return f"❌ 发布异常: {e}"
+
+
+# ==================================================
+# Tab 2: 热点探测
+# ==================================================
+
+
+def search_hotspots(keyword, sort_by, mcp_url):
+ """搜索小红书热门内容"""
+ if not keyword:
+ return "❌ 请输入搜索关键词", ""
+ try:
+ client = MCPClient(mcp_url)
+ result = client.search_feeds(keyword, sort_by=sort_by)
+ if "error" in result:
+ return f"❌ 搜索失败: {result['error']}", ""
+ text = result.get("text", "无结果")
+ return "✅ 搜索完成", text
+ except Exception as e:
+ logger.error("热点搜索失败: %s", e)
+ return f"❌ 搜索失败: {e}", ""
+
+
+def analyze_and_suggest(api_key, base_url, model, keyword, search_result):
+ """AI 分析热点并给出建议"""
+ if not search_result:
+ return "❌ 请先搜索", "", ""
+ try:
+ svc = LLMService(api_key, base_url, model)
+ analysis = svc.analyze_hotspots(search_result)
+
+ topics = "\n".join(f"• {t}" for t in analysis.get("hot_topics", []))
+ patterns = "\n".join(f"• {p}" for p in analysis.get("title_patterns", []))
+ suggestions = "\n".join(
+ f"**{s['topic']}** - {s['reason']}"
+ for s in analysis.get("suggestions", [])
+ )
+ structure = analysis.get("content_structure", "")
+
+ summary = (
+ f"## 🔥 热门选题\n{topics}\n\n"
+ f"## 📝 标题套路\n{patterns}\n\n"
+ f"## 📐 内容结构\n{structure}\n\n"
+ f"## 💡 推荐选题\n{suggestions}"
+ )
+ return "✅ 分析完成", summary, keyword
+ except Exception as e:
+ logger.error("热点分析失败: %s", e)
+ return f"❌ 分析失败: {e}", "", ""
+
+
+def generate_from_hotspot(api_key, base_url, model, topic_from_hotspot, style, search_result):
+ """基于热点分析生成文案"""
+ if not topic_from_hotspot:
+ return "", "", "", "", "❌ 请先选择或输入选题"
+ try:
+ svc = LLMService(api_key, base_url, model)
+ data = svc.generate_copy_with_reference(
+ topic=topic_from_hotspot,
+ style=style,
+ reference_notes=search_result[:2000], # 截断防止超长
+ )
+ tags = data.get("tags", [])
+ return (
+ data.get("title", ""),
+ data.get("content", ""),
+ data.get("sd_prompt", ""),
+ ", ".join(tags),
+ "✅ 基于热点的文案已生成",
+ )
+ except Exception as e:
+ return "", "", "", "", f"❌ 生成失败: {e}"
+
+
+# ==================================================
+# Tab 3: 评论管家
+# ==================================================
+
+# ---- 共用: 笔记列表缓存 ----
+
+# 主动评论缓存
+_cached_proactive_entries: list[dict] = []
+# 我的笔记评论缓存
+_cached_my_note_entries: list[dict] = []
+
+
+def _fetch_and_cache(keyword, mcp_url, cache_name="proactive"):
+ """通用: 获取笔记列表并缓存"""
+ global _cached_proactive_entries, _cached_my_note_entries
+ try:
+ client = MCPClient(mcp_url)
+ if keyword and keyword.strip():
+ entries = client.search_feeds_parsed(keyword.strip())
+ src = f"搜索「{keyword.strip()}」"
+ else:
+ entries = client.list_feeds_parsed()
+ src = "首页推荐"
+
+ if cache_name == "proactive":
+ _cached_proactive_entries = entries
+ else:
+ _cached_my_note_entries = entries
+
+ if not entries:
+ return gr.update(choices=[], value=None), f"⚠️ 从{src}未找到笔记"
+
+ choices = []
+ for i, e in enumerate(entries):
+ title_short = (e["title"] or "无标题")[:28]
+ label = f"[{i+1}] {title_short} | @{e['author'] or '未知'} | ❤ {e['likes']}"
+ choices.append(label)
+
+ return (
+ gr.update(choices=choices, value=choices[0]),
+ f"✅ 从{src}获取 {len(entries)} 条笔记",
+ )
+ except Exception as e:
+ if cache_name == "proactive":
+ _cached_proactive_entries = []
+ else:
+ _cached_my_note_entries = []
+ return gr.update(choices=[], value=None), f"❌ {e}"
+
+
+def _pick_from_cache(selected, cache_name="proactive"):
+ """通用: 从缓存中提取选中条目的 feed_id / xsec_token / title"""
+ cache = _cached_proactive_entries if cache_name == "proactive" else _cached_my_note_entries
+ if not selected or not cache:
+ return "", "", ""
+ try:
+ idx = int(selected.split("]")[0].replace("[", "")) - 1
+ e = cache[idx]
+ return e["feed_id"], e["xsec_token"], e.get("title", "")
+ except (ValueError, IndexError):
+ return "", "", ""
+
+
+# ---- 模块 A: 主动评论他人 ----
+
+def fetch_proactive_notes(keyword, mcp_url):
+ return _fetch_and_cache(keyword, mcp_url, "proactive")
+
+
+def on_proactive_note_selected(selected):
+ return _pick_from_cache(selected, "proactive")
+
+
+def load_note_for_comment(feed_id, xsec_token, mcp_url):
+ """加载目标笔记详情 (标题+正文+已有评论), 用于 AI 分析"""
+ if not feed_id or not xsec_token:
+ return "❌ 请先选择笔记", "", "", ""
+ try:
+ client = MCPClient(mcp_url)
+ result = client.get_feed_detail(feed_id, xsec_token, load_all_comments=True)
+ if "error" in result:
+ return f"❌ {result['error']}", "", "", ""
+ full_text = result.get("text", "")
+ # 尝试分离正文和评论
+ if "评论" in full_text:
+ parts = full_text.split("评论", 1)
+ content_part = parts[0].strip()
+ comments_part = "评论" + parts[1] if len(parts) > 1 else ""
+ else:
+ content_part = full_text[:500]
+ comments_part = ""
+ return "✅ 笔记内容已加载", content_part[:800], comments_part[:1500], full_text
+ except Exception as e:
+ return f"❌ {e}", "", "", ""
+
+
+def ai_generate_comment(api_key, base_url, model, persona,
+ post_title, post_content, existing_comments):
+ """AI 生成主动评论"""
+ if not api_key or not base_url:
+ return "⚠️ 请先配置 API Key", "❌ LLM 未配置"
+ if not model:
+ return "⚠️ 请先连接 LLM", "❌ 未选模型"
+ if not post_title and not post_content:
+ return "⚠️ 请先加载笔记内容", "❌ 无笔记内容"
+ try:
+ svc = LLMService(api_key, base_url, model)
+ comment = svc.generate_proactive_comment(
+ persona, post_title, post_content[:600], existing_comments[:800]
+ )
+ return comment, "✅ 评论已生成"
+ except Exception as e:
+ logger.error(f"AI 评论生成失败: {e}")
+ return f"生成失败: {e}", f"❌ {e}"
+
+
+def send_comment(feed_id, xsec_token, comment_content, mcp_url):
+ """发送评论到别人的笔记"""
+ if not all([feed_id, xsec_token, comment_content]):
+ return "❌ 缺少必要参数 (笔记ID / token / 评论内容)"
+ try:
+ client = MCPClient(mcp_url)
+ result = client.post_comment(feed_id, xsec_token, comment_content)
+ if "error" in result:
+ return f"❌ {result['error']}"
+ return "✅ 评论已发送！"
+ except Exception as e:
+ return f"❌ {e}"
+
+
+# ---- 模块 B: 回复我的笔记评论 ----
+
+def fetch_my_notes(keyword, mcp_url):
+ return _fetch_and_cache(keyword, mcp_url, "my_notes")
+
+
+def on_my_note_selected(selected):
+ return _pick_from_cache(selected, "my_notes")
+
+
+def fetch_my_note_comments(feed_id, xsec_token, mcp_url):
+ """获取我的笔记的评论列表"""
+ if not feed_id or not xsec_token:
+ return "❌ 请先选择笔记", ""
+ try:
+ client = MCPClient(mcp_url)
+ result = client.get_feed_detail(feed_id, xsec_token, load_all_comments=True)
+ if "error" in result:
+ return f"❌ {result['error']}", ""
+ return "✅ 评论加载完成", result.get("text", "暂无评论")
+ except Exception as e:
+ return f"❌ {e}", ""
+
+
+def ai_reply_comment(api_key, base_url, model, persona, post_title, comment_text):
+ """AI 生成评论回复"""
+ if not api_key or not base_url:
+ return "⚠️ 请先在全局设置中填写 API Key 和 Base URL", "❌ LLM 未配置"
+ if not model:
+ return "⚠️ 请先连接 LLM 并选择模型", "❌ 未选择模型"
+ if not comment_text:
+ return "请输入需要回复的评论内容", "⚠️ 请输入评论"
+ try:
+ svc = LLMService(api_key, base_url, model)
+ reply = svc.generate_reply(persona, post_title, comment_text)
+ return reply, "✅ 回复已生成"
+ except Exception as e:
+ logger.error(f"AI 回复生成失败: {e}")
+ return f"生成失败: {e}", f"❌ {e}"
+
+
+def send_reply(feed_id, xsec_token, reply_content, mcp_url):
+ """发送评论回复"""
+ if not all([feed_id, xsec_token, reply_content]):
+ return "❌ 缺少必要参数"
+ try:
+ client = MCPClient(mcp_url)
+ result = client.post_comment(feed_id, xsec_token, reply_content)
+ if "error" in result:
+ return f"❌ 回复失败: {result['error']}"
+ return "✅ 回复已发送"
+ except Exception as e:
+ return f"❌ 发送失败: {e}"
+
+
+# ==================================================
+# Tab 4: 数据看板
+# ==================================================
+
+
+# 全局缓存: 最近获取的用户列表
+_cached_user_entries: list[dict] = []
+
+
+def fetch_user_list_from_feeds(mcp_url):
+ """从推荐列表中提取用户列表, 供数据看板使用"""
+ global _cached_user_entries
+ try:
+ client = MCPClient(mcp_url)
+ entries = client.list_feeds_parsed()
+
+ # 去重: 按 user_id
+ seen = set()
+ users = []
+ for e in entries:
+ uid = e.get("user_id", "")
+ if uid and uid not in seen:
+ seen.add(uid)
+ users.append({
+ "user_id": uid,
+ "xsec_token": e.get("xsec_token", ""),
+ "nickname": e.get("author", "未知"),
+ })
+
+ _cached_user_entries = users
+
+ if not users:
+ return gr.update(choices=[], value=None), "⚠️ 未找到用户信息"
+
+ choices = [
+ f"@{u['nickname']} ({u['user_id'][:8]}...)"
+ for u in users
+ ]
+ return (
+ gr.update(choices=choices, value=choices[0]),
+ f"✅ 发现 {len(users)} 位用户，请在下拉框中选择",
+ )
+ except Exception as e:
+ _cached_user_entries = []
+ return gr.update(choices=[], value=None), f"❌ {e}"
+
+
+def on_user_selected(selected_user):
+ """用户下拉框选择回调, 自动填充 user_id 和 xsec_token"""
+ global _cached_user_entries
+ if not selected_user or not _cached_user_entries:
+ return gr.update(), gr.update()
+
+ # 匹配 "(user_id_prefix...)"
+ for u in _cached_user_entries:
+ if u["user_id"][:8] in selected_user:
+ return u["user_id"], u["xsec_token"]
+
+ return gr.update(), gr.update()
+
+
+def fetch_user_data(user_id, xsec_token, mcp_url):
+ """获取用户主页数据"""
+ if not user_id or not xsec_token:
+ return "❌ 请填写用户 ID 和 xsec_token", ""
+ try:
+ client = MCPClient(mcp_url)
+ result = client.get_user_profile(user_id, xsec_token)
+ if "error" in result:
+ return f"❌ 获取失败: {result['error']}", ""
+ return "✅ 数据加载完成", result.get("text", "无数据")
+ except Exception as e:
+ return f"❌ 获取数据失败: {e}", ""
+
+
+def fetch_homepage_feeds(mcp_url):
+ """获取首页推荐"""
+ try:
+ client = MCPClient(mcp_url)
+ result = client.list_feeds()
+ if "error" in result:
+ return f"❌ {result['error']}", ""
+ return "✅ 推荐列表已加载", result.get("text", "无数据")
+ except Exception as e:
+ return f"❌ {e}", ""
+
+
+# ==================================================
+# UI 构建
+# ==================================================
+
+config = cfg.all
+
+with gr.Blocks(
+ title="小红书 AI 爆文工坊 V2.0",
+ theme=gr.themes.Soft(),
+ css="""
+ .status-ok { color: #16a34a; font-weight: bold; }
+ .status-err { color: #dc2626; font-weight: bold; }
+ footer { display: none !important; }
+ """,
+) as app:
+ gr.Markdown(
+ "# 🍒 小红书 AI 爆文生产工坊 V2.0\n"
+ "> 灵感 → 文案 → 绘图 → 发布 → 运营，一站式全闭环"
+ )
+
+ # 全局状态
+ state_images = gr.State([])
+ state_search_result = gr.State("")
+
+ # ============ 全局设置栏 ============
+ with gr.Accordion("⚙️ 全局设置 (自动保存)", open=False):
+ with gr.Row():
+ api_key = gr.Textbox(
+ label="LLM API Key", value=config["api_key"],
+ type="password", scale=2,
+ )
+ base_url = gr.Textbox(
+ label="LLM Base URL", value=config["base_url"], scale=2,
+ )
+ mcp_url = gr.Textbox(
+ label="MCP Server URL", value=config["mcp_url"], scale=2,
+ )
+ with gr.Row():
+ sd_url = gr.Textbox(
+ label="SD WebUI URL", value=config["sd_url"], scale=2,
+ )
+ persona = gr.Textbox(
+ label="博主人设（评论回复用）",
+ value=config["persona"], scale=3,
+ )
+ with gr.Row():
+ btn_connect_llm = gr.Button("🔗 连接 LLM", size="sm")
+ btn_connect_sd = gr.Button("🎨 连接 SD", size="sm")
+ btn_check_mcp = gr.Button("📡 检查 MCP", size="sm")
+ with gr.Row():
+ llm_model = gr.Dropdown(
+ label="LLM 模型", value=config["model"],
+ allow_custom_value=True, interactive=True, scale=2,
+ )
+ sd_model = gr.Dropdown(
+ label="SD 模型", allow_custom_value=True,
+ interactive=True, scale=2,
+ )
+ status_bar = gr.Markdown("🔄 等待连接...")
+
+ # ============ Tab 页面 ============
+ with gr.Tabs():
+ # -------- Tab 1: 内容创作 --------
+ with gr.Tab("✨ 内容创作"):
+ with gr.Row():
+ # 左栏：输入
+ with gr.Column(scale=1):
+ gr.Markdown("### 💡 构思")
+ topic = gr.Textbox(label="笔记主题", placeholder="例如：优衣库早春穿搭")
+ style = gr.Dropdown(
+ ["好物种草", "干货教程", "情绪共鸣", "生活Vlog", "测评避雷", "知识科普"],
+ label="风格", value="好物种草",
+ )
+ btn_gen_copy = gr.Button("✨ 第一步：生成文案", variant="primary")
+
+ gr.Markdown("---")
+ gr.Markdown("### 🎨 绘图参数")
+ with gr.Accordion("高级设置", open=False):
+ neg_prompt = gr.Textbox(
+ label="反向提示词", value=DEFAULT_NEGATIVE, lines=2,
+ )
+ steps = gr.Slider(15, 50, value=25, step=1, label="步数")
+ cfg_scale = gr.Slider(1, 15, value=7, step=0.5, label="CFG Scale")
+ btn_gen_img = gr.Button("🎨 第二步：生成图片", variant="primary")
+
+ # 中栏：文案编辑
+ with gr.Column(scale=1):
+ gr.Markdown("### 📝 文案编辑")
+ res_title = gr.Textbox(label="标题 (≤20字)", interactive=True)
+ res_content = gr.TextArea(
+ label="正文 (可手动修改)", lines=12, interactive=True,
+ )
+ res_prompt = gr.TextArea(
+ label="绘图提示词", lines=3, interactive=True,
+ )
+ res_tags = gr.Textbox(
+ label="话题标签 (逗号分隔)", interactive=True,
+ placeholder="穿搭, 春季, 好物种草",
+ )
+
+ # 右栏：预览 & 发布
+ with gr.Column(scale=1):
+ gr.Markdown("### 🖼️ 视觉预览")
+ gallery = gr.Gallery(label="AI 生成图片", columns=2, height=300)
+ local_images = gr.File(
+ label="📁 上传本地图片（可混排）",
+ file_count="multiple",
+ file_types=["image"],
+ )
+
+ gr.Markdown("### 🚀 发布")
+ schedule_time = gr.Textbox(
+ label="定时发布 (可选, ISO8601格式)",
+ placeholder="如 2026-02-08T18:00:00+08:00，留空=立即发布",
+ )
+ with gr.Row():
+ btn_export = gr.Button("📂 导出本地", variant="secondary")
+ btn_publish = gr.Button("🚀 发布到小红书", variant="primary")
+ publish_msg = gr.Markdown("")
+
+ # -------- Tab 2: 热点探测 --------
+ with gr.Tab("🔥 热点探测"):
+ gr.Markdown("### 搜索热门内容 → AI 分析趋势 → 一键借鉴创作")
+ with gr.Row():
+ with gr.Column(scale=1):
+ hot_keyword = gr.Textbox(
+ label="搜索关键词", placeholder="如：春季穿搭",
+ )
+ hot_sort = gr.Dropdown(
+ ["综合", "最新", "最多点赞", "最多评论", "最多收藏"],
+ label="排序", value="综合",
+ )
+ btn_search = gr.Button("🔍 搜索", variant="primary")
+ search_status = gr.Markdown("")
+
+ with gr.Column(scale=2):
+ search_output = gr.TextArea(
+ label="搜索结果", lines=12, interactive=False,
+ )
+
+ with gr.Row():
+ btn_analyze = gr.Button("🧠 AI 分析热点趋势", variant="primary")
+ analysis_status = gr.Markdown("")
+ analysis_output = gr.Markdown(label="分析报告")
+ topic_from_hot = gr.Textbox(
+ label="选择/输入创作选题", placeholder="基于分析选一个方向",
+ )
+
+ with gr.Row():
+ hot_style = gr.Dropdown(
+ ["好物种草", "干货教程", "情绪共鸣", "生活Vlog", "测评避雷"],
+ label="风格", value="好物种草",
+ )
+ btn_gen_from_hot = gr.Button("✨ 基于热点生成文案", variant="primary")
+
+ with gr.Row():
+ hot_title = gr.Textbox(label="生成的标题", interactive=True)
+ hot_content = gr.TextArea(label="生成的正文", lines=8, interactive=True)
+ with gr.Row():
+ hot_prompt = gr.TextArea(label="绘图提示词", lines=3, interactive=True)
+ hot_tags = gr.Textbox(label="标签", interactive=True)
+ hot_gen_status = gr.Markdown("")
+ btn_sync_to_create = gr.Button(
+ "📋 同步到「内容创作」Tab → 绘图 & 发布",
+ variant="primary",
+ )
+
+ # -------- Tab 3: 评论管家 --------
+ with gr.Tab("💬 评论管家"):
+ gr.Markdown("### 智能评论管理：主动评论引流 & 自动回复粉丝")
+
+ with gr.Tabs():
+ # ======== 子 Tab A: 主动评论他人 ========
+ with gr.Tab("✍️ 主动评论引流"):
+ gr.Markdown(
+ "> **流程**：搜索/浏览笔记 → 选择目标 → 加载内容 → "
+ "AI 分析笔记+已有评论自动生成高质量评论 → 一键发送"
+ )
+
+ # 笔记选择器
+ with gr.Row():
+ pro_keyword = gr.Textbox(
+ label="🔍 搜索关键词 (留空则获取推荐)",
+ placeholder="穿搭、美食、旅行…",
+ )
+ btn_pro_fetch = gr.Button("🔍 获取笔记", variant="primary")
+ with gr.Row():
+ pro_selector = gr.Dropdown(
+ label="📋 选择目标笔记",
+ choices=[], interactive=True,
+ )
+ pro_fetch_status = gr.Markdown("")
+
+ # 隐藏字段
+ with gr.Row():
+ pro_feed_id = gr.Textbox(label="笔记 ID", interactive=False)
+ pro_xsec_token = gr.Textbox(label="xsec_token", interactive=False)
+ pro_title = gr.Textbox(label="标题", interactive=False)
+
+ # 加载内容 & AI 分析
+ btn_pro_load = gr.Button("📖 加载笔记内容", variant="secondary")
+ pro_load_status = gr.Markdown("")
+
+ with gr.Row():
+ with gr.Column(scale=1):
+ pro_content = gr.TextArea(
+ label="📄 笔记正文摘要", lines=8, interactive=False,
+ )
+ with gr.Column(scale=1):
+ pro_comments = gr.TextArea(
+ label="💬 已有评论", lines=8, interactive=False,
+ )
+ # 隐藏: 完整文本
+ pro_full_text = gr.Textbox(visible=False)
+
+ gr.Markdown("---")
+ with gr.Row():
+ with gr.Column(scale=1):
+ btn_pro_ai = gr.Button(
+ "🤖 AI 智能生成评论", variant="primary", size="lg",
+ )
+ pro_ai_status = gr.Markdown("")
+ with gr.Column(scale=2):
+ pro_comment_text = gr.TextArea(
+ label="✏️ 评论内容 (可手动修改)", lines=3,
+ interactive=True,
+ placeholder="点击左侧按钮自动生成，也可手动编写",
+ )
+ with gr.Row():
+ btn_pro_send = gr.Button("📩 发送评论", variant="primary")
+ pro_send_status = gr.Markdown("")
+
+ # ======== 子 Tab B: 回复我的评论 ========
+ with gr.Tab("💌 回复粉丝评论"):
+ gr.Markdown(
+ "> **流程**：选择我的笔记 → 加载评论 → "
+ "粘贴要回复的评论 → AI 生成回复 → 一键发送"
+ )
+
+ # 笔记选择器
+ with gr.Row():
+ my_keyword = gr.Textbox(
+ label="🔍 搜索我的笔记关键词 (留空获取推荐)",
+ placeholder="我发布过的笔记关键词…",
+ )
+ btn_my_fetch = gr.Button("🔍 获取笔记", variant="primary")
+ with gr.Row():
+ my_selector = gr.Dropdown(
+ label="📋 选择我的笔记",
+ choices=[], interactive=True,
+ )
+ my_fetch_status = gr.Markdown("")
+
+ with gr.Row():
+ my_feed_id = gr.Textbox(label="笔记 ID", interactive=False)
+ my_xsec_token = gr.Textbox(label="xsec_token", interactive=False)
+ my_title = gr.Textbox(label="笔记标题", interactive=False)
+
+ btn_my_load_comments = gr.Button("📥 加载评论", variant="primary")
+ my_comment_status = gr.Markdown("")
+
+ my_comments_display = gr.TextArea(
+ label="📋 粉丝评论列表", lines=12, interactive=False,
+ )
+
+ gr.Markdown("---")
+ gr.Markdown("#### 📝 回复评论")
+ with gr.Row():
+ with gr.Column(scale=1):
+ my_target_comment = gr.TextArea(
+ label="要回复的评论内容", lines=3,
+ placeholder="从上方评论列表中复制粘贴要回复的评论",
+ )
+ btn_my_ai_reply = gr.Button(
+ "🤖 AI 生成回复", variant="secondary",
+ )
+ my_reply_gen_status = gr.Markdown("")
+ with gr.Column(scale=1):
+ my_reply_content = gr.TextArea(
+ label="回复内容 (可修改)", lines=3,
+ interactive=True,
+ )
+ btn_my_send_reply = gr.Button(
+ "📩 发送回复", variant="primary",
+ )
+ my_reply_status = gr.Markdown("")
+
+ # -------- Tab 4: 账号登录 --------
+ with gr.Tab("🔐 账号登录"):
+ gr.Markdown(
+ "### 小红书账号登录\n"
+ "> 点击获取二维码 → 用小红书 App 扫码 → 确认登录 → 检查状态"
+ )
+ with gr.Row():
+ with gr.Column(scale=1):
+ gr.Markdown(
+ "**操作步骤：**\n"
+ "1. 确保 MCP 服务已启动\n"
+ "2. 点击「获取登录二维码」\n"
+ "3. 用小红书 App 扫码并确认\n"
+ "4. 点击「检查登录状态」验证\n\n"
+ "⚠️ 登录后不要在其他网页端登录同一账号，否则会被踢出"
+ )
+ btn_get_qrcode = gr.Button(
+ "📱 获取登录二维码", variant="primary", size="lg",
+ )
+ btn_check_login = gr.Button(
+ "🔍 检查登录状态", variant="secondary", size="lg",
+ )
+ login_status = gr.Markdown("🔄 等待操作...")
+
+ with gr.Column(scale=1):
+ qr_image = gr.Image(
+ label="扫码登录", height=350, width=350,
+ )
+
+ # -------- Tab 5: 数据看板 --------
+ with gr.Tab("📊 数据看板"):
+ gr.Markdown("### 账号数据概览")
+
+ # ---- 用户选择器 ----
+ gr.Markdown("#### 🔍 快速选择用户 (从推荐列表提取)")
+ with gr.Row():
+ btn_fetch_users = gr.Button(
+ "👥 从推荐获取用户", variant="primary",
+ )
+ user_selector = gr.Dropdown(
+ label="选择用户 (自动填充下方 ID)",
+ choices=[], interactive=True,
+ )
+ user_fetch_status = gr.Markdown("")
+
+ gr.Markdown("---")
+ with gr.Row():
+ with gr.Column(scale=1):
+ data_user_id = gr.Textbox(label="用户 ID")
+ data_xsec_token = gr.Textbox(label="xsec_token")
+ btn_load_profile = gr.Button("📊 加载用户数据", variant="primary")
+ data_status = gr.Markdown("")
+
+ gr.Markdown("---")
+ btn_load_feeds = gr.Button("🏠 查看首页推荐", variant="secondary")
+ feeds_status = gr.Markdown("")
+
+ with gr.Column(scale=2):
+ profile_display = gr.TextArea(
+ label="用户信息 & 笔记数据", lines=15,
+ interactive=False,
+ )
+ feeds_display = gr.TextArea(
+ label="首页推荐", lines=10, interactive=False,
+ )
+
+ # ==================================================
+ # 事件绑定
+ # ==================================================
+
+ # ---- 全局设置 ----
+ btn_connect_llm.click(
+ fn=connect_llm, inputs=[api_key, base_url],
+ outputs=[llm_model, status_bar],
+ )
+ btn_connect_sd.click(
+ fn=connect_sd, inputs=[sd_url],
+ outputs=[sd_model, status_bar],
+ )
+ btn_check_mcp.click(
+ fn=check_mcp_status, inputs=[mcp_url],
+ outputs=[status_bar],
+ )
+
+ # ---- Tab 1: 内容创作 ----
+ btn_gen_copy.click(
+ fn=generate_copy,
+ inputs=[api_key, base_url, llm_model, topic, style],
+ outputs=[res_title, res_content, res_prompt, res_tags, status_bar],
+ )
+
+ btn_gen_img.click(
+ fn=generate_images,
+ inputs=[sd_url, res_prompt, neg_prompt, sd_model, steps, cfg_scale],
+ outputs=[gallery, state_images, status_bar],
+ )
+
+ btn_export.click(
+ fn=one_click_export,
+ inputs=[res_title, res_content, state_images],
+ outputs=[publish_msg],
+ )
+
+ btn_publish.click(
+ fn=publish_to_xhs,
+ inputs=[res_title, res_content, res_tags, state_images,
+ local_images, mcp_url, schedule_time],
+ outputs=[publish_msg],
+ )
+
+ # ---- Tab 2: 热点探测 ----
+ btn_search.click(
+ fn=search_hotspots,
+ inputs=[hot_keyword, hot_sort, mcp_url],
+ outputs=[search_status, search_output],
+ )
+ # 搜索结果同步到 state
+ search_output.change(
+ fn=lambda x: x, inputs=[search_output], outputs=[state_search_result],
+ )
+
+ btn_analyze.click(
+ fn=analyze_and_suggest,
+ inputs=[api_key, base_url, llm_model, hot_keyword, search_output],
+ outputs=[analysis_status, analysis_output, topic_from_hot],
+ )
+
+ btn_gen_from_hot.click(
+ fn=generate_from_hotspot,
+ inputs=[api_key, base_url, llm_model, topic_from_hot, hot_style, search_output],
+ outputs=[hot_title, hot_content, hot_prompt, hot_tags, hot_gen_status],
+ )
+
+ # 同步热点文案到内容创作 Tab
+ btn_sync_to_create.click(
+ fn=lambda t, c, p, tg: (t, c, p, tg, "✅ 已同步到「内容创作」，可切换 Tab 继续绘图和发布"),
+ inputs=[hot_title, hot_content, hot_prompt, hot_tags],
+ outputs=[res_title, res_content, res_prompt, res_tags, status_bar],
+ )
+
+ # ---- Tab 3: 评论管家 ----
+
+ # == 子 Tab A: 主动评论引流 ==
+ btn_pro_fetch.click(
+ fn=fetch_proactive_notes,
+ inputs=[pro_keyword, mcp_url],
+ outputs=[pro_selector, pro_fetch_status],
+ )
+ pro_selector.change(
+ fn=on_proactive_note_selected,
+ inputs=[pro_selector],
+ outputs=[pro_feed_id, pro_xsec_token, pro_title],
+ )
+ btn_pro_load.click(
+ fn=load_note_for_comment,
+ inputs=[pro_feed_id, pro_xsec_token, mcp_url],
+ outputs=[pro_load_status, pro_content, pro_comments, pro_full_text],
+ )
+ btn_pro_ai.click(
+ fn=ai_generate_comment,
+ inputs=[api_key, base_url, llm_model, persona,
+ pro_title, pro_content, pro_comments],
+ outputs=[pro_comment_text, pro_ai_status],
+ )
+ btn_pro_send.click(
+ fn=send_comment,
+ inputs=[pro_feed_id, pro_xsec_token, pro_comment_text, mcp_url],
+ outputs=[pro_send_status],
+ )
+
+ # == 子 Tab B: 回复粉丝评论 ==
+ btn_my_fetch.click(
+ fn=fetch_my_notes,
+ inputs=[my_keyword, mcp_url],
+ outputs=[my_selector, my_fetch_status],
+ )
+ my_selector.change(
+ fn=on_my_note_selected,
+ inputs=[my_selector],
+ outputs=[my_feed_id, my_xsec_token, my_title],
+ )
+ btn_my_load_comments.click(
+ fn=fetch_my_note_comments,
+ inputs=[my_feed_id, my_xsec_token, mcp_url],
+ outputs=[my_comment_status, my_comments_display],
+ )
+ btn_my_ai_reply.click(
+ fn=ai_reply_comment,
+ inputs=[api_key, base_url, llm_model, persona,
+ my_title, my_target_comment],
+ outputs=[my_reply_content, my_reply_gen_status],
+ )
+ btn_my_send_reply.click(
+ fn=send_reply,
+ inputs=[my_feed_id, my_xsec_token, my_reply_content, mcp_url],
+ outputs=[my_reply_status],
+ )
+
+ # ---- Tab 4: 账号登录 ----
+ btn_get_qrcode.click(
+ fn=get_login_qrcode,
+ inputs=[mcp_url],
+ outputs=[qr_image, login_status],
+ )
+ btn_check_login.click(
+ fn=check_login,
+ inputs=[mcp_url],
+ outputs=[login_status],
+ )
+
+ # ---- Tab 5: 数据看板 ----
+ # 从推荐获取用户列表
+ btn_fetch_users.click(
+ fn=fetch_user_list_from_feeds,
+ inputs=[mcp_url],
+ outputs=[user_selector, user_fetch_status],
+ )
+ # 选择用户 -> 自动填充 user_id / xsec_token
+ user_selector.change(
+ fn=on_user_selected,
+ inputs=[user_selector],
+ outputs=[data_user_id, data_xsec_token],
+ )
+
+ btn_load_profile.click(
+ fn=fetch_user_data,
+ inputs=[data_user_id, data_xsec_token, mcp_url],
+ outputs=[data_status, profile_display],
+ )
+
+ btn_load_feeds.click(
+ fn=fetch_homepage_feeds,
+ inputs=[mcp_url],
+ outputs=[feeds_status, feeds_display],
+ )
+
+ # ---- 启动时自动刷新 SD ----
+ app.load(fn=connect_sd, inputs=[sd_url], outputs=[sd_model, status_bar])
+
+
+# ==================================================
+if __name__ == "__main__":
+ logger.info("🍒 小红书 AI 爆文工坊 V2.0 启动中...")
+ app.launch(inbrowser=True, share=False)
diff --git a/main_v1_backup.py b/main_v1_backup.py
new file mode 100644
index 0000000..abbcd12
--- /dev/null
+++ b/main_v1_backup.py
@@ -0,0 +1,264 @@
+import gradio as gr
+import requests
+import json
+import base64
+import io
+import os
+import time
+import re
+import shutil
+import platform
+import subprocess
+from PIL import Image
+
+# ================= 0. 基础配置与工具 =================
+
+# 强制不走代理连接本地 SD
+os.environ['NO_PROXY'] = '127.0.0.1,localhost'
+
+CONFIG_FILE = "config.json"
+OUTPUT_DIR = "xhs_workspace"
+os.makedirs(OUTPUT_DIR, exist_ok=True)
+
+class ConfigManager:
+ @staticmethod
+ def load():
+ if os.path.exists(CONFIG_FILE):
+ try:
+ with open(CONFIG_FILE, 'r', encoding='utf-8') as f:
+ return json.load(f)
+ except:
+ pass
+ return {
+ "api_key": "",
+ "base_url": "https://api.openai.com/v1",
+ "sd_url": "http://127.0.0.1:7860",
+ "model": "gpt-3.5-turbo"
+ }
+
+ @staticmethod
+ def save(config_data):
+ with open(CONFIG_FILE, 'w', encoding='utf-8') as f:
+ json.dump(config_data, f, indent=4, ensure_ascii=False)
+
+# ================= 1. 核心逻辑功能 =================
+
+def get_llm_models(api_key, base_url):
+ if not api_key or not base_url:
+ return gr.update(choices=[]), "⚠️ 请先填写配置"
+ try:
+ url = f"{base_url.rstrip('/')}/models"
+ headers = {"Authorization": f"Bearer {api_key}"}
+ response = requests.get(url, headers=headers, timeout=10)
+ if response.status_code == 200:
+ data = response.json()
+ models = [item['id'] for item in data.get('data', [])]
+
+ # 保存配置
+ cfg = ConfigManager.load()
+ cfg['api_key'] = api_key
+ cfg['base_url'] = base_url
+ ConfigManager.save(cfg)
+
+ # 修复警告：允许自定义值
+ return gr.update(choices=models, value=models[0] if models else None), f"✅ 已连接，加载 {len(models)} 个模型"
+ return gr.update(), f"❌ 连接失败: {response.status_code}"
+ except Exception as e:
+ return gr.update(), f"❌ 错误: {e}"
+
+def generate_copy(api_key, base_url, model, topic, style):
+ if not api_key: return "", "", "", "❌ 缺 API Key"
+
+ # --- 核心修改：优化了 Prompt，增加字数和违禁词限制 ---
+ system_prompt = """
+ 你是一个小红书爆款内容专家。请根据用户主题生成内容。
+
+ 【标题规则】(严格执行)：
+ 1. 长度限制：必须控制在 18 字以内（含Emoji），绝对不能超过 20 字！
+ 2. 格式要求：Emoji + 爆点关键词 + 核心痛点。
+ 3. 禁忌：禁止使用“第一”、“最”、“顶级”等绝对化广告法违禁词。
+ 4. 风格：二极管标题（震惊/后悔/必看/避雷/哭了），具有强烈的点击欲望。
+
+ 【正文规则】：
+ 1. 口语化，多用Emoji，分段清晰，不堆砌长句。
+ 2. 结尾必须有 5 个以上相关话题标签(#)。
+
+ 【绘图 Prompt】：
+ 生成对应的 Stable Diffusion 英文提示词，强调：masterpiece, best quality, 8k, soft lighting, ins style。
+
+ 返回 JSON 格式：
+ {"title": "...", "content": "...", "sd_prompt": "..."}
+ """
+
+ try:
+ headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+ payload = {
+ "model": model,
+ "messages": [
+ {"role": "system", "content": system_prompt},
+ {"role": "user", "content": f"主题：{topic}\n风格：{style}"}
+ ],
+ "response_format": {"type": "json_object"}
+ }
+ resp = requests.post(f"{base_url.rstrip('/')}/chat/completions", headers=headers, json=payload, timeout=60)
+
+ content = resp.json()['choices'][0]['message']['content']
+ content = re.sub(r'```json\s*|```', '', content).strip()
+ data = json.loads(content)
+
+ # --- 双重保险：Python 强制截断 ---
+ title = data.get('title', '')
+ # 如果 LLM 不听话超过了20字，强制截断并保留前19个字+省略号，或者直接保留前20个
+ if len(title) > 20:
+ title = title[:20]
+
+ return title, data.get('content', ''), data.get('sd_prompt', ''), "✅ 文案生成完毕"
+ except Exception as e:
+ return "", "", "", f"❌ 生成失败: {e}"
+
+def get_sd_models(sd_url):
+ try:
+ resp = requests.get(f"{sd_url}/sdapi/v1/sd-models", timeout=3)
+ if resp.status_code == 200:
+ models = [m['title'] for m in resp.json()]
+ return gr.update(choices=models, value=models[0] if models else None), "✅ SD 已连接"
+ return gr.update(choices=[]), "❌ SD 连接失败"
+ except:
+ return gr.update(choices=[]), "❌ SD 未启动或端口错误"
+
+def generate_images(sd_url, prompt, neg_prompt, model, steps, cfg):
+ if not model: return None, "❌ 未选择模型"
+
+ # 切换模型
+ try:
+ requests.post(f"{sd_url}/sdapi/v1/options", json={"sd_model_checkpoint": model})
+ except:
+ pass # 忽略切换错误，继续尝试生成
+
+ payload = {
+ "prompt": prompt,
+ "negative_prompt": neg_prompt,
+ "steps": steps,
+ "cfg_scale": cfg,
+ "width": 768,
+ "height": 1024,
+ "batch_size": 2
+ }
+
+ try:
+ resp = requests.post(f"{sd_url}/sdapi/v1/txt2img", json=payload, timeout=120)
+ images = []
+ for i in resp.json()['images']:
+ img = Image.open(io.BytesIO(base64.b64decode(i)))
+ images.append(img)
+ return images, "✅ 图片生成完毕"
+ except Exception as e:
+ return None, f"❌ 绘图失败: {e}"
+
+def one_click_export(title, content, images):
+ if not title: return "❌ 无法导出：没有标题"
+
+ safe_title = re.sub(r'[\\/*?:"<>|]', "", title)[:20]
+ folder_name = f"{int(time.time())}_{safe_title}"
+ folder_path = os.path.join(OUTPUT_DIR, folder_name)
+ os.makedirs(folder_path, exist_ok=True)
+
+ with open(os.path.join(folder_path, "文案.txt"), "w", encoding="utf-8") as f:
+ f.write(f"{title}\n\n{content}")
+
+ if images:
+ for idx, img in enumerate(images):
+ img.save(os.path.join(folder_path, f"图{idx+1}.png"))
+
+ try:
+ if platform.system() == "Windows":
+ os.startfile(folder_path)
+ elif platform.system() == "Darwin":
+ subprocess.call(["open", folder_path])
+ else:
+ subprocess.call(["xdg-open", folder_path])
+ return f"✅ 已导出至: {folder_path}"
+ except:
+ return f"✅ 已导出: {folder_path}"
+
+# ================= 2. UI 界面构建 =================
+
+cfg = ConfigManager.load()
+
+with gr.Blocks(title="小红书全自动工作台", theme=gr.themes.Soft()) as app:
+ gr.Markdown("## 🍒 小红书 AI 爆文生产工坊")
+
+ state_images = gr.State([])
+
+ with gr.Row():
+ with gr.Column(scale=1):
+ with gr.Accordion("⚙️ 系统设置 (自动保存)", open=True):
+ api_key = gr.Textbox(label="LLM API Key", value=cfg['api_key'], type="password")
+ base_url = gr.Textbox(label="Base URL", value=cfg['base_url'])
+ sd_url = gr.Textbox(label="SD URL", value=cfg['sd_url'])
+
+ with gr.Row():
+ btn_connect = gr.Button("🔗 连接并获取模型", size="sm")
+ btn_refresh_sd = gr.Button("🔄 刷新 SD", size="sm")
+
+ # 修复点 1：允许自定义值，防止报错
+ llm_model = gr.Dropdown(label="选择 LLM 模型", value=cfg['model'], allow_custom_value=True, interactive=True)
+ sd_model = gr.Dropdown(label="选择 SD 模型", allow_custom_value=True, interactive=True)
+ status_bar = gr.Markdown("等待就绪...")
+
+ gr.Markdown("### 💡 内容构思")
+ topic = gr.Textbox(label="笔记主题", placeholder="例如：优衣库早春穿搭")
+ style = gr.Dropdown(["好物种草", "干货教程", "情绪共鸣", "生活Vlog"], label="风格", value="好物种草")
+ btn_step1 = gr.Button("✨ 第一步：生成文案方案", variant="primary")
+
+ with gr.Column(scale=1):
+ gr.Markdown("### 📝 文案确认")
+ # 修复点 2：去掉了 show_copy_button 参数，兼容旧版 Gradio
+ res_title = gr.Textbox(label="标题 (AI生成)", interactive=True)
+ res_content = gr.TextArea(label="正文 (AI生成)", lines=10, interactive=True)
+ res_prompt = gr.TextArea(label="绘图提示词", lines=4, interactive=True)
+
+ with gr.Accordion("🎨 绘图参数", open=False):
+ neg_prompt = gr.Textbox(label="反向词", value="nsfw, lowres, bad anatomy, text, error")
+ steps = gr.Slider(15, 50, value=25, label="步数")
+ cfg_scale = gr.Slider(1, 15, value=7, label="相关性 (CFG)")
+
+ btn_step2 = gr.Button("🎨 第二步：开始绘图", variant="primary")
+
+ with gr.Column(scale=1):
+ gr.Markdown("### 🖼️ 视觉结果")
+ gallery = gr.Gallery(label="生成预览", columns=1, height="auto")
+ btn_export = gr.Button("📂 一键导出 (文案+图片)", variant="stop")
+ export_msg = gr.Markdown("")
+
+ # ================= 3. 事件绑定 =================
+
+ btn_connect.click(fn=get_llm_models, inputs=[api_key, base_url], outputs=[llm_model, status_bar])
+ btn_refresh_sd.click(fn=get_sd_models, inputs=[sd_url], outputs=[sd_model, status_bar])
+
+ btn_step1.click(
+ fn=generate_copy,
+ inputs=[api_key, base_url, llm_model, topic, style],
+ outputs=[res_title, res_content, res_prompt, status_bar]
+ )
+
+ def on_img_gen(sd_url, p, np, m, s, c):
+ imgs, msg = generate_images(sd_url, p, np, m, s, c)
+ return imgs, imgs, msg
+
+ btn_step2.click(
+ fn=on_img_gen,
+ inputs=[sd_url, res_prompt, neg_prompt, sd_model, steps, cfg_scale],
+ outputs=[gallery, state_images, status_bar]
+ )
+
+ btn_export.click(
+ fn=one_click_export,
+ inputs=[res_title, res_content, state_images],
+ outputs=[export_msg]
+ )
+
+ app.load(fn=get_sd_models, inputs=[sd_url], outputs=[sd_model, status_bar])
+
+if __name__ == "__main__":
+ app.launch(inbrowser=True)
\ No newline at end of file
diff --git a/mcp.md b/mcp.md
new file mode 100644
index 0000000..b5e0e75
--- /dev/null
+++ b/mcp.md
@@ -0,0 +1,857 @@
+# xiaohongshu-mcp
+
+
+[](#contributors-)
+
+
+[](./DONATIONS.md)
+[](./DONATIONS.md)
+[](https://hub.docker.com/r/xpzouying/xiaohongshu-mcp)
+
+MCP for 小红书/xiaohongshu.com。
+
+- 我的博客文章：[haha.ai/xiaohongshu-mcp](https://www.haha.ai/xiaohongshu-mcp)
+
+**遇到任何问题，务必要先看 [各种疑难杂症](https://github.com/xpzouying/xiaohongshu-mcp/issues/56)**。
+
+上面的 **疑难杂症** 列表后，还是解决不了你的部署问题，那么强烈推荐使用我写的另外一个工具：[xpzouying/x-mcp](https://github.com/xpzouying/x-mcp)，这个工具不需要你进行部署，只需要通过浏览器插件就能驱动你的 MCP，对于非技术同学来说更加友好。
+
+## Star History
+
+[](https://www.star-history.com/#xpzouying/xiaohongshu-mcp&Timeline)
+
+## 赞赏支持
+
+本项目所有的赞赏都会用于慈善捐赠。所有的慈善捐赠记录，请参考 [DONATIONS.md](./DONATIONS.md)。
+
+**捐赠时，请备注 MCP 以及名字。**
+如需更正/撤回署名，请开 Issue 或通过邮箱联系。
+
+**支付宝（不展示二维码）：**
+
+通过支付宝向 **xpzouying@gmail.com** 赞赏。
+
+**微信：**
+
+
+
+## 项目简介
+
+**主要功能**
+
+> 💡 **提示：** 点击下方功能标题可展开查看视频演示
+
+
+1. 登录和检查登录状态
+
+第一步必须，小红书需要进行登录。可以检查当前登录状态。
+
+**登录演示：**
+
+https://github.com/user-attachments/assets/8b05eb42-d437-41b7-9235-e2143f19e8b7
+
+**检查登录状态演示：**
+
+https://github.com/user-attachments/assets/bd9a9a4a-58cb-4421-b8f3-015f703ce1f9
+
+
+
+
+2. 发布图文内容
+
+支持发布图文内容到小红书，包括标题、内容描述和图片。
+
+**图片支持方式：**
+
+支持两种图片输入方式：
+
+1. **HTTP/HTTPS 图片链接**
+
+ ```
+ ["https://example.com/image1.jpg", "https://example.com/image2.png"]
+ ```
+
+2. **本地图片绝对路径**（推荐）
+ ```
+ ["/Users/username/Pictures/image1.jpg", "/home/user/images/image2.png"]
+ ```
+
+**为什么推荐使用本地路径：**
+
+- ✅ 稳定性更好，不依赖网络
+- ✅ 上传速度更快
+- ✅ 避免图片链接失效问题
+- ✅ 支持更多图片格式
+
+**发布图文帖子演示：**
+
+https://github.com/user-attachments/assets/8aee0814-eb96-40af-b871-e66e6bbb6b06
+
+
+
+
+3. 发布视频内容
+
+支持发布视频内容到小红书，包括标题、内容描述和本地视频文件。
+
+**视频支持方式：**
+
+仅支持本地视频文件绝对路径：
+
+```
+"/Users/username/Videos/video.mp4"
+```
+
+**功能特点：**
+
+- ✅ 支持本地视频文件上传
+- ✅ 自动处理视频格式转换
+- ✅ 支持标题、内容描述和标签
+- ✅ 等待视频处理完成后自动发布
+
+**注意事项：**
+
+- 仅支持本地视频文件，不支持 HTTP 链接
+- 视频处理时间较长，请耐心等待
+- 建议视频文件大小不超过 1GB
+
+
+
+
+4. 搜索内容
+
+根据关键词搜索小红书内容。
+
+**搜索帖子演示：**
+
+https://github.com/user-attachments/assets/03c5077d-6160-4b18-b629-2e40933a1fd3
+
+
+
+
+5. 获取推荐列表
+
+获取小红书首页推荐内容列表。
+
+**获取推荐列表演示：**
+
+https://github.com/user-attachments/assets/110fc15d-46f2-4cca-bdad-9de5b5b8cc28
+
+
+
+
+6. 获取帖子详情（包括互动数据和评论）
+
+获取小红书帖子的完整详情，包括：
+
+- 帖子内容（标题、描述、图片等）
+- 用户信息
+- 互动数据（点赞、收藏、分享、评论数）
+- 评论列表及子评论
+
+**⚠️ 重要提示：**
+
+- 需要提供帖子 ID 和 xsec_token（两个参数缺一不可）
+- 这两个参数可以从 Feed 列表或搜索结果中获取
+- 必须先登录才能使用此功能
+
+**获取帖子详情演示：**
+
+https://github.com/user-attachments/assets/76a26130-a216-4371-a6b3-937b8fda092a
+
+
+
+
+7. 发表评论到帖子
+
+支持自动发表评论到小红书帖子。
+
+**功能说明：**
+
+- 自动定位评论输入框
+- 输入评论内容并发布
+- 支持 HTTP API 和 MCP 工具调用
+
+**⚠️ 重要提示：**
+
+- 需要先登录才能使用此功能
+- 需要提供帖子 ID、xsec_token 和评论内容
+- 这些参数可以从 Feed 列表或搜索结果中获取
+
+**发表评论演示：**
+
+https://github.com/user-attachments/assets/cc385b6c-422c-489b-a5fc-63e92c695b80
+
+
+
+
+8. 获取用户个人主页
+
+获取小红书用户的个人主页信息，包括用户基本信息和笔记内容。
+
+**功能说明：**
+
+- 获取用户基本信息（昵称、简介、头像等）
+- 获取关注数、粉丝数、获赞量统计
+- 获取用户发布的笔记内容列表
+- 支持 HTTP API 和 MCP 工具调用
+
+**⚠️ 重要提示：**
+
+- 需要先登录才能使用此功能
+- 需要提供用户 ID 和 xsec_token
+- 这些参数可以从 Feed 列表或搜索结果中获取
+
+**返回信息包括：**
+
+- 用户基本信息：昵称、简介、头像、认证状态
+- 统计数据：关注数、粉丝数、获赞量、笔记数
+- 笔记列表：用户发布的所有公开笔记
+
+
+
+**小红书基础运营知识**
+
+- **标题：（非常重要）小红书要求标题不超过 20 个字**
+- **正文：（非常重要）：正文不能超过 1000 个字**
+- 当前支持图文发送以及视频发送：从推荐的角度看，图文的流量会比视频以及纯文字的更好。
+- （低优先级）可以考虑纯文字的支持。1. 个人感觉纯文字会大大增加运营的复杂度；2. 纯文字在我的使用场景的价值较低。
+- Tags：现已支持。添加合适的 Tags 能带来更多的流量。
+- 根据本人实操，小红书每天的发帖量应该是 **50 篇**。
+- **（非常重要）小红书的同一个账号不允许在多个网页端登录**，如果你登录了当前 xiaohongshu-mcp 后，就不要再在其他的网页端登录该账号，否则就会把当前 MCP 的账号“踢出登录”。你可以使用移动 App 端进行查看当前账号信息。
+
+**风险说明**
+
+1. 该项目是在自己的另外一个项目的基础上开源出来的，原来的项目稳定运行一年多，没有出现过封号的情况，只有出现过 Cookies 过期需要重新登录。
+2. 我是使用 Claude Code 接入，稳定自动化运营数周后，验证没有问题后开源。
+
+该项目是基于学习的目的，禁止一切违法行为。
+
+**实操结果**
+
+第一天点赞/收藏数达到了 999+，
+
+
+
+
+
+一周左右的成果
+
+
+
+## 1. 使用教程
+
+### 1.1. 快速开始（推荐）
+
+**方式一：下载预编译二进制文件**
+
+直接从 [GitHub Releases](https://github.com/xpzouying/xiaohongshu-mcp/releases) 下载对应平台的二进制文件：
+
+**主程序（MCP 服务）：**
+
+- **macOS Apple Silicon**: `xiaohongshu-mcp-darwin-arm64`
+- **macOS Intel**: `xiaohongshu-mcp-darwin-amd64`
+- **Windows x64**: `xiaohongshu-mcp-windows-amd64.exe`
+- **Linux x64**: `xiaohongshu-mcp-linux-amd64`
+
+**登录工具：**
+
+- **macOS Apple Silicon**: `xiaohongshu-login-darwin-arm64`
+- **macOS Intel**: `xiaohongshu-login-darwin-amd64`
+- **Windows x64**: `xiaohongshu-login-windows-amd64.exe`
+- **Linux x64**: `xiaohongshu-login-linux-amd64`
+
+使用步骤：
+
+```bash
+# 1. 首先运行登录工具
+chmod +x xiaohongshu-login-darwin-arm64
+./xiaohongshu-login-darwin-arm64
+
+# 2. 然后启动 MCP 服务
+chmod +x xiaohongshu-mcp-darwin-arm64
+./xiaohongshu-mcp-darwin-arm64
+```
+
+**⚠️ 重要提示**：首次运行时会自动下载无头浏览器（约 150MB），请确保网络连接正常。后续运行无需重复下载。
+
+**方式二：源码编译**
+
+
+源码编译安装详情
+
+依赖 Golang 环境，安装方法请参考 [Golang 官方文档](https://go.dev/doc/install)。
+
+设置 Go 国内源的代理，
+
+```bash
+# 配置 GOPROXY 环境变量，以下三选一
+
+# 1. 七牛 CDN
+go env -w GOPROXY=https://goproxy.cn,direct
+
+# 2. 阿里云
+go env -w GOPROXY=https://mirrors.aliyun.com/goproxy/,direct
+
+# 3. 官方
+go env -w GOPROXY=https://goproxy.io,direct
+```
+
+
+
+**方式三：使用 Docker 容器（最简单）**
+
+
+Docker 部署详情
+
+使用 Docker 部署是最简单的方式，无需安装任何开发环境。
+
+**1. 从 Docker Hub 拉取镜像（推荐）**
+
+我们提供了预构建的 Docker 镜像，可以直接从 Docker Hub 拉取使用：
+
+```bash
+# 拉取最新镜像
+docker pull xpzouying/xiaohongshu-mcp
+```
+
+Docker Hub 地址：[https://hub.docker.com/r/xpzouying/xiaohongshu-mcp](https://hub.docker.com/r/xpzouying/xiaohongshu-mcp)
+
+**2. 使用 Docker Compose 启动（推荐）**
+
+我们提供了配置好的 `docker-compose.yml` 文件，可以直接使用：
+
+```bash
+# 下载 docker-compose.yml
+wget https://raw.githubusercontent.com/xpzouying/xiaohongshu-mcp/main/docker/docker-compose.yml
+
+# 或者如果已经克隆了项目，进入 docker 目录
+cd docker
+
+# 启动服务
+docker compose up -d
+
+# 查看日志
+docker compose logs -f
+
+# 停止服务
+docker compose stop
+```
+
+**3. 自己构建镜像（可选）**
+
+```bash
+# 在项目根目录运行
+docker build -t xpzouying/xiaohongshu-mcp .
+```
+
+**4. 配置说明**
+
+Docker 版本会自动：
+
+- 配置 Chrome 浏览器和中文字体
+- 挂载 `./data` 用于存储 cookies
+- 挂载 `./images` 用于存储发布的图片
+- 暴露 18060 端口供 MCP 连接
+
+详细使用说明请参考：[Docker 部署指南](./docker/README.md)
+
+
+
+Windows 遇到问题首先看这里：[Windows 安装指南](./docs/windows_guide.md)
+
+### 1.2. 登录
+
+第一次需要手动登录，需要保存小红书的登录状态。
+
+**使用二进制文件**：
+
+```bash
+# 运行对应平台的登录工具
+./xiaohongshu-login-darwin-arm64
+```
+
+**使用源码**：
+
+```bash
+go run cmd/login/main.go
+```
+
+### 1.3. 启动 MCP 服务
+
+启动 xiaohongshu-mcp 服务。
+
+**使用二进制文件**：
+
+```bash
+# 默认：无头模式，没有浏览器界面
+./xiaohongshu-mcp-darwin-arm64
+
+# 非无头模式，有浏览器界面
+./xiaohongshu-mcp-darwin-arm64 -headless=false
+```
+
+**使用源码**：
+
+```bash
+# 默认：无头模式，没有浏览器界面
+go run .
+
+# 非无头模式，有浏览器界面
+go run . -headless=false
+```
+
+## 1.4. 验证 MCP
+
+```bash
+npx @modelcontextprotocol/inspector
+```
+
+
+
+运行后，打开红色标记的链接，配置 MCP inspector，输入 `http://localhost:18060/mcp` ，点击 `Connect` 按钮。
+
+
+
+按照上面配置 MCP inspector 后，点击 `List Tools` 按钮，查看所有的 Tools。
+
+## 1.5. 使用 MCP 发布
+
+### 检查登录状态
+
+
+
+### 发布图文
+
+示例中是从 https://unsplash.com/ 中随机找了个图片做测试。
+
+
+
+### 搜索内容
+
+使用搜索功能，根据关键词搜索小红书内容：
+
+
+
+## 2. MCP 客户端接入
+
+本服务支持标准的 Model Context Protocol (MCP)，可以接入各种支持 MCP 的 AI 客户端。
+
+### 2.1. 快速开始
+
+#### 启动 MCP 服务
+
+```bash
+# 启动服务（默认无头模式）
+go run .
+
+# 或者有界面模式
+go run . -headless=false
+```
+
+服务将运行在：`http://localhost:18060/mcp`
+
+#### 验证服务状态
+
+```bash
+# 测试 MCP 连接
+curl -X POST http://localhost:18060/mcp \
+ -H "Content-Type: application/json" \
+ -d '{"jsonrpc":"2.0","method":"initialize","params":{},"id":1}'
+```
+
+#### Claude Code CLI 接入
+
+```bash
+# 添加 HTTP MCP 服务器
+claude mcp add --transport http xiaohongshu-mcp http://localhost:18060/mcp
+
+# 检查 MCP 是否添加成功（确保 MCP 已经启动的前提下，运行下面命令）
+claude mcp list
+```
+
+### 2.2. 支持的客户端
+
+
+Claude Code CLI
+
+官方命令行工具，已在上面快速开始部分展示：
+
+```bash
+# 添加 HTTP MCP 服务器
+claude mcp add --transport http xiaohongshu-mcp http://localhost:18060/mcp
+
+# 检查 MCP 是否添加成功（确保 MCP 已经启动的前提下，运行下面命令）
+claude mcp list
+```
+
+
+
+
+Cursor
+
+#### 配置文件的方式
+
+创建或编辑 MCP 配置文件：
+
+**项目级配置**（推荐）：
+在项目根目录创建 `.cursor/mcp.json`：
+
+```json
+{
+ "mcpServers": {
+ "xiaohongshu-mcp": {
+ "url": "http://localhost:18060/mcp",
+ "description": "小红书内容发布服务 - MCP Streamable HTTP"
+ }
+ }
+}
+```
+
+**全局配置**：
+在用户目录创建 `~/.cursor/mcp.json` (同样内容)。
+
+#### 使用步骤
+
+1. 确保小红书 MCP 服务正在运行
+2. 保存配置文件后，重启 Cursor
+3. 在 Cursor 聊天中，工具应该自动可用
+4. 可以通过聊天界面的 "Available Tools" 查看已连接的 MCP 工具
+
+**Demo**
+
+插件 MCP 接入：
+
+
+
+调用 MCP 工具：（以检查登录状态为例）
+
+
+
+
+
+
+VSCode
+
+#### 方法一：使用命令面板配置
+
+1. 按 `Ctrl/Cmd + Shift + P` 打开命令面板
+2. 运行 `MCP: Add Server` 命令
+3. 选择 `HTTP` 方式。
+4. 输入地址： `http://localhost:18060/mcp`，或者修改成对应的 Server 地址。
+5. 输入 MCP 名字： `xiaohongshu-mcp`。
+
+#### 方法二：直接编辑配置文件
+
+**工作区配置**（推荐）：
+在项目根目录创建 `.vscode/mcp.json`：
+
+```json
+{
+ "servers": {
+ "xiaohongshu-mcp": {
+ "url": "http://localhost:18060/mcp",
+ "type": "http"
+ }
+ },
+ "inputs": []
+}
+```
+
+**查看配置**：
+
+
+
+1. 确认运行状态。
+2. 查看 `tools` 是否正确检测。
+
+**Demo**
+
+以搜索帖子内容为例：
+
+
+
+
+
+
+Google Gemini CLI
+
+在 `~/.gemini/settings.json` 或项目目录 `.gemini/settings.json` 中配置：
+
+```json
+{
+ "mcpServers": {
+ "xiaohongshu": {
+ "httpUrl": "http://localhost:18060/mcp",
+ "timeout": 30000
+ }
+ }
+}
+```
+
+更多信息请参考 [Gemini CLI MCP 文档](https://google-gemini.github.io/gemini-cli/docs/tools/mcp-server.html)
+
+
+
+
+MCP Inspector
+
+调试工具，用于测试 MCP 连接：
+
+```bash
+# 启动 MCP Inspector
+npx @modelcontextprotocol/inspector
+
+# 在浏览器中连接到：http://localhost:18060/mcp
+```
+
+使用步骤：
+
+- 使用 MCP Inspector 测试连接
+- 测试 Ping Server 功能验证连接
+- 检查 List Tools 是否返回 6 个工具
+
+
+
+
+Cline
+
+Cline 是一个强大的 AI 编程助手，支持 MCP 协议集成。
+
+#### 配置方法
+
+在 Cline 的 MCP 设置中添加以下配置：
+
+```json
+{
+ "xiaohongshu-mcp": {
+ "url": "http://localhost:18060/mcp",
+ "type": "streamableHttp",
+ "autoApprove": [],
+ "disabled": false
+ }
+}
+```
+
+#### 使用步骤
+
+1. 确保小红书 MCP 服务正在运行（`http://localhost:18060/mcp`）
+2. 在 Cline 中打开 MCP 设置
+3. 添加上述配置到 MCP 服务器列表
+4. 保存配置并重启 Cline
+5. 在对话中可以直接使用小红书相关功能
+
+#### 配置说明
+
+- `url`: MCP 服务地址
+- `type`: 使用 `streamableHttp` 类型以获得更好的性能
+- `autoApprove`: 可配置自动批准的工具列表（留空表示手动批准）
+- `disabled`: 设置为 `false` 启用此 MCP 服务
+
+#### 使用示例
+
+配置完成后，可以在 Cline 中直接使用自然语言操作小红书：
+
+```
+帮我检查小红书登录状态
+```
+
+```
+帮我发布一篇关于春天的图文到小红书，使用这张图片：/path/to/spring.jpg
+```
+
+```
+搜索小红书上关于"美食"的内容
+```
+
+
+
+
+其他支持 HTTP MCP 的客户端
+
+任何支持 HTTP MCP 协议的客户端都可以连接到：`http://localhost:18060/mcp`
+
+基本配置模板：
+
+```json
+{
+ "name": "xiaohongshu-mcp",
+ "url": "http://localhost:18060/mcp",
+ "type": "http"
+}
+```
+
+
+
+### 2.3. 可用 MCP 工具
+
+连接成功后，可使用以下 MCP 工具：
+
+- `check_login_status` - 检查小红书登录状态（无参数）
+- `publish_content` - 发布图文内容到小红书（必需：title, content, images）
+ - `images`: 支持 HTTP 链接或本地绝对路径，推荐使用本地路径
+- `publish_with_video` - 发布视频内容到小红书（必需：title, content, video）
+ - `video`: 仅支持本地视频文件绝对路径
+- `list_feeds` - 获取小红书首页推荐列表（无参数）
+- `search_feeds` - 搜索小红书内容（需要：keyword）
+- `get_feed_detail` - 获取帖子详情（需要：feed_id, xsec_token）
+- `post_comment_to_feed` - 发表评论到小红书帖子（需要：feed_id, xsec_token, content）
+- `user_profile` - 获取用户个人主页信息（需要：user_id, xsec_token）
+
+### 2.4. 使用示例
+
+使用 Claude Code 发布内容到小红书：
+
+**示例 1：使用 HTTP 图片链接**
+
+```
+帮我写一篇帖子发布到小红书上，
+配图为：https://cn.bing.com/th?id=OHR.MaoriRock_EN-US6499689741_UHD.jpg&w=3840
+图片是："纽西兰陶波湖的Ngātoroirangi矿湾毛利岩雕（© Joppi/Getty Images）"
+
+使用 xiaohongshu-mcp 进行发布。
+```
+
+**示例 2：使用本地图片路径（推荐）**
+
+```
+帮我写一篇关于春天的帖子发布到小红书上，
+使用这些本地图片：
+- /Users/username/Pictures/spring_flowers.jpg
+- /Users/username/Pictures/cherry_blossom.jpg
+
+使用 xiaohongshu-mcp 进行发布。
+```
+
+**示例 3：发布视频内容**
+
+```
+帮我写一篇关于美食制作的视频发布到小红书上，
+使用这个本地视频文件：
+- /Users/username/Videos/cooking_tutorial.mp4
+
+使用 xiaohongshu-mcp 的视频发布功能。
+```
+
+
+
+**发布结果：**
+
+
+
+### 2.5. 💬 MCP 使用常见问题解答
+
+---
+
+**Q:** 为什么检查登录用户名显示 `xiaghgngshu-mcp`？
+**A:** 用户名是写死的。
+
+---
+
+**Q:** 显示发布成功后，但实际上没有显示？
+**A:** 排查步骤如下：
+1. 使用 **非无头模式** 重新发布一次。
+2. 更换 **不同的内容** 重新发布。
+3. 登录网页版小红书，查看账号是否被 **风控限制网页版发布**。
+4. 检查 **图片大小** 是否过大。
+5. 确认 **图片路径中没有中文字符**。
+6. 若使用网络图片地址，请确认 **图片链接可正常访问**。
+
+---
+
+**Q:** 在设备上运行 MCP 程序出现闪退如何解决？
+**A:**
+1. 建议 **从源码安装**。
+2. 或使用 **Docker 安装 xiaohongshu-mcp**，教程参考：
+ - [使用 Docker 安装 xiaohongshu-mcp](https://github.com/xpzouying/xiaohongshu-mcp#:~:text=%E6%96%B9%E5%BC%8F%E4%B8%89%EF%BC%9A%E4%BD%BF%E7%94%A8%20Docker%20%E5%AE%B9%E5%99%A8%EF%BC%88%E6%9C%80%E7%AE%80%E5%8D%95%EF%BC%89)
+ - [X-MCP 项目页面](https://github.com/xpzouying/x-mcp/)
+
+---
+
+**Q:** 使用 `http://localhost:18060/mcp` 进行 MCP 验证时提示无法连接？
+**A:**
+- 在 **Docker 环境** 下，请使用
+ 👉 [http://host.docker.internal:18060/mcp](http://host.docker.internal:18060/mcp)
+- 在 **非 Docker 环境** 下，请使用 **本机 IPv4 地址** 访问。
+
+---
+
+## 3. 🌟 实战案例展示 (Community Showcases)
+
+> 💡 **强烈推荐查看**：这些都是社区贡献者的真实使用案例，包含详细的配置步骤和实战经验！
+
+### 📚 完整教程列表
+
+1. **[n8n 完整集成教程](./examples/n8n/README.md)** - 工作流自动化平台集成
+2. **[Cherry Studio 完整配置教程](./examples/cherrystudio/README.md)** - AI 客户端完美接入
+3. **[Claude Code + Kimi K2 接入教程](./examples/claude-code/claude-code-kimi-k2.md)** - Claude Code 门槛太高，那么就接入 Kimi 国产大模型吧～
+4. **[AnythingLLM 完整指南](./examples/anythingLLM/readme.md)** - AnythingLLM 是一款 all-in-one 多模态 AI 客户端，支持 workflow 定义，支持多种大模型和插件扩展。
+
+> 🎯 **提示**: 点击上方链接查看详细的图文教程，快速上手各种集成方案！
+>
+> 📢 **欢迎贡献**: 如果你有新的集成案例，欢迎提交 PR 分享给社区！
+
+## 4. 小红书 MCP 互助群
+
+**重要：在群里问问题之前，请一定要先仔细看完 README 文档以及查看 Issues。**
+
+
+
+| 【飞书 3 群】：扫码进入 | 【微信群 13 群】：扫码进入 |
+| ------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------- |
+| 
| 
|
+
+
+
+
+## 🙏 致谢贡献者 ✨
+
+感谢以下所有为本项目做出贡献的朋友！（排名不分先后）
+
+
+
+
+
+
+
+
+
+
+
+### ✨ 特别感谢
+
+| 贡献者 |
+| --------------------------------------------------------------------------------------------------------------------------- |
+| [
@wanpengxie](https://github.com/wanpengxie) |
+
+本项目遵循 [all-contributors](https://github.com/all-contributors/all-contributors) 规范。欢迎任何形式的贡献！
\ No newline at end of file
diff --git a/mcp_client.py b/mcp_client.py
new file mode 100644
index 0000000..13c1cd0
--- /dev/null
+++ b/mcp_client.py
@@ -0,0 +1,321 @@
+"""
+小红书 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 # 秒
+
+
+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) -> dict:
+ """发送 JSON-RPC 请求到 MCP 服务"""
+ payload = {
+ "jsonrpc": "2.0",
+ "method": method,
+ "params": params or {},
+ "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()
+ 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 通知
+ self._call("notifications/initialized", {})
+ self._initialized = True
+ return result
+ return {"status": "already_initialized"}
+
+ def _call_tool(self, tool_name: str, arguments: dict = None) -> dict:
+ """调用 MCP 工具"""
+ 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", ""))
+
+ # ---------- 帖子详情 ----------
+
+ 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 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,
+ })
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..2fc6975
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,3 @@
+gradio>=4.0.0
+requests>=2.28.0
+Pillow>=9.0.0
diff --git a/sd_service.py b/sd_service.py
new file mode 100644
index 0000000..750dc6a
--- /dev/null
+++ b/sd_service.py
@@ -0,0 +1,145 @@
+"""
+Stable Diffusion 服务模块
+封装对 SD WebUI API 的调用，支持 txt2img 和 img2img
+"""
+import requests
+import base64
+import io
+import logging
+from PIL import Image
+
+logger = logging.getLogger(__name__)
+
+SD_TIMEOUT = 180 # 图片生成可能需要较长时间
+
+# 默认反向提示词
+DEFAULT_NEGATIVE = (
+ "nsfw, lowres, bad anatomy, bad hands, text, error, missing fingers, "
+ "extra digit, fewer digits, cropped, worst quality, low quality, "
+ "normal quality, jpeg artifacts, signature, watermark, blurry"
+)
+
+
+class SDService:
+ """Stable Diffusion WebUI API 封装"""
+
+ def __init__(self, sd_url: str = "http://127.0.0.1:7860"):
+ self.sd_url = sd_url.rstrip("/")
+
+ def check_connection(self) -> tuple[bool, str]:
+ """检查 SD 服务是否可用"""
+ try:
+ resp = requests.get(f"{self.sd_url}/sdapi/v1/sd-models", timeout=5)
+ if resp.status_code == 200:
+ count = len(resp.json())
+ return True, f"SD 已连接，{count} 个模型可用"
+ return False, f"SD 返回异常状态: {resp.status_code}"
+ except requests.exceptions.ConnectionError:
+ return False, "SD WebUI 未启动或端口错误"
+ except Exception as e:
+ return False, f"SD 连接失败: {e}"
+
+ def get_models(self) -> list[str]:
+ """获取 SD 模型列表"""
+ resp = requests.get(f"{self.sd_url}/sdapi/v1/sd-models", timeout=5)
+ resp.raise_for_status()
+ return [m["title"] for m in resp.json()]
+
+ def switch_model(self, model_name: str):
+ """切换 SD 模型"""
+ try:
+ requests.post(
+ f"{self.sd_url}/sdapi/v1/options",
+ json={"sd_model_checkpoint": model_name},
+ timeout=60,
+ )
+ except Exception as e:
+ logger.warning("模型切换失败: %s", e)
+
+ def txt2img(
+ self,
+ prompt: str,
+ negative_prompt: str = DEFAULT_NEGATIVE,
+ model: str = None,
+ steps: int = 25,
+ cfg_scale: float = 7.0,
+ width: int = 768,
+ height: int = 1024,
+ batch_size: int = 2,
+ seed: int = -1,
+ ) -> list[Image.Image]:
+ """文生图"""
+ if model:
+ self.switch_model(model)
+
+ payload = {
+ "prompt": prompt,
+ "negative_prompt": negative_prompt,
+ "steps": steps,
+ "cfg_scale": cfg_scale,
+ "width": width,
+ "height": height,
+ "batch_size": batch_size,
+ "seed": seed,
+ }
+
+ resp = requests.post(
+ f"{self.sd_url}/sdapi/v1/txt2img",
+ json=payload,
+ timeout=SD_TIMEOUT,
+ )
+ resp.raise_for_status()
+
+ images = []
+ for img_b64 in resp.json().get("images", []):
+ img = Image.open(io.BytesIO(base64.b64decode(img_b64)))
+ images.append(img)
+ return images
+
+ def img2img(
+ self,
+ init_image: Image.Image,
+ prompt: str,
+ negative_prompt: str = DEFAULT_NEGATIVE,
+ denoising_strength: float = 0.6,
+ steps: int = 25,
+ cfg_scale: float = 7.0,
+ ) -> list[Image.Image]:
+ """图生图（参考图修改）"""
+ # 将 PIL Image 转为 base64
+ buf = io.BytesIO()
+ init_image.save(buf, format="PNG")
+ init_b64 = base64.b64encode(buf.getvalue()).decode("utf-8")
+
+ payload = {
+ "init_images": [init_b64],
+ "prompt": prompt,
+ "negative_prompt": negative_prompt,
+ "denoising_strength": denoising_strength,
+ "steps": steps,
+ "cfg_scale": cfg_scale,
+ "width": init_image.width,
+ "height": init_image.height,
+ }
+
+ resp = requests.post(
+ f"{self.sd_url}/sdapi/v1/img2img",
+ json=payload,
+ timeout=SD_TIMEOUT,
+ )
+ resp.raise_for_status()
+
+ images = []
+ for img_b64 in resp.json().get("images", []):
+ img = Image.open(io.BytesIO(base64.b64decode(img_b64)))
+ images.append(img)
+ return images
+
+ def get_lora_models(self) -> list[str]:
+ """获取可用的 LoRA 模型列表"""
+ try:
+ resp = requests.get(f"{self.sd_url}/sdapi/v1/loras", timeout=5)
+ resp.raise_for_status()
+ return [lora["name"] for lora in resp.json()]
+ except Exception:
+ return []
