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

链路:ffmpeg 抽帧(均匀采样) → 帧内联进多模态 messages → 走现有文本模型路由 → 纯文本分镜稿。

两个「本来以为要新建、其实已经有」的前提:
  1. **读图能力**:拆视频是多模态(帧图 + 文本)。默认语言模型现在可能是纯文本豆包,
     **不能再用 get_default_model(TEXT)**。固定钉 Gemini 3.1 Pro 官转
     (``gemini-3.1-pro-preview``,展示名带「官转」优先)。
  2. **ffmpeg**:第 5 阶段导出早就依赖它,已装进后端镜像(见 Dockerfile)。

**没有音轨**:帧里读不到口播,只能读画面上的字幕。这是刻意取舍——接语音转写要另开火山 ASR 服务、
另加一套凭证与计价,而带货参考片绝大多数带硬字幕,且口播词下游本来就要按用户自己的商品重写。
skill 里已要求模型「无字幕就如实写缺失,不许编口播词」。
"""

from __future__ import annotations

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

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 分钟。带货参考片远短于此;更长的帧采样密度不够,拆出来也是错的

# 抽帧:每 5 秒一帧,夹在 4~12 帧之间。12 帧 × 512px JPEG ≈ 0.5 MB base64,单次请求扛得住
SECONDS_PER_FRAME = 5
MIN_FRAMES = 4
MAX_FRAMES = 12
FRAME_WIDTH = 512  # 帧宽上限;分镜拆解看的是构图与景别,不需要原分辨率
FRAME_QUALITY = 5  # ffmpeg -q:v,2(最好)~31(最差)

_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")


# --------------------------------------------------------------------------- #
# 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。输入是一条电商短视频按时间均匀抽出的截帧。"
        "逐镜还原分镜,每镜写清主体/动作/场景/景别/运镜/光线氛围/商品露出七要素,"
        "台词只抄画面上的字幕,看不见的不要编。输出中文纯文本。"
    )


# --------------------------------------------------------------------------- #
# 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 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 frames_from_upload(upload) -> tuple[list[VideoFrame], float]:
    """校验上传文件 → 落临时盘 → 探时长 → 抽帧。临时文件退出即删。"""
    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"
    with tempfile.NamedTemporaryFile(suffix=suffix) as tmp:
        for chunk in upload.chunks():
            tmp.write(chunk)
        tmp.flush()
        duration = probe_duration(tmp.name)
        if duration > MAX_DURATION_SECONDS:
            raise VideoDigestError(
                f"视频不能超过 {MAX_DURATION_SECONDS // 60} 分钟,请剪出要参考的那一段再传"
            )
        return extract_frames(tmp.name, plan_frame_times(duration)), duration


# --------------------------------------------------------------------------- #
# 组装多模态消息
# --------------------------------------------------------------------------- #
def build_digest_messages(
    frames: list[VideoFrame],
    duration: float,
    *,
    product_hint: str = "",
) -> list[dict]:
    """system = 拆解 skill;user = 时间戳 + 帧图交替,让模型知道每张图在原片的第几秒。"""
    head = [
        f"这是一条时长约 {round(duration)} 秒的电商带货短视频,",
        f"按时间顺序均匀抽了 {len(frames)} 帧。每帧图前面标了它在原片中的时间点。",
    ]
    if product_hint:
        head.append(f"用户接下来想用这条片子的结构去拍自己的商品:{product_hint}。")
    head.append("请按技能里的输出格式还原它的分镜。")

    content: list[dict] = [{"type": "text", "text": "".join(head)}]
    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 validate_digest_text(text: str) -> str:
    """模型偶尔吐空 / 吐一句道歉。判空后交给路由层重试或切模型,别把废稿塞给用户。"""
    cleaned = (text or "").strip()
    if len(cleaned) < 80 or "【" not in cleaned:
        raise ValueError("视频拆解结果不完整")
    return cleaned


# 拆视频必须会看图。默认文本模型现在常是纯文本豆包,传帧图会直接失败。
# 钉 Gemini 3.1 Pro;展示名带「官转」的优先(中转站官方 Gemini 通道)。
DIGEST_VISION_MODEL_NAME = "gemini-3.1-pro-preview"


def resolve_digest_model_config():
    """视频提炼用的多模态文本模型:Gemini 3.1 Pro 官转。找不到不回落默认语言模型。"""
    from apps.ai.models import ModelConfig

    qs = (
        ModelConfig.objects.select_related("provider")
        .filter(
            capability=ModelConfig.Capability.TEXT,
            status=ModelConfig.Status.ACTIVE,
            provider__status="active",
        )
    )

    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:
        blob = _blob(model)
        name_hit = "gemini-3.1" in model.name.lower() or "gemini-3.1" in (model.display_name or "").lower()
        label_hit = "gemini 3.1" in (model.display_name or "").lower()
        if not (name_hit or label_hit or model.name == DIGEST_VISION_MODEL_NAME):
            continue
        # 官转 > 精确模型名 > 其它 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) -> 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.services.ledger import charge_reserved_credit

    frames, duration = frames_from_upload(upload)

    model_config = resolve_digest_model_config()
    if model_config is None:
        raise VideoDigestError("视频提炼需要 Gemini 3.1 Pro(会看图),当前没有启用,请联系管理员")

    product = getattr(project, "product", None)
    messages = build_digest_messages(
        frames,
        duration,
        product_hint=" · ".join(
            filter(None, [getattr(product, "title", ""), getattr(product, "category", "")])
        ),
    )

    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={
            "model": model_config.name,
            "endpoint": model_config.endpoint,
            "duration_seconds": round(duration, 2),
            "frame_count": len(frames),
            "frame_times": [f.at_seconds for f in frames],
        },
    )
    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=False,
            structured_output=False,
            business_operation="video_digest",
            temperature=0.4,
            validate_text=validate_digest_text,
            request_summary={"duration_seconds": round(duration, 2), "frame_count": len(frames)},
        )
        _text, _response, digest = routed.value
    except Exception as exc:  # noqa: BLE001
        _fail_digest_task(task, reservation, str(exc))
        raise

    with transaction.atomic():
        task.status = AITask.Status.SUCCEEDED
        task.response_payload = {"digest": digest[:8000]}
        task.actual_cost = task.estimated_cost
        task.completed_at = timezone.now()
        task.save(update_fields=["status", "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),
        "duration": round(duration, 1),
        "task_id": str(task.id),
    }


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