df1856d793
- 产品编辑入口统一走 ProductFormFull(卖点/风格/人群/品牌词全字段); 修复开任务页 <form> 套 <form> 致"编辑产品"报错、改不了、跳回首个产品 - dashboard 入口卡片对齐实际路由: 系统管理(/config) 与 工作配置(/settings) 分开; settings ?tab=products 直达改用挂载后读 URL, 消除 hydration mismatch - 新增用户管理(users API/admin service/改密页) + alembic 022/023/024 - 上线部署: Dockerfile / docker-compose.prod+https / nginx https / .env.example - A8 三套正交叙事(痛点/场景/成分背书) + beige 调色去AI化 + 飞轮 text_import 高权重信号 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
109 lines
4.3 KiB
Python
109 lines
4.3 KiB
Python
"""
|
||
app/workers/replenish_task.py — 文案后台补充任务（S1: 先展示后台补铁律）
|
||
|
||
合格文案(passed且score>=QUALITY_PASS_SCORE(80,红线)且非hard_block)不足用户目标条数时,
|
||
run_generation_pipeline 已先展示合格的; 此任务异步补齐到目标数或达上限。
|
||
只接收 task_id(基石B)。复用 pipeline_io.run_text_generation 的生成+过滤。
|
||
"""
|
||
|
||
import logging
|
||
|
||
from app.workers.celery_app import celery_app
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 补充上限：避免合格率低时无限补。最多补 MAX_REPLENISH_ROUNDS 轮。
|
||
MAX_REPLENISH_ROUNDS = 3
|
||
|
||
|
||
@celery_app.task(
|
||
bind=True,
|
||
name="app.workers.tasks.replenish_text_candidates",
|
||
max_retries=2,
|
||
default_retry_delay=30,
|
||
queue="generation",
|
||
)
|
||
def replenish_text_candidates(self, task_id: int) -> dict:
|
||
"""后台补充合格文案到目标条数（或达补充上限）。
|
||
|
||
每轮调用 run_text_generation 续生成,过滤后累加入库;
|
||
已合格数 >= text_count 即停; 达 MAX_REPLENISH_ROUNDS 仍不足也停(不卡用户)。
|
||
"""
|
||
from app.workers.tasks import _get_db, _push_event_sync
|
||
from app.workers.pipeline_io import run_text_generation
|
||
from app.workers.pipeline_steps import (
|
||
load_task_and_product,
|
||
decrypt_user_key,
|
||
decrypt_codeproxy_key,
|
||
build_clients_and_clear_key,
|
||
build_product_dict,
|
||
load_flywheel_context,
|
||
)
|
||
from app.models.task import TextCandidate
|
||
|
||
db = _get_db()
|
||
try:
|
||
task, product = load_task_and_product(db, task_id)
|
||
if not task:
|
||
logger.warning("补充任务找不到 task_id=%s", task_id)
|
||
return {"task_id": task_id, "replenished": 0, "reason": "task_not_found"}
|
||
|
||
# 已合格数（按入库的合格候选计）
|
||
existing = db.query(TextCandidate).filter(
|
||
TextCandidate.task_id == task_id
|
||
).count()
|
||
target = task.text_count
|
||
if existing >= target:
|
||
return {"task_id": task_id, "replenished": 0, "reason": "already_enough"}
|
||
|
||
workspace_id = task.workspace_id
|
||
plain_key = decrypt_user_key(db, task.operator_id, workspace_id)
|
||
alt_key = decrypt_codeproxy_key(db, task.operator_id, workspace_id)
|
||
clients = build_clients_and_clear_key(plain_key, alt_key)
|
||
plain_key = None # 用完即清除，基石B
|
||
alt_key = None
|
||
product_dict = build_product_dict(product)
|
||
flywheel_fragment, _ = load_flywheel_context(
|
||
db, workspace_id, task.product_id, product_dict
|
||
)
|
||
|
||
seq = 9000 # 补充事件 seq 段，避开主流程
|
||
rounds = 0
|
||
# 真实入库合格数以 DB count 为准，绝不用 len(raw)(那是原始生成数含被过滤的低分条)。
|
||
# task47 实测踩坑：len(raw)=2 但 saved=0(都没过90),误判达标提前停,库里仍缺。
|
||
current = existing
|
||
while current < target and rounds < MAX_REPLENISH_ROUNDS:
|
||
rounds += 1
|
||
_, _, _, fail_reason, _ = run_text_generation(
|
||
db, clients, task, product_dict, flywheel_fragment,
|
||
_push_event_sync, workspace_id, seq,
|
||
)
|
||
# run_text_generation 内部已把本轮合格候选写库,这里重查真实库内条数
|
||
db.expire_all()
|
||
current = db.query(TextCandidate).filter(
|
||
TextCandidate.task_id == task_id
|
||
).count()
|
||
seq += 100
|
||
logger.info(
|
||
"补充轮 %d: task_id=%s 库内合格=%d/%d",
|
||
rounds, task_id, current, target,
|
||
)
|
||
# 评分通道不可用时再补也是空烧 token(每轮还各 35s backoff)——立即停,
|
||
# 等通道恢复用户手动重试。区别于"质量不达标"(那个值得多补几轮)。
|
||
if fail_reason == "scoring_unavailable":
|
||
logger.error("补充中止: task_id=%s 评分通道不可用,停止空跑重试", task_id)
|
||
break
|
||
|
||
return {
|
||
"task_id": task_id,
|
||
"replenished": current - existing,
|
||
"rounds": rounds,
|
||
"final": current,
|
||
"target": target,
|
||
}
|
||
except Exception as exc:
|
||
logger.error("replenish_text_candidates failed: task_id=%s err=%s", task_id, exc)
|
||
raise self.retry(exc=exc)
|
||
finally:
|
||
db.close()
|