beige/backend/app/api/v1/task_actions.py

"""
app/api/v1/task_actions.py — 任务操作端点
选文案 / 选图 / 导入文案 / 重新生成 / 提交审核 / 偏好上下文
DTOs 和辅助函数从 tasks.py 导入。
"""

import logging
from typing import Annotated

from fastapi import APIRouter, Depends
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.core.response import ok, raise_not_found, raise_state_invalid
from app.middleware.workspace_guard import CurrentUser, require_write_permission, require_admin
from app.models.task import GenerationTask, ImageCandidate, TextCandidate

from app.api.v1.tasks import (
    SelectCandidateRequest,
    ImportTextRequest,
    _fmt_text,
    _fmt_image,
    _check_task_ownership,
)

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/tasks", tags=["tasks"])


@router.post("/{task_id}/text-candidates/select")
def select_text(
    task_id: int, body: SelectCandidateRequest,
    current_user: Annotated[CurrentUser, Depends(require_write_permission)] = None,
    db: Session = Depends(get_db),
):
    """选文案（飞轮信号 text_select +3）。"""
    from app.services.flywheel_service import record_signal
    task = _check_task_ownership(
        db.query(GenerationTask).filter(GenerationTask.id == task_id).first(),
        current_user.workspace_id,
    )
    tc = db.query(TextCandidate).filter(
        TextCandidate.id == body.candidate_id, TextCandidate.task_id == task_id
    ).first()
    if not tc:
        raise_not_found("文案候选不存在")
    # A/B/C 同套单选：避免旧选择累积导致审核/打包拿错套别。
    db.query(TextCandidate).filter(
        TextCandidate.task_id == task_id,
        TextCandidate.strategy == tc.strategy,
        TextCandidate.id != tc.id,
    ).update({"is_selected": False})
    tc.is_selected = True
    db.commit()
    record_signal(db, current_user, task, "text_select", candidate_id=tc.id, angle_label=tc.angle_label)
    return ok({"selected": body.candidate_id})


@router.post("/{task_id}/image-candidates/select")
def select_image(
    task_id: int, body: SelectCandidateRequest,
    current_user: Annotated[CurrentUser, Depends(require_write_permission)] = None,
    db: Session = Depends(get_db),
):
    """选图（飞轮信号 image_select +3）。"""
    import json
    from app.services.flywheel_service import record_signal
    from app.constants.enums import IMAGE_STRATEGY_ANGLE
    task = _check_task_ownership(
        db.query(GenerationTask).filter(GenerationTask.id == task_id).first(),
        current_user.workspace_id,
    )
    ic = db.query(ImageCandidate).filter(
        ImageCandidate.id == body.candidate_id, ImageCandidate.task_id == task_id
    ).first()
    if not ic:
        raise_not_found("图片候选不存在")
    # 同套同角色只保留一张已选图；重生后选择新图会自动取消旧图。
    db.query(ImageCandidate).filter(
        ImageCandidate.task_id == task_id,
        ImageCandidate.strategy == ic.strategy,
        ImageCandidate.role == ic.role,
        ImageCandidate.id != ic.id,
    ).update({"is_selected": False})
    ic.is_selected = True
    db.commit()
    # R7 断点2：选图带 strategy → 映射叙事角度标签进飞轮，与文案 angle_label 并轨；
    # strategy 原值另存 signal_meta，便于后续按套别复盘。angle_label=None 不再空转。
    angle = IMAGE_STRATEGY_ANGLE.get(ic.strategy or "")
    meta = json.dumps({"strategy": ic.strategy, "role": ic.role}, ensure_ascii=False) if ic.strategy else None
    record_signal(
        db, current_user, task, "image_select",
        candidate_id=ic.id, angle_label=angle, signal_meta=meta,
    )
    return ok({"selected": body.candidate_id})


@router.post("/{task_id}/import-text")
def import_text(
    task_id: int, body: ImportTextRequest,
    current_user: Annotated[CurrentUser, Depends(require_write_permission)] = None,
    db: Session = Depends(get_db),
):
    """轨B：导入外部文案直接进候选池，跳过 AI 生成（洞2）。
    按已导入条数 %3 轮转分配 strategy(A/B/C)，让导入轨「去生图」时能出三套正交配图
    (生图按 strategy 匹配文案，NULL 套永远匹配不上)。
    导入=客户实跑验证过的范本,记 text_import 飞轮信号(权重高于改稿/AI选稿,倩倩姐2026-06-26)。"""
    from app.constants.enums import IMAGE_STRATEGY_ANGLE
    from app.services.flywheel_service import record_signal
    task = _check_task_ownership(
        db.query(GenerationTask).filter(GenerationTask.id == task_id).first(),
        current_user.workspace_id,
    )
    # 已导入条数 %3 → A/B/C 轮转：第1条A 第2条B 第3条C 第4条A…，三套均匀分布
    existing = db.query(TextCandidate).filter(TextCandidate.task_id == task_id).count()
    strategy = ("A", "B", "C")[existing % 3]
    tc = TextCandidate(
        workspace_id=current_user.workspace_id, task_id=task_id,
        source="import", content=body.content, angle_label=body.angle_label,
        strategy=strategy,
    )
    db.add(tc)
    db.commit()
    db.refresh(tc)
    # 飞轮信号：导入文案=最值得学习的范本。角度优先用户填的,缺则按套别兜底,让偏好聚合学到角度。
    # 飞轮辅助不阻塞主路径:记信号失败只 log,导入文案(tc 已 commit)照常返回。
    try:
        record_signal(
            db, current_user, task, "text_import",
            candidate_id=tc.id,
            angle_label=body.angle_label or IMAGE_STRATEGY_ANGLE.get(strategy),
        )
    except Exception:
        logger.warning(
            "导入文案飞轮信号记录失败(不阻塞导入): task_id=%s candidate_id=%s",
            task_id, tc.id,
        )
    return ok(_fmt_text(tc))


class RegenerateRequest(BaseModel):
    """R2 重生参数（全 None=整体重生，兼容旧调用）。
    strategy='A'/'B'/'C' 指定单套；role 指定单张(需配 strategy)；custom_prompt 人工追加提示词。"""
    strategy: str | None = Field(default=None, pattern="^[ABC]$")
    role: str | None = None
    custom_prompt: str | None = Field(default=None, max_length=500)


@router.post("/{task_id}/regenerate")
def regenerate(
    task_id: int,
    body: RegenerateRequest | None = None,
    current_user: Annotated[CurrentUser, Depends(require_write_permission)] = None,
    db: Session = Depends(get_db),
):
    """重新生成（飞轮信号 regenerate -1）。
    body 全空=整体重生；带 strategy=单套重生；带 strategy+role=单张重生；custom_prompt=人工提示词。"""
    from app.services.flywheel_service import record_signal
    from app.services.task_service import enqueue_generation
    task = _check_task_ownership(
        db.query(GenerationTask).filter(GenerationTask.id == task_id).first(),
        current_user.workspace_id,
    )
    body = body or RegenerateRequest()
    if str(task.status) == "generating" or getattr(task.status, "value", None) == "generating":
        raise_state_invalid("任务正在生成中，请等待本轮完成后再重生")
    import redis as redis_lib
    from app.core.config import get_settings

    try:
        r = redis_lib.from_url(get_settings().REDIS_URL, decode_responses=True)
    except Exception as exc:
        logger.warning("检查重生锁失败，继续提交: task_id=%s err=%s", task_id, exc)
    else:
        if r.exists(f"pipeline:lock:{task_id}"):
            raise_state_invalid("任务正在重生中，请等待本轮完成后再操作")
    # role 必须配 strategy（单张重生需知道是哪套的哪张）
    if body.role and not body.strategy:
        raise_state_invalid("指定单张重生(role)时必须同时指定 strategy(套别)")
    record_signal(db, current_user, task, "regenerate")
    enqueue_generation(task.id, regen_strategy=body.strategy,
                       regen_role=body.role, custom_prompt=body.custom_prompt)
    scope = "单张" if body.role else ("单套" if body.strategy else "整体")
    return ok({"task_id": task_id, "status": "regenerating", "scope": scope})


@router.post("/{task_id}/generate-images")
def generate_images(
    task_id: int,
    current_user: Annotated[CurrentUser, Depends(require_write_permission)] = None,
    db: Session = Depends(get_db),
):
    """导入轨「去生图」：复用库内导入文案，跳过文案生成，只生图（首次出图）。
    复用 regenerate 的 generating 守卫 + redis 锁防重复烧钱。"""
    from app.services.task_service import enqueue_generation
    task = _check_task_ownership(
        db.query(GenerationTask).filter(GenerationTask.id == task_id).first(),
        current_user.workspace_id,
    )
    if str(task.status) == "generating" or getattr(task.status, "value", None) == "generating":
        raise_state_invalid("任务正在生成中，请等待本轮完成")
    # 必须库内有导入文案才放行生图，避免空语境生图（质量崩）
    has_text = db.query(TextCandidate).filter(TextCandidate.task_id == task_id).count()
    if not has_text:
        raise_state_invalid("无可用文案，请先导入文案再去生图")
    import redis as redis_lib
    from app.core.config import get_settings

    try:
        r = redis_lib.from_url(get_settings().REDIS_URL, decode_responses=True)
    except Exception as exc:
        logger.warning("检查生图锁失败，继续提交: task_id=%s err=%s", task_id, exc)
    else:
        if r.exists(f"pipeline:lock:{task_id}"):
            raise_state_invalid("任务正在生成中，请等待本轮完成")
    enqueue_generation(task.id, reuse_text=True)
    return ok({"task_id": task_id, "status": "generating"})


@router.post("/{task_id}/submit-review")
def submit_review(
    task_id: int,
    current_user: Annotated[CurrentUser, Depends(require_write_permission)] = None,
    db: Session = Depends(get_db),
):
    """提交审核（状态 pending_selection → pending_review）。"""
    task = _check_task_ownership(
        db.query(GenerationTask).filter(GenerationTask.id == task_id).first(),
        current_user.workspace_id,
    )
    if task.status != "pending_selection":
        raise_state_invalid("只有 pending_selection 状态可提审")
    _validate_strategy_selection(db, task)
    task.status = "pending_review"
    db.commit()
    return ok({"task_id": task_id, "status": "pending_review"})


def _validate_strategy_selection(db: Session, task: GenerationTask) -> None:
    """提审前最低门槛：至少 1 条已选文案 + 至少 1 张已选图。

    三套 A/B/C 是「挑着选」——用户可只要其中一套、丢弃某套、或某套重新生成，
    不强制三套齐全、不强制每套选满（倩倩姐 2026-06-26 拍板）。
    """
    from app.core.response import raise_business

    selected_text_count = (
        db.query(TextCandidate.id)
        .filter(TextCandidate.task_id == task.id, TextCandidate.is_selected == True)
        .count()
    )
    selected_image_count = (
        db.query(ImageCandidate.id)
        .filter(ImageCandidate.task_id == task.id, ImageCandidate.is_selected == True)
        .count()
    )

    parts = []
    if selected_text_count < 1:
        parts.append("请至少选择 1 条文案")
    if selected_image_count < 1:
        parts.append("请至少选择 1 张图片")
    if parts:
        raise_business("；".join(parts))


@router.get("/{task_id}/preference/context")
def get_preference_context(
    task_id: int,
    current_user: Annotated[CurrentUser, Depends(require_write_permission)] = None,
    db: Session = Depends(get_db),
):
    """取本任务飞轮偏好上下文（最近选中样本+打回原因）。"""
    task = _check_task_ownership(
        db.query(GenerationTask).filter(GenerationTask.id == task_id).first(),
        current_user.workspace_id,
    )
    from app.services.flywheel_service import get_preference_context
    # 取产品档案供冷启动基线（与生产链同口径）
    from app.models.product import Product
    _p = db.query(Product).filter(Product.id == task.product_id).first()
    product_dict = {
        "custom_prompt": getattr(_p, "custom_prompt", "") or "",
        "style_tone": getattr(_p, "style_tone", "") or "",
    } if _p else {}
    ctx = get_preference_context(db, current_user.workspace_id, task.product_id, product_dict)
    return ok(ctx)


class EditTextRequest(BaseModel):
    """改稿请求：candidate_id + 用户改后的新文案。"""
    candidate_id: int
    content: str = Field(min_length=1, max_length=5000)


@router.post("/{task_id}/text-candidates/{candidate_id}/edit")
def edit_text(
    task_id: int, candidate_id: int, body: EditTextRequest,
    current_user: Annotated[CurrentUser, Depends(require_write_permission)] = None,
    db: Session = Depends(get_db),
):
    """改稿（飞轮信号 text_edit +5，最强真实信号）。

    用户真改了文案的字=最在乎这条方向。回写 content + edited=True，
    并写 text_edit 飞轮偏好事件。倩倩姐2026-06-16拍板，先激进入飞轮跑一周再校准。
    """
    from app.services.flywheel_service import record_signal
    task = _check_task_ownership(
        db.query(GenerationTask).filter(GenerationTask.id == task_id).first(),
        current_user.workspace_id,
    )
    tc = db.query(TextCandidate).filter(
        TextCandidate.id == candidate_id, TextCandidate.task_id == task_id
    ).first()
    if not tc:
        raise_not_found("文案候选不存在")
    tc.content = body.content
    tc.edited = True
    db.commit()
    record_signal(db, current_user, task, "text_edit", candidate_id=tc.id, angle_label=tc.angle_label)
    return ok({"candidate_id": candidate_id, "edited": True})


@router.delete("/{task_id}/text-candidates/{candidate_id}")
def delete_text_candidate(
    task_id: int, candidate_id: int,
    current_user: Annotated[CurrentUser, Depends(require_admin)] = None,
    db: Session = Depends(get_db),
):
    """删除单条文案候选（仅管理员，物理删；倩倩姐2026-06-30拍板：删除权限只给管理员）。"""
    _check_task_ownership(
        db.query(GenerationTask).filter(GenerationTask.id == task_id).first(),
        current_user.workspace_id,
    )
    tc = db.query(TextCandidate).filter(
        TextCandidate.id == candidate_id, TextCandidate.task_id == task_id
    ).first()
    if not tc:
        raise_not_found("文案候选不存在")
    db.delete(tc)
    db.commit()
    return ok({"deleted": candidate_id})


@router.delete("/{task_id}/image-candidates/{candidate_id}")
def delete_image_candidate(
    task_id: int, candidate_id: int,
    current_user: Annotated[CurrentUser, Depends(require_admin)] = None,
    db: Session = Depends(get_db),
):
    """删除单张图片候选（仅管理员，物理删；倩倩姐2026-06-30拍板：删除权限只给管理员）。"""
    _check_task_ownership(
        db.query(GenerationTask).filter(GenerationTask.id == task_id).first(),
        current_user.workspace_id,
    )
    ic = db.query(ImageCandidate).filter(
        ImageCandidate.id == candidate_id, ImageCandidate.task_id == task_id
    ).first()
    if not ic:
        raise_not_found("图片候选不存在")
    db.delete(ic)
    db.commit()
    return ok({"deleted": candidate_id})