baseline: Clover 独立仓库首次基线提交

将 Clover 从上层产品包旧仓库中独立出来,建立专属版本控制。
当前状态=纵切片端到端已打通(登录→选品→出文出图→审核→下载包),
M1文案质量去套路化已验收。此提交作为后续按核销清单逐条修复的基线。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
yangqianqian
2026-06-16 11:30:22 +08:00
commit 6a2632da70
253 changed files with 27467 additions and 0 deletions

View File

@@ -0,0 +1,13 @@
# app/services/
业务逻辑层占位,按模块:
- auth_service.py # JWT签发/验证
- workspace_service.py # workspace权限校验
- product_service.py # 产品档案业务逻辑
- task_service.py # 任务状态机流转
- review_service.py # 审核流转+飞轮信号写入
- preference_aggregator.py # 飞轮实时聚合最近50条→prompt片段
- preference_collector.py # 三信号入口写入 preference_events
- banned_word_checker.py # 违禁词三级扫描(🟢改写/🟡提示/🔴拦截)
- package_exporter.py # 生成达人素材交付包
- image_postprocessor.py # 去水印后处理(重编码+削像素水印)

View File

View File

@@ -0,0 +1,5 @@
"""
AI 引擎包
扒自:上线版 worker/src/copy.js + image.js2026-06-09
重写为 Python逻辑对照JS版防走样
"""

View File

@@ -0,0 +1,104 @@
"""
_score_prompt.py — AI 评委 prompt让模型真读文案不机械找词
评判标准忠于《富贵情绪营销理论》原文(口播一手来源),标实战补充出处。
6维满分分布倩倩姐2026-06-15拍板与 llm_scorer._DIM_MAX / constants.AI_DIM_WEIGHTS 三处同步):
痛点人群精准18 / 情绪张力18 / 买点转化18 / 开头钩子15 / 标题点击力13 / 真实感13 = 95
+ 合规5机械硬拦不进AI评委= 100
"真实感"替换旧"产品聚焦一件事(16)":富贵"很少提产品/前70%干货后30%植入"独立升维。
"""
from __future__ import annotations
# ── 评委人设 ──────────────────────────────────────────────
SCORER_PERSONA = """你是一位资深小红书内容操盘手,深谙富贵情绪营销理论。
你的本事是:扫一眼就知道一条笔记能不能打动目标用户、会不会被划走。
你不数关键词、不看有没有出现某个固定词——你读的是【这条文案对真实用户有没有杀伤力】。
你按下面6个维度给文案打分每一维都要给出【具体理由】指出好在哪/差在哪/怎么改,
理由必须针对这条文案的真实内容,不准说放之四海皆准的空话。"""
# ── 6 维评判标准按权重降序合规第7维由代码机械硬拦不在此────
# 标准依据:括号内标注[富贵原文]或[实战补充],前者来自口播一手来源,权威最高。
SCORING_DIMENSIONS = """
【维度1·痛点人群精准满分18】[富贵原文]
判断:"说的就是我"——文案描述的处境/困扰,目标用户读了会不会对号入座。
好:具体到某类人的某个真实生活瞬间,让人觉得被看穿,落在"我的大问题/我的处境/我的渴望"上。
差:泛泛而谈谁都能套(如"适合所有想变美的女生"或PUA用户、拿别人的惨状吓唬人。
依据:富贵"我的大问题→处境→渴望"内容骨架;人群越具体穿透力越强;"用户被宠成爹你PUA他不好使"
【维度2·情绪张力满分18】[富贵原文·第一性原理]
判断:有没有"成果/后果"双向情绪,而不是平铺直叙报卖点。
· 后果=过去没用它,产生了什么糟糕处境(勾起恐惧/懊悔)
· 成果=用了它之后,会变成什么样(给出期待/乐观)
好:同时有"后果路径(过去的痛)+成果路径(未来的好)",有一句能戳中人、读完有情绪余温。
差:全程客观介绍产品、无情绪、像说明书;或只单向吓唬、或只空喊美好。
依据:富贵"营销第一性原理就是情绪,没有情绪什么内容都不转化""成果是未来、后果是过去,要做这两种情绪"
【维度3·买点转化满分18】[富贵原文]
判断:产品卖点有没有翻译成用户能感知的场景化利益(人话),而不是品牌视角的功效/参数。
好:用户能想象到的使用场景和结果(如"出门前最后一步、同事问我今天气色真好")。
差:品牌语言/参数罗列(如"采用XX技术""含XX成分"),用户无感。
依据:富贵"卖点是品牌视角、买点是用户视角""用户买的不是产品,是使用场景背后被解决的问题"
【维度4·开头钩子满分15】[富贵原文]
判断:开头能不能让人停下来、想继续读。
好:开头第一句就直击用户的大问题/痛处/具体场景代入,每句都打要害。
差:开头是套话或铺垫半天不进正题,没有任何抓人的点。
依据:富贵"内容就是锋利的刀,一定要插用户的心窝子"
【维度5·标题点击力满分13】[富贵原文+实战补充]
判断:标题有没有让目标用户想点的诱因。
好:标题带具体人群/场景/痛点/情绪钩子,一眼觉得"和我有关、我想看",最好来自用户真实说法。
差:只有产品名、平淡无钩子、或太像广告。
依据:富贵"热评就是标题,从真实评论里抓用户的话"[原文];标题善用痛点/人群/效果/情绪词[实战补充]。
【维度6·真实感满分13】[富贵原文]
判断:整条文案是不是像真人分享,而不是品牌广告或功效说明书。
好:价值/场景/感受为主,产品自然带出;前段是干货或真实经历,后段才软性推荐;不开头就报规格价格。
差:通篇硬广、产品功效罗列当主体、开头就卖、语气像文案模板而非真实人设。
依据:富贵"我们很少提产品""前70%是价值/场景后30%才是产品";用户能感受到"这不像广告"
""".strip()
# ── 真实感已升为维度6满分13此处保留空字符串避免调用方引用报错 ──
REALNESS_NOTE = "" # 升为 SCORING_DIMENSIONS 维度6不再重复附加
# ── 输出格式约束 ──────────────────────────────────────────
SCORER_OUTPUT_FORMAT = """
读完这条文案后严格返回纯JSON对象不要markdown代码块、不要多余文字格式
{
"dims": [
{"item":"痛点人群精准","score":<0-18整数>,"reason":"<针对本条的具体理由30字内>"},
{"item":"情绪张力","score":<0-18>,"reason":"..."},
{"item":"买点转化","score":<0-18>,"reason":"..."},
{"item":"开头钩子","score":<0-15>,"reason":"..."},
{"item":"标题点击力","score":<0-13>,"reason":"..."},
{"item":"真实感","score":<0-13>,"reason":"..."}
],
"verdict":"<优秀|合格|不合格>",
"summary":"<一句话总评,说清这条最大的优点和最该改的点>"
}
打分要敢拉开差距:平庸文案该给中低分,不要清一色高分;优秀的地方也别吝啬给高分。
""".strip()
# 合规维度满分(机械硬拦,不进 AI 评委)
COMPLIANCE_MAX = 5
# AI 评委 6 维满分合计(用于把 0-95 折算/校验;+合规5=100
AI_DIMS_MAX = 95
def build_score_prompt(copy: dict, product: dict | None = None) -> str:
"""组装单条文案的评委 prompt。copy={title,content,...}product 提供品牌/品类语境。"""
title = str(copy.get("title", "")).strip()
content = str(copy.get("content", "")).strip()
ctx = ""
if product:
name = product.get("name") or product.get("title") or ""
brand = product.get("brand_keyword") or product.get("brand") or ""
cat = product.get("category") or ""
bits = [b for b in (f"产品:{name}", f"品牌词:{brand}", f"品类:{cat}") if b.split("", 1)[1]]
if bits:
ctx = "【产品语境】\n" + "\n".join(bits) + "\n\n"
return (
f"{SCORING_DIMENSIONS}\n\n"
f"{ctx}【待评文案】\n标题:{title}\n正文:{content}\n\n"
f"{SCORER_OUTPUT_FORMAT}"
)

View File

@@ -0,0 +1,92 @@
"""
_scoring_dims.py — 五维打分逻辑(单一职责:计算层)
词表常量 + 每维打分函数,由 text_scoring.score_copy 调用
"""
from __future__ import annotations
import re
from typing import Any
from .constants import SCORE_WEIGHTS, BANNED_WORDS_DEFAULT, BANNED_VISUAL_WORDS, INTERNAL_COPY_HINTS
_EMOTION_WORDS = ["谁懂", "绝了", "姐妹", "宝子", "挖到", "救星", "离不开",
"闭眼入", "", "不踩雷", "后悔没早买"]
_SCENE_WORDS = ["通勤", "上班", "办公室", "工位", "宿舍", "出门", "旅行",
"居家", "运动", "带娃", "约会", "熬夜", "换季", "饭后",
"早餐", "下午", "加班", "外食", "日常", "早八", "学生党", "新手", "宝妈"]
_CATEGORY_WORDS: dict[str, list[str]] = {
"美妆护肤": ["", "", "质地", "服帖", "清透", "水润", "自然", "上脸", "底妆"],
"个护护理": ["", "护理", "", "", "吸收", "滋润", "粗糙", "倒刺", "随身", "质地"],
"食品饮品": ["口感", "入口", "味道", "冲泡", "", "工位", "", "不腻", "清爽"],
"营养健康": ["成分", "日常", "坚持", "习惯", "补给", "安心", "配方"],
"家居生活": ["收纳", "省事", "清洁", "桌面", "厨房", "租房", "细节", "高频"],
"服饰穿搭": ["上身", "版型", "面料", "通勤", "显瘦", "搭配", "质感"],
"通用好物": ["实用", "场景", "细节", "省事", "日常", "高频"],
}
def _has_any(text: str, words: list[str]) -> bool:
return any(w.lower() in text.lower() for w in words)
def _cat_words(category: str) -> list[str]:
return _CATEGORY_WORDS.get(category, _CATEGORY_WORDS["通用好物"])
def score_title(title: str, cat_words: list[str], w: dict) -> dict:
ts = 0
if 6 <= len(title) <= 34: ts += 8
if re.search(r"[0-9一二三四五六七八九十几]|最近|这支|这个|每天|早八|学生党|新手|宝妈|懒人|伪素颜", title): ts += 5
if _has_any(title, _SCENE_WORDS): ts += 6
if _has_any(title, cat_words): ts += 5
if re.search(r"[!?]|救星|不踩雷|闭眼入|会回购|被问|惊喜|加分|常备|省心|实用|别乱选|放心|有气色", title): ts += 6
if re.search(r"救星|绝了|挖到|偷偷|被问|离不开|后悔|拿捏|香|包里|回购|常备|工位|换季", title): ts += 4
ts = min(ts, w["title"])
return {"item": "标题吸引力", "score": ts, "max": w["title"],
"note": "标题有明确人群/场景钩子" if ts >= 18 else "建议补充人群、场景或强钩子"}
def score_emotion(full: str, w: dict) -> dict:
es = 0
if _has_any(full, _EMOTION_WORDS): es += 10
if _has_any(full, _SCENE_WORDS): es += 8
if re.search(r"不假白|不卡|不搓泥|没底气|纠结|救|不黏|不腻|踩雷|麻烦|翻车", full): es += 7
if re.search(r"姐妹|宝子|室友|同事|我自己|实测|亲测|上脸|出门|工位|宿舍|家里", full): es += 5
if re.search(r"[✅✨🌿💧📦🔍🧡🪞🧴🍃🥹😭👍]", full): es += 3
es = min(es, w["emotion"])
return {"item": "情绪共鸣", "score": es, "max": w["emotion"],
"note": "口语感和痛点表达较充分" if es >= 18 else "建议增加真实痛点和口语化表达"}
def score_selling(copy: dict, full: str, selling_points: list, w: dict) -> dict:
bs = 0
matched = [pt for pt in selling_points if any(kw in full for kw in str(pt).split()[:3])]
if len(matched) >= 1: bs += 7
if len(matched) >= 2: bs += 4
if re.search(r"方便|自然|清爽|质地|口感|成分|实测|亲测|场景|随身|省事|高频|性价比|好用|适合|推荐", full): bs += 8
if copy.get("buyingPoint") or re.search(r"分钟|出门|通勤|办公室|宿舍|居家|换季|工位|旅行", full): bs += 9
if copy.get("coverTitle") or copy.get("imageBrief"): bs += 4
bs = min(bs, w["selling"])
return {"item": "买点表达", "score": bs, "max": w["selling"],
"note": "卖点已转成用户可感知买点" if bs >= 18 else "建议把功能卖点翻译成使用场景和结果"}
def score_keyword(copy: dict, tags: str, keywords: list, w: dict) -> dict:
ks = 0
matched = [k for k in keywords if str(k).replace("#", "") in tags + str(copy.get("content",""))]
if len(matched) >= 1: ks += 6
if len(matched) >= 2: ks += 5
if len(copy.get("tags", [])) >= 3: ks += 5
if "#" in tags: ks += 4
if len(copy.get("tags", [])) >= 5: ks += 4
ks = min(ks, w["keyword"])
note = f"覆盖:{''.join(matched)}" if matched else "建议补充品类词和长尾词"
return {"item": "关键词覆盖", "score": ks, "max": w["keyword"], "note": note}
def score_compliance(full: str, bwords: list[str], w: dict) -> tuple[dict, list[str]]:
found_banned = [bw for bw in bwords if bw.lower() in full.lower()]
found_hints = [hw for hw in INTERNAL_COPY_HINTS if hw in full]
cs = 0 if (found_banned or found_hints) else w["compliance"]
note = (f"含禁用词:{''.join(found_banned)}" if found_banned
else (f"正文混入内部提示:{''.join(found_hints)}" if found_hints else "未发现禁用词"))
return {"item": "合规性", "score": cs, "max": w["compliance"], "note": note}, found_banned + found_hints

View File

@@ -0,0 +1,64 @@
"""
_storyboard_data.py — 品类证明策略数据(纯数据,不含逻辑)
品类来自 product.category不枚举兜底用"通用好物"
"""
from __future__ import annotations
# 视觉违禁词替换规则(扒 sanitizeImagePlanText
SANITIZE_RULES: list[tuple[str, str]] = [
(r"before\s*&\s*after", "质地与肤感说明"),
(r"before\s*/?\s*after", "质地与肤感说明"),
(r"\bbefore\b", "质地状态"),
(r"\bafter\b", "上脸肤感"),
(r"使用前后|用前用后|用前后|前后对比|使用前|使用后", "质地/场景/肤感说明"),
(r"功效对比|效果对比|改善对比", "质地/场景说明对比"),
(r"肤色变白|皮肤变白|变白|美白", "自然光泽感"),
(r"瑕疵消失|斑点消失|痘印消失|消除瑕疵|祛斑", "妆感更服帖"),
(r"治疗前后|治疗后|医美前后|治愈|修复受损", "日常使用场景说明"),
]
# 品类证明策略不写死枚举product.category 匹配,兜底通用)
PROOF_STRATEGIES: dict[str, dict] = {
"个护护理": {
"overlay_tpl": "{point}看得见",
"visual": "手部/身体局部使用证明:少量点涂、推开后吸收状态、真实纹理和自然光",
"asset_use": "优先使用实拍/参考图中的手部、干纹、涂抹、随身场景",
"forbidden": "不要变白、祛斑、医学效果、before/after字样不要和封面同构图",
},
"美妆护肤": {
"overlay_tpl": "{point}看得见",
"visual": "肤感/质地证明:手背、脸颊局部或质地微距,展示推开前后真实状态",
"asset_use": "优先使用实拍/参考图中的手背、上脸、质地素材",
"forbidden": "不要变白、祛斑、医学效果、before/after字样",
},
"食品饮品": {
"overlay_tpl": "{point}一眼懂",
"visual": "冲泡/开袋/入口证明:展示包装、杯中状态、质地颜色,真实桌面光线",
"asset_use": "产品图保证包装准确,参考图用于杯子、开袋、冲泡、居家场景",
"forbidden": "不要涂抹、不要护肤肤感、不要医疗健康承诺",
},
"营养健康": {
"overlay_tpl": "看清{point}",
"visual": "理性证明页:包装、成分表、使用场景和每日习惯卡片",
"asset_use": "产品图和说明图用于成分/包装准确,参考图用于日常使用场景",
"forbidden": "不要治疗、改善疾病、速效、医生背书、前后对比",
},
"家居生活": {
"overlay_tpl": "{point}真省事",
"visual": "使用过程证明:展示痛点场景、产品介入和使用过程细节",
"asset_use": "参考图用于真实家居环境,产品图保证外观准确",
"forbidden": "不要护肤涂抹,不要虚假夸大结果",
},
"服饰穿搭": {
"overlay_tpl": "{point}有细节",
"visual": "上身/材质证明:展示面料纹理、版型细节或普通身材上身局部",
"asset_use": "参考图用于上身/搭配/材质,产品图保证款式颜色准确",
"forbidden": "不要护肤涂抹,不要过度精修模特感",
},
"通用好物": {
"overlay_tpl": "{point}清晰可见",
"visual": "产品使用场景证明:真实道具/场景,展示产品细节和使用过程",
"asset_use": "产品图保证准确,参考图用于场景辅助",
"forbidden": "不要夸大效果,不要硬广式价格牌",
},
}

View File

@@ -0,0 +1,209 @@
"""
_text_prompt.py — 文案 prompt 组装 / JSON 解析 / 本地模板兜底
方法层(全品类共用):人设/变量池/5步框架/四段结构/反AI味规则
数据层(每产品各异):由 product dict 动态注入,代码不出现具体品牌/成分名
"""
from __future__ import annotations
import json
import random
import re
# ── Q4 佛系反推销人设全品类共用数据层产品名从product动态注入──────────
_PERSONA = """你是一个日常生活分享博主,不是品牌推广号。
核心人设:佛系、不推销、真实记录日常。
内核:不主动劝买,靠真实体验让读者自己动心;写的是生活,产品只是生活的一部分。
比例70% 写生活场景和使用感受30% 提产品,绝不颠倒。
收尾铁律:每条结尾方式必须不同,禁止使用"东西放这了/买不买跟我没关系"这类被用滥的固定句式。"""
# ── Q1 随机变量池 ABC反同质化核心每次随机抽组合N条不撞──────────────
# 方法层:框架固定,内容可扩展,绝不写死
_POOL_A_IDENTITY: list[str] = [
"上班族早八妆前随手抹",
"宿舍懒人护肤三分钟搞定",
"敏感肌妈妈哄完娃才有五分钟",
"学生党第一次用高价护肤品",
"素颜出门前最后一步",
]
_POOL_B_EMOTION: list[str] = [
"看到镜子里发现气色暗了一周",
"闺蜜问你最近皮肤怎么这么好",
"出门被催快点根本来不及叠瓶",
]
_POOL_C_FLAW: list[str] = [
"第一次用量太少了没啥感觉",
"包装简单到以为是山寨",
"价格摆在那以为会很油很厚",
]
def _pick_combo() -> dict[str, str]:
"""随机抽一组 ABC 变量每次生成调用一次N条各不相同"""
return {
"identity": random.choice(_POOL_A_IDENTITY),
"emotion": random.choice(_POOL_B_EMOTION),
"flaw": random.choice(_POOL_C_FLAW),
}
# ── Q5 negative词prompt级别负向约束不让AI写进正文 ─────────────────────
_NEGATIVE_WORDS = (
"神器、福音、救急单品、遮羞布、日常维稳、精简底妆、"
"不仅而且、焕发、守护、尽享、日常维稳、"
"按头安利、绝绝子、闭眼冲、杀疯了、YYDS"
)
# ── emoji 规则适度有表情倩倩姐2026-06-08拍板像真人发的小红书──────────
_EMOJI_RULES = """
【emoji表情适度有表情必须遵守
- 卖点小标题前加 emoji✅ 卖点1 / ✨ 卖点2 / 🌿 卖点3每条卖点1个符合语义
- 正文段落可点缀少量 emoji 烘托情绪(如 🥹 😭 🤍 💛每段最多1-2个不堆砌
- 结尾话题标签前后带表情,如 "#好物分享 🛒"
- emoji 服务情绪和分点,不要每句都加;整条正文 emoji 总量控制在 6-12 个
- 常用小红书 emoji 池:✅✨🌿💧🪞🧴📦🔍💛🤍🥹😭🛒(按语义选,不乱用)
""".strip()
# ── Q2 5步框架 + Q3 四段结构 ─────────────────────────────────────────────
_STRUCTURE_RULES = """
【5步框架必须严格遵循
① Hook暴击低价/痛点:第一句戳中场景或价格锚点,吊足读者好奇
② 痛点共鸣2-3句描写使用前的真实困境用上面抽到的起因情绪A·B
③ 救星登场:自然带出产品,口吻是"碰巧发现/朋友安利/囤货时顺手",不是"推荐给你"
④ 卖点罗列每条加✅小标题3条以内卖点翻译成使用感受不是功效列表
⑤ 收尾(每条必须从下方策略池随机选一种,同批次不得重复同一种,禁止"东西放这了/买不买跟我没关系"此类固定句式):
【收尾策略池·每条选不同策略】
A·留白式感受只说自己现在的状态不提买不买"反正现在素颜出门我不慌了"
B·反问读者把感受抛回给读者"你们护肤有没有那种一用就回不去的东西?"
C·场景延续把故事延伸到未来某个细节"下次同事再问我皮肤的事我就知道说啥了"
D·克制回购暗示轻描淡写说自己行为"第一罐用完了,已经在备第二罐"
E·纯记录收笔像日记最后一句不引导不评价"大概就这样,记录一下"
F·引导搜索仅在有品牌词时使用自然提一句"感兴趣可以搜搜『{品牌词}",不催单
【正文四段结构(必须)】
段1·痛点引入描写使用前的困境/触发场景身份场景A·起因情绪B
段2·实测记录真实使用过程带上小缺点C真实感来源
段3·种草核心产品带来的变化用感受描述而非功效声称
段4·引导收尾从收尾策略池随机选一种佛系口吻不强推末尾带1-2个相关话题标签
【字数】正文350-400字不含标题tags3-5处段落空行增强可读性
""".strip()
# ── Q8 标题公式5类结构每批次覆盖不同类型不重复──────────────────────
_TITLE_FORMULA = """
【标题公式5类每条用不同类型禁止同批重复
肤质型:「{肤质}+用了{产品}+{感受}」例→"油皮用了素颜霜整个夏天不脱妆"
价格型:「{价格锚点}+{产品}+{功效感受}」例→"三位数买到大牌平替,用完第一罐回购第二罐"
功效型:「{使用场景}+{产品}+{可感知变化}」例→"早起素颜出门靠这罐,同事问我最近皮肤怎么了"
夸张型:「疑问/感叹+{夸张感受}+{产品}」例→"这什么神奇产品,涂上去感觉素颜也能出门了"
标题党型:「{反常识/意外信息}+真相是{翻转}」例→"被人说皮肤变好了,没说的是我用了它一个月"
硬性约束标题≤20字禁止出现"绝绝子/YYDS/杀疯了";不直接写功效词(美白/祛斑等)。
""".strip()
_ANTI_AI = f"""
【反AI味必须遵守】
- 禁止固定开篇套话:不许"姐妹们/宝子们/今天给大家分享"开头
- 禁止以下AI味词出现在正文{_NEGATIVE_WORDS}
- 禁止人群重叠:{'{count}'}条文案中身份场景不能重复靠变量池ABC保证
- 禁止场景重复:同批次文案不能都是"早上上班"或都是"学生宿舍"
- 避免生硬推销词:按头安利/绝绝子/闭眼冲 不能出现
""".strip()
# ── 系统 prompt合并Q2/Q3/Q4/Q7/Q8数据层走build_prompt动态注入──────────
COPY_SYSTEM = f"""{_PERSONA}
合规红线(全品类通用):
- 禁用"美白、祛斑、速效、医用、药妆",不得暗示治疗或改善疾病
- 不得承诺效果不得出现before/after变白对比暗示
- 图文合规避免社交App界面、点赞评论等截图元素
{_ANTI_AI.format(count="N")}
{_TITLE_FORMULA}
{_EMOJI_RULES}
{_STRUCTURE_RULES}
返回纯JSON数组每条字段
title标题≤20字/ content正文严格350-400字按emoji规则适度带表情/ tagslist3-5个#话题)
angle本条角度标签/ coverTitle封面大字≤10字/ imageBrief配图方向
硬性格式只输出JSON不要markdown代码块字符串内用中文引号「」。"""
def build_prompt(product: dict, count: int, extra_rules: str = "") -> str:
"""
组装文案生成 user_prompt。
数据层product 动态注入name/selling_points/style_tone/text_angles/custom_prompt
方法层:已在 COPY_SYSTEM 固定,这里只注入产品数据+随机变量
"""
name = product.get("name", "产品")
selling = "".join(product.get("selling_points") or ["核心卖点待录入"])
style = product.get("style_tone", "素人日常分享风")
angles = product.get("text_angles") or []
custom = (product.get("custom_prompt") or "").strip()
brand_kw = (product.get("brand_keyword") or "").strip()
# Q1每条抽一组随机变量传给模型作角色约束
combos = [_pick_combo() for _ in range(count)]
combos_text = "\n".join(
f"{i+1}条:身份场景=「{c['identity']}」·起因情绪=「{c['emotion']}」·小缺点=「{c['flaw']}"
for i, c in enumerate(combos)
)
angle_hint = f"文案角度要覆盖:{''.join(angles)}(每条用不同角度)。" if angles else ""
brand_rule = f"每条正文和标题中植入品牌词「{brand_kw}」一次(自然融入,不生硬)。" if brand_kw else ""
lines = [
f"产品:{name}",
f"核心卖点(必须翻译成用户能感知的生活化利益,禁止直接列功效词;翻译范例:'烟酰胺''熬夜后第二天脸不那么黄了''高保湿''涂上去一整天都没搓泥拔干'{selling}",
f"风格调性:{style}",
angle_hint,
brand_rule,
custom,
f"\n【Q1随机变量池·每条身份/起因/小缺点各不相同,严格按下方分配使用】",
combos_text,
extra_rules,
f"\n请严格按5步框架+四段结构生成 {count}每条350-400字返回纯JSON数组。",
]
return "\n".join(l for l in lines if l)
def parse_json_array(raw: str) -> list[dict]:
"""从模型输出提取 JSON 数组(容错 markdown 包裹)"""
text = re.sub(r"```(?:json)?", "", raw).strip()
start, end = text.find("["), text.rfind("]")
if start == -1 or end == -1:
return []
try:
return json.loads(text[start:end + 1])
except json.JSONDecodeError:
return []
def build_local_drafts(product: dict, count: int) -> list[dict]:
"""本地模板兜底(保证永不空手,遵循四段结构)
角度从 text_angles 循环取保证N条角度各不同不被 dedupe_copies 吞掉)
"""
name = product.get("name", "产品")
points = product.get("selling_points") or ["使用方便", "真实可感知"]
# 从产品档案取角度池,不够就用通用角度兜底,保证每条都不同
_fallback_angles = ["生活场景型", "成分分析型", "使用感受型", "性价比型", "痛点切入型"]
angles_pool = product.get("text_angles") or _fallback_angles
for i in range(count):
c = _pick_combo()
# 循环取不同角度(角度相同的两条会被 dedupe_copies 过滤掉,所以必须不重复)
angle = angles_pool[i % len(angles_pool)]
yield {
"title": f"发现一个{name}{angle}用法",
"content": (
f"{c['identity']}{c['emotion']}\n\n"
f"用了一段时间,{points[i % len(points)]}这点最让我意外。\n\n"
f"说个小缺点:{c['flaw']},后来才摸到感觉。\n\n"
f"反正现在用顺手了。✅"
),
"tags": [f"#{name}", "#真实测评", "#好物分享"],
"angle": angle,
"coverTitle": f"{name}{angle}",
"imageBrief": "封面产品近景,内页核心卖点+真实使用场景。",
}

View File

@@ -0,0 +1,125 @@
"""
违禁词三级处理(扒 copy.js sanitizePlanningText 扩展为三级)
🟢 auto_fix = 自动改写replacement 字段给出替换词)
🟡 soft_warn = 软提示(返回建议词,不阻塞)
🔴 hard_block= 硬拦截(直接返回 None拦住发布
词库来自数据库 banned_words 表level + replacement 字段),
DB 未配时用本模块内置默认词库作冷启动。
"""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from typing import Literal
BannedLevel = Literal["auto_fix", "soft_warn", "hard_block"]
@dataclass
class BannedWordEntry:
word: str
level: BannedLevel
replacement: str | None = None # auto_fix 时提供替换词
# ── 默认词库(北哥回填解读与落点 §4.3,数据库未配时使用)─
DEFAULT_BANNED_WORDS: list[BannedWordEntry] = [
# 功效违禁auto_fix改写成合规表达对应北哥"提亮肤色感/改善暗沉观感"
BannedWordEntry("美白", "auto_fix", "提亮肤色感"),
BannedWordEntry("祛斑", "auto_fix", "改善暗沉观感"),
# 功效违禁hard_block无法合规改写直接拦截
BannedWordEntry("速效", "hard_block"),
BannedWordEntry("医用", "hard_block"),
BannedWordEntry("药妆", "hard_block"),
BannedWordEntry("强效焕白", "hard_block"),
# 保证性词soft_warn
BannedWordEntry("绝对", "soft_warn"),
BannedWordEntry("第一名", "soft_warn"),
BannedWordEntry("再也不", "soft_warn"),
# 夸张词soft_warn
BannedWordEntry("杀疯了", "soft_warn"),
BannedWordEntry("秒杀", "soft_warn"),
BannedWordEntry("震撼", "soft_warn"),
# AI 味词auto_fix置换为口语表达同时在 _NEGATIVE_WORDS prompt负向约束里已禁止AI写进正文
BannedWordEntry("神器", "auto_fix", "好用的"),
BannedWordEntry("福音", "auto_fix", "适合的"),
BannedWordEntry("救急单品", "auto_fix", "随手备用的"),
BannedWordEntry("遮羞布", "auto_fix", "底妆感"), # 北哥原文补录
BannedWordEntry("不仅而且", "auto_fix", ",另外"),
BannedWordEntry("焕发", "auto_fix", "呈现"),
BannedWordEntry("守护", "auto_fix", ""),
BannedWordEntry("尽享", "auto_fix", "使用"),
BannedWordEntry("日常维稳", "auto_fix", "日常保养"),
BannedWordEntry("精简底妆", "auto_fix", "轻便底妆"),
# 视觉违禁hard_block文案含这些词不许过
BannedWordEntry("前后对比", "hard_block"),
BannedWordEntry("使用前后", "hard_block"),
BannedWordEntry("变白", "auto_fix", "自然光泽感"),
BannedWordEntry("瑕疵消失", "auto_fix", "妆感更服帖"),
]
@dataclass
class CheckResult:
text: str # 原文soft_warn/hard_block 场景下保持原文)
fixed_text: str | None # auto_fix 后的文本;其他级别为 None
status: Literal["pass", "auto_fixed", "soft_warn", "hard_block"]
found: list[dict] = field(default_factory=list)
# found 每项: {"word": str, "level": BannedLevel, "replacement": str|None}
def check_and_fix(
text: str,
entries: list[BannedWordEntry] | None = None,
) -> CheckResult:
"""
对一段文本做三级违禁词扫描。
entries优先用 DB 词条,为 None 时用默认词库。
"""
word_list = entries if entries is not None else DEFAULT_BANNED_WORDS
found: list[dict] = []
working = text
# 先扫描所有命中
for entry in word_list:
if entry.word.lower() in working.lower():
found.append({
"word": entry.word,
"level": entry.level,
"replacement": entry.replacement,
})
if not found:
return CheckResult(text=text, fixed_text=None, status="pass", found=[])
# 有 hard_block → 直接拦截
if any(f["level"] == "hard_block" for f in found):
return CheckResult(text=text, fixed_text=None, status="hard_block", found=found)
# 只有 soft_warn → 软提示,不改文字
if any(f["level"] == "soft_warn" for f in found) and \
all(f["level"] in ("soft_warn", "auto_fix") for f in found):
# 仍执行 auto_fix 改写,但结果状态是 soft_warn优先级高
for f in found:
if f["level"] == "auto_fix" and f["replacement"] is not None:
working = re.sub(re.escape(f["word"]), f["replacement"], working, flags=re.IGNORECASE)
return CheckResult(text=text, fixed_text=working, status="soft_warn", found=found)
# 只有 auto_fix → 自动改写,返回 fixed_text
for f in found:
if f["level"] == "auto_fix" and f["replacement"] is not None:
working = re.sub(re.escape(f["word"]), f["replacement"], working, flags=re.IGNORECASE)
return CheckResult(text=text, fixed_text=working, status="auto_fixed", found=found)
def build_entries_from_db(rows: list[dict]) -> list[BannedWordEntry]:
"""把 DB banned_words 行转成 BannedWordEntry 列表"""
return [
BannedWordEntry(
word=r["word"],
level=r["level"],
replacement=r.get("replacement"),
)
for r in rows
if r.get("word") and r.get("level") in ("auto_fix", "soft_warn", "hard_block")
]

View File

@@ -0,0 +1,139 @@
"""
AI 引擎中心常量
扒自worker/src/copy.js + image.js 上线版
业务参数不写死基石A——分值权重可由产品档案配置覆盖
"""
# ── 合规红线 ──────────────────────────────────────────────
# 初始默认词库(数据库 banned_words 表可覆盖updatable=True
BANNED_WORDS_DEFAULT = ["美白", "祛斑", "速效", "医用", "药妆"]
BANNED_VISUAL_WORDS = [
"前后对比", "使用前后", "用前用后",
"before", "after", "变白", "瑕疵消失", "治疗前后",
]
# 内部提示词(不能混入正文 content 字段)
INTERNAL_COPY_HINTS = [
"配图建议", "图片方向", "内页规划", "适合做成",
"不要做促销海报", "配图说明", "封面建议",
]
# ── 机械五维打分基线仅轨B导入文案/降级回退用轨A已切 AI 评委 llm_score_copy─────
# 历史注轨A原用此5维2026-06-15切至7维AI评委(6维+合规)后此处只作轨B+回退占位。
# 关键词维度(keyword20)因 products 表无 keywords 字段导致 matched 恒空已知不准;
# 轨A不再依赖此权重轨B展示参考可接受。不按此调分真过关靠北哥抽检。
SCORE_WEIGHTS = {
"title": 25,
"emotion": 25,
"selling": 25,
"keyword": 20,
"compliance": 5,
}
# ── AI 评委 7 维满分分布倩倩姐2026-06-15拍板·与 llm_scorer._DIM_MAX/_score_prompt 三处同步)──
# 6维AI读分(痛点18+情绪18+买点18+钩子15+标题13+真实感13=95) + 合规5 = 100
# "真实感"=富贵"很少提产品/前70%干货后30%植入"原则,替换旧机械维度"产品聚焦一件事(16)"
AI_DIM_WEIGHTS = {
"痛点人群精准": 18,
"情绪张力": 18,
"买点转化": 18,
"开头钩子": 15,
"标题点击力": 13,
"真实感": 13,
"compliance": 5, # 机械硬拦,不进 AI 评委
}
# 过线分。倩倩姐2026-06-15拍板80是临时观察值(AI评委给分克制84文案实为合格)。
# 倩倩姐2026-06-15再次拍板维持80临时线不准擅自调85。方向=提生成质量顶分数,不降标准。
# 真过关靠北哥抽检;提质量方向=优化生成 prompt不靠提高门槛凑数。
QUALITY_PASS_SCORE = 80
# ── 文案去重阈值 ──────────────────────────────────────────
DEDUP_TITLE_THRESHOLD = 0.82 # 标题相似度≥此值判重
DEDUP_TITLE_CONTENT_TITLE = 0.65 # 标题+正文联合判重时的标题阈值
DEDUP_TITLE_CONTENT_BODY = 0.72 # 标题+正文联合判重时的正文阈值
# ── 自动优化循环 ──────────────────────────────────────────
MAX_OPTIMIZE_ROUNDS = 2 # 最多重生成轮次
# ── storyboard 分镜角色(枚举不写死数量)────────────────
# Q6: 北哥6张套路顺序 ①封面痛点大字 ②单品特写+品牌词 ③成分 ④质地 ⑤上脸对比 ⑥促单
PAGE_ROLES = [
{"role": "hook", "name": "封面痛点大字", "focus": "负责点击:强情绪大字标题压痛点,产品露出,真实生活场景,像用户主动分享,不像广告海报"},
{"role": "product_closeup", "name": "单品特写", "focus": "负责种草锚点:单品高清特写+品牌词自然植入第2/6张都带品牌词强化记忆"},
{"role": "ingredient", "name": "成分拆解", "focus": "负责信任:核心成分信息、作用说明,避免医疗化和绝对化表达,信息清晰可信"},
{"role": "texture", "name": "质地展示", "focus": "负责种草:质地近景、涂抹过程、肤感说明,真实手部/桌面/日常光线"},
{"role": "applied_proof", "name": "上脸对比", "focus": "负责证明:可感知上脸效果,展示涂抹前后质地变化(不做肤色变白/瑕疵消失等违规暗示第5张"},
{"role": "closer", "name": "促单收尾", "focus": "负责转化:转化句+品牌词引导搜索品牌词成交软性收尾不硬广第6张再带一次品牌词"},
# 扩展角色8张链路用
{"role": "pain_scene", "name": "痛点共鸣", "focus": "负责共鸣:展示目标人群的真实困扰和使用前情境,但不做功效前后对比"},
{"role": "social_proof","name": "信任背书", "focus": "负责背书:多人反馈、囤货、复购等真实社交证据"},
{"role": "scenario", "name": "多场景演示", "focus": "负责代入:多场景使用展示,不做夸大效果承诺"},
{"role": "tutorial", "name": "使用教程", "focus": "负责降低门槛:简洁步骤、用量、注意事项"},
]
PAGE_ROLE_MAP = {r["role"]: r for r in PAGE_ROLES}
# ── 生图风格预设(扒 image.js STYLE_PROMPTS:26-29──────────
# 按 style 参数选小红书风格调性,注入 base_prompt 的"视觉风格"行
STYLE_PROMPTS = {
"xiaohongshu_cover": "小红书种草风独立3:4图文海报/素材图1024×1536构图明亮干净真实实拍质感醒目中文短标题文字在安全区内",
"comparison": "小红书说明对比风独立3:4图文海报/素材图1024×1536构图质地/场景/肤感左右或上下对比,信息层级清晰",
"ingredient": "小红书成分科普风独立3:4图文海报/素材图1024×1536构图成分卡片布局浅色商务美妆风避免医疗化表达",
}
STYLE_DEFAULT = "xiaohongshu_cover"
# ── 叙事链路说明(扒 image.js planImageSet narrativeText:677-681──
# 按图数告诉模型整组图的种草节奏,让每张各司其职不雷同
NARRATIVE_BY_COUNT = {
3: "3张极速链路第1张负责点击第2张是按品类变化的核心证明页第3张负责软性转化。",
6: "6张标准种草链路封面点击、单品特写带品牌词、成分信任、质地种草、上脸证明、促单转化每张画面和文字各司其职不重复。",
8: "8张沉浸测评链路点击、痛点共鸣、单品特写、成分、质地、上脸证明、背书、软性转化。",
}
# ── 3套正交叙事策略倩倩姐2026-06-15起草北哥过目版──────────────
# A痛点先行/B场景先行/C成分背书先行三套正交轴拉开差异
# 每套叙事链路注入 base_prompt 叙事链路段,替换 NARRATIVE_BY_COUNT 默认值
NARRATIVE_BY_STRATEGY = {
"A": (
'【套A·痛点先行】整组基调紧迫感、强对比、情绪共鸣文字短促带感叹号直戳"脸黄显疲惫""素颜不敢出门"'
'6张链路①痛点暴击封面强情绪大字直击暗黄/素颜焦虑)→ ②暗黄脸实拍对比(感叹号+对比词制造紧迫感)'
'→ ③单品特写+品牌词 → ④成分为什么能救暗黄(成分拆解+信任) → ⑤上脸提亮实证 → ⑥"别再顶着黄脸早八"软性促单。'
),
"B": (
'【套B·场景先行】整组基调轻松、生活化、代入感突出"快/省时/伪素颜自由",点到性价比不堆砌。'
'6张链路"早八来不及"场景封面(生活场景钩子) → ②手忙脚乱通勤场景(代入早八焦虑)'
'→ ③一抹搞定单品特写+品牌词 → ④养肤成分让你敢素颜 → ⑤30秒上脸效果 → ⑥"伪素颜自由+平价"软性促单。'
),
"C": (
'【套C·成分背书先行】整组基调专业、可信、真实测评感强调成分逻辑+前后对比+像有用户实证背书。'
'6张链路①成分权威封面核心成分信息锚定信任 → ②核心成分图解(作用说明+清晰可信)'
'→ ③单品特写+品牌词 → ④使用前后时间线对比 → ⑤真实上脸细节 → ⑥"成分党闭眼入"软性促单。'
),
}
# ── 生图通道 ──────────────────────────────────────────────
IMAGE_RETRY_ATTEMPTS = 3
IMAGE_RETRY_BACKOFF_BASE = 2.0 # 指数退避底数(秒)
IMAGE_SIZE_DEFAULT = "1024x1536"
# ── 生图合规负向约束(方法层常量,全品类共用,可扩展)───────────────────────
# 追加到每个 base_prompt 末尾,防模型脑补违禁词/真实品牌到包装
IMAGE_NEGATIVE_CONSTRAINTS = (
"【包装合规硬性禁止——必须严格遵守】"
"①包装/瓶身/标签上禁止出现任何违禁词:美白/whitening/祛斑/brightening/"
"医用/medical/drug/药妆/速效/instant中英文全禁"
"②禁止脑补任何真实品牌名或logo如水密码/WETCODE/兰蔻/SK-II等"
"产品包装只允许出现用户传入的指定品牌词,未传则画无字素瓶;"
"③英文功效词(ANTI-AGING/TONE-UP/BRIGHTENING/FIRMING等)禁止印在包装;"
"④如果提供了产品参考图,包装文字以参考图为准,不得自行添加或修改任何文字。"
"⑤背景纯净:禁止出现电子设备/笔记本电脑/键盘/手机/桌面杂物等无关物体(参考图若含此类背景一律不沿用),"
"只保留浅色简洁台面或产品定制场景,主体聚焦产品本身。"
)
# ── 飞轮信号权重(初始默认,北哥可校准)────────────────
FLYWHEEL_WEIGHTS = {
"text_select": 3,
"image_select": 3,
"approve": 5,
"reject_with_reason": -3,
"regenerate": -1,
}
FLYWHEEL_LOOKBACK = 50 # 聚合最近N条事件
FLYWHEEL_COLD_START = 5 # 信号不足N条时用产品档案冷启动

View File

@@ -0,0 +1,225 @@
"""
gemini_factory.py — 每任务构建独立的 AI client 实例
解决全局单例问题(扒 banana gemini_service.py __init__改造为每任务局部实例
铁律基石B
- 调用方只传 task_id不传 key
- 本模块在 worker 内部查库 → Fernet 解密 → 构建 client
- 解密结果只活在局部变量,函数返回后即销毁
- 绝不打印 / 记录 / 传递明文 key
"""
from __future__ import annotations
import asyncio
import logging
import os
from dataclasses import dataclass, field
from typing import Any
import httpx
logger = logging.getLogger(__name__)
@dataclass
class AIClients:
"""
一个任务专用的 AI client 集合。
worker 在任务开始时构建,任务结束后释放(局部变量,不存 Redis/DB
"""
# httpx AsyncClient 懒加载 + 按事件循环缓存Celery 每任务多次 asyncio.run
# 持久 client 会绑死到首个已关闭的 loop → "Event loop is closed"。
# 故只存 token/base按当前运行 loop 缓存 clientloop 变了就重建。
_gpt_token: str | None = field(default=None, repr=False)
_gpt_base: str | None = field(default=None, repr=False)
_gpt_client: httpx.AsyncClient | None = field(default=None, repr=False)
_gpt_client_loop_id: int | None = field(default=None, repr=False)
# 备用 OpenAI 兼容中转站codeproxyapiports 503 时真正切过去(独立 base+key
_alt_token: str | None = field(default=None, repr=False)
_alt_base: str | None = field(default=None, repr=False)
# 多 base/token 的 client 池key=(base,token的id),按 loop 失效重建
_client_pool: dict = field(default_factory=dict, repr=False)
_pool_loop_id: int | None = field(default=None, repr=False)
_gemini_key: str | None = field(default=None, repr=False) # 局部变量不打印
_model_image: str = "gpt-image-2"
_model_text: str = "claude-sonnet-4-5" # apiports无gpt-4o-mini,文案用claude中文质量好
def _client(self) -> httpx.AsyncClient:
"""主通道(apiports) client按当前事件循环缓存"""
return self._client_for(self._gpt_base, self._gpt_token)
def _client_for(self, base: str | None, token: str | None) -> httpx.AsyncClient:
"""按 (base, token) 返回可用 clientloop 变化则整池重建(避免跨 loop 复用)"""
if not token:
raise RuntimeError("GPT client 未初始化(缺 token")
loop_id = id(asyncio.get_running_loop())
if self._pool_loop_id != loop_id:
self._client_pool = {}
self._pool_loop_id = loop_id
ck = (base or "", token)
if ck not in self._client_pool:
self._client_pool[ck] = httpx.AsyncClient(
headers={"Authorization": f"Bearer {token}"},
base_url=base or None,
timeout=120.0,
)
return self._client_pool[ck]
# ── ImageClient 协议实现(供 image_gen.py 使用)────────
def _gpt_target(self, provider: str | None) -> tuple[str, str | None, httpx.AsyncClient]:
"""按 provider 选 (base, token, client)codeproxy 走备用站独立 base+key"""
if provider == "codeproxy" and self._alt_token:
base = (self._alt_base or os.environ.get("CODEPROXY_BASE_URL") or "").rstrip("/")
return base, self._alt_token, self._client_for(base, self._alt_token)
base = (os.environ.get("IMAGE_API_BASE") or os.environ.get("APIPORTS_BASE_URL") or "").rstrip("/")
return base, self._gpt_token, self._client_for(self._gpt_base, self._gpt_token)
async def gpt_edits(self, prompt: str, reference_images: list[bytes], size: str, provider: str | None = None) -> bytes:
"""GPT edits endpoint带产品参考图禁纯文生图"""
import io
files: list[tuple] = [("prompt", (None, prompt))]
for i, img in enumerate(reference_images):
files.append(("image[]", (f"ref_{i}.png", io.BytesIO(img), "image/png")))
files.append(("size", (None, size)))
files.append(("model", (None, self._model_image)))
base, _, client = self._gpt_target(provider)
resp = await client.post(f"{base}/images/edits", files=files, timeout=120.0)
resp.raise_for_status()
return _extract_image_bytes(resp.json())
async def gpt_generate(self, prompt: str, size: str, provider: str | None = None) -> bytes:
"""GPT 纯文生图(仅 ALLOW_TEXT_ONLY_IMAGE=true 时用)"""
base, _, client = self._gpt_target(provider)
payload = {"model": self._model_image, "prompt": prompt, "n": 1, "size": size}
resp = await client.post(f"{base}/images/generations", json=payload, timeout=120.0)
resp.raise_for_status()
return _extract_image_bytes(resp.json())
async def gemini_generate(self, prompt: str, reference_images: list[bytes], model: str) -> bytes:
"""Gemini 生图(备用通道)"""
if not self._gemini_key:
raise RuntimeError("Gemini key 未初始化")
gemini_base = os.environ.get("GEMINI_API_URL", "https://generativelanguage.googleapis.com/v1beta")
url = f"{gemini_base}/models/{model}:generateContent?key={self._gemini_key}"
parts: list[dict] = [{"text": prompt}]
for img in reference_images:
import base64
parts.append({"inline_data": {"mime_type": "image/png", "data": base64.b64encode(img).decode()}})
payload = {"contents": [{"role": "user", "parts": parts}], "generationConfig": {"responseModalities": ["IMAGE", "TEXT"]}}
async with httpx.AsyncClient() as client:
resp = await client.post(url, json=payload, timeout=120.0)
resp.raise_for_status()
return _extract_gemini_image(resp.json())
async def chat_complete(self, messages: list[dict], model: str | None = None, max_tokens: int = 4096, temperature: float = 0.75) -> str:
"""文字生成(文案生成用)"""
base = (os.environ.get("IMAGE_API_BASE") or os.environ.get("APIPORTS_BASE_URL") or "").rstrip("/")
payload = {"model": model or self._model_text, "messages": messages, "max_tokens": max_tokens, "temperature": temperature}
# 单批≤4条文案正常 40-55s 返回apiports 网关 ~60s 上限。客户端超时设 75s
# 略高于网关上限即可过长如180s会在 apiports 卡顿时干等,拖慢整体。
timeout = float(os.environ.get("TEXT_LLM_TIMEOUT", "75"))
resp = await self._client().post(f"{base}/chat/completions", json=payload, timeout=timeout)
resp.raise_for_status()
return resp.json()["choices"][0]["message"]["content"] or ""
async def gpt_vision_analyze(self, prompt: str, images: list[bytes], model: str | None = None) -> str:
"""
GPT/Claude vision 读产品图,返回 JSON 字符串。
messages content 混合 text + image_url(base64)OpenAI vision 格式。
model 默认最强档claude-opus-4-8绝不偷降级。
最多传 4 张图,避免超 token。
"""
import base64
content: list[dict] = [{"type": "text", "text": prompt}]
for img in images[:4]:
b64 = base64.b64encode(img).decode()
content.append({
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{b64}"},
})
used_model = model or os.environ.get("MODEL_TEXT", "claude-opus-4-8")
messages = [{"role": "user", "content": content}]
base = (os.environ.get("IMAGE_API_BASE") or os.environ.get("APIPORTS_BASE_URL") or "").rstrip("/")
payload = {
"model": used_model,
"messages": messages,
"max_tokens": 2048,
"temperature": 0.2,
}
resp = await self._client().post(f"{base}/chat/completions", json=payload, timeout=90.0)
resp.raise_for_status()
return resp.json()["choices"][0]["message"]["content"] or ""
# duck-type: text_variants._call_llm 用的属性
@property
def _model(self) -> str:
return self._model_text
async def aclose(self) -> None:
# client 可能绑在已关闭的 loopCelery 多次 asyncio.runaclose 也可能报
# "Event loop is closed",吞掉即可——进程级连接随 loop 关闭自然释放。
for c in list(self._client_pool.values()):
try:
await c.aclose()
except Exception:
pass
self._client_pool = {}
self._pool_loop_id = None
self._gpt_client = None
self._gpt_client_loop_id = None
def build_ai_clients(plain_key: str, gemini_key: str | None = None) -> AIClients:
"""
用解密后的明文 key 构建 AIClients。
只在 Celery worker 函数体内调用plain_key 是局部变量。
httpx client 不在此预创建(避免绑死到调用方 loop首次 await 时按 loop 懒建。
调用完成后 caller 负责 await clients.aclose()。
"""
gpt_base = (
os.environ.get("IMAGE_API_BASE") # 旧变量名
or os.environ.get("APIPORTS_BASE_URL") # .env 实际变量名
or ""
).rstrip("/")
# 备用站 codeproxy系统级 key非用户录入apiports 503 时切过去保生图成功
alt_base = (os.environ.get("CODEPROXY_BASE_URL") or "").rstrip("/")
alt_token = os.environ.get("CODEPROXY_KEY") or None
return AIClients(
_gpt_token=plain_key,
_gpt_base=gpt_base or None,
_alt_base=alt_base or None,
_alt_token=alt_token,
_gemini_key=gemini_key,
_model_image=os.environ.get("IMAGE_MODEL") or os.environ.get("MODEL_IMAGE", "gpt-image-2"),
_model_text=os.environ.get("MODEL_TEXT", "claude-opus-4-8"),
)
# ── 图片响应解析工具 ─────────────────────────────────────
def _extract_image_bytes(resp_json: dict) -> bytes:
"""从 OpenAI images API 响应提取图片 bytesb64 或 url"""
import base64
data = resp_json.get("data", [{}])
if not data:
raise ValueError("图片 API 返回空 data")
item = data[0]
if "b64_json" in item:
return base64.b64decode(item["b64_json"])
if "url" in item:
resp = httpx.get(item["url"], timeout=30.0)
resp.raise_for_status()
return resp.content
raise ValueError(f"无法解析图片响应:{list(item.keys())}")
def _extract_gemini_image(resp_json: dict) -> bytes:
"""从 Gemini generateContent 响应提取图片 bytes"""
import base64
candidates = resp_json.get("candidates", [])
for cand in candidates:
parts = cand.get("content", {}).get("parts", [])
for part in parts:
if "inlineData" in part:
return base64.b64decode(part["inlineData"]["data"])
raise ValueError("Gemini 响应中未找到图片数据")

View File

@@ -0,0 +1,175 @@
"""
生图通道 — gpt-image-2 主edits 带产品图) / Gemini 备 + 重试退避
扒自worker/src/image.js generateOneImage / requestProviderImage / imageProviderOrder
新增asyncio 重试退避上线版缺的banana 有 _retry 思路)
铁律:
- IMAGE_PROVIDER_PRIMARY/FALLBACK 走环境变量,不写死
- GPT 主通道必须有产品参考图,无图报错(禁纯文生图防产品跑偏)
- key 不在本模块,由 worker 传入构造好的 async HTTP client
"""
from __future__ import annotations
import asyncio
import logging
import os
from typing import Any, Protocol
from .constants import IMAGE_RETRY_ATTEMPTS, IMAGE_RETRY_BACKOFF_BASE, IMAGE_SIZE_DEFAULT
from .storyboard import plan_image_set, sanitize_text
logger = logging.getLogger(__name__)
class ImageClient(Protocol):
"""worker 注入的图片生成客户端协议(隔离 key 细节)"""
async def gpt_edits(
self, prompt: str, reference_images: list[bytes], size: str, provider: str | None = None
) -> bytes: ...
async def gpt_generate(self, prompt: str, size: str, provider: str | None = None) -> bytes: ...
async def gemini_generate(
self, prompt: str, reference_images: list[bytes], model: str
) -> bytes: ...
def _image_provider_order() -> list[str]:
"""从环境变量读主备顺序(扒 imageProviderOrder"""
primary = os.environ.get("IMAGE_PROVIDER_PRIMARY", "gpt").lower()
fallback = os.environ.get("IMAGE_PROVIDER_FALLBACK", "gemini").lower()
seen: list[str] = []
for p in [primary, fallback]:
if p and p not in seen:
seen.append(p)
return seen
def _gemini_models() -> list[str]:
"""Gemini fallback 模型列表(多模型依次重试)"""
env_val = os.environ.get("GEMINI_IMAGE_MODELS", "gemini-2.0-flash-preview-image-generation,imagen-3.0-generate-002")
return [m.strip() for m in env_val.split(",") if m.strip()]
async def _retry(coro_fn, attempts: int = IMAGE_RETRY_ATTEMPTS, backoff: float = IMAGE_RETRY_BACKOFF_BASE) -> Any:
"""指数退避重试(扒 banana _retry 思路)"""
last_exc: Exception | None = None
for i in range(attempts):
try:
return await coro_fn()
except Exception as exc:
last_exc = exc
if i < attempts - 1:
wait = backoff ** i
logger.warning("生图失败第%d次,%.1fs后重试%s", i + 1, wait, exc)
await asyncio.sleep(wait)
raise RuntimeError(f"重试{attempts}次均失败") from last_exc
async def _request_gpt(client: ImageClient, prompt: str, reference_images: list[bytes], provider: str | None = None) -> bytes:
if reference_images:
return await client.gpt_edits(prompt, reference_images, IMAGE_SIZE_DEFAULT, provider)
# 无产品参考图时降级为纯文生图(需 ALLOW_TEXT_ONLY_IMAGE=true 或 M2阶段
allow_text_only = os.environ.get("ALLOW_TEXT_ONLY_IMAGE", "true").lower() == "true"
if allow_text_only:
logger.warning("无产品参考图,降级为纯文生图(可能产品跑偏,建议前端上传参考图)")
return await client.gpt_generate(prompt, IMAGE_SIZE_DEFAULT, provider)
raise ValueError("GPT 主通道缺产品图:禁止纯文生图以免产品跑偏(设 ALLOW_TEXT_ONLY_IMAGE=true 可解锁)")
async def _request_gemini(client: ImageClient, prompt: str, reference_images: list[bytes]) -> bytes:
errors: list[str] = []
for model in _gemini_models():
try:
return await client.gemini_generate(prompt, reference_images, model)
except Exception as exc:
errors.append(f"{model}: {exc}")
raise RuntimeError("Gemini 全部模型失败:" + "".join(errors))
async def generate_one_image(
client: ImageClient,
prompt: str,
reference_images: list[bytes] | None = None,
) -> bytes:
"""
主入口:按主备顺序依次尝试,每个 provider 内部有重试退避。
返回图片 bytesPNG/JPEG
"""
refs = reference_images or []
providers = _image_provider_order()
errors: list[str] = []
for provider in providers:
try:
# apiports/codeproxy/openai 都是 OpenAI 兼容中转站,走 gpt 协议,
# 但传 provider 进去 → client 按 provider 切到对应中转站的 base+key。
# 这才是真主备apiports 503 → codeproxy 用独立 base+key 顶上。
if provider in ("apiports", "codeproxy", "openai"):
img = await _retry(lambda p=provider: _request_gpt(client, prompt, refs, p))
elif provider == "gpt":
img = await _retry(lambda: _request_gpt(client, prompt, refs, None))
elif provider == "gemini":
img = await _retry(lambda: _request_gemini(client, prompt, refs))
else:
raise ValueError(f"未知图片通道:{provider}")
return img
except Exception as exc:
errors.append(f"{provider}: {exc}")
logger.warning("图片通道 %s 失败,尝试下一个:%s", provider, exc)
raise RuntimeError("所有图片通道均失败:" + "".join(errors))
async def generate_storyboard_images(
client: ImageClient,
note: dict,
product: dict,
image_count: int = 3,
reference_images: list[bytes] | None = None,
analysis: dict | None = None,
strategy: str | None = None,
) -> list[dict]:
"""
按 storyboard 逐张生图asyncio.gather 并发),返回每张结果列表。
strategy: None=默认叙事,'A'/'B'/'C'=三套正交叙事策略
每项:{role, name, image_bytes, error}
"""
plan = plan_image_set(note, product, image_count, analysis, strategy=strategy)
storyboard = plan["storyboard"]
base_prompt = plan["base_prompt"]
async def _gen_one(item: dict) -> dict:
# 逐图 prompt 9 字段(扒 promptFromStoryboard:323-334每张差异化
brand_line = ""
if item.get("brand_keyword"):
brand_line = f"品牌词=「{item['brand_keyword']}」,{item.get('brand_keyword_rule','自然植入')}"
per_prompt = (
f"{base_prompt}\n"
f"本张名称={item['name']}"
f"本张目标={item.get('goal') or item['focus']}"
f"图上主文字=「{sanitize_text(item.get('overlay_text',''), 20)}」。"
f"使用卖点={item.get('selling_point','')}"
f"文案依据={item.get('source_basis','')}"
f"画面主体={item.get('visual_strategy','')}"
f"素材使用={item.get('asset_use','')}"
f"{brand_line}"
f"禁止事项={item.get('forbidden','')}"
"排版要求独立小红书3:4图文海报画面完整标题只出现一次不与其他页重复"
"中文文字少而清晰,主标题+最多3个短点位可自然用✅✨🌿💧🪞🧴📦🔍种草符号但不堆砌"
"不要生成App截图或笔记详情页界面。"
)
try:
img_bytes = await generate_one_image(client, per_prompt, reference_images)
return {"role": item["role"], "name": item["name"], "image_bytes": img_bytes, "error": None}
except Exception as exc:
logger.error("分镜 %s 生图失败: %s", item["role"], exc)
return {"role": item["role"], "name": item["name"], "image_bytes": None, "error": str(exc)}
# 限并发apiports 图片接口有 QPS 限制6 张全并发会撞 429/503
concurrency = int(os.environ.get("IMAGE_CONCURRENCY", "2"))
sem = asyncio.Semaphore(max(1, concurrency))
async def _gen_guarded(item: dict) -> dict:
async with sem:
return await _gen_one(item)
results = await asyncio.gather(*(_gen_guarded(item) for item in storyboard))
return list(results)

View File

@@ -0,0 +1,177 @@
"""
图片后处理去AI化主路
对齐大卫 xhs-tool/backend/infrastructure/imagePostProcess.js运营实测去AI化版
主路 = 尺寸可选(±2%容差内不resize) + SynthID破除(可选) + 高保真重编码去元数据。
诚实声明C2PA 元数据可去除;私有像素水印(如 SynthID只能削弱不保证 100% 清除。
"""
from __future__ import annotations
import io
import logging
import os
logger = logging.getLogger(__name__)
try:
from PIL import Image, ImageEnhance, ImageOps
_PILLOW_OK = True
except ImportError:
_PILLOW_OK = False
logger.warning("Pillow 未安装image_postprocessor 不可用")
# 比例映射表,对齐大卫 RATIO_MAP。key 为字符串如 '3:4'
RATIO_MAP: dict[str, tuple[int, int]] = {
"1:1": (1024, 1024),
"3:4": (1024, 1536), # gpt-image-2 原生尺寸,默认
"4:3": (1536, 1024),
"9:16": (864, 1536),
"16:9": (1536, 864),
}
# ±2% 容差内不做 resize避免无谓重采样对齐大卫 diff > 0.02 才 resize
_RATIO_TOLERANCE = 0.02
def _need_resize(actual_w: int, actual_h: int, target_w: int, target_h: int) -> bool:
"""判断实际比例与目标比例差距是否超出容差。"""
actual_ratio = actual_w / actual_h
target_ratio = target_w / target_h
diff = abs(actual_ratio - target_ratio) / target_ratio
return diff > _RATIO_TOLERANCE
def process_image(
image_bytes: bytes,
aspect_ratio: str = "3:4",
resample_strength: int = 1, # 0=不重采样, 1=轻采样(默认), 2=重采样
) -> bytes:
"""
处理单张图片。
参数:
image_bytes — 原始图片 bytesPNG/JPEG/WebP 等)
aspect_ratio — 目标比例,取 RATIO_MAP 的 key默认 '3:4'=1024×1536
resample_strength — 轻重采样削像素水印0/1/2默认 1=轻采样
返回 JPEG bytes无 EXIF/C2PA/XMP 元数据)。
失败时降级返回原图 bytes不抛异常对齐大卫 catch 返回原图)。
"""
if not _PILLOW_OK:
logger.error("Pillow 未安装,跳过后处理,返回原图")
return image_bytes
try:
img = Image.open(io.BytesIO(image_bytes)).convert("RGB")
actual_w, actual_h = img.size
target = RATIO_MAP.get(aspect_ratio)
# --- Step1: 尺寸对齐±2% 容差内跳过 resize---
if target:
tw, th = target
if _need_resize(actual_w, actual_h, tw, th):
img = ImageOps.fit(img, (tw, th), method=Image.LANCZOS)
logger.debug("resize %dx%d%dx%d (ratio=%s)", actual_w, actual_h, tw, th, aspect_ratio)
# --- Step2: resample_strength 削像素水印(可选,默认轻采样)---
img = _apply_resample(img, resample_strength)
# --- Step3: SynthID 破除SYNTHID_HARD_MODE=1 才开,默认关)---
if os.environ.get("SYNTHID_HARD_MODE") == "1" and target:
img = _apply_synthid_break(img, target)
# --- Step4: 高保真 JPEG 重编码,去所有元数据 ---
buf = io.BytesIO()
img.save(
buf,
format="JPEG",
quality=100,
subsampling=0, # 4:4:4 chroma
optimize=True,
# 不传 exif/icc_profile/xmp = 不写入任何元数据
)
result = buf.getvalue()
logger.debug("后处理完成 %d B → %d B (ratio=%s)", len(image_bytes), len(result), aspect_ratio)
return result
except Exception as exc:
logger.warning("图片后处理失败,降级返回原图: %s", exc)
return image_bytes
def _apply_resample(img: "Image.Image", strength: int) -> "Image.Image":
"""
轻/重采样削像素级水印resample_strength 控制)。
0 — 不采样,仅靠重编码去元数据。
1 — 轻采样缩98%再回原尺寸,保视觉质量,削弱像素水印(对齐旧逻辑)。
2 — 重采样:两次缩放,削弱更多,轻微质量损失。
"""
if strength < 1:
return img
w, h = img.size
img = img.resize((int(w * 0.98), int(h * 0.98)), Image.LANCZOS)
img = img.resize((w, h), Image.LANCZOS)
if strength >= 2:
img = img.resize((int(w * 0.96), int(h * 0.96)), Image.LANCZOS)
img = img.resize((w, h), Image.LANCZOS)
return img
def _apply_synthid_break(img: "Image.Image", target: tuple[int, int]) -> "Image.Image":
"""
SynthID 破除SYNTHID_HARD_MODE=1 时调用):
对齐大卫逻辑 — 缩到(w-2,h-2)再裁掉1px边 + 亮度*1.005/饱和*0.998。
诚实声明:只能削弱 SynthID不保证 100% 清除。
"""
tw, th = target
img = ImageOps.fit(img, (tw - 2, th - 2), method=Image.LANCZOS)
# 裁掉1px边消除边缘水印残留
img = img.crop((1, 1, tw - 3, th - 3))
# 微调亮度/饱和(对齐大卫 modulate brightness/saturation
img = ImageEnhance.Brightness(img).enhance(1.005)
img = ImageEnhance.Color(img).enhance(0.998)
return img
def batch_process(
images: list[bytes],
aspect_ratio: str = "3:4",
resample_strength: int = 1,
) -> list[dict]:
"""
批量后处理。返回 [{index, data, error}],单张失败不阻塞其余。
"""
results = []
for i, img_bytes in enumerate(images):
try:
processed = process_image(img_bytes, aspect_ratio=aspect_ratio,
resample_strength=resample_strength)
results.append({"index": i, "data": processed, "error": None})
except Exception as exc:
logger.error("图片[%d]后处理失败: %s", i, exc)
results.append({"index": i, "data": img_bytes, "error": str(exc)})
return results
async def gemini_rewatermark_fallback(
client: "Any", # GeminiClient由 worker 注入
image_bytes: bytes,
) -> bytes:
"""
备选路Gemini 重绘去水印。
⚠️ 对海报中文大字有改字风险,仅特殊场景启用。
"""
prompt = (
"Remove all watermarks, text overlays, and digital signatures from this image. "
"Reconstruct any covered areas naturally to match the surrounding content. "
"Return a clean version of the same image without any watermarks."
)
try:
result = await client.gemini_generate(
prompt, [image_bytes], "gemini-2.0-flash-preview-image-generation"
)
return result
except Exception as exc:
logger.error("Gemini 去水印失败,降级返回原图: %s", exc)
return image_bytes

View File

@@ -0,0 +1,96 @@
"""
llm_scorer.py — AI 评委打分入口(让模型真读文案,替代机械找词)
合规第7维仍走机械硬拦(score_compliance)AI 读前6维给分+理由。
任何异常/解析失败 → 回退旧机械 score_copy绝不卡链路。
"""
from __future__ import annotations
import asyncio
import json
import logging
import re
from typing import Any
from .constants import BANNED_WORDS_DEFAULT, BANNED_VISUAL_WORDS, QUALITY_PASS_SCORE
from ._scoring_dims import score_compliance
from .text_scoring import score_copy
from ._score_prompt import SCORER_PERSONA, build_score_prompt, COMPLIANCE_MAX
logger = logging.getLogger(__name__)
# 6 个 AI 维度满分倩倩姐2026-06-15拍板·与 constants.AI_DIM_WEIGHTS/_score_prompt 三处同步)
# 痛点18+情绪18+买点18+钩子15+标题13+真实感13=95+合规5=100
# "真实感"替换旧"产品聚焦一件事",对齐富贵"很少提产品/前70%干货后30%植入"原则
_DIM_MAX = {
"痛点人群精准": 18, "情绪张力": 18, "买点转化": 18,
"开头钩子": 15, "标题点击力": 13, "真实感": 13,
}
# 评委合规相关默认权重(仅供 score_compliance 复用其内部硬拦逻辑)
_COMPLIANCE_W = {"compliance": COMPLIANCE_MAX}
def _parse_verdict(raw: str) -> dict | None:
"""从模型输出里抠出 JSON 对象,失败返 None。"""
s = raw.strip()
m = re.search(r"\{.*\}", s, re.DOTALL)
if not m:
return None
try:
obj = json.loads(m.group(0))
return obj if isinstance(obj, dict) and isinstance(obj.get("dims"), list) else None
except (json.JSONDecodeError, ValueError):
return None
async def llm_score_copy(
client: Any,
copy: dict[str, Any],
source: dict[str, Any],
banned_words: list[str] | None = None,
pass_score: int = QUALITY_PASS_SCORE,
) -> dict[str, Any]:
"""AI 评委读 1 条文案 → 6维分+理由,合规机械硬拦。返回与 score_copy 同结构。"""
bwords = list(set((banned_words or []) + BANNED_WORDS_DEFAULT + BANNED_VISUAL_WORDS))
full = f"{copy.get('title','')}\n{copy.get('content','')}\n{' '.join(str(t) for t in copy.get('tags',[]))}"
dim_comp, found_all = score_compliance(full, bwords, _COMPLIANCE_W)
prompt = build_score_prompt(copy, source)
raw = ""
backoff = [5, 10, 20]
for attempt in range(4):
try:
raw = await client.chat_complete(
messages=[{"role": "system", "content": SCORER_PERSONA},
{"role": "user", "content": prompt}],
model=client._model, max_tokens=1500, temperature=0.3,
)
break
except Exception as exc: # noqa: BLE001 — 含 httpx.HTTPStatusError 503/429
status = getattr(getattr(exc, "response", None), "status_code", 0)
if status in (503, 429) and attempt < 3:
await asyncio.sleep(backoff[min(attempt, 2)])
continue
logger.warning("AI评委调用失败回退机械打分: %s", exc)
return score_copy(copy, source, banned_words, pass_score=pass_score)
verdict = _parse_verdict(raw)
if not verdict:
logger.warning("AI评委输出解析失败回退机械打分。raw[:120]=%s", raw[:120])
return score_copy(copy, source, banned_words, pass_score=pass_score)
details: list[dict] = []
for d in verdict["dims"]:
item = str(d.get("item", "")).strip()
if item not in _DIM_MAX: # 只收白名单6维模型偶尔多吐"总分"等噪声项,丢弃
continue
mx = _DIM_MAX[item]
sc = max(0, min(mx, int(round(float(d.get("score", 0))))))
details.append({"item": item, "score": sc, "max": mx, "note": str(d.get("reason", ""))[:60]})
details.append(dim_comp)
total = max(0, min(100, sum(d["score"] for d in details)))
passed = (total >= pass_score) and not found_all
return {
"score": total, "score_detail": details, "passed": passed,
"banned_words_found": found_all,
"verdict": verdict.get("verdict", ""), "summary": str(verdict.get("summary", ""))[:120],
}

View File

@@ -0,0 +1,132 @@
"""
package_exporter.py — 达人素材交付包生成
架构方案§五 1A步骤5按笔记分文件夹 + 图(01/02/03) + 文案.txt + 发布清单 + 合规说明
路径规则uploads/packages/{workspace_id}/{task_id}/note_{n}/
"""
from __future__ import annotations
import json
import logging
import os
import zipfile
from datetime import datetime
from pathlib import Path
from typing import Any
logger = logging.getLogger(__name__)
# 文件夹结构
# uploads/packages/{workspace_id}/{task_id}/
# note_01/
# 01_hook.jpg # 按 seq 序号命名防传错序
# 02_proof.jpg
# 文案.txt # 标题 + 正文 + 标签
# note_02/
# ...
# 📋发布清单.txt
# ✅合规说明.txt
# package.zip # 最终打包文件
def build_delivery_package(
workspace_id: int,
task_id: int,
notes: list[dict], # 每条笔记,含 text_candidate + image_candidates
base_path: str = "uploads/packages",
) -> str:
"""
打包交付,返回 zip 文件的本地路径。
notes 格式:[{
"title": str, "content": str, "tags": list[str],
"images": [{"seq": int, "role": str, "data": bytes}],
"banned_word_status": str, # 合规说明用
}]
"""
package_dir = Path(base_path) / str(workspace_id) / str(task_id)
package_dir.mkdir(parents=True, exist_ok=True)
note_dirs: list[Path] = []
for idx, note in enumerate(notes, start=1):
note_dir = package_dir / f"note_{idx:02d}"
note_dir.mkdir(exist_ok=True)
note_dirs.append(note_dir)
# ── 图片文件(按 seq 序号命名)
for img in sorted(note.get("images", []), key=lambda x: x.get("seq", 0)):
seq = img.get("seq", idx)
role = img.get("role", "img")
fname = f"{seq:02d}_{role}.jpg"
img_data = img.get("data", b"")
if img_data:
(note_dir / fname).write_bytes(img_data)
# ── 文案.txt标题 + 正文 + 标签,达人可直接复制)
tags = note.get("tags") or []
body = note.get("content", "")
# 正文末尾如果 LLM 已写入 #话题 标签,不再重复追加(避免重复)
body_has_tags = bool(tags) and any(
t.strip("#") in body for t in tags if t
)
copy_lines = [
f"【标题】{note.get('title', '')}",
"",
body,
]
if tags and not body_has_tags:
copy_lines += ["", " ".join(tags)]
(note_dir / "文案.txt").write_text("\n".join(copy_lines), encoding="utf-8")
# ── 发布清单.txt
checklist_lines = [
"📋 发布清单",
f"生成时间:{datetime.now().strftime('%Y-%m-%d %H:%M')}",
f"任务ID{task_id}",
"",
]
for idx, note in enumerate(notes, start=1):
title = note.get("title", f"笔记{idx}")
n_images = len(note.get("images", []))
checklist_lines.append(f"note_{idx:02d} 标题:{title[:30]} 图片数:{n_images}")
checklist_lines += [
"",
"发布注意事项:",
"- 每条笔记图片按 01/02/03 顺序上传,避免传错序",
"- 文案.txt 中标题/正文/标签已区分,复制对应部分",
"- 品牌词已植入,请勿删除",
"- 不要添加链接(种品牌词,引导天猫搜索成交)",
]
(package_dir / "📋发布清单.txt").write_text("\n".join(checklist_lines), encoding="utf-8")
# ── 合规说明.txt
compliance_lines = [
"✅ 合规说明",
f"生成时间:{datetime.now().strftime('%Y-%m-%d %H:%M')}",
"",
"本批次内容已完成以下合规处理:",
"1. 违禁词扫描(美白/祛斑/速效/医用/药妆等)",
"2. 视觉违禁词处理(前后对比/变白等)",
"3. 图片去水印处理C2PA元数据已清除",
"",
"各笔记合规状态:",
]
for idx, note in enumerate(notes, start=1):
status = note.get("banned_word_status", "pass")
status_label = {"pass": "✅通过", "auto_fixed": "✅自动改写", "soft_warn": "⚠️软提示", "hard_block": "❌硬拦截"}.get(status, status)
compliance_lines.append(f"note_{idx:02d}{status_label}")
compliance_lines += [
"",
"C2PA元数据可去除私有像素水印只能削弱不保证100%清除。",
"如有合规疑问,请联系运营团队。",
]
(package_dir / "✅合规说明.txt").write_text("\n".join(compliance_lines), encoding="utf-8")
# ── 打 zip
zip_path = package_dir / "package.zip"
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
for note_dir in note_dirs:
for fpath in sorted(note_dir.iterdir()):
zf.write(fpath, arcname=f"{note_dir.name}/{fpath.name}")
zf.write(package_dir / "📋发布清单.txt", arcname="📋发布清单.txt")
zf.write(package_dir / "✅合规说明.txt", arcname="✅合规说明.txt")
logger.info("delivery package built: %s (notes=%d)", zip_path, len(notes))
return str(zip_path)

View File

@@ -0,0 +1,145 @@
"""
偏好飞轮聚合preference_aggregator
扒自Clover架构方案.md §偏好飞轮怎么转 + PRD §8
三层继承L1 公司品牌基线 > L2 矩阵号人设(二期)> L3 个人手感
聚合最近 FLYWHEEL_LOOKBACK 条 events → prompt 片段注入文案生成
关键:
- 按 product_id 分开学(素颜霜偏好不串精华)
- 信号不足 FLYWHEEL_COLD_START 条时,用产品档案冷启动
- 返回结构对齐 API契约 GET /tasks/{id}/preference/context
"""
from __future__ import annotations
import logging
from collections import Counter
from typing import Any
from .constants import FLYWHEEL_LOOKBACK, FLYWHEEL_COLD_START
logger = logging.getLogger(__name__)
def aggregate_preference_context(
events: list[dict],
product: dict,
workspace_id: int,
product_id: int,
) -> dict:
"""
输入:最近 preference_events 行(已按 workspace_id+product_id 过滤)
输出:{recent_preference, reject_reasons, injected_count, prompt_fragment}
prompt_fragment 直接注入文案生成 prompt
"""
# 按 product_id 过滤(防串货)
relevant = [
e for e in events
if e.get("workspace_id") == workspace_id and e.get("product_id") == product_id
][:FLYWHEEL_LOOKBACK]
injected_count = len(relevant)
if injected_count < FLYWHEEL_COLD_START:
# 冷启动:用产品档案静态基线
return _cold_start(product, injected_count)
# ── 统计最常选角度text_select + approve 信号)
angle_counts: Counter = Counter()
reject_reasons: list[str] = []
for e in relevant:
sig_type = e.get("signal_type", "")
angle = str(e.get("angle_label", "")).strip()
weight = int(e.get("signal_weight", 1))
if sig_type in ("text_select", "approve") and angle:
angle_counts[angle] += weight
elif sig_type == "reject_with_reason":
reason = str(e.get("reason", "")).strip()
if reason:
reject_reasons.append(reason)
# 取权重最高的角度
top_angles = [a for a, _ in angle_counts.most_common(3)]
# 取最近3条打回原因
recent_rejects = reject_reasons[-3:] if reject_reasons else []
# ── 拼 prompt 片段三层继承L1>L2>L3一期只跑L1+L3
prompt_fragment = _build_prompt_fragment(top_angles, recent_rejects, product)
# ── 人类可读摘要(前端"本次已注入"显示)
if top_angles:
pref_summary = f"最近偏好角度:{''.join(top_angles)}(已选{injected_count}次信号)"
else:
pref_summary = f"已注入{injected_count}条偏好信号"
return {
"recent_preference": pref_summary,
"reject_reasons": recent_rejects,
"injected_count": injected_count,
"prompt_fragment": prompt_fragment, # 注入 generate_text_variants extra_rules
}
def _cold_start(product: dict, injected_count: int) -> dict:
"""信号不足时用产品档案基线"""
angles = product.get("text_angles") or []
style = product.get("style_tone", "素人分享风")
fragment = ""
if angles:
fragment = f"优先覆盖以下文案角度:{''.join(angles[:3])}。风格调性:{style}"
return {
"recent_preference": f"冷启动(历史信号{injected_count}条,不足{FLYWHEEL_COLD_START}条),使用产品档案基线",
"reject_reasons": [],
"injected_count": injected_count,
"prompt_fragment": fragment,
}
def _build_prompt_fragment(
top_angles: list[str],
reject_reasons: list[str],
product: dict,
) -> str:
"""
组装注入文案 prompt 的片段
越积累越精准1次=全靠基线10次=知道偏好角度30次=措辞从"供参考"升为明确指令
"""
lines: list[str] = []
if top_angles:
lines.append(f"【偏好角度参考】历史选择偏好:{''.join(top_angles)},请优先采用这些角度方向。")
if reject_reasons:
formatted = "".join(f"{r}" for r in reject_reasons)
lines.append(f"【打回原因参考】以下问题请主动规避:{formatted}")
# L1 品牌基线(产品档案 custom_prompt
custom = (product.get("custom_prompt") or "").strip()
if custom:
lines.append(f"【品牌基线】{custom}")
return "\n".join(lines)
def collect_preference_event(
signal_type: str,
user_id: int,
workspace_id: int,
product_id: int,
angle_label: str = "",
reason: str = "",
weights: dict[str, int] | None = None,
) -> dict:
"""
构造 preference_event 行(由业务接口内部调用,不暴露给前端)
返回待插 DB 的字段 dict
"""
from .constants import FLYWHEEL_WEIGHTS
w_map = weights or FLYWHEEL_WEIGHTS
weight = w_map.get(signal_type, 0)
return {
"signal_type": signal_type,
"signal_weight": weight,
"user_id": user_id,
"workspace_id": workspace_id,
"product_id": product_id,
"angle_label": angle_label,
"reason": reason,
"data_ownership": "client_data", # 原始行为信号归客户PRD §3 data_ownership
}

View File

@@ -0,0 +1,109 @@
"""
prompt_composer.py — 统一 prompt 组装入口≤100行
扒自banana prompts/service.py + worker/src/copy.js prompt 逻辑
Lead 指名接口compose_variants / compose_preference_context
组装逻辑委托:
_text_prompt.py → build_prompt (文案 prompt 主体)
preference_aggregator.py → aggregate_preference_context (飞轮上下文)
原则prompt 组装从这里进,不散落在 text_variants / generate_text_variants 里。
"""
from __future__ import annotations
import logging
from typing import Any
from ._text_prompt import build_prompt, COPY_SYSTEM
from .preference_aggregator import aggregate_preference_context
logger = logging.getLogger(__name__)
# ── 主接口 ────────────────────────────────────────────────────────────────────
def compose_variants(
product: dict,
count: int,
flywheel_context: str = "",
extra_rules: str = "",
) -> tuple[str, str]:
"""
一次出 count 角度文案的完整 prompt。
返回 (system_prompt, user_prompt)。
飞轮片段追加到 user_prompt 末尾(不改 system避免覆盖质量红线
参数:
product — 产品档案 dictname/selling_points/text_angles/custom_prompt 等)
count — 需要几条
flywheel_context— 由 compose_preference_context 返回的 prompt_fragment
extra_rules — 额外规则(优化循环重生成时传 hint
"""
combined_extra = "\n".join(filter(None, [flywheel_context, extra_rules]))
user_prompt = build_prompt(product, count, extra_rules=combined_extra)
logger.debug(
"compose_variants: product=%s count=%d flywheel_len=%d",
product.get("name", "?"), count, len(flywheel_context),
)
return COPY_SYSTEM, user_prompt
def compose_preference_context(
events: list[dict],
product: dict,
workspace_id: int,
product_id: int,
) -> dict:
"""
聚合偏好事件 → 可注入 prompt 的飞轮上下文。
返回结构(对齐 API契约 GET /tasks/{id}/preference/context
{
recent_preference: str, # 人类可读摘要(前端"本次已注入"显示)
reject_reasons: list, # 最近打回原因
injected_count: int, # 有效信号数
prompt_fragment: str, # 注入 compose_variants flywheel_context 的字符串
}
信号不足 FLYWHEEL_COLD_START 条时用产品档案冷启动。
按 workspace_id + product_id 双维过滤(素颜霜偏好不串精华)。
"""
return aggregate_preference_context(events, product, workspace_id, product_id)
# ── 辅助:解析模型返回的 JSON给 text_variants 调用,集中不散) ──────────────
def parse_model_output(raw: str) -> list[dict]:
"""从 LLM 原始输出提取 JSON 数组(容错 markdown 包裹)"""
from ._text_prompt import parse_json_array
return parse_json_array(raw)
# ── 辅助:图片 prompt 组装入口(预留,联调时填充)─────────────────────────────
def compose_image_prompt(
role_name: str,
visual_system: dict,
product: dict,
extra: str = "",
) -> str:
"""
单张分镜 prompt 组装(供 image_gen.generate_one_image 调用)。
TODO: 联调后从 storyboard.plan_image_set 取 base_prompt 注入。
role_name — 分镜角色hook / pain_scene / closer 等)
visual_system— build_visual_system 返回的视觉系统 dict
extra — 追加约束(飞轮图片偏好片段,二期接入)
"""
name = product.get("name", "产品")
style = visual_system.get("style", "")
palette = visual_system.get("color_palette", "")
base = visual_system.get("base_prompt", "")
lines = [
f"[{role_name}] 为产品「{name}」生成种草图。",
base and f"视觉基调:{base}",
style and f"摄影风格:{style}",
palette and f"色调:{palette}",
extra,
]
return "\n".join(l for l in lines if l)

View File

@@ -0,0 +1,197 @@
"""
storyboard 分镜引擎
扒自worker/src/image.js
- getNarrativeRoles按图数取分镜角色
- proof_strategy按品类定证明页策略品类不写死走数据驱动
- build_visual_system成组视觉统一
- plan_image_set组装最终分镜计划
"""
from __future__ import annotations
import re
from .constants import (
PAGE_ROLE_MAP, IMAGE_NEGATIVE_CONSTRAINTS,
STYLE_PROMPTS, STYLE_DEFAULT, NARRATIVE_BY_COUNT, NARRATIVE_BY_STRATEGY,
)
# sanitize_text 移至 templates腾行数此处 re-export 供 image_gen 沿用 import
from .storyboard_templates import role_template, proof_strategy, sanitize_text # noqa: F401
def clamp_count(value: int, fallback: int = 6, lo: int = 1, hi: int = 8) -> int:
try:
return max(lo, min(hi, int(value)))
except (TypeError, ValueError):
return fallback
def short_selling_points(points, fallback: str = "") -> str:
"""3个短卖点拼成 a / b / c扒 shortSellingPoints:112-120"""
src = points if isinstance(points, list) else str(points or "").split("")
clean = [sanitize_text(p, 18) for p in src if sanitize_text(p, 18)][:3]
return " / ".join(clean) if clean else sanitize_text(fallback, 28)
def short_tags(tags, keywords=None) -> str:
"""标签去#截断拼成 #a #b扒 shortTags:48-54"""
merged = list(tags or []) + list(keywords or [])
out = []
for t in merged:
c = sanitize_text(str(t), 12).lstrip("#")
if c:
out.append(f"#{c}")
return " ".join(out[:5])
def analyze_copy_for_image(note: dict, product: dict) -> dict:
"""
从文案+产品提取生图锚点(扒 analyzeCopyForImage:129-148
给每张图填 audience/pain/scene/hook让画面有真实代入而非空泛。
"""
text = f"{note.get('title','')}{note.get('coverTitle','')}{note.get('content','')}"
tags = [sanitize_text(str(t).lstrip('#'), 12) for t in (note.get("tags") or [])]
audience = sanitize_text(
product.get("target_audience")
or next((t for t in tags if re.search(r"党|人|妈妈|女生|学生|通勤|上班|办公室", t)), "")
or "目标用户", 18)
scene = sanitize_text(
next((t for t in tags if re.search(r"通勤|宿舍|上课|约会|出门|办公室|旅行|居家|工位", t)), "")
or "日常自然光场景", 18)
pain = sanitize_text(
next((w for w in re.split(r"[、,,。;;!?\n]", text)
if re.search(r"暗沉|没气色|假白|卡粉|搓泥|油|干|赶时间|预算|麻烦", w)), "")
or "日常使用痛点", 18)
hook = sanitize_text(note.get("coverTitle") or note.get("title") or f"{audience}{scene}", 18)
return {"audience": audience, "scene": scene, "pain": pain, "hook": hook}
def get_narrative_roles(image_count: int = 6) -> list[dict]:
"""
按图数返回分镜角色列表(扒 getNarrativeRolesQ6对齐北哥6张套路
≤3 张:极速链路 hook / applied_proof / closer
≤6 张:北哥标准链路 ①封面痛点大字 ②单品特写+品牌词 ③成分 ④质地 ⑤上脸对比 ⑥促单
>6 张:沉浸链路 + pain_scene / scenario / social_proof
"""
count = clamp_count(image_count)
m = PAGE_ROLE_MAP
if count <= 3:
sequence = ["hook", "applied_proof", "closer"]
elif count <= 6:
# Q6北哥6张标准顺序——品牌词在②(product_closeup)和⑥(closer)两次出现
sequence = ["hook", "product_closeup", "ingredient", "texture", "applied_proof", "closer"]
else:
# 8张沉浸链路在北哥6张基础上插入 pain_scene / social_proof
sequence = ["hook", "pain_scene", "product_closeup", "ingredient", "texture", "applied_proof", "social_proof", "closer"]
return [m[r] for r in sequence[:count] if r in m]
# ── proofStrategy 已移至 storyboard_templates.proof_strategy腾行数超200拆
def build_visual_system(product: dict, analysis: dict | None = None) -> dict:
"""
成组视觉统一(扒 buildVisualSystem
analysis 来自 product.js 分析结果visualIdentity可空
"""
identity = (analysis or {}).get("visualIdentity", {})
palette = (
"".join(identity["colorPalette"][:5])
if isinstance(identity.get("colorPalette"), list) and identity["colorPalette"]
else "提取产品包装主色,搭配浅色真实生活背景"
)
return {
"palette": palette,
"typography": identity.get("typographyStyle", "主标题清晰黑体或手写感标题,辅助文字便签/勾选标注,字重颜色保持同一体系"),
"sticker": identity.get("stickerLanguage", "少量箭头、放大镜、勾选、小表情、便签,不使用促销按钮"),
"layout": identity.get("layoutStyle", "同一组图片保持色调、光线、产品露出方式一致,每张图承担不同叙事角色"),
"texture": identity.get("materialTexture", "产品包装、质地、手背/上脸肤感要真实自然"),
"package_details": identity.get("packageDetails", "如果提供产品图,必须还原包装颜色、瓶身形状、标签方向和主视觉"),
"xhs_style_preset": identity.get("xhsStylePreset", "真实测评风/手写安利风/清单便签风"),
"symbol_system": identity.get("symbolSystem", "中等密度小红书种草符号:✅ ✨ 🌿 💧 🪞 🧴 📦 🔍 💛每张最多2-4个"),
"quality_rules": [
"同组图片字体体系相对一致,但不要像固定模板",
"每张压图文字必须服务当前叙事角色,不能重复封面标题",
"护肤品优先出现手背涂抹、质地微距、自然上脸局部或真实生活场景",
"人物真实自然有轻微皮肤纹理和生活感不要AI精修美女",
"禁止乱码、错别字、App底栏、Like评论分享、硬广价格牌、虚假功效before/after",
],
}
def plan_image_set(note: dict, product: dict, image_count: int = 3, analysis: dict | None = None, strategy: str | None = None) -> dict:
"""
组装分镜计划(主入口)
strategy: None=默认按图数叙事,'A'/'B'/'C'=三套正交叙事策略
返回:{requested_count, storyboard, visual_system, base_prompt}
storyboard 每项:{role, name, focus, overlay_text, prompt_for_item}
"""
count = clamp_count(image_count)
roles = get_narrative_roles(count)
visual = build_visual_system(product, analysis)
category = product.get("category", "通用好物")
points = product.get("selling_points") or ["核心买点"]
src = analyze_copy_for_image(note, product) # 文案锚点audience/pain/scene/hook
storyboard = []
brand_kw = sanitize_text(product.get("brand_keyword") or "", 12)
brand_roles = {"product_closeup", "closer"} # Q6第2/6张带品牌词
for i, role in enumerate(roles):
point = sanitize_text(points[i % len(points)], 18)
tpl = role_template(role["role"])
proof = proof_strategy(category, point) # 仅 applied_proof 用品类证明
# 填模板占位:每角色画面/文字各不同修缩水根因——不再全角色共用proof
fill = {
"audience": src["audience"], "pain": src["pain"], "scene": src["scene"],
"hook": src["hook"], "point": point, "brand": brand_kw or product.get("name", "产品"),
"proof_overlay": proof.get("overlay", point),
"proof_visual": proof.get("visual", ""),
"proof_forbidden": proof.get("forbidden", ""),
}
item = {
"role": role["role"],
"name": role["name"],
"focus": role["focus"],
"goal": sanitize_text(tpl["goal"].format(**fill), 40),
"overlay_text": sanitize_text(tpl["overlay"].format(**fill), 20),
"visual_strategy": sanitize_text(tpl["visual"].format(**fill), 120),
"source_basis": sanitize_text(tpl["basis"].format(**fill), 60),
"selling_point": point,
"asset_use": proof.get("asset_use", "产品图保证包装准确,参考图用于真实场景"),
"forbidden": sanitize_text(tpl["forbidden"].format(**fill), 60),
}
if brand_kw and role["role"] in brand_roles:
item["brand_keyword"] = brand_kw
item["brand_keyword_rule"] = "品牌词自然融入画面文字或产品露出,不做广告牌感"
storyboard.append(item)
# base_prompt 全局规则(扒 planImageSet basePrompt:682-690
product_name = product.get("name", "产品")
cover_title = sanitize_text(note.get("coverTitle") or note.get("title") or "", 18)
style_text = STYLE_PROMPTS.get(product.get("style_tone") or STYLE_DEFAULT, STYLE_PROMPTS[STYLE_DEFAULT])
# strategy非空时取对应正交叙事否则按图数取默认链路
narrative = NARRATIVE_BY_STRATEGY.get(strategy) if strategy else None
if not narrative:
narrative = NARRATIVE_BY_COUNT.get(count, NARRATIVE_BY_COUNT[6])
sp_text = short_selling_points(points, cover_title)
tag_text = short_tags(note.get("tags") or [], product.get("keywords") or [])
base_prompt = (
f"生成一张小红书可上传的独立3:4图文海报/素材图目标比例1024×1536可直接上传的独立图片不是提示词不是App截图。"
f"产品:{product_name}。短标题:{cover_title}"
f"短卖点:{sp_text}。短标签:{tag_text}"
f"叙事链路:{narrative}"
f"视觉风格:{style_text}"
f"成组视觉:主色={visual['palette']};字体={visual['typography']};贴纸={visual['sticker']}"
f"符号系统={visual['symbol_system']};产品还原={visual['package_details']}"
"重要限制中文文字少而清晰每张只允许一个主标题同一句话禁止重复出现正文点位最多3条"
"四周留安全边距文字不贴边不被裁切真实自然像实拍素材后排版降低AI味"
"禁止生成小红书App界面截图、Like/评论/分享/底栏/头像等社交元素;"
"禁止肤色变白、瑕疵消失、治疗前后等视觉暗示,允许安全的未推开/推开后质地状态对比;"
"如果提供产品图,产品是不可修改的真实商品锚点,禁止改名、换包装、混入其他产品。"
f"\n{IMAGE_NEGATIVE_CONSTRAINTS}"
)
return {
"requested_count": count,
"storyboard": storyboard,
"visual_system": visual,
"base_prompt": base_prompt,
}

View File

@@ -0,0 +1,174 @@
"""
角色差异化分镜模板
扒自worker/src/image.js buildImageStoryboard storyboardByRole(222-321)
每个分镜角色有【各自不同】的画面/文字/构图策略,这是"小红书风格不雷同"的根因。
之前缩水6张共用同一个品类 proof 策略 → 图全长一样。
模板里 {占位} 在 storyboard.py 运行时按文案/卖点填充。
字段含义(对齐 promptFromStoryboard 9 字段):
goal 本张目标(这张图要让用户产生什么动作/情绪)
overlay 图上主文字模板(每张不同,不重复封面标题)
visual 画面主体(构图、景别、道具、光线——这是不雷同的关键)
basis 文案依据(这张图从文案哪里来,给模型锚点)
forbidden 本张禁止事项
"""
from __future__ import annotations
import re
# ── sanitize扒 sanitizeImagePlanText防违禁视觉词进 prompt
_SANITIZE_RULES: list[tuple[str, str]] = [
(r"before\s*&\s*after", "质地与肤感说明"),
(r"before\s*/?\s*after", "质地与肤感说明"),
(r"\bbefore\b", "质地状态"),
(r"\bafter\b", "上脸肤感"),
(r"使用前后|用前用后|用前后|前后对比|使用前|使用后", "质地/场景/肤感说明"),
(r"功效对比|效果对比|改善对比", "质地/场景说明对比"),
(r"肤色变白|皮肤变白|变白|美白", "自然光泽感"),
(r"瑕疵消失|斑点消失|痘印消失|消除瑕疵|祛斑", "妆感更服帖"),
(r"治疗前后|治疗后|医美前后|治愈|修复受损", "日常使用场景说明"),
]
def sanitize_text(value: str, max_len: int = 56) -> str:
s = str(value)
for pattern, repl in _SANITIZE_RULES:
s = re.sub(pattern, repl, s, flags=re.IGNORECASE)
return re.sub(r"\s+", " ", s).strip()[:max_len]
# 北哥6张标准套 + 8张扩展角色每角色独立画面策略
ROLE_STORYBOARD_TPL: dict[str, dict] = {
"hook": {
"goal": "{audience}因为{pain}停下划走,产生点开欲",
"overlay": "{hook}",
"visual": "自然光生活场景,手持产品或产品在桌面前景,真实肤感/手部细节像iPhone随手实拍的封面不是海报",
"basis": "来自选中文案标题、人群{audience}、痛点{pain}",
"forbidden": "不要价格、不要重复后续卖点、不要App界面、不要广告海报感",
},
"product_closeup": {
"goal": "建立单品记忆锚点,让用户记住是哪个产品",
"overlay": "{brand}",
"visual": "单品高清特写居中,干净浅色台面,柔和顶光,瓶身/包装/标签清晰可读,品牌词自然出现在画面或瓶身",
"basis": "来自产品名和品牌词第2张和第6张都要带品牌词强化记忆",
"forbidden": "不要堆多个产品、不要花哨背景抢主体、不要改包装文字",
},
"ingredient": {
"goal": "用成分/配方信息建立信任,但不医疗化",
"overlay": "看清{point}",
"visual": "成分卡片式布局,产品+成分图标/短说明,浅色商务美妆风,信息层级清楚",
"basis": "来自卖点里的成分/功效点,理性表达不夸大",
"forbidden": "不要治疗/改善疾病承诺、不要医生背书、不要绝对化",
},
"texture": {
"goal": "让用户看到{point}的真实质感证据",
"overlay": "{point}看得见",
"visual": "手背或指尖涂抹质地微距,产品放在旁边,自然光,保留真实皮肤纹理,能看清延展和肤感",
"basis": "来自卖点里的质地/肤感描述",
"forbidden": "不要生成变白效果、不要医疗化对比、不要和封面同构图",
},
"applied_proof": {
"goal": "用可感知的上脸/使用证据证明{point}",
"overlay": "{proof_overlay}",
"visual": "{proof_visual}",
"basis": "来自核心卖点{point}和用户对效果的关注",
"forbidden": "{proof_forbidden}",
},
"closer": {
"goal": "用囤货/省钱情报/搜索暗号完成软性转化",
"overlay": "这波真的会囤 {brand}",
"visual": "拆箱、囤货角或产品放在日常物品旁,真实分享氛围,轻量搜索/品牌词暗号提示,再带一次品牌词引导成交",
"basis": "来自价格心智/选择理由,但不做硬广",
"forbidden": "不要大促价格牌、不要购买按钮、不要红黄电商风",
},
# ── 8张扩展角色 ──
"pain_scene": {
"goal": "让用户共鸣{pain}",
"overlay": "{pain}真的懂",
"visual": "{scene}里的真实困扰场景,产品作为解决方案线索出现,不做使用前后对比",
"basis": "来自文案痛点和目标人群",
"forbidden": "不要夸大焦虑、不要before/after",
},
"social_proof": {
"goal": "补足信任背书,让内容不像单方面推销",
"overlay": "身边人都在问",
"visual": "产品在包里/桌面/宿舍囤货角,配简短手写感反馈气泡,真实随手拍",
"basis": "来自评论区语言/选择理由,缺评论时用低调口碑表达",
"forbidden": "不要假造大量头像评论、不要App评论区截图",
},
"scenario": {
"goal": "展示{scene}以外的多场景使用代入",
"overlay": "这些场景都能用",
"visual": "2-3个生活小场景拼贴宿舍/通勤包/办公桌,产品贯穿其中,统一光线",
"basis": "来自目标人群的多场景使用需求",
"forbidden": "不要电商详情页拼贴、不要夸大效果",
},
"tutorial": {
"goal": "降低使用门槛,告诉用户怎么用",
"overlay": "三步就上手",
"visual": "三步手势教程:取量、点涂/使用、收尾,干净背景,产品在画面内",
"basis": "来自文案里的快速/懒人使用场景",
"forbidden": "不要复杂说明书、不要过多文字",
},
}
def role_template(role: str) -> dict:
"""取角色模板,未知角色用 applied_proof 兜底(和源头一致)"""
return ROLE_STORYBOARD_TPL.get(role, ROLE_STORYBOARD_TPL["applied_proof"])
# ── proofStrategy按品类定 applied_proof 证明页策略,扒 image.js:163-208
# 品类来自 product.category不硬编码枚举无匹配走"通用好物"兜底
PROOF_STRATEGIES: dict[str, dict] = {
"个护护理": {
"overlay_tpl": "{point}看得见",
"visual": "手部/身体局部使用证明:少量点涂、推开后吸收状态、真实纹理和自然光;产品只做辅助露出",
"asset_use": "优先使用实拍/参考图中的手部、干纹、涂抹、随身场景;产品图保证包装准确",
"forbidden": "不要变白、祛斑、医学效果、before/after字样不要和封面同构图",
},
"美妆护肤": {
"overlay_tpl": "{point}看得见",
"visual": "肤感/质地证明:手背、脸颊局部或质地微距,展示推开前后真实状态,保留皮肤纹理和自然光",
"asset_use": "优先使用实拍/参考图中的手背、上脸、质地素材;产品图辅助露出",
"forbidden": "不要变白、祛斑、医学效果、before/after字样不要和封面同构图",
},
"食品饮品": {
"overlay_tpl": "{point}一眼懂",
"visual": "冲泡/开袋/入口证明:展示包装、杯中状态、质地颜色或一口口感,真实桌面光线",
"asset_use": "产品图保证包装准确,参考图用于杯子、开袋、冲泡、办公室/居家场景",
"forbidden": "不要涂抹、不要护肤肤感、不要医疗健康承诺",
},
"营养健康": {
"overlay_tpl": "看清{point}",
"visual": "理性证明页:包装、成分表、使用场景和每日习惯卡片,信息清晰但不做治疗承诺",
"asset_use": "产品图和说明图用于成分/包装准确,参考图用于日常使用场景",
"forbidden": "不要治疗、改善疾病、速效、医生背书、前后对比",
},
"家居生活": {
"overlay_tpl": "{point}真省事",
"visual": "使用过程证明:展示痛点场景、产品介入和使用过程细节,强调顺手/收纳/效率",
"asset_use": "参考图用于真实家居环境,产品图保证外观准确",
"forbidden": "不要护肤涂抹,不要虚假夸大结果",
},
"服饰穿搭": {
"overlay_tpl": "{point}有细节",
"visual": "上身/材质证明:展示面料纹理、版型细节或普通身材上身局部,真实自然",
"asset_use": "参考图用于上身/搭配/材质,产品图保证款式颜色准确",
"forbidden": "不要护肤涂抹,不要过度精修模特感",
},
"通用好物": {
"overlay_tpl": "{point}清晰可见",
"visual": "产品使用场景证明:真实道具/场景,展示产品细节和使用过程",
"asset_use": "产品图保证准确,参考图用于场景辅助",
"forbidden": "不要夸大效果,不要硬广式价格牌",
},
}
def proof_strategy(category: str, point: str) -> dict:
"""取品类证明策略,无匹配用通用兜底(扒 proofStrategy"""
s = PROOF_STRATEGIES.get(category, PROOF_STRATEGIES["通用好物"]).copy()
s["overlay"] = s.pop("overlay_tpl", "{point}").format(point=point)
return s

View File

@@ -0,0 +1,93 @@
"""
text_scoring.py — 五维打分接口 + 去重≤100行
打分维度逻辑见 _scoring_dims.py
"""
from __future__ import annotations
import re
from difflib import SequenceMatcher
from typing import Any
from .constants import (
BANNED_WORDS_DEFAULT, BANNED_VISUAL_WORDS,
SCORE_WEIGHTS, QUALITY_PASS_SCORE,
DEDUP_TITLE_THRESHOLD, DEDUP_TITLE_CONTENT_TITLE, DEDUP_TITLE_CONTENT_BODY,
)
from ._scoring_dims import (
_cat_words, score_title, score_emotion, score_selling,
score_keyword, score_compliance,
)
def score_copy(
copy: dict[str, Any],
source: dict[str, Any],
banned_words: list[str] | None = None,
weights: dict[str, int] | None = None,
pass_score: int = QUALITY_PASS_SCORE,
) -> dict[str, Any]:
"""
五维打分标题25 / 情绪25 / 买点25 / 关键词20 / 合规5
返回:{score, score_detail, passed, banned_words_found}
"""
w = weights or SCORE_WEIGHTS
bwords = list(set((banned_words or []) + BANNED_WORDS_DEFAULT + BANNED_VISUAL_WORDS))
title = str(copy.get("title", ""))
content = str(copy.get("content", ""))
tags = " ".join(str(t) for t in copy.get("tags", []))
full = f"{title}\n{content}\n{tags}\n{copy.get('imageBrief','')}"
selling_points = source.get("selling_points", []) or []
keywords = source.get("keywords", []) or []
category = source.get("category", "通用好物")
cat_w = _cat_words(category)
dim_title = score_title(title, cat_w, w)
dim_emotion = score_emotion(full, w)
dim_selling = score_selling(copy, full, selling_points, w)
dim_keyword = score_keyword(copy, tags, keywords, w)
dim_compliance, found_all = score_compliance(full, bwords, w)
details = [dim_title, dim_emotion, dim_selling, dim_keyword, dim_compliance]
total = max(0, min(100, sum(d["score"] for d in details)))
passed = (total >= pass_score) and not found_all
return {"score": total, "score_detail": details, "passed": passed, "banned_words_found": found_all}
def _sim(a: str, b: str) -> float:
return SequenceMatcher(None, a, b).ratio()
def _copy_signature(copy: dict) -> str:
content = str(copy.get("content", ""))
opening = re.sub(r"\s+", "", content[:30])
return f"{copy.get('title', '')}|{opening}"
def is_similar_copy(a: dict, b: dict) -> bool:
"""同质化判重标题≥0.82 OR 标题≥0.65且正文≥0.72"""
t = _sim(str(a.get("title", "")), str(b.get("title", "")))
if t >= DEDUP_TITLE_THRESHOLD:
return True
if t >= DEDUP_TITLE_CONTENT_TITLE:
if _sim(str(a.get("content",""))[:200], str(b.get("content",""))[:200]) >= DEDUP_TITLE_CONTENT_BODY:
return True
return False
def dedupe_copies(copies: list[dict], previous: list[dict] | None = None) -> list[dict]:
"""本轮内互去重 + 与历史去重 + angle 去重"""
history = previous or []
kept: list[dict] = []
used_angles: set[str] = set()
for c in copies:
sig = _copy_signature(c)
if any(_copy_signature(h) == sig for h in history): continue
if any(is_similar_copy(c, h) for h in history): continue
if any(is_similar_copy(c, k) for k in kept): continue
angle = str(c.get("angle", "")).strip()
if angle and angle in used_angles: continue
if angle: used_angles.add(angle)
kept.append(c)
return kept

View File

@@ -0,0 +1,186 @@
"""
text_variants.py — 文案双轨主入口≤100行
轨A: generate_text_variants — 调 LLM 出 N 角度 JSON
轨B: text_import_handler — 导入外部文案进候选池
prompt 组装/解析见 _text_prompt.py评分/去重见 text_scoring.py
"""
from __future__ import annotations
import asyncio
import logging
import os
from typing import Any
from .constants import MAX_OPTIMIZE_ROUNDS
from ._text_prompt import COPY_SYSTEM, build_prompt, parse_json_array, build_local_drafts
from .text_scoring import score_copy, dedupe_copies
from .llm_scorer import llm_score_copy
from .banned_word_checker import check_and_fix, build_entries_from_db, CheckResult
logger = logging.getLogger(__name__)
async def _call_llm(client: Any, prompt: str, max_tokens: int = 8192) -> str:
"""统一 LLM 调用client 由 worker 注入,隔离 key。
G1坑修复AIClients 没有 .chat.completions正确方法是 .chat_complete()
S8: 503/429 指数退避重试最多3次2^attempt 秒),其他异常直接降级返 ''
max_tokens 由调用方按批量缩放opus 会尽量填满输出空间8192 token 的生成
单批 >60s 必撞 apiports 网关上限返 503(task46 实测每请求恰挂 ~61s)。实测单条
max_tokens=1500~2500 仅 16~18s。故按条数动态收墙钟压进 60s 网关窗口内。
"""
import httpx
# 倩倩姐2026-06-13拍板"加大重试+拉长退避"apiports负载波动时单条opus也会被
# 拖过60s返503短退避(1/2/4s)赶不开高负载窗口。故重试5次、退避拉长到最长30s
# 给中转站负载回落留时间。墙钟换稳定(MVP免费阶段可接受)。
max_attempts = 5
backoff = [5, 10, 20, 30] # 第1~4次重试前等待秒数拉长跨过apiports高负载窗口
for attempt in range(max_attempts):
try:
return await client.chat_complete(
messages=[
{"role": "system", "content": COPY_SYSTEM},
{"role": "user", "content": prompt},
],
model=client._model,
max_tokens=max_tokens,
temperature=0.75,
)
except httpx.HTTPStatusError as exc:
status = exc.response.status_code if exc.response is not None else 0
if status in (503, 429) and attempt < max_attempts - 1:
wait = backoff[min(attempt, len(backoff) - 1)]
logger.warning(
"LLM 返回 %s,第%d/%d次重试,等待 %ds: %s",
status, attempt + 1, max_attempts - 1, wait, exc,
)
await asyncio.sleep(wait)
continue
logger.error("LLM HTTP错误(不可重试或已达上限): %s: %s", type(exc).__name__, exc)
return ""
except Exception as exc:
# 其他异常(超时/网络断开等)不重试,直接降级
logger.error("LLM 调用失败: %s: %s", type(exc).__name__, exc)
return ""
return ""
# apiports 网关单次响应有 ~60s 上限claude 一次生成 >4 条长文案会超时返 503。
# 故分批:每批最多 4 条,串行调用合并。批大小可经 TEXT_BATCH_SIZE 调。
TEXT_BATCH_SIZE = int(os.environ.get("TEXT_BATCH_SIZE", "4"))
async def _generate_one_batch(llm_client: Any, product: dict, batch_n: int, extra: str) -> list[dict]:
"""生成一批 batch_n 条含解析重试最多2次。失败返回空列表。
max_tokens 按条数缩放(每条约 1800 token封顶 8192),压进 apiports 60s 网关窗口。"""
batch_max_tokens = min(8192, max(1800, batch_n * 1800))
for attempt in range(2):
raw = await _call_llm(llm_client, build_prompt(product, batch_n, extra_rules=extra), batch_max_tokens)
parsed = parse_json_array(raw)
if parsed:
return parsed
logger.warning("文案批(%d条)第%d次解析失败%s", batch_n, attempt + 1,
",重试" if attempt == 0 else ",放弃本批")
return []
async def _generate_in_batches(llm_client: Any, product: dict, count: int, extra: str) -> list[dict]:
"""把 count 条按 TEXT_BATCH_SIZE 分批,串行调用合并。
串行而非并发opus 单批就慢(~300s)且 apiports 限并发,多批 gather 会触发
大面积 503 雪崩(task45 实测)。故改串行,墙钟换稳定。"""
sizes: list[int] = []
remaining = count
while remaining > 0:
n = min(TEXT_BATCH_SIZE, remaining)
sizes.append(n)
remaining -= n
collected: list[dict] = []
for n in sizes:
r = await _generate_one_batch(llm_client, product, n, extra)
collected.extend(r)
return collected
async def generate_text_variants(
llm_client: Any,
product: dict,
count: int,
previous_copies: list[dict] | None = None,
banned_word_rows: list[dict] | None = None,
flywheel_context: str = "",
) -> list[dict]:
"""轨A一次出 count 条不同角度文案,三层兜底,自动优化循环"""
banned_entries = build_entries_from_db(banned_word_rows or [])
extra = flywheel_context
copies: list[dict] = await _generate_in_batches(llm_client, product, count, extra)
if not copies:
copies = list(build_local_drafts(product, count)) # generator → list
candidates: list[dict] = []
for c in copies:
ban: CheckResult = check_and_fix(
f"{c.get('title','')} {c.get('content','')}",
banned_entries or None,
)
scored = await llm_score_copy(llm_client, c, product, [e.word for e in banned_entries])
c.update({"source": "ai", "score": scored["score"], "score_detail": scored["score_detail"],
"passed": scored["passed"], "banned_word_status": ban.status,
"verdict": scored.get("verdict", ""), "summary": scored.get("summary", "")})
if ban.status == "auto_fixed" and ban.fixed_text:
c["content"] = ban.fixed_text
candidates.append(c)
failed = [c for c in candidates if not c["passed"] and c["banned_word_status"] != "hard_block"]
# 优化轮默认关闭apiports 60s 网关限制下优化轮的 _call_llm 常需白等 60s 才 503
# 严重拖慢出文案(实测 +100s+)。质量优化等北哥 prompt 方案到位再开(架构已留位)。
optimize_enabled = os.environ.get("TEXT_OPTIMIZE_ENABLED", "false").lower() == "true"
rounds = MAX_OPTIMIZE_ROUNDS if optimize_enabled else 0
for _ in range(rounds):
if not failed:
break
# 优化轮也受 60s 网关上限约束:一次最多重生成 TEXT_BATCH_SIZE 条
batch_failed = failed[:TEXT_BATCH_SIZE]
hint = "\n".join(
f"标题「{c['title']}{c['score']}分,需改进:" +
"".join(d["note"] for d in c.get("score_detail", []) if d["score"] < d["max"] * 0.72)
for c in batch_failed
)
raw2 = await _call_llm(llm_client, build_prompt(
product, len(batch_failed),
extra_rules=f"以下文案未达标,请重新生成并改进:\n{hint}\n不要重复已有标题和角度。",
), min(8192, max(1800, len(batch_failed) * 1800)))
if not raw2:
# LLM 失败(如 503/超时):优化是锦上添花,原始候选已够用,不再耗时重试
logger.warning("文案优化轮 LLM 失败,沿用原始候选不再重试")
break
for nc in parse_json_array(raw2):
sc2 = await llm_score_copy(llm_client, nc, product, [e.word for e in banned_entries])
nc.update({"source": "ai", "score": sc2["score"], "score_detail": sc2["score_detail"],
"passed": sc2["passed"], "banned_word_status": "pass",
"verdict": sc2.get("verdict", ""), "summary": sc2.get("summary", "")})
candidates.append(nc)
failed = [c for c in candidates if not c["passed"]]
return dedupe_copies(candidates, previous_copies or [])[:count]
def text_import_handler(
raw_text: str,
product: dict,
banned_word_rows: list[dict] | None = None,
) -> dict:
"""轨B导入外部文案豆包等直接进候选池source=import"""
banned_entries = build_entries_from_db(banned_word_rows or [])
lines = raw_text.strip().splitlines()
title = lines[0].strip() if lines else ""
content = "\n".join(lines[1:]).strip() if len(lines) > 1 else raw_text.strip()
candidate: dict = {"title": title, "content": content, "tags": [], "angle": "import",
"buyingPoint": "", "coverTitle": title, "imageBrief": "", "source": "import"}
ban = check_and_fix(f"{title} {content}", banned_entries or None)
# 轨B(导入外部文案)走机械 score_copy 而非 AI 评委:导入的是用户自带成品,评分仅作
# 参考展示不卡发布;且本函数同步、改 await 会扩大到调用方。AI 评委只用于轨A生成链路。
scored = score_copy(candidate, product, [e.word for e in banned_entries])
candidate.update({"score": scored["score"], "score_detail": scored["score_detail"],
"passed": scored["passed"], "banned_word_status": ban.status})
return candidate

View File

@@ -0,0 +1,71 @@
"""
app/services/auth_service.py — 认证 service
密码哈希校验、用户查找、响应格式化。
路由层不含业务逻辑,全在此。
"""
import logging
from passlib.context import CryptContext
from sqlalchemy.orm import Session
from app.core.response import raise_unauthorized
from app.middleware.workspace_guard import CurrentUser
from app.models.user import User
from app.models.workspace import WorkspaceMember
logger = logging.getLogger(__name__)
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(plain: str) -> str:
return pwd_context.hash(plain)
def verify_password(plain: str, hashed: str) -> bool:
return pwd_context.verify(plain, hashed)
def authenticate_user(
db: Session, username: str, password: str
) -> tuple[User, int, str]:
"""
验证用户名+密码,返回 (user, workspace_id, role)。
失败抛 CloverHTTPException 40101。
"""
user = db.query(User).filter(
User.username == username, User.is_active == True
).first()
if not user or not verify_password(password, user.hashed_password):
raise_unauthorized("用户名或密码错误")
# 取用户所在的第一个 workspace手动建账号场景只有一个
member = (
db.query(WorkspaceMember)
.filter(WorkspaceMember.user_id == user.id)
.first()
)
if not member:
raise_unauthorized("用户未加入任何 workspace请联系管理员")
# 记录登录
try:
from app.models.user import LoginRecord
db.add(LoginRecord(user_id=user.id))
db.commit()
except Exception:
logger.warning("Failed to write login_record for user=%s", user.id)
db.rollback()
return user, member.workspace_id, member.role
def build_user_response(user: User, workspace_id: int, role: str) -> dict:
"""格式化用户响应体契约§4 DTO"""
return {
"id": user.id,
"username": user.username,
"email": user.email,
"current_workspace_id": workspace_id,
"role": role,
}

View File

@@ -0,0 +1,121 @@
"""
app/services/flywheel_service.py — 飞轮信号写入 + 偏好上下文聚合
preference_collector三信号入口选文案/选图/审核)写入 preference_events。
preference_aggregator查最近50条 → 最常选角度 + 打回原因近3条原文拼 prompt。
飞轮不暴露独立埋点端点,只由业务接口内部调用(契约红线)。
"""
import logging
from typing import Any
from sqlalchemy import desc, func
from sqlalchemy.orm import Session
from app.constants.enums import SIGNAL_WEIGHTS, DataOwnership, SignalType
from app.middleware.workspace_guard import CurrentUser
from app.models.flywheel import PreferenceEvent
from app.models.task import GenerationTask
logger = logging.getLogger(__name__)
# 实时聚合窗口最近50条事件
_AGGREGATION_WINDOW = 50
# 冷启动阈值不足5条信号用产品档案冷启动
_COLD_START_THRESHOLD = 5
def record_signal(
db: Session,
current_user: CurrentUser,
task: GenerationTask,
signal_type: str,
candidate_id: int | None = None,
angle_label: str | None = None,
reason: str | None = None,
) -> None:
"""
写入飞轮信号。
workspace_id + product_id 都必须有基石C + 按产品分开学)。
signal_weight 用枚举默认值,北哥可校准。
data_ownership 默认 client_data选择行为归客户
"""
weight = SIGNAL_WEIGHTS.get(signal_type, 0)
event = PreferenceEvent(
workspace_id=current_user.workspace_id,
product_id=task.product_id,
task_id=task.id,
user_id=current_user.user_id,
signal_type=signal_type,
signal_weight=weight,
candidate_id=candidate_id,
angle_label=angle_label,
reason=reason,
data_ownership=DataOwnership.CLIENT_DATA,
)
try:
db.add(event)
db.commit()
logger.info(
"Flywheel signal: type=%s weight=%s user=%s product=%s",
signal_type, weight, current_user.user_id, task.product_id,
)
except Exception:
db.rollback()
logger.error(
"Failed to write preference_event: type=%s user=%s",
signal_type, current_user.user_id,
)
raise
def get_preference_context(
db: Session, workspace_id: int, product_id: int
) -> dict[str, Any]:
"""
实时聚合偏好上下文最近50条 events
返回recent_preference摘要 + reject_reasons近3条 + injected_count。
不足5条 → 冷启动提示(产品档案兜底,由 AIE prompt 层读 products.custom_prompt
按 workspace_id + product_id 严格过滤不串数据基石C
"""
recent = (
db.query(PreferenceEvent)
.filter(
PreferenceEvent.workspace_id == workspace_id,
PreferenceEvent.product_id == product_id,
)
.order_by(desc(PreferenceEvent.created_at))
.limit(_AGGREGATION_WINDOW)
.all()
)
if len(recent) < _COLD_START_THRESHOLD:
return {
"recent_preference": "信号不足,使用产品档案基线(冷启动)",
"reject_reasons": [],
"injected_count": len(recent),
}
# 统计最常被选中的角度
angle_counts: dict[str, int] = {}
for ev in recent:
if ev.signal_type in (SignalType.TEXT_SELECT, SignalType.APPROVE) and ev.angle_label:
angle_counts[ev.angle_label] = angle_counts.get(ev.angle_label, 0) + 1
top_angles = sorted(angle_counts.items(), key=lambda x: x[1], reverse=True)[:3]
if top_angles:
pref_desc = "".join(f"{a}(已选{c}次)" for a, c in top_angles)
preference_summary = f"最近偏好:{pref_desc}"
else:
preference_summary = "暂无明显角度偏好"
# 取最近3条打回原因原文不做 AI 归纳契约§3
reject_reasons = [
ev.reason for ev in recent
if ev.signal_type == SignalType.REJECT_WITH_REASON and ev.reason
][:3]
return {
"recent_preference": preference_summary,
"reject_reasons": reject_reasons,
"injected_count": len(recent),
}

View File

@@ -0,0 +1,86 @@
"""
app/services/task_service.py — 任务创建 service
校验有无 key → 建 GenerationTask → 只推 task_id 入队,绝不传 key基石B
"""
import logging
from sqlalchemy.orm import Session
from app.core.response import raise_business
from app.middleware.workspace_guard import CurrentUser
from app.models.task import GenerationTask
from app.models.workspace import UserApiKey
logger = logging.getLogger(__name__)
def _check_user_has_key(db: Session, user_id: int, workspace_id: int) -> None:
"""校验用户在此 workspace 是否有可用 API Keyopenai/apiports均可没有则引导去配置。"""
key = (
db.query(UserApiKey)
.filter(
UserApiKey.user_id == user_id,
UserApiKey.workspace_id == workspace_id,
UserApiKey.provider.in_(["openai", "apiports"]), # G6坑修复接受主备通道名
)
.first()
)
if not key:
raise_business("尚未配置 API Key请先在设置中录入")
def create_generation_task(
db: Session,
current_user: CurrentUser,
body, # CreateTaskRequest
) -> GenerationTask:
"""
建 GenerationTask 并推 Celery 队列。
只传 task_id绝不传 key基石B
"""
if body.track == "ai":
# 轨A先检查有没有 key
_check_user_has_key(db, current_user.user_id, current_user.workspace_id)
# 禁降级铁律:本次产品入镜(need_product_image=True)时,产品必须已上传参考图,
# 否则拒绝建任务(不允许降级纯文生图,防产品包装跑偏/过抽检失败)。
need_img = getattr(body, "need_product_image", True)
if need_img:
from app.models.product import Product
product = db.query(Product).filter(
Product.id == body.product_id,
Product.workspace_id == current_user.workspace_id,
).first()
if not product:
raise_business("产品不存在")
if not (product.image_path or "").strip():
raise_business("该产品未上传参考图,无法生成产品入镜内容;请先到产品库上传产品图,或关闭「产品入镜」开关")
task = GenerationTask(
workspace_id=current_user.workspace_id,
product_id=body.product_id,
operator_id=current_user.user_id,
theme=body.theme,
text_count=body.text_count,
image_count=body.image_count,
track=body.track,
need_product_image=need_img,
status="pending",
)
db.add(task)
db.commit()
db.refresh(task)
logger.info("GenerationTask created: id=%s ws=%s", task.id, current_user.workspace_id)
if body.track == "ai":
enqueue_generation(task.id)
return task
def enqueue_generation(task_id: int) -> None:
"""只推 task_id 入队,绝不推 key基石B"""
from app.workers.tasks import run_generation_pipeline
run_generation_pipeline.delay(task_id)
logger.info("Enqueued task_id=%s", task_id)