diff --git a/backend/app/api/v1/benchmarks.py b/backend/app/api/v1/benchmarks.py index b0b405b..f042ba0 100644 --- a/backend/app/api/v1/benchmarks.py +++ b/backend/app/api/v1/benchmarks.py @@ -33,10 +33,18 @@ class BannedWordCreate(BaseModel): def _fmt_benchmark(b: BenchmarkNote) -> dict: + import json + features = None + if b.features_json: + try: + features = json.loads(b.features_json) + except Exception: + features = None return { "id": b.id, "product_id": b.product_id, "screenshot_url": b.screenshot_url, "highlights": b.highlights, "link_url": b.link_url, + "features": features, "analyze_status": b.analyze_status, "created_at": b.created_at.isoformat(), } @@ -84,7 +92,66 @@ def create_benchmark( return ok(_fmt_benchmark(b)) -# ── 违禁词库 ──────────────────────────────────────────────── +@router.post("/benchmarks/{benchmark_id}/analyze") +async def analyze_benchmark_note( + benchmark_id: int, + current_user: Annotated[CurrentUser, Depends(require_admin)] = None, + db: Session = Depends(get_db), +): + """ + 第2环：对已存标杆笔记跑 8 维拆解。 + 输入源=screenshot_url 截图(vision) + highlights 手填亮点(参考)。 + 结果写入 features_json，analyze_status 流转 analyzing→done/failed。 + key 链路遵基石B：plain_key 只活局部，立即清零。 + """ + import json + from app.core.response import raise_business + from app.models.workspace import UserApiKey + from app.utils.fernet_utils import decrypt_key + from app.services.ai_engine.gemini_factory import build_ai_clients + from app.services.ai_engine.benchmark_analyzer import analyze_benchmark + + b = db.query(BenchmarkNote).filter( + BenchmarkNote.id == benchmark_id, + BenchmarkNote.workspace_id == current_user.workspace_id, + ).first() + if not b: + raise_not_found("标杆笔记不存在") + if not b.screenshot_url and not b.highlights: + raise_business("该标杆无截图也无手填亮点，无法分析") + + # 读截图字节（screenshot_url 存的是上传后的绝对/相对路径） + screenshot = None + if b.screenshot_url: + try: + with open(b.screenshot_url, "rb") as f: + screenshot = f.read() + except Exception as e: + logger.warning("标杆截图读取失败 id=%s: %s", benchmark_id, e) + + api_key_row = db.query(UserApiKey).filter( + UserApiKey.user_id == current_user.user_id, + UserApiKey.workspace_id == current_user.workspace_id, + UserApiKey.provider.in_(["openai", "apiports"]), + ).first() + if not api_key_row: + raise_business("未配置 API Key，请先在设置中录入") + plain_key = decrypt_key(api_key_row.encrypted_key) + clients = build_ai_clients(plain_key) + plain_key = None + + b.analyze_status = "analyzing"; db.commit() + try: + result = await analyze_benchmark(clients, screenshot, b.highlights) + finally: + await clients.aclose() + + b.features_json = json.dumps(result, ensure_ascii=False) + b.analyze_status = "done" if result.get("source") != "fallback" else "failed" + db.commit(); db.refresh(b) + logger.info("benchmark analyzed id=%s status=%s source=%s", + benchmark_id, b.analyze_status, result.get("source")) + return ok({**_fmt_benchmark(b), "features": result, "analyze_status": b.analyze_status}) @router.get("/banned-words") def list_banned_words( page: int = 1, page_size: int = 50, diff --git a/backend/app/services/ai_engine/_benchmark_prompt.py b/backend/app/services/ai_engine/_benchmark_prompt.py new file mode 100644 index 0000000..e725ba6 --- /dev/null +++ b/backend/app/services/ai_engine/_benchmark_prompt.py @@ -0,0 +1,25 @@ +""" +app/services/ai_engine/_benchmark_prompt.py — 第2环标杆8维分析 prompt + +8维草案来源（倩倩姐2026-06-16）：banana不变量/可变量方法论 + 俊达拆解中心配方 + 3小红书链接。 +方法论核心：分析目的不是打分，是提炼"仿写时必须保留的配方"（不变量），喂给第5环文案+第6环配图对标。 +输入=爆款笔记截图（GPT vision 读图）+ 运营手填亮点（highlights）。 +""" + +BENCHMARK_8DIM_PROMPT = """你是小红书爆款拆解专家。请把这篇爆款笔记拆成"可复用配方"，供后续对标仿写。 +硬性规则：只分析本次提供的截图与亮点描述，不臆造、不沿用历史案例、不默认任何品类。 +看不清或无法判断的维度，填"截图未体现"，不要编。 + +请从以下8个维度拆解，返回JSON（仅JSON，无其他文字）： +{ + "title_formula": "标题用了什么套路（提问式/数字式/悬念式/对比式/利益前置等）+大致字数", + "opening_hook": "开头前1-3句如何抓人（痛点直击/反差/利益前置/场景代入等），摘录关键句", + "content_structure": "正文骨架顺序（如 痛点-方案-效果 / 故事-产品-号召 / 测评-对比-结论）", + "selling_point_style": "卖点怎么呈现（几个卖点、轻重排布、是否数据化/场景化、植入方式）", + "emotion_tone": "情绪语气（亲切/专业/俏皮/种草感/姐妹分享等）+ emoji使用规律（频率/类型/位置）", + "target_audience": "这篇打给谁（年龄段/使用场景/核心痛点画像），一句话", + "topic_tags": "话题标签策略（选了哪类tag、几个、大词与精准小词的配比）", + "visual_rhythm": "视觉配图配方（封面风格、构图、色调、光线、配图排布节奏、是否压字/大字报模板）", + "confidence": 0.9, + "source": "vision" +}""" diff --git a/backend/app/services/ai_engine/benchmark_analyzer.py b/backend/app/services/ai_engine/benchmark_analyzer.py new file mode 100644 index 0000000..c6c17fc --- /dev/null +++ b/backend/app/services/ai_engine/benchmark_analyzer.py @@ -0,0 +1,66 @@ +""" +app/services/ai_engine/benchmark_analyzer.py — 第2环 标杆8维分析引擎 + +复用 products.analyze_product_image 的 vision 范式，解耦成 service。 +输入：爆款截图 bytes + 手填亮点；输出：8维配方 JSON（写入 benchmark_notes.features_json）。 +key 链路遵基石B：plain_key 只活在调用方局部，本模块只收已构建的 clients。 +""" +import json +import logging +import re + +from app.services.ai_engine._benchmark_prompt import BENCHMARK_8DIM_PROMPT + +logger = logging.getLogger(__name__) + +# 8维字段顺序（与 prompt JSON key 一致；analyze_status 流转 done 时校验齐全） +DIM_KEYS = [ + "title_formula", "opening_hook", "content_structure", "selling_point_style", + "emotion_tone", "target_audience", "topic_tags", "visual_rhythm", +] + + +def parse_benchmark_json(raw: str) -> dict: + """从模型返回文本提取 JSON（容错 markdown ```json 包裹）。""" + cleaned = re.sub(r"```json\s*", "", raw) + cleaned = re.sub(r"```\s*", "", cleaned) + m = re.search(r"\{[\s\S]*\}", cleaned) + if not m: + raise ValueError("标杆分析返回中未找到 JSON") + return json.loads(m.group()) + + +def fallback_benchmark(highlights: str | None = None) -> dict: + """vision 失败兜底：保证 8 维不空手，标 source=fallback 供前端提示。""" + note = highlights or "截图未体现" + return { + "title_formula": note, "opening_hook": note, "content_structure": note, + "selling_point_style": note, "emotion_tone": note, "target_audience": note, + "topic_tags": note, "visual_rhythm": note, + "confidence": 0.4, "source": "fallback", + } + + +async def analyze_benchmark(clients, screenshot: bytes | None, highlights: str | None) -> dict: + """ + 跑 8 维分析。截图走 vision；无截图但有手填亮点则把亮点拼进 prompt 兜底。 + 返回结构化 8 维 dict（含 confidence/source）。失败不抛，返回 fallback。 + """ + prompt = BENCHMARK_8DIM_PROMPT + if highlights: + prompt = f"{prompt}\n\n运营补充的亮点描述（作为分析参考）：{highlights}" + + images = [screenshot] if screenshot else [] + try: + if not images: + # 无截图：纯文本兜底（gpt_vision_analyze 支持空图走文本） + logger.info("analyze_benchmark: 无截图，走手填亮点文本分析") + raw = await clients.gpt_vision_analyze(prompt, images) + result = parse_benchmark_json(raw) + result["source"] = "vision" if images else "text" + return result + except Exception as e: + logger.warning("analyze_benchmark failed, fallback: %s", e) + result = fallback_benchmark(highlights) + result["warning"] = f"标杆分析失败，已兜底：{e}" + return result