SupplyAgent/agents/find_agent/support/qwen_video_analysis.py

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

当前 find_agent 明确不使用视频理解，本模块未注册到 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 = 120.0
DEFAULT_MAX_DURATION_SECONDS = 180.0
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."""


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(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)


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,
) -> tuple[str, dict[str, Any]]:
    meta: dict[str, Any] = {
        "original_video_url": video_url,
        "truncated": False,
    }
    if max_duration_seconds is None:
        return video_url, meta

    ffmpeg = _require_ffmpeg()

    duration = await asyncio.to_thread(_probe_duration, ffmpeg, video_url)
    if duration is not None:
        meta["original_duration_seconds"] = duration
        if duration <= max_duration_seconds:
            logger.info(
                "video within limit, skip truncate: duration=%.1fs max=%.1fs url=%s",
                duration,
                max_duration_seconds,
                video_url,
            )
            return video_url, meta
        _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)

        if duration is None:
            duration = await asyncio.to_thread(_probe_duration, ffmpeg, str(source_path))
            if duration is not None:
                meta["original_duration_seconds"] = duration
            if duration is not None and duration <= max_duration_seconds:
                logger.info(
                    "downloaded video within limit, skip truncate: duration=%.1fs",
                    duration,
                )
                _safe_unlink(source_path)
                return video_url, meta

        _ensure_oss_configured()
        await asyncio.to_thread(
            _truncate_video,
            ffmpeg,
            source_path,
            clipped_path,
            max_duration_seconds,
        )
        _safe_unlink(source_path)

        analysis_url = await asyncio.to_thread(
            _upload_clip,
            clipped_path,
            video_url,
            max_duration_seconds,
        )
        _safe_unlink(clipped_path)

    meta["truncated"] = True
    meta["analysis_video_url"] = analysis_url
    meta["max_duration_seconds"] = max_duration_seconds
    logger.info(
        "video truncated: original_duration=%s max=%.1fs analysis_url=%s",
        meta.get("original_duration_seconds"),
        max_duration_seconds,
        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,
) -> str:
    """
    历史视频解析函数，不注册为 find_agent 工具。

    通过阿里云百炼平台调用 qwen3.7-plus 模型，分析视频 URL 并返回文字描述。
    需要设置环境变量 DASHSCOPE_API_KEY。超长视频会先截断再解析。

    Args:
        video_url: 视频地址（需公网可访问的 mp4 等格式）
        prompt: 解析提示词，默认 "描述这段视频的内容"
        fps: 视频抽帧频率，默认 2（每秒采样 2 帧）
        model: 模型名称，默认 "qwen3.7-plus"
        timeout: 请求超时时间（秒），默认 300
        max_duration_seconds: 解析前最长保留秒数，默认 180（3 分钟）；
            传 None 表示不截断
        download_timeout: 下载原视频超时时间（秒），默认 120

    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
    )

    try:
        analysis_url, truncate_meta = await _prepare_analysis_url(
            video_url,
            max_duration_seconds,
            request_download_timeout,
        )
        content = await asyncio.to_thread(
            _analyze_video_sync,
            analysis_url,
            prompt,
            fps,
            model,
            request_timeout,
        )

        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 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 ValueError as e:
        logger.error("qwen_video_analyze config error: %s", e)
        return _error_result(str(e))
    except httpx.TimeoutException as e:
        logger.warning(
            "qwen_video_analyze download timed out: video_url=%s error=%s",
            video_url,
            e,
        )
        return _error_result(
            f"下载视频超时: {e}",
            error_code="tool_timeout",
            retryable=True,
        )
    except httpx.HTTPError as e:
        logger.error(
            "qwen_video_analyze download error: video_url=%s error=%s",
            video_url,
            e,
            exc_info=True,
        )
        return _error_result(f"下载视频失败: {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())