"""千问视频解析的内部实现。

视频在交给模型前会先下载，最多保留前 30 秒，并上传到项目 OSS。
本模块仍由调用方按需使用，不在此处决定是否注册到 Agent。
"""
from __future__ import annotations

import asyncio
import hashlib
import importlib
import json
import logging
import os
import re
import shutil
import subprocess
import tempfile
import time
from pathlib import Path
from typing import Any, Optional

import httpx
from dotenv import load_dotenv
from openai import APIStatusError, APITimeoutError, OpenAI

from supply_agent.paths import find_project_root
from supply_infra.config import get_infra_settings
from supply_infra.oss.client import OssClient

logger = logging.getLogger(__name__)

# DASHSCOPE_BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
DASHSCOPE_BASE_URL = "https://llm-33b86fznnpci2exm.cn-beijing.maas.aliyuncs.com/compatible-mode/v1"
DEFAULT_MODEL = "qwen3.7-plus"
DEFAULT_PROMPT = "描述这段视频的内容"
DEFAULT_FPS = 2.0
DEFAULT_TIMEOUT = 300.0
DEFAULT_DOWNLOAD_TIMEOUT = 60.0
DEFAULT_UPLOAD_TIMEOUT = 60.0
DEFAULT_MAX_DURATION_SECONDS = 30.0
DEFAULT_TOOL_TIMEOUT = 480.0
DOWNLOAD_ATTEMPTS = 2
FFPROBE_TIMEOUT_SECONDS = 30.0
FFMPEG_TIMEOUT_SECONDS = 300.0

_DURATION_RE = re.compile(
    r"Duration:\s*(\d+):(\d+):(\d+(?:\.\d+)?)",
    re.IGNORECASE,
)

_env_loaded = False


class ToolTimeoutError(RuntimeError):
    """A bounded local tool operation exceeded its deadline."""


class VideoDownloadError(RuntimeError):
    """The source video could not be downloaded within the allowed attempts."""

    def __init__(self, message: str, *, timed_out: bool = False) -> None:
        super().__init__(message)
        self.timed_out = timed_out


def _ensure_env_loaded() -> None:
    """从项目根目录加载 .env，使 os.getenv 能读到其中的变量。"""
    global _env_loaded
    if _env_loaded:
        return
    load_dotenv(find_project_root() / ".env")
    _env_loaded = True


def _get_client() -> OpenAI:
    _ensure_env_loaded()
    api_key = os.getenv("DASHSCOPE_API_KEY")
    if not api_key:
        raise ValueError("未设置环境变量 DASHSCOPE_API_KEY")
    return OpenAI(api_key=api_key, base_url=DASHSCOPE_BASE_URL)


def _resolve_ffmpeg() -> str | None:
    ffmpeg = shutil.which("ffmpeg")
    if ffmpeg:
        return ffmpeg
    try:
        module = importlib.import_module("imageio_ffmpeg")
        return module.get_ffmpeg_exe()
    except ImportError:
        return None


def _resolve_ffprobe() -> str | None:
    return shutil.which("ffprobe")


def _require_ffmpeg() -> str:
    ffmpeg = _resolve_ffmpeg()
    if not ffmpeg:
        raise ValueError(
            "截断视频需要 ffmpeg。本地可执行 pip install imageio-ffmpeg，"
            "或 brew install ffmpeg；部署镜像已内置系统 ffmpeg"
        )
    return ffmpeg


def _ensure_oss_configured() -> None:
    settings = get_infra_settings()
    if not settings.aliyun_oss_access_key_id or not settings.aliyun_oss_access_key_secret:
        raise ValueError(
            "转存视频需要配置 ALIYUN_OSS_ACCESS_KEY_ID 和 ALIYUN_OSS_ACCESS_KEY_SECRET"
        )


def _parse_duration_text(text: str) -> float | None:
    match = _DURATION_RE.search(text)
    if not match:
        return None
    hours, minutes, seconds = match.groups()
    return int(hours) * 3600 + int(minutes) * 60 + float(seconds)


def _probe_duration(ffmpeg: str, target: str) -> float | None:
    ffprobe = _resolve_ffprobe()
    if ffprobe:
        try:
            result = subprocess.run(
                [
                    ffprobe,
                    "-v",
                    "error",
                    "-show_entries",
                    "format=duration",
                    "-of",
                    "default=noprint_wrappers=1:nokey=1",
                    target,
                ],
                capture_output=True,
                text=True,
                check=False,
                timeout=FFPROBE_TIMEOUT_SECONDS,
            )
        except subprocess.TimeoutExpired:
            logger.warning(
                "ffprobe timed out after %.0fs: %s",
                FFPROBE_TIMEOUT_SECONDS,
                target,
            )
        else:
            if result.returncode == 0:
                raw = result.stdout.strip()
                if raw:
                    try:
                        return float(raw)
                    except ValueError:
                        pass

    try:
        result = subprocess.run(
            [ffmpeg, "-i", target],
            capture_output=True,
            text=True,
            check=False,
            timeout=FFPROBE_TIMEOUT_SECONDS,
        )
    except subprocess.TimeoutExpired:
        logger.warning(
            "ffmpeg duration probe timed out after %.0fs: %s",
            FFPROBE_TIMEOUT_SECONDS,
            target,
        )
        return None
    return _parse_duration_text(result.stderr)


def _truncate_video(
    ffmpeg: str,
    input_path: Path,
    output_path: Path,
    max_duration_seconds: float,
) -> None:
    duration_text = str(max_duration_seconds)
    copy_cmd = [
        ffmpeg,
        "-y",
        "-i",
        str(input_path),
        "-t",
        duration_text,
        "-c",
        "copy",
        "-movflags",
        "+faststart",
        str(output_path),
    ]
    try:
        result = subprocess.run(
            copy_cmd,
            capture_output=True,
            text=True,
            check=False,
            timeout=FFMPEG_TIMEOUT_SECONDS,
        )
    except subprocess.TimeoutExpired:
        logger.warning(
            "ffmpeg stream copy timed out after %.0fs",
            FFMPEG_TIMEOUT_SECONDS,
        )
        result = None
    if (
        result is not None
        and result.returncode == 0
        and output_path.is_file()
        and output_path.stat().st_size > 0
    ):
        return

    logger.warning(
        "ffmpeg stream copy failed, fallback to re-encode: %s",
        ((result.stderr or result.stdout)[-500:] if result is not None else "timeout"),
    )
    encode_cmd = [
        ffmpeg,
        "-y",
        "-i",
        str(input_path),
        "-t",
        duration_text,
        "-c:v",
        "libx264",
        "-preset",
        "fast",
        "-c:a",
        "aac",
        "-movflags",
        "+faststart",
        str(output_path),
    ]
    try:
        encode_result = subprocess.run(
            encode_cmd,
            capture_output=True,
            text=True,
            check=False,
            timeout=FFMPEG_TIMEOUT_SECONDS,
        )
    except subprocess.TimeoutExpired:
        raise ToolTimeoutError(
            f"ffmpeg 截断视频超时（{FFMPEG_TIMEOUT_SECONDS:.0f}秒）"
        ) from None
    if encode_result.returncode != 0 or not output_path.is_file():
        detail = (encode_result.stderr or encode_result.stdout)[-500:]
        raise RuntimeError(f"ffmpeg 截断视频失败: {detail}")


async def _download_video_once(url: str, dest: Path, timeout: float) -> None:
    async with httpx.AsyncClient(
        timeout=timeout,
        trust_env=False,
        follow_redirects=True,
        headers={"User-Agent": "curl/8.6.0", "Accept": "*/*"},
    ) as client:
        async with client.stream("GET", url) as response:
            response.raise_for_status()
            with dest.open("wb") as file_obj:
                async for chunk in response.aiter_bytes():
                    file_obj.write(chunk)


async def _download_video(url: str, dest: Path, timeout: float) -> None:
    """下载视频，失败后只重试一次。

    httpx 自身的 timeout 是分阶段超时；外层 asyncio.timeout 确保整次
    下载也有硬截止时间，不会被持续滴流的服务器无限拖住。
    """
    last_error: BaseException | None = None
    timed_out = False

    for attempt in range(1, DOWNLOAD_ATTEMPTS + 1):
        _safe_unlink(dest)
        try:
            async with asyncio.timeout(timeout):
                await _download_video_once(url, dest, timeout)
            if not dest.is_file() or dest.stat().st_size <= 0:
                raise httpx.DecodingError("downloaded video is empty")
            return
        except (TimeoutError, httpx.TimeoutException) as exc:
            last_error = exc
            timed_out = True
        except (httpx.HTTPError, OSError) as exc:
            last_error = exc

        logger.warning(
            "video download attempt %d/%d failed: url=%s error=%s",
            attempt,
            DOWNLOAD_ATTEMPTS,
            url,
            last_error,
        )

    _safe_unlink(dest)
    reason = str(last_error) or type(last_error).__name__
    raise VideoDownloadError(
        f"下载视频失败，已尝试 {DOWNLOAD_ATTEMPTS} 次: {reason}",
        timed_out=timed_out,
    ) from last_error


def _safe_unlink(path: Path) -> None:
    """删除临时视频文件，失败时仅记录日志。"""
    try:
        if path.is_file():
            path.unlink()
    except OSError as exc:
        logger.warning("failed to delete temp video file %s: %s", path, exc)


def _upload_clip(local_path: Path, source_url: str, max_duration_seconds: float) -> str:
    url_hash = hashlib.sha256(source_url.encode()).hexdigest()[:16]
    client = OssClient()
    object_key = client.object_key(
        "video_clips",
        f"{url_hash}_{int(max_duration_seconds)}s.mp4",
    )
    return client.upload_file(local_path, object_key)


async def _prepare_analysis_url(
    video_url: str,
    max_duration_seconds: float | None,
    download_timeout: float,
    upload_timeout: float = DEFAULT_UPLOAD_TIMEOUT,
) -> tuple[str, dict[str, Any]]:
    effective_max_duration = min(
        max_duration_seconds
        if max_duration_seconds is not None
        else DEFAULT_MAX_DURATION_SECONDS,
        DEFAULT_MAX_DURATION_SECONDS,
    )
    if effective_max_duration <= 0:
        raise ValueError("max_duration_seconds 必须大于 0")

    meta: dict[str, Any] = {
        "original_video_url": video_url,
        "truncated": False,
        "converted_to_oss": False,
        "max_duration_seconds": effective_max_duration,
    }

    ffmpeg = _require_ffmpeg()
    _ensure_oss_configured()

    with tempfile.TemporaryDirectory(prefix="qwen_video_") as tmpdir:
        source_path = Path(tmpdir) / "source.mp4"
        clipped_path = Path(tmpdir) / "clipped.mp4"

        await _download_video(video_url, source_path, download_timeout)
        duration = await asyncio.to_thread(_probe_duration, ffmpeg, str(source_path))
        if duration is not None:
            meta["original_duration_seconds"] = duration

        upload_path = source_path
        if duration is None or duration > effective_max_duration:
            await asyncio.to_thread(
                _truncate_video,
                ffmpeg,
                source_path,
                clipped_path,
                effective_max_duration,
            )
            upload_path = clipped_path
            meta["truncated"] = True
        else:
            logger.info(
                "downloaded video within limit; uploading the complete file: "
                "duration=%.1fs max=%.1fs",
                duration,
                effective_max_duration,
            )

        try:
            analysis_url = await asyncio.wait_for(
                asyncio.to_thread(
                    _upload_clip,
                    upload_path,
                    video_url,
                    effective_max_duration,
                ),
                timeout=upload_timeout,
            )
        except TimeoutError:
            raise ToolTimeoutError(
                f"OSS 上传视频超时（{upload_timeout:.0f}秒）"
            ) from None

    meta["converted_to_oss"] = True
    meta["analysis_video_url"] = analysis_url
    logger.info(
        "video prepared: original_duration=%s max=%.1fs truncated=%s analysis_url=%s",
        meta.get("original_duration_seconds"),
        effective_max_duration,
        meta["truncated"],
        analysis_url,
    )
    return analysis_url, meta


def _analyze_video_sync(
    video_url: str,
    prompt: str,
    fps: float,
    model: str,
    timeout: float,
) -> str:
    client = _get_client()
    completion = client.chat.completions.create(
        model=model,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "video_url",
                        "video_url": {"url": video_url},
                        "fps": fps,
                    },
                    {"type": "text", "text": prompt},
                ],
            }
        ],
        timeout=timeout,
    )
    return completion.choices[0].message.content or ""


def _success_result(
    video_url: str,
    prompt: str,
    content: str,
    model: str,
    duration_ms: int,
    *,
    extra: dict[str, Any] | None = None,
) -> str:
    payload: dict[str, Any] = {
        "title": "视频解析结果",
        "video_url": video_url,
        "prompt": prompt,
        "model": model,
        "content": content,
        "output": content,
        "duration_ms": duration_ms,
    }
    if extra:
        payload.update(extra)
    return json.dumps(payload, ensure_ascii=False)


def _error_result(
    error: str,
    *,
    title: str = "视频解析失败",
    error_code: str | None = None,
    retryable: bool | None = None,
) -> str:
    payload: dict[str, Any] = {"error": error, "title": title}
    if error_code:
        payload["error_code"] = error_code
    if retryable is not None:
        payload["retryable"] = retryable
    return json.dumps(payload, ensure_ascii=False)


def _classify_api_error(exc: Exception) -> tuple[str, str, str, bool]:
    """将上游 API 异常归类为 Agent 可理解的错误。"""
    raw = str(exc)
    lowered = raw.lower()
    if "data_inspection_failed" in lowered or "inappropriate content" in lowered:
        return (
            "content_inspection_failed",
            "视频内容审核未通过",
            "视频内容未通过模型侧安全审核，无法解析画面；请改依据标题、互动数据和画像继续判断，不要重复调用本工具。",
            False,
        )
    return "api_error", "视频解析失败", raw, True


def verify_truncation_runtime(*, check_oss: bool = False) -> dict[str, Any]:
    """
    校验视频截断运行时依赖是否可用。

    部署后可执行:
        python scripts/verify_video_truncate_runtime.py
    """
    ffmpeg = _require_ffmpeg()
    ffprobe = _resolve_ffprobe()

    with tempfile.TemporaryDirectory(prefix="qwen_video_verify_") as tmpdir:
        source_path = Path(tmpdir) / "source.mp4"
        clipped_path = Path(tmpdir) / "clipped.mp4"
        subprocess.run(
            [
                ffmpeg,
                "-hide_banner",
                "-loglevel",
                "error",
                "-f",
                "lavfi",
                "-i",
                "testsrc=duration=3:size=160x120:rate=10",
                "-t",
                "3",
                "-c:v",
                "libx264",
                "-pix_fmt",
                "yuv420p",
                str(source_path),
            ],
            check=True,
            capture_output=True,
            text=True,
            timeout=FFMPEG_TIMEOUT_SECONDS,
        )
        _truncate_video(ffmpeg, source_path, clipped_path, 1.0)
        if not clipped_path.is_file() or clipped_path.stat().st_size <= 0:
            raise RuntimeError("ffmpeg 截断验证失败：输出文件为空")

    oss_configured = False
    if check_oss:
        _ensure_oss_configured()
        oss_configured = True

    return {
        "ffmpeg": ffmpeg,
        "ffprobe": ffprobe,
        "oss_configured": oss_configured,
        "status": "ok",
    }


async def qwen_video_analyze(
    video_url: str,
    prompt: str = DEFAULT_PROMPT,
    fps: float = DEFAULT_FPS,
    model: str = DEFAULT_MODEL,
    timeout: Optional[float] = None,
    max_duration_seconds: Optional[float] = DEFAULT_MAX_DURATION_SECONDS,
    download_timeout: Optional[float] = None,
    tool_timeout: Optional[float] = None,
) -> str:
    """
    通过阿里云百炼平台调用 qwen3.7-plus 模型，分析视频 URL 并返回文字描述。
    需要设置环境变量 DASHSCOPE_API_KEY。所有视频都会先下载并转存到项目 OSS；
    超过 30 秒时只保留前 30 秒，不超过时保留完整视频。

    Args:
        video_url: 视频地址（需公网可访问的 mp4 等格式）
        prompt: 解析提示词，默认 "描述这段视频的内容"
        fps: 视频抽帧频率，默认 2（每秒采样 2 帧）
        model: 模型名称，默认 "qwen3.7-plus"
        timeout: 请求超时时间（秒），默认 300
        max_duration_seconds: 兼容参数，最大值固定限制为 30 秒；None 也按 30 秒处理
        download_timeout: 单次下载原视频的整体超时时间（秒），默认 60；
            下载失败后只重试一次
        tool_timeout: 整个工具的硬超时时间（秒），默认 480

    Returns:
        JSON 字符串，包含 content（解析文本）和 output（同 content，供 LLM 阅读）。
    """
    start_time = time.time()
    request_timeout = timeout if timeout is not None else DEFAULT_TIMEOUT
    request_download_timeout = (
        download_timeout if download_timeout is not None else DEFAULT_DOWNLOAD_TIMEOUT
    )
    request_tool_timeout = tool_timeout if tool_timeout is not None else DEFAULT_TOOL_TIMEOUT

    if not video_url.strip():
        return _error_result("视频地址不能为空", error_code="invalid_argument", retryable=False)
    if request_timeout <= 0 or request_download_timeout <= 0 or request_tool_timeout <= 0:
        return _error_result(
            "timeout、download_timeout 和 tool_timeout 必须大于 0",
            error_code="invalid_argument",
            retryable=False,
        )

    try:
        async with asyncio.timeout(request_tool_timeout):
            analysis_url, truncate_meta = await _prepare_analysis_url(
                video_url,
                max_duration_seconds,
                request_download_timeout,
            )
            try:
                content = await asyncio.wait_for(
                    asyncio.to_thread(
                        _analyze_video_sync,
                        analysis_url,
                        prompt,
                        fps,
                        model,
                        request_timeout,
                    ),
                    timeout=request_timeout,
                )
            except TimeoutError:
                raise ToolTimeoutError(
                    f"视频模型请求超时（{request_timeout:.0f}秒）"
                ) from None

        duration_ms = int((time.time() - start_time) * 1000)
        logger.info(
            "qwen_video_analyze completed: video_url=%s analysis_url=%s truncated=%s model=%s duration_ms=%d",
            video_url,
            analysis_url,
            truncate_meta.get("truncated"),
            model,
            duration_ms,
        )
        return _success_result(
            analysis_url,
            prompt,
            content,
            model,
            duration_ms,
            extra=truncate_meta,
        )

    except VideoDownloadError as e:
        logger.warning(
            "qwen_video_analyze download abandoned: video_url=%s error=%s",
            video_url,
            e,
        )
        return _error_result(
            str(e),
            error_code="download_timeout" if e.timed_out else "download_failed",
            retryable=False,
        )
    except ToolTimeoutError as e:
        logger.warning(
            "qwen_video_analyze local operation timed out: video_url=%s error=%s",
            video_url,
            e,
        )
        return _error_result(
            str(e),
            error_code="tool_timeout",
            retryable=True,
        )
    except TimeoutError:
        logger.warning(
            "qwen_video_analyze exceeded overall timeout: video_url=%s timeout=%s",
            video_url,
            request_tool_timeout,
        )
        return _error_result(
            f"视频解析工具超时（{request_tool_timeout:.0f}秒）",
            error_code="tool_timeout",
            retryable=True,
        )
    except ValueError as e:
        logger.error("qwen_video_analyze config error: %s", e)
        return _error_result(str(e))
    except APITimeoutError as e:
        logger.warning(
            "qwen_video_analyze model request timed out: video_url=%s error=%s",
            video_url,
            e,
        )
        return _error_result(
            f"视频模型请求超时: {e}",
            error_code="tool_timeout",
            retryable=True,
        )
    except APIStatusError as e:
        error_code, title, message, retryable = _classify_api_error(e)
        logger.warning(
            "qwen_video_analyze api rejected: video_url=%s code=%s status=%s",
            video_url,
            error_code,
            e.status_code,
        )
        return _error_result(
            message,
            title=title,
            error_code=error_code,
            retryable=retryable,
        )
    except Exception as e:
        logger.error(
            "qwen_video_analyze error: video_url=%s error=%s",
            video_url,
            e,
            exc_info=True,
        )
        return _error_result(str(e))


async def main() -> None:
    test_url = os.getenv("TEST_VIDEO_URL", "https://www.douyin.com/aweme/v1/play/?video_id=v0200fg10000d4lbj67og65hqvt943u0&ratio=1080p&line=0")
    # test_url = os.getenv("TEST_VIDEO_URL", "http://rescdn.yishihui.com/longvideo/transcode/video/vpc/20260713/b2507233bc8fe040003d19b34c2c7d73.mp4")

    result_json = await qwen_video_analyze(video_url=test_url)
    result = json.loads(result_json)
    if "error" in result:
        print(f"解析失败: {result['error']}")
    else:
        print(result["content"])


if __name__ == "__main__":
    asyncio.run(main())
