"""上传视频提炼 —— 参考视频 → 可人工逐镜编辑的中文分镜稿。

优先把**完整视频(含音轨)**内联给 Gemini 3.1 Pro,跟官网「直接上传视频」同一条能力:
模型能看全片、听口播,不会只拿到稀疏静帧。文件太大塞不进请求时,才退回 ffmpeg 抽帧。

拆视频必须会看视频/图。默认语言模型现在可能是纯文本豆包,不能用 get_default_model(TEXT)。
固定钉 Gemini 3.1 Pro 官转(``gemini-3.1-pro-preview``,展示名带「官转」优先)。
"""

from __future__ import annotations

import base64
import json
import logging
import math
import re
import shutil
import subprocess
import tempfile
import uuid
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path

logger = logging.getLogger(__name__)

from django.conf import settings

# 上传限制:超了直接 400,不进 ffmpeg,也不花模型钱
ALLOWED_SUFFIXES = (".mp4", ".mov", ".m4v", ".webm")
MAX_UPLOAD_BYTES = 200 * 1024 * 1024  # 200 MB
MAX_DURATION_SECONDS = 180  # 3 分钟。带货参考片远短于此;更长的帧采样密度不够,拆出来也是错的

# 官网直接传视频的上限大约是请求 20MB;base64 会胀到 4/3,所以原文件卡在 15MB。
INLINE_VIDEO_MAX_BYTES = 15 * 1024 * 1024
_SUFFIX_MIME = {
    ".mp4": "video/mp4",
    ".m4v": "video/mp4",
    ".mov": "video/quicktime",
    ".webm": "video/webm",
}

# 抽帧只在整段视频塞不进请求时启用。
SECONDS_PER_FRAME = 2
MIN_FRAMES = 8
MAX_FRAMES = 36
FRAME_WIDTH = 768
FRAME_QUALITY = 3
DIGEST_MAX_TOKENS = 12288
_SHOT_MARK = re.compile(r"【(?:镜头\s*\d+|第\s*\d+\s*镜)】")

_FFMPEG_TIMEOUT = 60


class VideoDigestError(ValueError):
    """用户可见的失败(文件不合格 / ffmpeg 读不动),一律 400。"""


@dataclass(frozen=True)
class VideoFrame:
    at_seconds: int
    jpeg: bytes

    def as_data_url(self) -> str:
        return "data:image/jpeg;base64," + base64.b64encode(self.jpeg).decode("ascii")


@dataclass(frozen=True)
class DigestVideo:
    mime: str
    data: bytes

    def as_data_url(self) -> str:
        return f"data:{self.mime};base64," + base64.b64encode(self.data).decode("ascii")


# --------------------------------------------------------------------------- #
# skill 加载
# --------------------------------------------------------------------------- #
def _skill_dir() -> Path:
    """与 script_agent._skill_dir 同源:优先 BASE_DIR/skills(镜像内),回落仓库根(本地旧布局)。"""
    base = Path(settings.BASE_DIR)
    for cand in (base / "skills", base.parent.parent / "skills"):
        if (cand / "video-shot-digest").is_dir():
            return cand / "video-shot-digest"
    return base / "skills" / "video-shot-digest"


@lru_cache(maxsize=1)
def load_digest_skill() -> str:
    main = _skill_dir() / "SKILL.md"
    if main.exists():
        return main.read_text(encoding="utf-8")
    # 兜底:skill 丢了也别整条链路挂掉,退化成一句话提示词(产出会明显变差,交接文档已注明须带 skills 目录)
    return (
        "你是分镜拆解 agent。输入是一条短视频的完整文件,含画面和音轨。"
        "必须覆盖全片,从 00:00 写到片尾,有几镜写几镜,不要概括成几大段。"
        "每镜按【镜头 01】写出时间、时长、景别、机位、运镜、画面、人物动作、人物表情、"
        "台词/旁白、音效、背景音乐、字幕、备注。没有就写无。输出中文纯文本。"
    )


# --------------------------------------------------------------------------- #
# ffmpeg:探时长 + 抽帧
# --------------------------------------------------------------------------- #
def _binary(name: str) -> str:
    found = shutil.which(name)
    if not found:
        raise VideoDigestError("服务器暂时无法解析视频,请稍后再试")
    return found


def probe_duration(path: str | Path) -> float:
    """ffprobe 读时长(秒)。读不到 = 不是能解的视频。"""
    try:
        out = subprocess.run(
            [
                _binary("ffprobe"), "-v", "error",
                "-print_format", "json", "-show_format",
                str(path),
            ],
            capture_output=True, timeout=_FFMPEG_TIMEOUT, check=True,
        ).stdout
        duration = float(json.loads(out)["format"]["duration"])
    except VideoDigestError:
        raise
    except Exception as exc:  # noqa: BLE001 — ffprobe 各种失败对用户是同一件事
        raise VideoDigestError("这个视频读不出来,请换一个 mp4 / mov 文件") from exc
    if duration <= 0:
        raise VideoDigestError("这个视频读不出来,请换一个 mp4 / mov 文件")
    return duration


def probe_video_size(path: str | Path) -> tuple[int, int]:
    """ffprobe 读画面宽高。读不到返回 0,0,不挡拆解。"""
    try:
        out = subprocess.run(
            [
                _binary("ffprobe"), "-v", "error",
                "-select_streams", "v:0",
                "-show_entries", "stream=width,height",
                "-of", "csv=p=0:s=x",
                str(path),
            ],
            capture_output=True, timeout=_FFMPEG_TIMEOUT, check=True,
        ).stdout.decode("utf-8", errors="replace").strip()
        width_s, height_s = out.split("x", 1)
        return max(0, int(width_s)), max(0, int(height_s))
    except Exception:  # noqa: BLE001
        return 0, 0


def extract_cover_jpeg(path: str | Path, duration: float) -> bytes:
    """抽一帧作历史封面。失败返回空字节,不挡拆解。"""
    try:
        ffmpeg = _binary("ffmpeg")
    except VideoDigestError:
        return b""
    at = max(0.0, min(float(duration) * 0.35, max(0.0, float(duration) - 0.15)))
    try:
        done = subprocess.run(
            [
                ffmpeg, "-v", "error", "-ss", f"{at:.2f}", "-i", str(path),
                "-frames:v", "1", "-vf", "scale=640:-2",
                "-q:v", "4", "-f", "image2", "-",
            ],
            capture_output=True, timeout=_FFMPEG_TIMEOUT, check=True,
        )
    except Exception:  # noqa: BLE001
        return b""
    return done.stdout or b""


def ratio_label(width: int, height: int) -> str:
    if not width or not height:
        return ""
    ratio = width / height
    if abs(ratio - 9 / 16) < 0.08:
        return "9:16 竖屏"
    if abs(ratio - 16 / 9) < 0.08:
        return "16:9 横屏"
    if abs(ratio - 1) < 0.08:
        return "1:1"
    return "横屏" if width > height else "竖屏"


def title_from_filename(name: str) -> str:
    stem = Path(name or "").stem.strip()
    return (stem or "参考视频")[:80]


def duration_clock(seconds: float) -> str:
    total = max(0, int(round(float(seconds or 0))))
    return f"{total // 60:02d}:{total % 60:02d}"


def shot_count(text: str) -> int:
    return len(_SHOT_MARK.findall(text or ""))


def plan_frame_times(duration: float) -> list[int]:
    """均匀采样时间点。取每段的**中点**,避开首尾黑场与片尾卡片。"""
    count = max(MIN_FRAMES, min(MAX_FRAMES, math.ceil(duration / SECONDS_PER_FRAME)))
    step = duration / count
    return [int(step * (i + 0.5)) for i in range(count)]


def extract_frames(path: str | Path, times: list[int]) -> list[VideoFrame]:
    """逐时间点抽一帧。``-ss`` 放在 ``-i`` 前走关键帧快速定位,每帧约几十毫秒。"""
    ffmpeg = _binary("ffmpeg")
    frames: list[VideoFrame] = []
    for at in times:
        try:
            done = subprocess.run(
                [
                    ffmpeg, "-v", "error", "-ss", str(at), "-i", str(path),
                    "-frames:v", "1", "-vf", f"scale={FRAME_WIDTH}:-2",
                    "-q:v", str(FRAME_QUALITY), "-f", "image2", "-",
                ],
                capture_output=True, timeout=_FFMPEG_TIMEOUT, check=True,
            )
        except Exception:  # noqa: BLE001 — 单帧抽失败(定位越界等)跳过,别拖垮整次提炼
            continue
        if done.stdout:
            frames.append(VideoFrame(at_seconds=at, jpeg=done.stdout))
    if not frames:
        raise VideoDigestError("没能从这个视频里取到画面,请换一个文件")
    return frames


def _write_upload(upload) -> tuple[str, str, int]:
    """校验后缀和体积,把上传落到临时文件。返回 (path, suffix, size)。调用方负责删除。"""
    name = (getattr(upload, "name", "") or "").lower()
    if not name.endswith(ALLOWED_SUFFIXES):
        raise VideoDigestError("只支持 mp4 / mov / m4v / webm 四种视频格式")
    size = getattr(upload, "size", 0) or 0
    if size > MAX_UPLOAD_BYTES:
        raise VideoDigestError(f"视频不能超过 {MAX_UPLOAD_BYTES // 1024 // 1024} MB,请压缩后再传")
    suffix = Path(name).suffix or ".mp4"
    tmp = tempfile.NamedTemporaryFile(suffix=suffix, delete=False)
    try:
        for chunk in upload.chunks():
            tmp.write(chunk)
        tmp.flush()
    finally:
        tmp.close()
    return tmp.name, suffix, Path(tmp.name).stat().st_size


def _materialize_upload(upload) -> tuple[str, str, int, float]:
    """校验 → 落盘 → 探时长。返回 (path, suffix, size, duration),调用方负责删文件。"""
    path, suffix, size = _write_upload(upload)
    try:
        duration = probe_duration(path)
    except Exception:
        Path(path).unlink(missing_ok=True)
        raise
    if duration > MAX_DURATION_SECONDS:
        Path(path).unlink(missing_ok=True)
        raise VideoDigestError(
            f"视频不能超过 {MAX_DURATION_SECONDS // 60} 分钟,请剪出要参考的那一段再传"
        )
    return path, suffix, size, duration


def _compress_video(path: str) -> bytes | None:
    """压到能内联的体积。失败返回 None,由调用方改抽帧。"""
    ffmpeg = _binary("ffmpeg")
    out = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
    out.close()
    try:
        done = subprocess.run(
            [
                ffmpeg, "-v", "error", "-y", "-i", path,
                "-c:v", "libx264", "-preset", "veryfast", "-crf", "28",
                "-vf", "scale='min(1280,iw)':-2",
                "-c:a", "aac", "-b:a", "64k",
                "-movflags", "+faststart", out.name,
            ],
            capture_output=True, timeout=_FFMPEG_TIMEOUT * 3,
        )
        if done.returncode != 0:
            return None
        data = Path(out.name).read_bytes()
        if not data or len(data) > INLINE_VIDEO_MAX_BYTES:
            return None
        return data
    except Exception:  # noqa: BLE001 — 压缩失败就抽帧,别挡住提炼
        return None
    finally:
        Path(out.name).unlink(missing_ok=True)


def _is_browser_playable(path: str) -> bool:
    """Chrome 播不了微信常见的 HEVC。H.264 + AAC/无音轨才直接存。"""
    try:
        out = subprocess.run(
            [
                _binary("ffprobe"), "-v", "error",
                "-show_entries", "stream=codec_name,codec_type",
                "-of", "json",
                str(path),
            ],
            capture_output=True, timeout=_FFMPEG_TIMEOUT, check=True,
        )
        streams = json.loads(out.stdout).get("streams") or []
    except Exception:  # noqa: BLE001
        return False
    video_ok = False
    audio_ok = True
    for stream in streams:
        kind = stream.get("codec_type")
        codec = str(stream.get("codec_name") or "").lower()
        if kind == "video":
            video_ok = codec in {"h264", "vp8", "vp9", "av1"}
        elif kind == "audio":
            audio_ok = codec in {"aac", "mp3", "opus", "vorbis"}
    return video_ok and audio_ok


def _prepare_browser_video(path: str, suffix: str) -> tuple[str, str, bool]:
    """转成浏览器能播的 H.264 AAC。失败退回原片。第三项表示调用方要删临时文件。"""
    ext = suffix if str(suffix).startswith(".") else f".{suffix or 'mp4'}"
    if _is_browser_playable(path):
        return path, ext, False
    ffmpeg = _binary("ffmpeg")
    out = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
    out.close()
    try:
        done = subprocess.run(
            [
                ffmpeg, "-v", "error", "-y", "-i", path,
                "-c:v", "libx264", "-preset", "veryfast", "-crf", "23",
                "-c:a", "aac", "-b:a", "128k",
                "-movflags", "+faststart", out.name,
            ],
            capture_output=True, timeout=_FFMPEG_TIMEOUT * 3,
        )
        if done.returncode != 0 or not Path(out.name).is_file() or Path(out.name).stat().st_size <= 0:
            Path(out.name).unlink(missing_ok=True)
            return path, ext, False
        return out.name, ".mp4", True
    except Exception:  # noqa: BLE001 — 转码失败仍存原片,总比历史卡不能播好
        Path(out.name).unlink(missing_ok=True)
        return path, ext, False


def _native_video(path: str, size: int, suffix: str) -> DigestVideo | None:
    mime = _SUFFIX_MIME.get(suffix.lower(), "video/mp4")
    if size <= INLINE_VIDEO_MAX_BYTES:
        return DigestVideo(mime=mime, data=Path(path).read_bytes())
    compressed = _compress_video(path)
    if compressed:
        return DigestVideo(mime="video/mp4", data=compressed)
    return None


def digest_input_from_upload(
    upload,
    *,
    keep_source: bool = False,
) -> tuple[DigestVideo | None, list[VideoFrame], float, dict]:
    """优先整段视频(含音轨);塞不进请求才抽帧。顺带抽出历史卡要用的封面/宽高。

    keep_source=True 时不删临时文件,调用方上传原片后再删。
    """
    path = ""
    keep = False
    original_name = Path(getattr(upload, "name", "") or "参考视频.mp4").name or "参考视频.mp4"
    try:
        path, suffix, size, duration = _materialize_upload(upload)
        width, height = probe_video_size(path)
        cover_jpeg = extract_cover_jpeg(path, duration)
        video = _native_video(path, size, suffix)
        frames = [] if video is not None else extract_frames(path, plan_frame_times(duration))
        extras = {
            "file_name": original_name,
            "file_size": size,
            "width": width,
            "height": height,
            "cover_jpeg": cover_jpeg,
            "suffix": suffix,
            "source_path": path if keep_source else "",
        }
        keep = keep_source
        return video, frames, duration, extras
    finally:
        if path and not keep:
            Path(path).unlink(missing_ok=True)


def frames_from_upload(upload) -> tuple[list[VideoFrame], float]:
    """校验上传文件 → 落临时盘 → 探时长 → 抽帧。单测与抽帧兜底用。"""
    path = ""
    try:
        path, _suffix, _size, duration = _materialize_upload(upload)
        return extract_frames(path, plan_frame_times(duration)), duration
    finally:
        if path:
            Path(path).unlink(missing_ok=True)


# --------------------------------------------------------------------------- #
# 组装多模态消息
# --------------------------------------------------------------------------- #
def build_digest_messages(
    frames: list[VideoFrame] | None = None,
    duration: float = 0,
    *,
    product_hint: str = "",
    video: DigestVideo | None = None,
    aspect_ratio: str = "",
    file_title: str = "",
) -> list[dict]:
    """system = 拆解 skill;user = 完整视频(优先)或抽帧。"""
    frames = frames or []
    measured = f"实测时长约 {round(duration)} 秒"
    if aspect_ratio:
        measured += f",画面比例 {aspect_ratio}"
    title_hint = f"文件名可参考片名:《{file_title}》。" if file_title else ""
    if video is not None:
        head = [
            f"这是一条{measured}的短视频**完整文件**(含画面和口播音轨)。{title_hint}",
            "请按技能还原全片导演分镜稿:从 00:00 写到片尾,有几镜写几镜。",
            "每镜必须写齐时间、时长、景别、机位、运镜、画面、人物动作、人物表情、台词/旁白、音效、背景音乐、字幕、备注。",
            "台词/旁白按听到的口播逐字写;画面上的花字写进字幕。",
        ]
    else:
        head = [
            f"这是一条{measured}的短视频,",
            f"按时间顺序均匀抽了 {len(frames)} 帧。每帧图前面标了它在原片中的时间点。{title_hint}",
            "请按技能还原**全片**导演分镜稿:从 00:00 写到片尾,有几镜写几镜。",
            "每镜必须写齐时间、时长、景别、机位、运镜、画面、人物动作、人物表情、台词/旁白、音效、背景音乐、字幕、备注。",
        ]
    if product_hint:
        head.append(f"用户接下来想用这条片子的结构去拍自己的商品:{product_hint}。")

    content: list[dict] = [{"type": "text", "text": "".join(head)}]
    if video is not None:
        data_url = video.as_data_url()
        # 官转(New-API)把 OpenAI image_url 的 data URI 转成 Gemini inline_data,
        # mime 从 data:video/mp4 头读取,模型按整段视频+音轨理解,等同官网直接上传。
        content.append({"type": "image_url", "image_url": {"url": data_url}})
    else:
        for frame in frames:
            content.append({"type": "text", "text": f"[第 {frame.at_seconds} 秒]"})
            content.append({"type": "image_url", "image_url": {"url": frame.as_data_url()}})
    return [
        {"role": "system", "content": load_digest_skill()},
        {"role": "user", "content": content},
    ]


def min_digest_shots(duration: float, frame_count: int) -> int:
    """短片至少 3 镜;20 秒以上约每 6 秒一镜,且不超过抽到的帧数。"""
    if duration < 20 and frame_count < 8:
        return 3
    by_time = max(4, int(duration // 6))
    if frame_count:
        return min(frame_count, by_time)
    return by_time


def validate_digest_text(text: str, *, duration: float = 0, frame_count: int = 0) -> str:
    """模型偶尔吐空、只写片头、或概括成几大段。废稿不塞给用户。"""
    cleaned = (text or "").strip()
    if len(cleaned) < 80 or "【" not in cleaned:
        raise ValueError("视频拆解结果不完整")
    shots = _SHOT_MARK.findall(cleaned)
    if not shots:
        raise ValueError("视频拆解结果不完整")
    if duration >= 20 or frame_count >= 8:
        needed = min_digest_shots(duration, frame_count)
        if len(shots) < needed:
            raise ValueError("视频拆解镜头过少,请重试")
    return cleaned


# 拆视频必须会看图。火山豆包直连读不了这组帧图,会在 ARK 上挂满 120s。
# 只认中转站的 Gemini 3.1 Pro;展示名带「官转」的优先。
DIGEST_VISION_MODEL_NAME = "gemini-3.1-pro-preview"


def _is_digest_vision_model(model) -> bool:
    from apps.ai.services import OFFICIAL_DIRECT_PROVIDERS

    provider_name = getattr(getattr(model, "provider", None), "name", "") or ""
    if provider_name in OFFICIAL_DIRECT_PROVIDERS:
        return False
    blob = f"{model.name} {model.display_name}".lower()
    return (
        model.name == DIGEST_VISION_MODEL_NAME
        or "gemini-3.1" in blob
        or "gemini 3.1" in blob
    )


def resolve_digest_model_config(preferred_id=None):
    """视频提炼用的多模态文本模型:Gemini 3.1 Pro 官转。找不到不回落默认语言模型。

    前端可传 model_config_id(跟生成脚本同一套下拉)。只有会看图的 Gemini 3.1 才认,
    选了豆包等纯文本模型仍钉回官转,避免拆帧直接失败。
    """
    from apps.ai.models import ModelConfig

    qs = (
        ModelConfig.objects.select_related("provider")
        .filter(
            capability=ModelConfig.Capability.TEXT,
            status=ModelConfig.Status.ACTIVE,
            provider__status="active",
        )
    )
    if preferred_id:
        chosen = qs.filter(pk=preferred_id).first()
        if chosen is not None and _is_digest_vision_model(chosen):
            return chosen

    def _blob(model) -> str:
        return " ".join(
            filter(
                None,
                [
                    model.name,
                    model.display_name,
                    getattr(model.provider, "name", ""),
                    getattr(model.provider, "display_name", ""),
                ],
            )
        )

    ranked = []
    for model in qs:
        if not _is_digest_vision_model(model):
            continue
        blob = _blob(model)
        # 官转 > 精确模型名 > 其它 Gemini 3.1
        score = 0
        if "官转" in blob:
            score += 100
        if model.name == DIGEST_VISION_MODEL_NAME:
            score += 20
        if getattr(model.provider, "name", "") == "yunqi_gemini":
            score += 5
        ranked.append((score, model.created_at, model))
    if not ranked:
        return None
    ranked.sort(key=lambda item: (-item[0], item[1]))
    return ranked[0][2]


# --------------------------------------------------------------------------- #
# 入口:一次真实的计费调用
# --------------------------------------------------------------------------- #
def digest_project_video(*, project, user, upload, model_config_id=None) -> dict:
    """上传视频 → 分镜稿。抽帧在建任务之前做,文件不合格不占积分。"""
    product = getattr(project, "product", None)
    return _digest_video(
        team=project.team,
        user=user,
        upload=upload,
        project=project,
        product_hint=" · ".join(
            filter(None, [getattr(product, "title", ""), getattr(product, "category", "")])
        ),
        model_config_id=model_config_id,
    )


def digest_team_video(*, team, user, upload, model_config_id=None) -> dict:
    """视频提炼页:不绑项目,计费挂当前团队。"""
    return _digest_video(
        team=team,
        user=user,
        upload=upload,
        project=None,
        product_hint="",
        model_config_id=model_config_id,
    )


def _store_digest_cover(*, team, jpeg: bytes) -> tuple[str, str]:
    """封面传到 TOS。失败返回空,历史卡走占位底。"""
    from io import BytesIO

    from apps.assets.storage import TosStorage

    if not jpeg:
        return "", ""
    key = f"teams/{team.id}/video-digest/{uuid.uuid4()}.jpg"
    storage = TosStorage()
    stored = storage.upload_fileobj(fileobj=BytesIO(jpeg), object_key=key, content_type="image/jpeg")
    return stored.object_key, storage.public_url(object_key=stored.object_key)


def _store_digest_video(*, team, path: str, suffix: str) -> tuple[str, str]:
    """原片(优先转成 H.264)传到 TOS,历史封面点击才能播。失败返回空,不挡拆解。"""
    from apps.assets.storage import TosStorage

    if not path or not Path(path).is_file():
        return "", ""
    upload_path, ext, ephemeral = _prepare_browser_video(path, suffix)
    try:
        if ext.lower() not in ALLOWED_SUFFIXES:
            ext = ".mp4"
        mime = _SUFFIX_MIME.get(ext.lower(), "video/mp4")
        key = f"teams/{team.id}/video-digest/{uuid.uuid4()}{ext.lower()}"
        storage = TosStorage()
        with Path(upload_path).open("rb") as fileobj:
            stored = storage.upload_fileobj(fileobj=fileobj, object_key=key, content_type=mime)
        return stored.object_key, storage.public_url(object_key=stored.object_key)
    finally:
        if ephemeral:
            Path(upload_path).unlink(missing_ok=True)


def _media_url_from_payload(req: dict, url_key: str, object_key_name: str, *, signed: bool = False) -> str:
    stored = str(req.get(url_key) or "").strip()
    key = str(req.get(object_key_name) or "").strip()
    if not key:
        return stored
    try:
        from apps.assets.storage import TosStorage

        storage = TosStorage()
        if signed:
            return storage.presigned_get_url(object_key=key, expires_in=6 * 3600)
        return storage.public_url(object_key=key)
    except Exception:  # noqa: BLE001
        return stored


def serialize_digest_history(task) -> dict:
    from django.utils import timezone

    from apps.ai.models import AITask

    req = task.request_payload or {}
    resp = task.response_payload or {}
    prompt = str(resp.get("prompt") or resp.get("digest") or "").strip()
    duration = float(req.get("duration_seconds") or 0)
    file_name = str(req.get("file_name") or "") or "参考视频.mp4"
    title = str(req.get("title") or "").strip() or title_from_filename(file_name)
    shots = int(req.get("shot_count") or 0) or shot_count(prompt)
    created = timezone.localtime(task.created_at) if task.created_at else timezone.now()
    return {
        "id": str(task.id),
        "title": title,
        "status": "已完成" if task.status == AITask.Status.SUCCEEDED else "失败",
        "duration": int(round(duration)),
        "duration_label": duration_clock(duration),
        "ratio": str(req.get("ratio") or ""),
        "shots": shots,
        "file_name": file_name,
        "cover_url": _media_url_from_payload(req, "cover_url", "cover_key"),
        "video_url": _media_url_from_payload(req, "video_url", "video_key", signed=True),
        "prompt": prompt,
        "created_at": task.created_at.isoformat() if task.created_at else "",
        "created_date": created.strftime("%Y-%m-%d"),
    }


def list_team_digest_history(*, team, limit: int = 50) -> list[dict]:
    from apps.ai.models import AITask

    qs = (
        AITask.objects.filter(
            team=team,
            task_type=AITask.Type.VIDEO_DIGEST,
            project__isnull=True,
            status=AITask.Status.SUCCEEDED,
            is_deleted=False,
            purged_at__isnull=True,
        )
        .order_by("-created_at")[: max(1, min(int(limit), 50))]
    )
    return [serialize_digest_history(task) for task in qs]


def save_digest_prompt(*, team, task_id, prompt: str) -> dict | None:
    from apps.ai.models import AITask

    cleaned = (prompt or "").strip()
    if not cleaned:
        raise VideoDigestError("提示词不能为空")
    task = AITask.objects.filter(
        id=task_id,
        team=team,
        task_type=AITask.Type.VIDEO_DIGEST,
        project__isnull=True,
        is_deleted=False,
        purged_at__isnull=True,
    ).first()
    if task is None:
        return None
    payload = dict(task.response_payload or {})
    payload["prompt"] = cleaned[:32000]
    task.response_payload = payload
    task.save(update_fields=["response_payload", "updated_at"])
    return serialize_digest_history(task)


def _digest_video(*, team, user, upload, project=None, product_hint="", model_config_id=None) -> dict:
    """上传视频 → 分镜稿。抽帧在建任务之前做,文件不合格不占积分。"""
    from django.db import transaction
    from django.utils import timezone

    from apps.ai.models import AITask
    from apps.ai.services import create_ai_task, execute_routed_text_request
    from apps.billing.pricing import quote_video_digest
    from apps.billing.services.ledger import charge_reserved_credit, reserve_credit

    video, frames, duration, extras = digest_input_from_upload(
        upload, keep_source=(project is None)
    )
    extras = extras or {}
    source_path = str(extras.get("source_path") or "")
    file_name = str(extras.get("file_name") or getattr(upload, "name", "") or "参考视频.mp4")
    width = int(extras.get("width") or 0)
    height = int(extras.get("height") or 0)
    file_size = int(extras.get("file_size") or getattr(upload, "size", 0) or 0)

    model_config = resolve_digest_model_config(preferred_id=model_config_id)
    if model_config is None:
        if source_path:
            Path(source_path).unlink(missing_ok=True)
        raise VideoDigestError("视频提炼需要 Gemini 3.1 Pro(会看图),当前没有启用,请联系管理员")
    logger.info(
        "video digest using %s:%s (%s) input=%s duration=%.1fs bytes=%s frames=%s",
        model_config.provider.name,
        model_config.name,
        model_config.display_name,
        "native_video" if video is not None else "frames",
        duration,
        len(video.data) if video is not None else 0,
        len(frames),
    )

    messages = build_digest_messages(
        frames,
        duration,
        product_hint=product_hint,
        video=video,
        aspect_ratio=ratio_label(width, height),
        file_title=title_from_filename(file_name),
    )
    request_payload = {
        "model": model_config.name,
        "endpoint": model_config.endpoint,
        "feature": "video_remix" if project is None else "video_digest",
        "duration_seconds": round(duration, 2),
        "input": "native_video" if video is not None else "frames",
        "frame_count": len(frames),
        "video_bytes": len(video.data) if video is not None else 0,
        "frame_times": [f.at_seconds for f in frames],
        "file_name": file_name,
        "title": title_from_filename(file_name),
        "file_size": file_size,
        "width": width,
        "height": height,
        "ratio": ratio_label(width, height),
    }

    quote = quote_video_digest(team=team, model_config=model_config)
    if quote.meta.get("rate"):
        request_payload = {**request_payload, "points_per_yuan_snapshot": quote.meta["rate"]}

    if project is not None:
        task = create_ai_task(
            project=project,
            user=user,
            task_type=AITask.Type.VIDEO_DIGEST,
            model_config=model_config,
            # 帧是几百 KB base64,绝不进 request_payload(会把 AITask 表撑爆),只记形状
            request_payload=request_payload,
            quote=quote,
        )
    else:
        try:
            with transaction.atomic():
                task = AITask.objects.create(
                    team=team,
                    created_by=user,
                    project=None,
                    task_type=AITask.Type.VIDEO_DIGEST,
                    status=AITask.Status.CREATED,
                    model_config=model_config,
                    idempotency_key=f"video_digest:{team.id}:{uuid.uuid4()}",
                    request_payload=request_payload,
                    estimated_cost=quote.points,
                    base_cost=quote.base_cost_yuan,
                )
                reserve_credit(team=team, user=user, task=task, amount=quote.points)
                task.status = AITask.Status.RESERVED
                task.save(update_fields=["status", "updated_at"])
        except ValueError as exc:
            if source_path:
                Path(source_path).unlink(missing_ok=True)
            if "insufficient credit" in str(exc).lower():
                raise VideoDigestError("团队余额不足，请充值后重试") from exc
            raise VideoDigestError(str(exc)) from exc
    reservation = task.credit_reservation
    try:
        task.status = AITask.Status.SUBMITTED
        task.submitted_at = timezone.now()
        task.save(update_fields=["status", "submitted_at", "updated_at"])
        routed = execute_routed_text_request(
            task=task,
            primary_model=model_config,
            messages=messages,
            streaming=True,
            structured_output=False,
            business_operation="video_digest",
            temperature=0.4,
            validate_text=lambda text: validate_digest_text(
                text, duration=duration, frame_count=len(frames) or (24 if video else 0)
            ),
            extra_body={"max_tokens": DIGEST_MAX_TOKENS},
            request_summary={
                "duration_seconds": round(duration, 2),
                "frame_count": len(frames),
                "input": "native_video" if video is not None else "frames",
            },
            allow_retry=False,
            allow_fallback=False,
        )
        _text, _response, digest = routed.value
    except Exception as exc:  # noqa: BLE001
        if source_path:
            Path(source_path).unlink(missing_ok=True)
        _fail_digest_task(task, reservation, str(exc))
        raise

    cover_key, cover_url, video_key, video_url = "", "", "", ""
    if project is None:
        try:
            cover_key, cover_url = _store_digest_cover(team=team, jpeg=extras.get("cover_jpeg") or b"")
        except Exception:  # noqa: BLE001 — 封面失败不挡拆解结果
            logger.warning("video digest cover upload failed", exc_info=True)
        try:
            video_key, video_url = _store_digest_video(
                team=team,
                path=source_path,
                suffix=str(extras.get("suffix") or ".mp4"),
            )
        except Exception:  # noqa: BLE001 — 原片失败仍可看提示词,封面不能播
            logger.warning("video digest source upload failed", exc_info=True)
        if source_path:
            Path(source_path).unlink(missing_ok=True)
            source_path = ""

    shots = shot_count(digest)
    request_payload = {
        **(task.request_payload or {}),
        **request_payload,
        "shot_count": shots,
        "cover_key": cover_key,
        "cover_url": cover_url,
        "video_key": video_key,
        "video_url": video_url,
    }

    with transaction.atomic():
        task.status = AITask.Status.SUCCEEDED
        task.request_payload = request_payload
        task.response_payload = {"digest": digest[:32000], "prompt": digest[:32000]}
        task.actual_cost = task.estimated_cost
        task.completed_at = timezone.now()
        task.save(
            update_fields=[
                "status",
                "request_payload",
                "response_payload",
                "actual_cost",
                "completed_at",
                "updated_at",
            ]
        )
        charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost)

    return {
        "text": digest,
        "chars": len(digest),
        "frames": len(frames),
        "input": "native_video" if video is not None else "frames",
        "duration": round(duration, 1),
        "task_id": str(task.id),
        "estimated_cost": str(task.estimated_cost),
        "title": request_payload.get("title") or title_from_filename(file_name),
        "file_name": file_name,
        "ratio": request_payload.get("ratio") or "",
        "shots": shots,
        "cover_url": cover_url,
        "video_url": video_url,
        "width": width,
        "height": height,
    }


def _fail_digest_task(task, reservation, message: str) -> None:
    from django.utils import timezone

    from apps.ai.models import AITask
    from apps.billing.services.ledger import release_credit

    try:
        task.status = AITask.Status.FAILED
        task.error_message = message[:2000]
        task.completed_at = timezone.now()
        task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
    finally:
        try:
            release_credit(reservation=reservation, reason=message[:200])
        except Exception:  # noqa: BLE001
            pass