✨ feat(content): 新增智能选题引擎、批量创作和图文协同优化
- 新增智能选题引擎 `TopicEngine`，整合热点数据与历史权重，提供多维度评分和创作角度建议 - 新增内容模板系统 `ContentTemplate`，支持从 JSON 文件加载模板并应用于文案生成 - 新增批量创作功能 `batch_generate_copy`，支持串行生成多篇文案并自动入草稿队列 - 升级文案质量流水线：实现 Prompt 分层架构（基础层 + 风格层 + 人设层）、LLM 自检与改写机制、深度去 AI 化后处理 - 优化图文协同：新增封面图策略选择、SD prompt 与文案语义联动、图文匹配度评估 - 集成数据闭环：在文案生成中自动注入 `AnalyticsService` 权重数据，实现发布 → 数据回收 → 优化创作的完整循环 - 更新 UI 组件：新增选题推荐展示区、批量创作折叠面板、封面图策略选择器和图文匹配度评分展示 ♻️ refactor(llm): 重构 Prompt 架构并增强去 AI 化处理 - 将 `PROMPT_COPYWRITING` 拆分为分层架构（基础层 + 风格层 + 人设层），提高维护性和灵活性 - 增强 `_humanize_content` 方法：新增语气词注入、标点不规范化、段落节奏打散和 emoji 密度控制 - 新增 `_self_check` 和 `_self_check_rewrite` 方法，实现文案 AI 痕迹自检与自动改写 - 新增 `evaluate_image_text_match` 方法，支持文案与 SD prompt 的语义匹配度评估（可选，失败不阻塞） - 新增封面图策略配置 `COVER_STRATEGIES` 和情感基调映射 `EMOTION_SD_MAP` 📝 docs(openspec): 归档内容创作优化提案和详细规格 - 新增 `openspec/changes/archive/2026-02-28-optimize-content-creation/` 目录，包含设计文档、提案、规格说明和任务清单 - 新增 `openspec/specs/` 下的批量创作、文案质量流水线、图文协同、服务内容和智能选题引擎规格文档 - 更新 `openspec/specs/services-content/spec.md`，反映新增的批量创作和智能选题入口函数 🔧 chore(config): 更新服务配置和 UI 集成 - 在 `services/content.py` 中集成权重数据自动注入逻辑，实现数据驱动创作 - 在 `ui/app.py` 中新增选题推荐、批量生成和图文匹配度评估的回调函数 - 在 `ui/tab_create.py` 中新增智能选题推荐区、批量创作面板和图文匹配度评估组件 - 修复 `services/sd_service.py` 中的头像文件路径问题，确保目录存在
This commit is contained in:
@@ -7,6 +7,7 @@ import re
|
||||
import logging
|
||||
|
||||
import gradio as gr
|
||||
from PIL import Image
|
||||
|
||||
from .config_manager import ConfigManager
|
||||
from .llm_service import LLMService
|
||||
|
||||
+201
-2
@@ -22,14 +22,39 @@ logger = logging.getLogger("autobot")
|
||||
cfg = ConfigManager()
|
||||
|
||||
def generate_copy(model, topic, style, sd_model_name, persona_text):
|
||||
"""生成文案（自动适配 SD 模型的 prompt 风格，支持人设）"""
|
||||
"""生成文案（自动适配 SD 模型，支持人设，自动注入权重数据）"""
|
||||
api_key, base_url, _ = _get_llm_config()
|
||||
if not api_key:
|
||||
return "", "", "", "", "❌ 请先配置并连接 LLM 提供商"
|
||||
try:
|
||||
svc = LLMService(api_key, base_url, model)
|
||||
persona = _resolve_persona(persona_text) if persona_text else None
|
||||
data = svc.generate_copy(topic, style, sd_model_name=sd_model_name, persona=persona)
|
||||
|
||||
# 尝试自动注入权重数据（数据闭环 9.1）
|
||||
data = None
|
||||
try:
|
||||
from .analytics_service import AnalyticsService
|
||||
analytics = AnalyticsService()
|
||||
if analytics.has_weights:
|
||||
weight_insights = analytics.weights_summary
|
||||
title_advice = analytics.get_title_advice()
|
||||
hot_tags = ", ".join(analytics.get_top_tags(8))
|
||||
data = svc.generate_weighted_copy(
|
||||
topic, style,
|
||||
weight_insights=weight_insights,
|
||||
title_advice=title_advice,
|
||||
hot_tags=hot_tags,
|
||||
sd_model_name=sd_model_name,
|
||||
persona=persona,
|
||||
)
|
||||
logger.info("使用加权文案生成路径（权重数据已注入）")
|
||||
except Exception as e:
|
||||
logger.debug("权重数据注入跳过: %s", e)
|
||||
|
||||
# 无权重或权重路径失败时，退回基础生成
|
||||
if data is None:
|
||||
data = svc.generate_copy(topic, style, sd_model_name=sd_model_name, persona=persona)
|
||||
|
||||
cfg.set("model", model)
|
||||
tags = data.get("tags", [])
|
||||
return (
|
||||
@@ -207,3 +232,177 @@ def publish_to_xhs(title, content, tags_str, images, local_images, mcp_url, sche
|
||||
logger.warning("临时文件清理失败 %s: %s", tmp_path, cleanup_err)
|
||||
|
||||
|
||||
# ========== 批量创作 ==========
|
||||
|
||||
def batch_generate_copy(
|
||||
model: str,
|
||||
topics: list[str],
|
||||
style: str,
|
||||
sd_model_name: str = "",
|
||||
persona_text: str = "",
|
||||
template_name: str = "",
|
||||
publish_queue=None,
|
||||
) -> tuple[list[dict], str]:
|
||||
"""
|
||||
批量生成多篇文案（串行），自动插入发布队列草稿
|
||||
|
||||
Args:
|
||||
model: LLM 模型名
|
||||
topics: 主题列表 (最多 10 个)
|
||||
style: 写作风格
|
||||
sd_model_name: SD 模型名
|
||||
persona_text: 人设文本
|
||||
template_name: 可选的模板名
|
||||
publish_queue: 可选的 PublishQueue 实例
|
||||
|
||||
Returns:
|
||||
(results_list, status_msg)
|
||||
"""
|
||||
if not topics:
|
||||
return [], "❌ 请输入至少一个主题"
|
||||
if len(topics) > 10:
|
||||
return [], "❌ 批量生成最多支持 10 个主题，请减少数量"
|
||||
|
||||
api_key, base_url, _ = _get_llm_config()
|
||||
if not api_key:
|
||||
return [], "❌ 请先配置并连接 LLM 提供商"
|
||||
|
||||
# 加载模板覆盖
|
||||
prompt_override = ""
|
||||
tags_preset = []
|
||||
if template_name:
|
||||
try:
|
||||
from .content_template import ContentTemplate
|
||||
ct = ContentTemplate()
|
||||
override = ct.apply_template(template_name)
|
||||
style = override.get("style") or style
|
||||
prompt_override = override.get("prompt_override", "")
|
||||
tags_preset = override.get("tags_preset", [])
|
||||
except Exception as e:
|
||||
logger.warning("模板加载失败，使用默认参数: %s", e)
|
||||
|
||||
svc = LLMService(api_key, base_url, model)
|
||||
persona = _resolve_persona(persona_text) if persona_text else None
|
||||
|
||||
results = []
|
||||
success_count = 0
|
||||
fail_count = 0
|
||||
|
||||
for idx, topic in enumerate(topics):
|
||||
topic = topic.strip()
|
||||
if not topic:
|
||||
continue
|
||||
|
||||
try:
|
||||
data = svc.generate_copy(
|
||||
topic, style,
|
||||
sd_model_name=sd_model_name,
|
||||
persona=persona,
|
||||
)
|
||||
|
||||
# 如有模板 prompt_override，它已通过风格参数间接生效
|
||||
# 合并模板标签
|
||||
tags = data.get("tags", [])
|
||||
if tags_preset:
|
||||
existing = set(tags)
|
||||
for t in tags_preset:
|
||||
if t not in existing:
|
||||
tags.append(t)
|
||||
data["tags"] = tags
|
||||
|
||||
data["batch_index"] = idx
|
||||
results.append(data)
|
||||
success_count += 1
|
||||
|
||||
# 自动入队为草稿
|
||||
if publish_queue:
|
||||
try:
|
||||
publish_queue.add(
|
||||
title=data.get("title", ""),
|
||||
content=data.get("content", ""),
|
||||
sd_prompt=data.get("sd_prompt", ""),
|
||||
tags=data.get("tags", []),
|
||||
topic=topic,
|
||||
style=style,
|
||||
persona=persona_text if persona_text else "",
|
||||
status="draft",
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning("批量草稿入队失败 #%d: %s", idx, e)
|
||||
|
||||
logger.info("批量生成 %d/%d 完成: %s", idx + 1, len(topics), topic[:20])
|
||||
|
||||
except Exception as e:
|
||||
logger.error("批量生成 %d/%d 失败 [%s]: %s", idx + 1, len(topics), topic[:20], e)
|
||||
results.append({
|
||||
"batch_index": idx,
|
||||
"topic": topic,
|
||||
"error": str(e),
|
||||
})
|
||||
fail_count += 1
|
||||
|
||||
status = f"✅ 批量生成完成: {success_count} 成功"
|
||||
if fail_count:
|
||||
status += f", {fail_count} 失败"
|
||||
if publish_queue and success_count:
|
||||
status += f" | {success_count} 篇已入草稿队列"
|
||||
|
||||
return results, status
|
||||
|
||||
|
||||
def generate_copy_with_topic_engine(
|
||||
model: str,
|
||||
style: str,
|
||||
sd_model_name: str = "",
|
||||
persona_text: str = "",
|
||||
count: int = 1,
|
||||
hotspot_data: dict = None,
|
||||
publish_queue=None,
|
||||
) -> tuple[list[dict], str]:
|
||||
"""
|
||||
使用智能选题引擎自动选题 + 生成文案
|
||||
|
||||
Args:
|
||||
model: LLM 模型名
|
||||
style: 写作风格
|
||||
sd_model_name: SD 模型名
|
||||
persona_text: 人设文本
|
||||
count: 生成篇数
|
||||
hotspot_data: 可选的热点分析数据
|
||||
publish_queue: 可选的 PublishQueue 实例
|
||||
|
||||
Returns:
|
||||
(results_list, status_msg)
|
||||
"""
|
||||
try:
|
||||
from .analytics_service import AnalyticsService
|
||||
from .topic_engine import TopicEngine
|
||||
|
||||
analytics = AnalyticsService()
|
||||
engine = TopicEngine(analytics)
|
||||
recommendations = engine.recommend_topics(count=count, hotspot_data=hotspot_data)
|
||||
|
||||
if not recommendations:
|
||||
return [], "❌ 选题引擎未找到推荐主题，请先进行热点搜索或积累数据"
|
||||
|
||||
topics = [r["topic"] for r in recommendations]
|
||||
results, status = batch_generate_copy(
|
||||
model=model,
|
||||
topics=topics,
|
||||
style=style,
|
||||
sd_model_name=sd_model_name,
|
||||
persona_text=persona_text,
|
||||
publish_queue=publish_queue,
|
||||
)
|
||||
|
||||
# 把选题推荐信息附加到结果
|
||||
for result in results:
|
||||
idx = result.get("batch_index", -1)
|
||||
if 0 <= idx < len(recommendations):
|
||||
result["topic_recommendation"] = recommendations[idx]
|
||||
|
||||
return results, status
|
||||
|
||||
except Exception as e:
|
||||
logger.error("智能选题生成失败: %s", e)
|
||||
return [], f"❌ 智能选题生成失败: {e}"
|
||||
@@ -0,0 +1,132 @@
|
||||
"""
|
||||
services/content_template.py
|
||||
内容模板系统 — 管理和应用创作模板
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger("autobot")
|
||||
|
||||
TEMPLATES_FILE = "templates.json"
|
||||
|
||||
# 内置默认模板 (templates.json 不存在时使用)
|
||||
DEFAULT_TEMPLATES = [
|
||||
{
|
||||
"name": "好物种草",
|
||||
"description": "适合分享好用的产品和购物推荐",
|
||||
"topic_pattern": "",
|
||||
"style": "好物种草",
|
||||
"prompt_override": "请以真实使用者的口吻分享产品体验，突出个人感受和使用前后对比，避免像广告文案。",
|
||||
"tags_preset": ["好物推荐", "真实测评", "分享好物"],
|
||||
},
|
||||
{
|
||||
"name": "日常分享",
|
||||
"description": "记录日常生活点滴、感悟和心情",
|
||||
"topic_pattern": "",
|
||||
"style": "日常分享",
|
||||
"prompt_override": "请以轻松随意的语气记录生活日常，像发朋友圈那样自然，多用短句和口语化表达。",
|
||||
"tags_preset": ["日常", "生活记录", "碎碎念"],
|
||||
},
|
||||
{
|
||||
"name": "攻略教程",
|
||||
"description": "分享经验技巧、教程和攻略指南",
|
||||
"topic_pattern": "",
|
||||
"style": "攻略教程",
|
||||
"prompt_override": "请以过来人的身份分享干货经验，用分步骤的方式让读者易懂，加入踩坑经历增加可信度。",
|
||||
"tags_preset": ["干货分享", "经验", "保姆级教程"],
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
class ContentTemplate:
|
||||
"""
|
||||
内容模板管理器
|
||||
|
||||
从 xhs_workspace/templates.json 加载模板，
|
||||
文件不存在时使用内置默认模板。
|
||||
"""
|
||||
|
||||
def __init__(self, workspace_dir: str = "xhs_workspace"):
|
||||
self.workspace_dir = workspace_dir
|
||||
self.templates_path = os.path.join(workspace_dir, TEMPLATES_FILE)
|
||||
self._templates: list[dict] = self._load_templates()
|
||||
|
||||
def _load_templates(self) -> list[dict]:
|
||||
"""加载模板列表"""
|
||||
if os.path.exists(self.templates_path):
|
||||
try:
|
||||
with open(self.templates_path, "r", encoding="utf-8") as f:
|
||||
templates = json.load(f)
|
||||
if isinstance(templates, list) and templates:
|
||||
logger.info("已从 %s 加载 %d 个模板", self.templates_path, len(templates))
|
||||
return templates
|
||||
except (json.JSONDecodeError, IOError) as e:
|
||||
logger.warning("模板文件加载失败，使用默认模板: %s", e)
|
||||
|
||||
logger.info("使用内置默认模板 (%d 个)", len(DEFAULT_TEMPLATES))
|
||||
return list(DEFAULT_TEMPLATES)
|
||||
|
||||
def save_templates(self):
|
||||
"""将当前模板保存到文件"""
|
||||
try:
|
||||
os.makedirs(self.workspace_dir, exist_ok=True)
|
||||
with open(self.templates_path, "w", encoding="utf-8") as f:
|
||||
json.dump(self._templates, f, ensure_ascii=False, indent=2)
|
||||
logger.info("模板已保存到 %s", self.templates_path)
|
||||
except IOError as e:
|
||||
logger.error("模板保存失败: %s", e)
|
||||
|
||||
@property
|
||||
def templates(self) -> list[dict]:
|
||||
"""获取所有模板"""
|
||||
return self._templates
|
||||
|
||||
def get_template_names(self) -> list[str]:
|
||||
"""获取模板名称列表"""
|
||||
return [t.get("name", "未命名") for t in self._templates]
|
||||
|
||||
def get_template(self, name: str) -> dict | None:
|
||||
"""按名称获取模板"""
|
||||
for t in self._templates:
|
||||
if t.get("name") == name:
|
||||
return t
|
||||
return None
|
||||
|
||||
def apply_template(self, template_name: str) -> dict:
|
||||
"""
|
||||
应用模板，返回用于文案生成的参数覆盖
|
||||
|
||||
Returns:
|
||||
dict with keys:
|
||||
- style: str
|
||||
- prompt_override: str (附加到 LLM prompt 的额外指令)
|
||||
- tags_preset: list[str] (标签默认值)
|
||||
"""
|
||||
template = self.get_template(template_name)
|
||||
if not template:
|
||||
logger.warning("模板 '%s' 不存在，返回空覆盖", template_name)
|
||||
return {"style": "", "prompt_override": "", "tags_preset": []}
|
||||
|
||||
return {
|
||||
"style": template.get("style", ""),
|
||||
"prompt_override": template.get("prompt_override", ""),
|
||||
"tags_preset": template.get("tags_preset", []),
|
||||
}
|
||||
|
||||
def add_template(self, template: dict):
|
||||
"""添加新模板"""
|
||||
required_fields = {"name", "description", "style"}
|
||||
if not required_fields.issubset(template.keys()):
|
||||
raise ValueError(f"模板缺少必要字段: {required_fields - template.keys()}")
|
||||
self._templates.append(template)
|
||||
self.save_templates()
|
||||
|
||||
def remove_template(self, name: str) -> bool:
|
||||
"""删除模板"""
|
||||
before = len(self._templates)
|
||||
self._templates = [t for t in self._templates if t.get("name") != name]
|
||||
if len(self._templates) < before:
|
||||
self.save_templates()
|
||||
return True
|
||||
return False
|
||||
+404
-82
@@ -12,7 +12,9 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
# ================= Prompt 模板 =================
|
||||
|
||||
PROMPT_COPYWRITING = """
|
||||
# ---- 分层 Prompt 架构：基础层 + 风格层 + 人设层 ----
|
||||
|
||||
PROMPT_BASE = """
|
||||
你是一个真实的小红书博主，正在用手机编辑一篇笔记。你不是内容专家，你只是一个想认真分享的普通人。
|
||||
|
||||
【你的写作状态】：
|
||||
@@ -55,14 +57,128 @@ PROMPT_COPYWRITING = """
|
||||
6. 避免完美的逻辑链条：不要每段都工工整整地推进论点，真人笔记是跳跃式的
|
||||
7. 偶尔口语化到"学渣"程度："就 很那个 你懂的" "属于是" "多少有点" "怎么说呢"
|
||||
8. 绝对不要用"然而" "此外" "因此" "尽管" "虽然...但是..."这些书面连接词
|
||||
"""
|
||||
|
||||
# ---- 风格层 Prompt ----
|
||||
|
||||
PROMPT_STYLE_GOODS = """
|
||||
【风格：好物种草】：
|
||||
你在分享一个让你很惊喜的东西。重点是真实体验感受，不是产品说明书。
|
||||
- 写出"发现宝藏"的兴奋感，但别太夸张
|
||||
- 可以先说你怎么发现/入手的（"刷到好多人推 忍不住下单了"）
|
||||
- 重点说使用感受，别罗列参数配置
|
||||
- 适当提一两个小缺点增加可信度
|
||||
- 价格相关的要自然带出（"百元价位能有这效果 我真的服"）
|
||||
"""
|
||||
|
||||
PROMPT_STYLE_DAILY = """
|
||||
【风格：日常分享】：
|
||||
你在分享生活中一个有意思/有感触的瞬间。核心是情绪共鸣。
|
||||
- 写得像发朋友圈，随意自然
|
||||
- 不需要有"干货"，纯粹分享感受就好
|
||||
- 可以碎碎念、跑题、突然感叹
|
||||
- 配图描述偏生活化场景（自拍、日常环境、随手拍）
|
||||
"""
|
||||
|
||||
PROMPT_STYLE_GUIDE = """
|
||||
【风格：攻略教程】：
|
||||
你在分享一个你研究了很久/踩了很多坑之后总结的经验。
|
||||
- 用"过来人"的语气，不是老师讲课
|
||||
- 开头可以用痛点引入（"之前踩了好多坑""终于搞明白了"）
|
||||
- 信息量要足但别太结构化，穿插个人经历和小吐槽
|
||||
- 可以用简单的分段，但别用"第一步、第二步"这种死板格式
|
||||
- 结尾可以加"有问题评论区问我"之类的互动引导
|
||||
"""
|
||||
|
||||
# 风格层映射
|
||||
PROMPT_STYLES = {
|
||||
"好物种草": PROMPT_STYLE_GOODS,
|
||||
"日常分享": PROMPT_STYLE_DAILY,
|
||||
"攻略教程": PROMPT_STYLE_GUIDE,
|
||||
"真实分享": PROMPT_STYLE_DAILY,
|
||||
"经验分享": PROMPT_STYLE_GUIDE,
|
||||
"种草安利": PROMPT_STYLE_GOODS,
|
||||
}
|
||||
|
||||
PROMPT_COPYWRITING_SUFFIX = """
|
||||
【绘图 Prompt】：
|
||||
{sd_prompt_guide}
|
||||
|
||||
【重要 - 图文语义联动】：
|
||||
生成 sd_prompt 时，必须从文案正文中提取具体的场景描述和关键词：
|
||||
- 如文案提到"咖啡馆翻书"，sd_prompt 必须包含 cozy cafe, reading book 等对应元素
|
||||
- 如文案提到"海边散步"，sd_prompt 必须包含 beach walking, seaside 等
|
||||
- 如文案是温柔/治愈风，sd_prompt 加入 soft lighting, warm tone, gentle atmosphere
|
||||
- 如文案是活力/运动风，sd_prompt 加入 bright colors, dynamic pose, energetic mood
|
||||
- 如文案是酷飒/高级风，sd_prompt 加入 cool tone, dramatic lighting, editorial style
|
||||
- 文案→图片的场景一致性是最重要的，不要凭空编造与文案无关的场景
|
||||
|
||||
返回 JSON 格式：
|
||||
{{"title": "...", "content": "...", "sd_prompt": "...", "tags": ["标签1", "标签2", ...]}}
|
||||
"""
|
||||
|
||||
# 保留旧变量名兼容（组合基础层 + 默认后缀）
|
||||
PROMPT_COPYWRITING = PROMPT_BASE + PROMPT_COPYWRITING_SUFFIX
|
||||
|
||||
PROMPT_SELF_CHECK = """
|
||||
你是一个专业的AI内容检测专家。请评估以下小红书笔记文案的"AI痕迹程度"。
|
||||
|
||||
【评估维度】（每项0-20分，总分0-100）：
|
||||
1. **书面化程度** (0-20)：是否使用了"然而""此外""综上所述"等书面连接词？句式是否过于规整？
|
||||
2. **逻辑完美度** (0-20)：段落逻辑是否过于顺畅完美？真人写作会有跳跃和碎片化
|
||||
3. **用词规范度** (0-20)：用词是否过于"正确"？真人会用网络语、口语、不规范表达
|
||||
4. **结构工整度** (0-20)：是否有明显的分点罗列、排比对仗？段落长度是否过于均匀？
|
||||
5. **情感自然度** (0-20)：情感表达是否像真人？还是像AI在"模拟"情感？
|
||||
|
||||
【待评估文案】：
|
||||
{content}
|
||||
|
||||
返回 JSON 格式：
|
||||
{{"ai_score": 总分(0-100), "feedback": "具体哪些地方暴露了AI痕迹，以及改进建议", "dimension_scores": {{"书面化": x, "逻辑完美": x, "用词规范": x, "结构工整": x, "情感自然": x}}}}
|
||||
"""
|
||||
|
||||
PROMPT_SELF_CHECK_REWRITE = """
|
||||
你是一个小红书文案优化专家。以下文案被检测出AI痕迹，请根据反馈进行改写，让它更像真人写的。
|
||||
|
||||
【原始文案】：
|
||||
{original_content}
|
||||
|
||||
【AI检测反馈】：
|
||||
{feedback}
|
||||
|
||||
【改写要求】：
|
||||
- 保留原始内容的核心信息和观点
|
||||
- 针对反馈中指出的AI痕迹进行修改
|
||||
- 不要改变标题和标签
|
||||
- 改写后的文案长度保持在 400-600 字
|
||||
- 让文案读起来更像一个真人在手机上随手写的
|
||||
|
||||
直接返回改写后的正文，不要有任何解释。
|
||||
"""
|
||||
|
||||
PROMPT_IMAGE_TEXT_MATCH = """
|
||||
你是一个图文内容质量评审专家。请评估以下小红书笔记文案与其配图 SD 绘图提示词之间的语义匹配度。
|
||||
|
||||
【文案正文】：
|
||||
{content}
|
||||
|
||||
【SD 绘图 Prompt】：
|
||||
{sd_prompt}
|
||||
|
||||
【评估维度】：
|
||||
1. 场景一致性：文案描述的场景是否在图片中有体现？
|
||||
2. 情感基调匹配：文案的情绪与图片氛围是否一致？
|
||||
3. 关键元素覆盖：文案中的核心事物（产品、地点、人物状态）是否在 prompt 中有对应描述？
|
||||
|
||||
返回 JSON 格式：
|
||||
{{"match_score": 0-100分, "suggestions": ["改进建议1", "改进建议2"]}}
|
||||
|
||||
评分标准：
|
||||
- 80-100: 高度匹配，图文呼应好
|
||||
- 50-79: 基本匹配，有可改进空间
|
||||
- 0-49: 匹配度低，建议重新生成
|
||||
"""
|
||||
|
||||
PROMPT_PERFORMANCE_ANALYSIS = """
|
||||
你是一个有实战经验的小红书运营数据分析师。下面是一个博主已发布的笔记数据，按互动量从高到低排列：
|
||||
|
||||
@@ -89,38 +205,22 @@ PROMPT_PERFORMANCE_ANALYSIS = """
|
||||
{{"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 = """
|
||||
你是一个真实的小红书博主，正在用手机编辑一篇笔记。
|
||||
|
||||
PROMPT_WEIGHTED_COPYWRITING_EXTRA = """
|
||||
【智能学习洞察——基于你过去笔记的数据分析】：
|
||||
{weight_insights}
|
||||
|
||||
【创作要求】：
|
||||
基于以上数据洞察，请创作一篇更容易获得高互动的笔记。要把数据分析的结论融入创作中，但写出来的内容要自然，不能看出是"为了数据而写"。
|
||||
|
||||
【标题规则】(严格执行)：
|
||||
1. 长度限制：必须控制在 18 字以内（含Emoji），绝对不能超过 20 字！
|
||||
2. 参考高互动标题的模式：{title_advice}
|
||||
3. 口语化，有情绪感，像发朋友圈
|
||||
4. 禁止广告法违禁词
|
||||
|
||||
【正文规则——像说话一样写】：
|
||||
1. 想象你在跟闺蜜/朋友面对面聊天
|
||||
2. 正文控制在 400-600 字
|
||||
3. 自然展开，不要分点罗列
|
||||
4. 可以有小情绪：吐槽、感叹、自嘲、开心炸裂
|
||||
5. emoji穿插在情绪高点，不要每句都有
|
||||
6. 绝对禁止 AI 痕迹书面用语
|
||||
【补充标题规则】：
|
||||
参考高互动标题的模式：{title_advice}
|
||||
|
||||
【推荐标签】：优先使用这些高权重标签 → {hot_tags}
|
||||
|
||||
【绘图 Prompt】：
|
||||
{sd_prompt_guide}
|
||||
|
||||
返回 JSON 格式：
|
||||
{{"title": "...", "content": "...", "sd_prompt": "...", "tags": ["标签1", "标签2", ...]}}
|
||||
"""
|
||||
|
||||
# 保留旧变量名兼容（加权创作也使用分层基础）
|
||||
PROMPT_WEIGHTED_COPYWRITING = PROMPT_BASE + PROMPT_WEIGHTED_COPYWRITING_EXTRA + PROMPT_COPYWRITING_SUFFIX
|
||||
|
||||
PROMPT_HOTSPOT_ANALYSIS = """
|
||||
你是一个有实战经验的小红书运营人。下面是搜索到的热门笔记信息：
|
||||
|
||||
@@ -237,45 +337,24 @@ PROMPT_PROACTIVE_COMMENT = """
|
||||
请直接输出一条评论，不要有任何解释或前缀。记住：你是一个真人，不是AI。
|
||||
"""
|
||||
|
||||
PROMPT_COPY_WITH_REFERENCE = """
|
||||
你是一个真实的小红书博主，正在参考一些热门笔记来写一篇自己的原创内容。
|
||||
你不是在写营销文案，你只是觉得这些笔记写得不错，想借鉴思路写一篇自己的体验分享。
|
||||
|
||||
PROMPT_COPY_WITH_REFERENCE_EXTRA = """
|
||||
【参考笔记】：
|
||||
{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", ...]}}
|
||||
【参考笔记创作指导】：
|
||||
- 学习参考笔记标题的情绪感和口语感，但内容完全原创
|
||||
- 开头可以直接说事，不需要"嗨大家好"之类的开场白
|
||||
- 中间夹杂个人感受和小吐槽
|
||||
- 挑2-3个最有感触的重点说，不要面面俱到
|
||||
- 可以适当提一两个小缺点增加可信度
|
||||
"""
|
||||
|
||||
# 保留旧变量名兼容（参考创作也使用分层基础）
|
||||
PROMPT_COPY_WITH_REFERENCE = PROMPT_BASE + PROMPT_COPY_WITH_REFERENCE_EXTRA + PROMPT_COPYWRITING_SUFFIX
|
||||
|
||||
|
||||
class LLMService:
|
||||
"""LLM API 服务封装"""
|
||||
@@ -419,6 +498,80 @@ class LLMService:
|
||||
|
||||
return base + chinese_aesthetic_guide + anti_detect_tips + persona_guide
|
||||
|
||||
# ========== 封面图策略 ==========
|
||||
|
||||
# 情感基调 → SD 氛围词映射
|
||||
EMOTION_SD_MAP = {
|
||||
"温柔": "soft lighting, warm color palette, gentle atmosphere, cozy mood",
|
||||
"治愈": "warm tone, soft focus, peaceful, comforting light, serene mood",
|
||||
"活力": "bright vivid colors, dynamic angle, energetic mood, sunlight",
|
||||
"酷飒": "cool tone, dramatic lighting, sharp contrast, cinematic, editorial",
|
||||
"甜美": "pastel colors, soft pink tone, dreamy, cute, romantic lighting",
|
||||
"高级": "neutral tone, minimalist, luxury, muted colors, sophisticated",
|
||||
"搞笑": "bright cheerful colors, exaggerated expression, fun, playful",
|
||||
"文艺": "film grain, muted vintage tone, nostalgic, soft natural light",
|
||||
}
|
||||
|
||||
# 封面图策略 → SD prompt 后缀 + 尺寸
|
||||
COVER_STRATEGIES = {
|
||||
"人物特写": {
|
||||
"sd_suffix": "portrait, face close-up, shallow depth of field, bokeh background, upper body shot",
|
||||
"width": 768,
|
||||
"height": 1024,
|
||||
},
|
||||
"场景展示": {
|
||||
"sd_suffix": "wide angle, environmental shot, product in context, lifestyle scene, natural setting",
|
||||
"width": 1024,
|
||||
"height": 768,
|
||||
},
|
||||
"对比图": {
|
||||
"sd_suffix": "before and after, side by side comparison, split view, clean background, product showcase",
|
||||
"width": 1024,
|
||||
"height": 1024,
|
||||
},
|
||||
"文字卡片": {
|
||||
"sd_suffix": "minimal background, clean simple design, solid color backdrop, text space, magazine layout",
|
||||
"width": 768,
|
||||
"height": 1024,
|
||||
},
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def get_cover_strategy(strategy_name: str) -> dict:
|
||||
"""获取封面图策略配置"""
|
||||
return LLMService.COVER_STRATEGIES.get(
|
||||
strategy_name,
|
||||
LLMService.COVER_STRATEGIES.get("人物特写")
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def get_emotion_atmosphere(emotion: str) -> str:
|
||||
"""根据情感基调获取 SD 氛围词"""
|
||||
return LLMService.EMOTION_SD_MAP.get(emotion, "")
|
||||
|
||||
# ========== 图文匹配度评估 ==========
|
||||
|
||||
def evaluate_image_text_match(self, content: str, sd_prompt: str) -> dict:
|
||||
"""
|
||||
评估文案与 SD prompt 的语义匹配度（可选，失败不阻塞）
|
||||
|
||||
Returns:
|
||||
dict with match_score (0-100), suggestions (list[str])
|
||||
失败时返回 {"match_score": -1, "suggestions": [], "skipped": True}
|
||||
"""
|
||||
prompt = PROMPT_IMAGE_TEXT_MATCH.format(content=content, sd_prompt=sd_prompt)
|
||||
try:
|
||||
raw = self._chat(prompt, "请评估图文匹配度", json_mode=True)
|
||||
result = self._parse_json(raw)
|
||||
return {
|
||||
"match_score": int(result.get("match_score", 0)),
|
||||
"suggestions": result.get("suggestions", []),
|
||||
"skipped": False,
|
||||
}
|
||||
except Exception as e:
|
||||
logger.warning("图文匹配度评估失败（已跳过）: %s", e)
|
||||
return {"match_score": -1, "suggestions": [], "skipped": True}
|
||||
|
||||
def _chat(self, system_prompt: str, user_message: str,
|
||||
json_mode: bool = True, temperature: float = 0.8) -> str:
|
||||
"""底层聊天接口（含空返回检测、json_mode 回退、模型降级）"""
|
||||
@@ -592,13 +745,55 @@ class LLMService:
|
||||
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}"
|
||||
@staticmethod
|
||||
def _build_layered_prompt(style: str, sd_guide: str, persona: str = None) -> str:
|
||||
"""构建分层 Prompt：基础层 → 风格层 → 人设层 → 后缀"""
|
||||
parts = [PROMPT_BASE]
|
||||
# 风格层（缺失时退回基础层）
|
||||
style_prompt = PROMPT_STYLES.get(style, "")
|
||||
if style_prompt:
|
||||
parts.append(style_prompt)
|
||||
# 人设层
|
||||
if persona:
|
||||
user_msg = f"【博主人设】：{persona}\n请以此人设的视角和风格创作。\n\n{user_msg}"
|
||||
parts.append(f"\n【博主人设】：{persona}\n请以此人设的视角和风格创作。\n")
|
||||
# 后缀（SD prompt 指导 + JSON 格式要求）
|
||||
parts.append(PROMPT_COPYWRITING_SUFFIX.format(sd_prompt_guide=sd_guide))
|
||||
return "\n".join(parts)
|
||||
|
||||
def _self_check(self, content: str) -> dict:
|
||||
"""对文案进行 AI 痕迹自检，返回 {ai_score, feedback, dimension_scores}"""
|
||||
try:
|
||||
prompt = PROMPT_SELF_CHECK.format(content=content)
|
||||
raw = self._chat(prompt, "请评估以上文案的AI痕迹程度",
|
||||
json_mode=True, temperature=0.3)
|
||||
result = self._parse_json(raw)
|
||||
# 确保字段完整
|
||||
return {
|
||||
"ai_score": int(result.get("ai_score", 50)),
|
||||
"feedback": result.get("feedback", ""),
|
||||
"dimension_scores": result.get("dimension_scores", {}),
|
||||
}
|
||||
except Exception as e:
|
||||
logger.warning("文案自检失败（跳过）: %s", e)
|
||||
return {"ai_score": 0, "feedback": "", "dimension_scores": {}}
|
||||
|
||||
def _self_check_rewrite(self, original_content: str, feedback: str) -> str:
|
||||
"""根据自检反馈改写文案"""
|
||||
try:
|
||||
prompt = PROMPT_SELF_CHECK_REWRITE.format(
|
||||
original_content=original_content, feedback=feedback
|
||||
)
|
||||
rewritten = self._chat(prompt, "请改写文案", json_mode=False, temperature=0.9)
|
||||
return rewritten.strip() if rewritten and rewritten.strip() else original_content
|
||||
except Exception as e:
|
||||
logger.warning("文案改写失败（使用原始文案）: %s", e)
|
||||
return original_content
|
||||
|
||||
def generate_copy(self, topic: str, style: str, sd_model_name: str = None, persona: str = None) -> dict:
|
||||
"""生成小红书文案（分层 Prompt 架构，含重试逻辑，自动适配SD模型，支持人设）"""
|
||||
sd_guide = self.get_sd_prompt_guide(sd_model_name, persona=persona)
|
||||
system_prompt = self._build_layered_prompt(style, sd_guide, persona=persona)
|
||||
user_msg = f"主题：{topic}\n风格：{style}"
|
||||
last_error = None
|
||||
for attempt in range(2):
|
||||
try:
|
||||
@@ -618,10 +813,30 @@ class LLMService:
|
||||
title = title[:20]
|
||||
data["title"] = title
|
||||
|
||||
# 自检机制：检测 AI 痕迹
|
||||
quality_meta = {"ai_score": 0, "self_check_passed": True, "rewritten": False}
|
||||
raw_content = data.get("content", "")
|
||||
if raw_content:
|
||||
check_result = self._self_check(raw_content)
|
||||
ai_score = check_result.get("ai_score", 0)
|
||||
quality_meta["ai_score"] = ai_score
|
||||
if ai_score >= 60:
|
||||
# AI 痕迹较重，触发改写
|
||||
quality_meta["self_check_passed"] = False
|
||||
feedback = check_result.get("feedback", "")
|
||||
rewritten = self._self_check_rewrite(raw_content, feedback)
|
||||
if rewritten != raw_content:
|
||||
data["content"] = rewritten
|
||||
quality_meta["rewritten"] = True
|
||||
logger.info("文案自检未通过 (ai_score=%d)，已改写", ai_score)
|
||||
else:
|
||||
logger.info("文案自检未通过 (ai_score=%d)，改写无变化", ai_score)
|
||||
|
||||
# 去 AI 化后处理
|
||||
if "content" in data:
|
||||
data["content"] = self._humanize_content(data["content"])
|
||||
|
||||
data["quality_meta"] = quality_meta
|
||||
return data
|
||||
|
||||
except (json.JSONDecodeError, ValueError) as e:
|
||||
@@ -636,15 +851,22 @@ class LLMService:
|
||||
|
||||
def generate_copy_with_reference(self, topic: str, style: str,
|
||||
reference_notes: str, sd_model_name: str = None, persona: str = None) -> dict:
|
||||
"""参考热门笔记生成文案（含重试逻辑，自动适配SD模型，支持人设）"""
|
||||
"""参考热门笔记生成文案（分层 Prompt，含重试逻辑，自动适配SD模型，支持人设）"""
|
||||
sd_guide = self.get_sd_prompt_guide(sd_model_name, persona=persona)
|
||||
prompt = PROMPT_COPY_WITH_REFERENCE.format(
|
||||
# 分层构建：基础层 + 风格层 + 参考笔记层 + 后缀
|
||||
ref_extra = PROMPT_COPY_WITH_REFERENCE_EXTRA.format(
|
||||
reference_notes=reference_notes, topic=topic, style=style,
|
||||
sd_prompt_guide=sd_guide,
|
||||
)
|
||||
user_msg = f"请创作关于「{topic}」的小红书笔记"
|
||||
style_prompt = PROMPT_STYLES.get(style, "")
|
||||
parts = [PROMPT_BASE]
|
||||
if style_prompt:
|
||||
parts.append(style_prompt)
|
||||
parts.append(ref_extra)
|
||||
if persona:
|
||||
user_msg = f"【博主人设】：{persona}\n请以此人设的视角和风格创作。\n\n{user_msg}"
|
||||
parts.append(f"\n【博主人设】：{persona}\n请以此人设的视角和风格创作。\n")
|
||||
parts.append(PROMPT_COPYWRITING_SUFFIX.format(sd_prompt_guide=sd_guide))
|
||||
prompt = "\n".join(parts)
|
||||
user_msg = f"请创作关于「{topic}」的小红书笔记"
|
||||
last_error = None
|
||||
for attempt in range(2):
|
||||
try:
|
||||
@@ -769,7 +991,7 @@ class LLMService:
|
||||
if t.startswith(prefix):
|
||||
t = t[len(prefix):].strip()
|
||||
|
||||
# ========== 第四层: 标点符号真人化 ==========
|
||||
# ========== 第四层: 标点符号真人化（增强版） ==========
|
||||
# AI 特征: 每句话都有完整标点 → 真人经常不加标点或只用逗号
|
||||
sentences = t.split('\n')
|
||||
humanized_lines = []
|
||||
@@ -777,16 +999,22 @@ class LLMService:
|
||||
if not line.strip():
|
||||
humanized_lines.append(line)
|
||||
continue
|
||||
# 随机去掉句末句号 (真人经常不打句号)
|
||||
if line.rstrip().endswith('。') and random.random() < 0.35:
|
||||
# 随机去掉句末句号 (真人经常不打句号) — 概率提高到 50%
|
||||
if line.rstrip().endswith('。') and random.random() < 0.50:
|
||||
line = line.rstrip()[:-1]
|
||||
# 随机把部分逗号替换成空格或什么都不加 (模拟打字不加标点)
|
||||
if random.random() < 0.15:
|
||||
# 只替换一个逗号
|
||||
# 随机把部分逗号替换成空格或什么都不加 (模拟打字不加标点) — 概率提高到 25%
|
||||
if random.random() < 0.25:
|
||||
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:]
|
||||
replacement = random.choice([' ', '', ' '])
|
||||
line = line[:pos] + replacement + line[pos+1:]
|
||||
# 随机把感叹号降级为句号 (AI 爱用感叹号)
|
||||
if line.rstrip().endswith('！') and random.random() < 0.20:
|
||||
line = line.rstrip()[:-1] + '。'
|
||||
# 随机删除句末标点 (真人有时就不加)
|
||||
if line.rstrip() and line.rstrip()[-1] in '。，,' and random.random() < 0.10:
|
||||
line = line.rstrip()[:-1]
|
||||
humanized_lines.append(line)
|
||||
t = '\n'.join(humanized_lines)
|
||||
|
||||
@@ -807,7 +1035,30 @@ class LLMService:
|
||||
paragraphs[idx] = connector + paragraphs[idx].lstrip()
|
||||
t = '\n\n'.join(paragraphs)
|
||||
|
||||
# ========== 第六层: 句子长度打散 ==========
|
||||
# ========== 第六层: 语气词注入 ==========
|
||||
# 真人说话带语气词 → AI 生成文本通常没有
|
||||
tone_particles_end = ['啊', '呢', '吧', '嘛', '呀', '哦', '啦', '噢']
|
||||
tone_particles_mid = ['嘿', '诶', '哈', '唉']
|
||||
lines = t.split('\n')
|
||||
particle_budget = random.randint(2, 4) # 全文最多注入 2-4 个
|
||||
injected = 0
|
||||
for i in range(len(lines)):
|
||||
if injected >= particle_budget:
|
||||
break
|
||||
line = lines[i].strip()
|
||||
if not line or len(line) < 6:
|
||||
continue
|
||||
# 句末语气词: 在没有标点或句号结尾的句子加语气词
|
||||
if random.random() < 0.15 and line[-1] not in '。！？!?~～…':
|
||||
lines[i] = lines[i].rstrip() + random.choice(tone_particles_end)
|
||||
injected += 1
|
||||
# 句首感叹词: 在段落开头偶尔加
|
||||
elif random.random() < 0.08 and not any(line.startswith(p) for p in tone_particles_mid):
|
||||
lines[i] = random.choice(tone_particles_mid) + ' ' + lines[i].lstrip()
|
||||
injected += 1
|
||||
t = '\n'.join(lines)
|
||||
|
||||
# ========== 第七层: 句子长度打散 ==========
|
||||
# AI 特征: 句子长度高度均匀 → 真人笔记长短参差不齐
|
||||
# 随机把一些长句用换行打散
|
||||
lines = t.split('\n')
|
||||
@@ -832,7 +1083,7 @@ class LLMService:
|
||||
final_lines.append(line)
|
||||
t = '\n'.join(final_lines)
|
||||
|
||||
# ========== 第七层: 随机注入微小不完美 ==========
|
||||
# ========== 第八层: 随机注入微小不完美 ==========
|
||||
# 真人打字偶尔有重复字、多余空格等
|
||||
if random.random() < 0.2:
|
||||
# 随机在某处加一个波浪号或省略号
|
||||
@@ -844,9 +1095,73 @@ class LLMService:
|
||||
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)
|
||||
# ========== 第九层: 段落节奏打散 ==========
|
||||
# AI 特征: 连续段落字数接近 → 真人笔记长短参差不齐
|
||||
paragraphs = t.split('\n\n')
|
||||
if len(paragraphs) >= 3:
|
||||
para_lens = [len(p.strip()) for p in paragraphs]
|
||||
for i in range(1, len(paragraphs) - 1):
|
||||
if para_lens[i] == 0:
|
||||
continue
|
||||
prev_len = para_lens[i - 1] if para_lens[i - 1] > 0 else 1
|
||||
# 如果连续两段字数差异 < 30%，尝试打散
|
||||
ratio = abs(para_lens[i] - prev_len) / max(para_lens[i], prev_len)
|
||||
if ratio < 0.30 and para_lens[i] > 20 and random.random() < 0.40:
|
||||
# 策略: 在中间段落找标点断开，制造长短不一
|
||||
p = paragraphs[i].strip()
|
||||
cut_pos = -1
|
||||
mid = len(p) // 3 # 在 1/3 处截断，制造不均匀
|
||||
for offset in range(0, mid):
|
||||
for check in [mid + offset, mid - offset]:
|
||||
if 0 < check < len(p) and p[check] in '，。！？、,!?':
|
||||
cut_pos = check
|
||||
break
|
||||
if cut_pos > 0:
|
||||
break
|
||||
if cut_pos > 0:
|
||||
paragraphs[i] = p[:cut_pos + 1] + '\n\n' + p[cut_pos + 1:].lstrip()
|
||||
t = '\n\n'.join(paragraphs)
|
||||
|
||||
# ========== 第十层: emoji 密度控制 ==========
|
||||
# 目标: 全文 6-12 个 emoji，分布不均匀，避免堆叠
|
||||
emoji_pattern = re.compile(
|
||||
r'[\U0001F300-\U0001F9FF\u2600-\u27BF\u2702-\u27B0'
|
||||
r'\U0001FA00-\U0001FA6F\U0001FA70-\U0001FAFF'
|
||||
r'\u231A-\u231B\u23E9-\u23F3\u23F8-\u23FA'
|
||||
r'\u25AA-\u25AB\u25B6\u25C0\u25FB-\u25FE'
|
||||
r'\u2614-\u2615\u2648-\u2653\u267F\u2693'
|
||||
r'\u26A1\u26AA-\u26AB\u26BD-\u26BE\u26C4-\u26C5'
|
||||
r'\u26D4\u26EA\u26F2-\u26F3\u26F5\u26FA\u26FD\u2934-\u2935]'
|
||||
)
|
||||
emojis_found = emoji_pattern.findall(t)
|
||||
emoji_count = len(emojis_found)
|
||||
target_min, target_max = 6, 12
|
||||
|
||||
if emoji_count > target_max:
|
||||
# 太多: 随机删除多余的
|
||||
excess = emoji_count - random.randint(target_min, target_max)
|
||||
if excess > 0:
|
||||
# 找到所有 emoji 位置，随机选择 excess 个删除
|
||||
positions = [m.start() for m in emoji_pattern.finditer(t)]
|
||||
remove_positions = set(random.sample(positions, min(excess, len(positions))))
|
||||
t = ''.join(c for idx, c in enumerate(t) if idx not in remove_positions)
|
||||
elif emoji_count < target_min and emoji_count > 0:
|
||||
# 太少: 在随机位置复制现有 emoji
|
||||
shortage = random.randint(target_min, target_min + 2) - emoji_count
|
||||
lines = t.split('\n')
|
||||
non_empty_lines = [i for i, l in enumerate(lines) if l.strip() and len(l.strip()) > 4]
|
||||
if non_empty_lines and emojis_found:
|
||||
for _ in range(min(shortage, len(non_empty_lines))):
|
||||
idx = random.choice(non_empty_lines)
|
||||
emoji_to_add = random.choice(emojis_found)
|
||||
# 在行末添加
|
||||
lines[idx] = lines[idx].rstrip() + emoji_to_add
|
||||
t = '\n'.join(lines)
|
||||
|
||||
# 去掉连续相同的 emoji 堆叠（超过 2 个相同的只保留 1 个）
|
||||
t = re.sub(r'([\U0001F300-\U0001F9FF\u2600-\u27BF])\1{1,}', r'\1', t)
|
||||
|
||||
# ========== 第十一层: 清理 ==========
|
||||
# 清理多余空行
|
||||
t = re.sub(r'\n{3,}', '\n\n', t)
|
||||
# 清理行首多余空格 (手机打字不会缩进)
|
||||
@@ -925,17 +1240,24 @@ class LLMService:
|
||||
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模型，支持人设）"""
|
||||
"""基于权重学习生成高互动潜力的文案（分层 Prompt，自动适配SD模型，支持人设）"""
|
||||
sd_guide = self.get_sd_prompt_guide(sd_model_name, persona=persona)
|
||||
prompt = PROMPT_WEIGHTED_COPYWRITING.format(
|
||||
# 分层构建：基础层 + 风格层 + 权重洞察层 + 后缀
|
||||
weighted_extra = PROMPT_WEIGHTED_COPYWRITING_EXTRA.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请创作一篇基于数据洞察的高质量小红书笔记"
|
||||
style_prompt = PROMPT_STYLES.get(style, "")
|
||||
parts = [PROMPT_BASE]
|
||||
if style_prompt:
|
||||
parts.append(style_prompt)
|
||||
parts.append(weighted_extra)
|
||||
if persona:
|
||||
user_msg = f"【博主人设】：{persona}\n请以此人设的视角和风格创作。\n\n{user_msg}"
|
||||
parts.append(f"\n【博主人设】：{persona}\n请以此人设的视角和风格创作。\n")
|
||||
parts.append(PROMPT_COPYWRITING_SUFFIX.format(sd_prompt_guide=sd_guide))
|
||||
prompt = "\n".join(parts)
|
||||
user_msg = f"主题：{topic}\n风格：{style}\n请创作一篇基于数据洞察的高质量小红书笔记"
|
||||
last_error = None
|
||||
for attempt in range(2):
|
||||
try:
|
||||
|
||||
@@ -19,7 +19,7 @@ logger = logging.getLogger(__name__)
|
||||
SD_TIMEOUT = 1800 # 图片生成可能需要较长时间
|
||||
|
||||
# 头像文件默认保存路径
|
||||
FACE_IMAGE_PATH = os.path.join(os.path.dirname(__file__), "assets", "faces", "my_face.png")
|
||||
FACE_IMAGE_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), "assets", "faces", "my_face.png")
|
||||
|
||||
# ==================== 多模型配置系统 ====================
|
||||
# 每个模型的最优参数、prompt 增强词、负面提示词、三档预设
|
||||
@@ -739,6 +739,7 @@ class SDService:
|
||||
def save_face_image(img: Image.Image, path: str = None) -> str:
|
||||
"""保存头像图片，返回保存路径"""
|
||||
path = path or FACE_IMAGE_PATH
|
||||
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||||
img = img.convert("RGB")
|
||||
img.save(path, format="PNG")
|
||||
logger.info("头像已保存: %s", path)
|
||||
|
||||
@@ -0,0 +1,462 @@
|
||||
"""
|
||||
services/topic_engine.py
|
||||
智能选题引擎 — 聚合热点数据 + 历史权重，推荐高潜力选题
|
||||
"""
|
||||
import logging
|
||||
import os
|
||||
import json
|
||||
import re
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
logger = logging.getLogger("autobot")
|
||||
|
||||
|
||||
class TopicEngine:
|
||||
"""
|
||||
智能选题推荐引擎
|
||||
|
||||
职责: 聚合热点探测结果与历史互动权重，为用户推荐高潜力选题。
|
||||
不直接访问 MCP / LLM，通过注入的 AnalyticsService 获取数据。
|
||||
"""
|
||||
|
||||
def __init__(self, analytics_service):
|
||||
"""
|
||||
Args:
|
||||
analytics_service: AnalyticsService 实例，提供权重和笔记数据
|
||||
"""
|
||||
self.analytics = analytics_service
|
||||
|
||||
# ========== 核心: 多维度评分 ==========
|
||||
|
||||
def score_topic(self, topic: str, hotspot_data: dict = None) -> dict:
|
||||
"""
|
||||
为单个候选主题计算综合评分
|
||||
|
||||
维度:
|
||||
- hotspot_score (0-40): 热点热度
|
||||
- weight_score (0-30): 历史互动权重
|
||||
- scarcity_score(0-20): 内容稀缺度
|
||||
- timeliness_score(0-10): 时效性
|
||||
|
||||
Args:
|
||||
topic: 候选主题文本
|
||||
hotspot_data: 可选的热点分析数据（包含 hot_topics, suggestions 等）
|
||||
|
||||
Returns:
|
||||
dict with total_score, hotspot_score, weight_score, scarcity_score, timeliness_score
|
||||
"""
|
||||
hotspot_score = self._calc_hotspot_score(topic, hotspot_data)
|
||||
weight_score = self._calc_weight_score(topic)
|
||||
scarcity_score = self._calc_scarcity_score(topic)
|
||||
timeliness_score = self._calc_timeliness_score(topic)
|
||||
|
||||
total = hotspot_score + weight_score + scarcity_score + timeliness_score
|
||||
|
||||
return {
|
||||
"total_score": total,
|
||||
"hotspot_score": hotspot_score,
|
||||
"weight_score": weight_score,
|
||||
"scarcity_score": scarcity_score,
|
||||
"timeliness_score": timeliness_score,
|
||||
}
|
||||
|
||||
# ========== 推荐主题列表 ==========
|
||||
|
||||
def recommend_topics(self, count: int = 5, hotspot_data: dict = None) -> list[dict]:
|
||||
"""
|
||||
推荐排序后的选题列表
|
||||
|
||||
逻辑:
|
||||
1. 收集候选主题 (热点 + 权重主题)
|
||||
2. 对每个主题评分
|
||||
3. 去重 (语义相近合并)
|
||||
4. 按总分降序取 top-N
|
||||
5. 为每个主题生成创作角度建议
|
||||
|
||||
Args:
|
||||
count: 返回推荐数量 (默认 5)
|
||||
hotspot_data: 可选的热点分析数据
|
||||
|
||||
Returns:
|
||||
list of dict, 每项包含:
|
||||
topic, score, reason, source, angles,
|
||||
score_detail (各维度分数)
|
||||
"""
|
||||
candidates = self._collect_candidates(hotspot_data)
|
||||
|
||||
if not candidates:
|
||||
logger.warning("选题引擎: 无候选主题可推荐")
|
||||
return []
|
||||
|
||||
# 评分
|
||||
scored = []
|
||||
for topic, source in candidates:
|
||||
detail = self.score_topic(topic, hotspot_data)
|
||||
scored.append({
|
||||
"topic": topic,
|
||||
"score": detail["total_score"],
|
||||
"source": source,
|
||||
"score_detail": detail,
|
||||
})
|
||||
|
||||
# 去重
|
||||
scored = self._deduplicate(scored)
|
||||
|
||||
# 排序
|
||||
scored.sort(key=lambda x: x["score"], reverse=True)
|
||||
scored = scored[:count]
|
||||
|
||||
# 生成 reason 和 angles
|
||||
for item in scored:
|
||||
item["reason"] = self._generate_reason(item)
|
||||
item["angles"] = self._generate_angles(item["topic"], item["source"])
|
||||
|
||||
return scored
|
||||
|
||||
# ========== 候选收集 ==========
|
||||
|
||||
def _collect_candidates(self, hotspot_data: dict = None) -> list[tuple[str, str]]:
|
||||
"""
|
||||
收集所有候选主题，返回 [(topic, source), ...]
|
||||
|
||||
source: "hotspot" | "weight" | "trend"
|
||||
"""
|
||||
candidates = []
|
||||
seen = set()
|
||||
|
||||
# 1. 从热点数据收集
|
||||
if hotspot_data:
|
||||
for topic in hotspot_data.get("hot_topics", []):
|
||||
topic_clean = self._clean_topic(topic)
|
||||
if topic_clean and topic_clean not in seen:
|
||||
candidates.append((topic_clean, "hotspot"))
|
||||
seen.add(topic_clean)
|
||||
|
||||
for suggestion in hotspot_data.get("suggestions", []):
|
||||
topic_clean = self._clean_topic(suggestion.get("topic", ""))
|
||||
if topic_clean and topic_clean not in seen:
|
||||
candidates.append((topic_clean, "hotspot"))
|
||||
seen.add(topic_clean)
|
||||
|
||||
# 2. 从权重数据收集
|
||||
topic_weights = self.analytics._weights.get("topic_weights", {})
|
||||
for topic, info in topic_weights.items():
|
||||
topic_clean = self._clean_topic(topic)
|
||||
if topic_clean and topic_clean not in seen:
|
||||
candidates.append((topic_clean, "weight"))
|
||||
seen.add(topic_clean)
|
||||
|
||||
# 3. 从分析历史提取趋势主题
|
||||
history = self.analytics._weights.get("analysis_history", [])
|
||||
for entry in history[-5:]:
|
||||
top_topic = entry.get("top_topic", "")
|
||||
if top_topic and top_topic not in seen:
|
||||
candidates.append((top_topic, "trend"))
|
||||
seen.add(top_topic)
|
||||
|
||||
return candidates
|
||||
|
||||
# ========== 评分子模块 ==========
|
||||
|
||||
def _calc_hotspot_score(self, topic: str, hotspot_data: dict = None) -> int:
|
||||
"""热点热度评分 (0-40)"""
|
||||
if not hotspot_data:
|
||||
return 0
|
||||
|
||||
score = 0
|
||||
|
||||
# 检查是否在热门主题中
|
||||
hot_topics = hotspot_data.get("hot_topics", [])
|
||||
for i, ht in enumerate(hot_topics):
|
||||
if self._topic_similar(topic, ht):
|
||||
# 排名越靠前分越高
|
||||
score = max(score, 40 - i * 5)
|
||||
break
|
||||
|
||||
# 检查是否在推荐建议中
|
||||
suggestions = hotspot_data.get("suggestions", [])
|
||||
for suggestion in suggestions:
|
||||
if self._topic_similar(topic, suggestion.get("topic", "")):
|
||||
score = max(score, 30)
|
||||
break
|
||||
|
||||
return min(40, score)
|
||||
|
||||
def _calc_weight_score(self, topic: str) -> int:
|
||||
"""历史互动权重评分 (0-30)"""
|
||||
topic_weights = self.analytics._weights.get("topic_weights", {})
|
||||
if not topic_weights:
|
||||
return 0
|
||||
|
||||
# 精确匹配
|
||||
if topic in topic_weights:
|
||||
weight = topic_weights[topic].get("weight", 0)
|
||||
# weight 原始范围 0-100，映射到 0-30
|
||||
return min(30, int(weight * 0.3))
|
||||
|
||||
# 模糊匹配
|
||||
best_score = 0
|
||||
for existing_topic, info in topic_weights.items():
|
||||
if self._topic_similar(topic, existing_topic):
|
||||
weight = info.get("weight", 0)
|
||||
best_score = max(best_score, min(30, int(weight * 0.3)))
|
||||
|
||||
return best_score
|
||||
|
||||
def _calc_scarcity_score(self, topic: str) -> int:
|
||||
"""
|
||||
内容稀缺度评分 (0-20)
|
||||
|
||||
近 7 天已发布 >= 2 篇的主题: scarcity_score <= 5
|
||||
"""
|
||||
notes = self.analytics._analytics_data.get("notes", {})
|
||||
seven_days_ago = (datetime.now() - timedelta(days=7)).isoformat()
|
||||
|
||||
recent_count = 0
|
||||
for nid, note in notes.items():
|
||||
collected = note.get("collected_at", "")
|
||||
if collected >= seven_days_ago:
|
||||
note_topic = note.get("topic", "")
|
||||
if self._topic_similar(topic, note_topic):
|
||||
recent_count += 1
|
||||
|
||||
if recent_count >= 2:
|
||||
return min(5, max(0, 5 - recent_count)) # 发的越多越低
|
||||
elif recent_count == 1:
|
||||
return 12 # 有一篇，中等稀缺
|
||||
else:
|
||||
return 20 # 完全空白，高稀缺
|
||||
|
||||
def _calc_timeliness_score(self, topic: str) -> int:
|
||||
"""
|
||||
时效性评分 (0-10)
|
||||
|
||||
基于主题是否包含时效性关键词（季节、节日等）
|
||||
"""
|
||||
now = datetime.now()
|
||||
month = now.month
|
||||
|
||||
# 季节关键词
|
||||
season_keywords = {
|
||||
"春": [2, 3, 4, 5],
|
||||
"夏": [5, 6, 7, 8],
|
||||
"秋": [8, 9, 10, 11],
|
||||
"冬": [11, 12, 1, 2],
|
||||
"早春": [2, 3],
|
||||
"初夏": [5, 6],
|
||||
"初秋": [8, 9],
|
||||
}
|
||||
|
||||
# 节日关键词
|
||||
festival_windows = {
|
||||
"情人节": (2, 10, 2, 18),
|
||||
"三八": (3, 1, 3, 12),
|
||||
"妇女节": (3, 1, 3, 12),
|
||||
"母亲节": (5, 5, 5, 15),
|
||||
"618": (6, 1, 6, 20),
|
||||
"七夕": (7, 20, 8, 15),
|
||||
"中秋": (9, 1, 9, 30),
|
||||
"国庆": (9, 25, 10, 10),
|
||||
"双十一": (10, 20, 11, 15),
|
||||
"双11": (10, 20, 11, 15),
|
||||
"双十二": (12, 1, 12, 15),
|
||||
"圣诞": (12, 15, 12, 28),
|
||||
"元旦": (12, 25, 1, 5),
|
||||
"年货": (1, 5, 2, 10),
|
||||
"春节": (1, 10, 2, 10),
|
||||
"开学": (8, 20, 9, 15),
|
||||
}
|
||||
|
||||
score = 5 # 基础分
|
||||
|
||||
# 季节匹配
|
||||
for keyword, months in season_keywords.items():
|
||||
if keyword in topic and month in months:
|
||||
score = max(score, 8)
|
||||
break
|
||||
|
||||
# 节日窗口匹配
|
||||
for keyword, (m1, d1, m2, d2) in festival_windows.items():
|
||||
if keyword in topic:
|
||||
start = datetime(now.year, m1, d1)
|
||||
end = datetime(now.year, m2, d2)
|
||||
# 处理跨年
|
||||
if start > end:
|
||||
if now >= start or now <= end:
|
||||
score = 10
|
||||
break
|
||||
elif start <= now <= end:
|
||||
score = 10
|
||||
break
|
||||
else:
|
||||
score = max(score, 3) # 不在窗口期但有时效关键词
|
||||
|
||||
return score
|
||||
|
||||
# ========== 去重 ==========
|
||||
|
||||
def _deduplicate(self, scored: list[dict]) -> list[dict]:
|
||||
"""
|
||||
去重: 语义相近的主题合并，保留分数较高者
|
||||
|
||||
例: "春季穿搭" 和 "早春穿搭" 合并为高分项
|
||||
"""
|
||||
if len(scored) <= 1:
|
||||
return scored
|
||||
|
||||
result = []
|
||||
merged_indices = set()
|
||||
|
||||
for i in range(len(scored)):
|
||||
if i in merged_indices:
|
||||
continue
|
||||
best = scored[i]
|
||||
for j in range(i + 1, len(scored)):
|
||||
if j in merged_indices:
|
||||
continue
|
||||
if self._topic_similar(scored[i]["topic"], scored[j]["topic"]):
|
||||
merged_indices.add(j)
|
||||
if scored[j]["score"] > best["score"]:
|
||||
best = scored[j]
|
||||
result.append(best)
|
||||
|
||||
return result
|
||||
|
||||
# ========== 辅助方法 ==========
|
||||
|
||||
@staticmethod
|
||||
def _clean_topic(topic: str) -> str:
|
||||
"""清理主题文本"""
|
||||
if not topic:
|
||||
return ""
|
||||
# 去除序号、emoji、多余空格
|
||||
t = re.sub(r'^[\d.、)\]】]+\s*', '', topic.strip())
|
||||
t = re.sub(r'[•·●]', '', t)
|
||||
return t.strip()
|
||||
|
||||
@staticmethod
|
||||
def _topic_similar(a: str, b: str) -> bool:
|
||||
"""
|
||||
判断两个主题是否语义相近 (简单规则匹配)
|
||||
|
||||
策略:
|
||||
1. 完全相同 → True
|
||||
2. 一方包含另一方 → True
|
||||
3. 去除修饰词后相同 → True
|
||||
4. 共享核心词比例 > 60% → True
|
||||
"""
|
||||
if not a or not b:
|
||||
return False
|
||||
|
||||
a_clean = a.strip().lower()
|
||||
b_clean = b.strip().lower()
|
||||
|
||||
# 完全相同
|
||||
if a_clean == b_clean:
|
||||
return True
|
||||
|
||||
# 包含关系
|
||||
if a_clean in b_clean or b_clean in a_clean:
|
||||
return True
|
||||
|
||||
# 去修饰词
|
||||
modifiers = ["早", "初", "晚", "新", "最", "超", "巨", "真的", "必看"]
|
||||
a_core = a_clean
|
||||
b_core = b_clean
|
||||
for mod in modifiers:
|
||||
a_core = a_core.replace(mod, "")
|
||||
b_core = b_core.replace(mod, "")
|
||||
if a_core and b_core and a_core == b_core:
|
||||
return True
|
||||
|
||||
# 核心词重叠
|
||||
# 按字分词 (中文简单分词)
|
||||
a_chars = set(a_clean)
|
||||
b_chars = set(b_clean)
|
||||
if len(a_chars) >= 2 and len(b_chars) >= 2:
|
||||
intersection = a_chars & b_chars
|
||||
union = a_chars | b_chars
|
||||
if len(intersection) / len(union) > 0.6:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def _generate_reason(item: dict) -> str:
|
||||
"""根据评分生成推荐理由"""
|
||||
detail = item.get("score_detail", {})
|
||||
parts = []
|
||||
|
||||
if detail.get("hotspot_score", 0) >= 25:
|
||||
parts.append("当前热点话题")
|
||||
if detail.get("weight_score", 0) >= 15:
|
||||
parts.append("历史互动表现好")
|
||||
if detail.get("scarcity_score", 0) >= 15:
|
||||
parts.append("内容空白可抢占")
|
||||
if detail.get("timeliness_score", 0) >= 8:
|
||||
parts.append("时效性强")
|
||||
|
||||
source = item.get("source", "")
|
||||
if source == "hotspot" and not parts:
|
||||
parts.append("热点趋势推荐")
|
||||
elif source == "weight" and not parts:
|
||||
parts.append("基于历史表现推荐")
|
||||
elif source == "trend" and not parts:
|
||||
parts.append("持续趋势主题")
|
||||
|
||||
if not parts:
|
||||
parts.append("综合推荐")
|
||||
|
||||
return "，".join(parts)
|
||||
|
||||
@staticmethod
|
||||
def _generate_angles(topic: str, source: str) -> list[str]:
|
||||
"""
|
||||
为主题生成 1-3 个创作角度建议
|
||||
|
||||
注意: 这里用规则生成，不调用 LLM
|
||||
"""
|
||||
angles = []
|
||||
|
||||
# 通用角度模板
|
||||
templates_by_type = {
|
||||
"穿搭": [
|
||||
f"从预算角度分享{topic}的平替选择",
|
||||
f"身材不同如何驾驭{topic}",
|
||||
f"一周{topic}不重样的实穿记录",
|
||||
],
|
||||
"美食": [
|
||||
f"零失败的{topic}详细做法",
|
||||
f"外卖 vs 自己做{topic}的对比",
|
||||
f"{topic}的隐藏吃法",
|
||||
],
|
||||
"护肤": [
|
||||
f"不同肤质的{topic}选择指南",
|
||||
f"踩雷vs回购：{topic}真实体验",
|
||||
f"平价替代大牌{topic}推荐",
|
||||
],
|
||||
"好物": [
|
||||
f"用了半年的{topic}真实测评",
|
||||
f"后悔没早买的{topic}清单",
|
||||
f"从使用场景出发推荐{topic}",
|
||||
],
|
||||
}
|
||||
|
||||
# 根据主题关键词匹配模板
|
||||
matched = False
|
||||
for keyword, templates in templates_by_type.items():
|
||||
if keyword in topic:
|
||||
angles = templates[:3]
|
||||
matched = True
|
||||
break
|
||||
|
||||
if not matched:
|
||||
# 通用角度
|
||||
angles = [
|
||||
f"个人真实体验分享{topic}",
|
||||
f"新手入门{topic}的详细攻略",
|
||||
f"关于{topic}的冷知识和避坑指南",
|
||||
]
|
||||
|
||||
# 限制每个角度不超过 30 字
|
||||
return [a[:30] for a in angles]
|
||||
Reference in New Issue
Block a user