from __future__ import annotations

import base64
import json
import mimetypes
import os
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Literal

import httpx
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.models.model_config import ModelConfig
from app.models.token_usage import TokenUsage
from app.services.upload_video_asset_service import resolve_upload_video_path
from app.services.resource_signed_url_service import build_resource_signed_url
from app.utils.id_gen import generate_id

AnalysisMode = Literal["full_breakdown", "summary_only"]


@dataclass(slots=True)
class ShotVideoAnalysisResult:
    result: dict[str, Any]
    raw_response: dict[str, Any]
    usage: dict[str, Any]


def _timeout_seconds() -> int:
    return int(getattr(settings, "SHOT_ANALYSIS_TIMEOUT_SECONDS", 180) or 180)


def _video_fps() -> float:
    return float(getattr(settings, "SHOT_ANALYSIS_VIDEO_FPS", 1.0) or 1.0)


def _split_min_seconds() -> float:
    return float(getattr(settings, "SHOT_SPLIT_MIN_SECONDS", 1) or 1)


def _split_max_seconds() -> float:
    return float(getattr(settings, "SHOT_SPLIT_MAX_SECONDS", 120) or 120)


def _resolve_local_file_path(file_url: str) -> str:
    if file_url.startswith("/uploads/") or file_url.startswith("uploads/"):
        return str(resolve_upload_video_path(file_url))
    return file_url


def build_file_url_or_data_uri(file_url: str, fallback_mime: str = "video/mp4") -> str:
    if file_url.startswith(("http://", "https://", "data:")):
        return file_url
    file_url_sign = build_resource_signed_url(resource_url=file_url, expire_seconds=86400)
    return f"{settings.BASE_URL}{file_url_sign}"

    # file_path = _resolve_local_file_path(file_url)
    # path = Path(file_path)
    # if not path.exists():
    #     raise FileNotFoundError(f"视频文件不存在: {file_path}")
    #
    # max_mb = float(getattr(settings, "SHOT_ANALYSIS_MAX_LOCAL_VIDEO_MB", 45) or 45)
    # size_mb = path.stat().st_size / 1024 / 1024
    # if size_mb > max_mb:
    #     raise ValueError(f"本地视频文件过大: {size_mb:.2f} MB，当前限制 {max_mb:g} MB")
    #
    # mime = mimetypes.guess_type(str(path))[0] or fallback_mime
    # with open(path, "rb") as f:
    #     b64 = base64.b64encode(f.read()).decode("utf-8")
    # return f"data:{mime};base64,{b64}"


def build_user_message(user_text: str, video_url: str) -> tuple[dict[str, Any], dict[str, Any]]:
    real_url = build_file_url_or_data_uri(video_url)
    content_parts = [
        {
            "type": "video_url",
            "video_url": {
                "url": real_url,
                "fps": _video_fps(),
            },
        },
        {"type": "text", "text": user_text},
    ]
    log_content_parts = [
        {
            "type": "video_url",
            "video_url": {
                "url": video_url,
                "fps": _video_fps(),
            },
        },
        {"type": "text", "text": user_text},
    ]
    return {"role": "user", "content": content_parts}, {"role": "user", "content": log_content_parts}


def build_video_analysis_system_prompt(*, mode: AnalysisMode) -> str:
    if mode == "summary_only":
        return """
你是专业的短视频内容分析师、广告素材拆解师。

你的任务：
1. 根据用户提供的视频附件，分析这个视频片段的整体内容。
2. 判断视频分类。
3. 判断视频可能面向的受众人群。
4. 必须输出严格 JSON 对象。
5. 不输出 Markdown。
6. 不输出解释文字。
7. 不返回 null，未知内容填“无”。

顶级字段只能包含：
- 原视频内容
- 原视频分类
- 原视频受众人群
- 拆镜内容剖析

summary_only 模式下“拆镜内容剖析”必须返回空数组。

安全规则：
1. 不要识别视频中人物身份。
2. 不要猜测真实姓名、联系方式、账号身份。
3. 如果视频是游戏录屏，只分析画面内容、玩法内容、玩家情绪表达、受众，不要编造不存在的剧情。
4. 如果视频包含广告内容，可以分析广告品类、目标用户、转化意图，但不要编造品牌信息。
""".strip()

    return f"""
你是专业的短视频内容分析师、广告素材拆解师、视频分镜分析师。

你的任务：
1. 根据用户提供的视频附件，分析原视频整体内容。
2. 判断原视频分类。
3. 判断原视频可能面向的受众人群。
4. 对视频进行拆镜内容剖析。
5. 必须输出严格 JSON 对象。
6. 不输出 Markdown。
7. 不输出解释文字。
8. 不返回 null，未知内容填“无”。

顶级字段只能包含：
- 原视频内容
- 原视频分类
- 原视频受众人群
- 拆镜内容剖析

拆镜内容剖析必须是数组。

每个拆镜片段必须包含：
- 拆镜开始秒
- 拆镜结束秒
- 拆镜时间节点
- 对应时间节点内的内容
- 分类
- 受众人群

拆镜时间规则：
1. 拆镜开始秒必须是数字，例如 0、15、26。
2. 拆镜结束秒必须是数字，例如 15、26、31。
3. 拆镜时间节点必须由拆镜开始秒和拆镜结束秒组成，例如“0-15秒”。
4. 禁止输出“-15秒”这种缺少开始秒的时间节点。
5. 禁止输出“15-秒”这种缺少结束秒的时间节点。
6. 每个拆镜片段时长不能低于 {_split_min_seconds():g} 秒。
7. 每个拆镜片段时长不能高于 {_split_max_seconds():g} 秒。
8. 如果某段内容不足 {_split_min_seconds():g} 秒，不要单独拆出来。
9. 如果单个连续内容超过 {_split_max_seconds():g} 秒，需要按语义变化继续拆分。
10. 如果没有明显镜头变化、场景变化、人物动作变化、剧情变化、字幕重点变化或语义变化，不要强行剖析。
11. 如果无法可靠拆镜，则“拆镜内容剖析”返回空数组。
12. 拆镜时间必须从 0 秒或视频中实际可识别的开始时间开始，不允许出现负数。
13. 拆镜结束秒必须大于拆镜开始秒。
14. 拆镜片段必须按时间顺序排列。

安全规则：
1. 不要识别视频中人物身份。
2. 不要猜测真实姓名、联系方式、账号身份。
3. 如果视频是游戏录屏，只分析画面内容、玩法内容、玩家情绪表达、受众，不要编造不存在的剧情。
4. 如果视频包含广告内容，可以分析广告品类、目标用户、转化意图，但不要编造品牌信息。
""".strip()


def build_video_analysis_user_text(*, mode: AnalysisMode) -> str:
    if mode == "summary_only":
        payload = {
            "任务": "请根据上传的视频片段附件，返回这个视频片段的内容分析 JSON。",
            "输出JSON格式": {
                "原视频内容": "概括这个视频片段整体内容，描述主要画面、主体、场景、动作、剧情或信息点",
                "原视频分类": "判断视频类型，例如：游戏视频、产品广告视频、剧情视频、口播讲解视频、教程视频、生活记录视频等",
                "原视频受众人群": "判断该片段更适合的人群",
                "拆镜内容剖析": [],
            },
            "返回要求": ["只返回 JSON 对象", "不要返回 Markdown", "不要返回解释文字", "不要返回代码块", "不要返回 null，未知填无"],
        }
        return json.dumps(payload, ensure_ascii=False)

    payload = {
        "任务": "请根据上传的视频附件，返回原视频内容分析和拆镜内容剖析 JSON。",
        "输出JSON格式": {
            "原视频内容": "概括原视频整体内容，描述主要画面、人物/主体、场景、动作、剧情或信息点",
            "原视频分类": "判断视频类型，例如：游戏视频、产品广告视频、剧情视频、口播讲解视频、教程视频、生活记录视频、直播切片视频、图文快闪视频等",
            "原视频受众人群": "判断该视频更适合的人群，例如：游戏玩家、年轻娱乐用户、潜在购买用户、同城社交用户等",
            "拆镜内容剖析": [
                {
                    "拆镜开始秒": 0,
                    "拆镜结束秒": 15,
                    "拆镜时间节点": "0-15秒",
                    "对应时间节点内的内容": "描述这个时间片段内发生了什么",
                    "分类": "判断这个片段的内容分类，例如：开场吸引、冲突铺垫、玩法展示、卖点展示、情绪爆发、行动引导、结果展示等",
                    "受众人群": "判断这个片段主要吸引的人群",
                }
            ],
        },
        "拆镜规则": [
            f"每个拆镜片段时长必须大于等于 {_split_min_seconds():g} 秒",
            f"每个拆镜片段时长必须小于等于 {_split_max_seconds():g} 秒",
            "拆镜开始秒必须是数字",
            "拆镜结束秒必须是数字",
            "拆镜开始秒不能是负数",
            "拆镜结束秒必须大于拆镜开始秒",
            "拆镜时间节点必须等于：拆镜开始秒-拆镜结束秒秒",
            "禁止输出“-15秒”",
            "禁止输出“15-秒”",
            "如果无法判断拆镜节点，拆镜内容剖析返回空数组",
        ],
        "返回要求": ["只返回 JSON 对象", "不要返回 Markdown", "不要返回解释文字", "不要返回代码块", "不要返回 null，未知填无"],
    }
    return json.dumps(payload, ensure_ascii=False)


def strip_json_code_fence(text: str) -> str:
    text = text.strip()
    if text.startswith("```"):
        text = re.sub(r"^```(?:json)?", "", text, flags=re.IGNORECASE).strip()
        text = re.sub(r"```$", "", text).strip()
    return text


def parse_model_json(content: str) -> dict[str, Any]:
    cleaned = strip_json_code_fence(content)
    data = json.loads(cleaned)
    if not isinstance(data, dict):
        raise ValueError(f"模型返回 JSON 不是对象类型: {type(data).__name__}")
    return data


def get_message_content_or_raise(data: dict[str, Any]) -> str:
    choices = data.get("choices") or []
    if not choices:
        raise RuntimeError(f"模型响应没有 choices: {json.dumps(data, ensure_ascii=False)}")
    choice = choices[0]
    finish_reason = choice.get("finish_reason")
    if finish_reason == "length":
        usage = data.get("usage", {})
        raise RuntimeError(f"模型输出被长度限制截断，finish_reason={finish_reason}, usage={json.dumps(usage, ensure_ascii=False)}")
    message = choice.get("message") or {}
    content = message.get("content", "")
    if not content:
        raise RuntimeError(f"模型响应 content 为空: {json.dumps(data, ensure_ascii=False)}")
    return content.strip()


def fill_none_with_wu(value: Any) -> Any:
    if value is None:
        return "无"
    if isinstance(value, str):
        return value if value.strip() else "无"
    if isinstance(value, list):
        return [fill_none_with_wu(item) for item in value]
    if isinstance(value, dict):
        return {k: fill_none_with_wu(v) for k, v in value.items()}
    return value


def first_present(data: dict[str, Any], keys: list[str]) -> Any:
    for key in keys:
        if key in data:
            return data.get(key)
    return None


def parse_number(value: Any) -> float | None:
    if isinstance(value, (int, float)):
        return float(value)
    text = str(value or "").strip()
    if not text:
        return None
    m = re.search(r"-?\d+(?:\.\d+)?", text)
    return float(m.group(0)) if m else None


def parse_time_value_to_seconds(value: str) -> float | None:
    value = str(value or "").strip()
    if not value:
        return None
    m = re.match(r"^(\d+(?:\.\d+)?)\s*(?:秒|s)?$", value, flags=re.IGNORECASE)
    if m:
        return float(m.group(1))
    parts = value.split(":")
    if len(parts) in (2, 3) and all(re.match(r"^\d+(?:\.\d+)?$", p.strip()) for p in parts):
        nums = [float(p.strip()) for p in parts]
        if len(nums) == 2:
            minute, second = nums
            return minute * 60 + second
        hour, minute, second = nums
        return hour * 3600 + minute * 60 + second
    return None


def parse_time_node_to_range(time_node: str) -> tuple[float, float] | None:
    text = str(time_node or "").strip()
    if not text:
        return None
    text = text.replace("—", "-").replace("–", "-").replace("－", "-")
    text = text.replace("到", "-").replace("至", "-").replace("~", "-").replace("～", "-")
    text = text.replace("第", "").replace("时间段", "").replace("：", ":")
    m = re.search(r"(\d{1,2}:\d{1,2}(?::\d{1,2})?)\s*-\s*(\d{1,2}:\d{1,2}(?::\d{1,2})?)", text)
    if m:
        start = parse_time_value_to_seconds(m.group(1))
        end = parse_time_value_to_seconds(m.group(2))
        if start is not None and end is not None and end > start:
            return start, end
    m = re.search(r"(\d+(?:\.\d+)?)\s*(?:秒|s)?\s*-\s*(\d+(?:\.\d+)?)\s*(?:秒|s)?", text, flags=re.IGNORECASE)
    if m:
        start = float(m.group(1))
        end = float(m.group(2))
        if end > start:
            return start, end
    m = re.search(r"^\s*-\s*(\d+(?:\.\d+)?)\s*(?:秒|s)?\s*$", text, flags=re.IGNORECASE)
    if m:
        end = float(m.group(1))
        if end > 0:
            return 0.0, end
    return None


def format_second(value: float) -> int | float:
    checked = float(value)
    return int(checked) if checked.is_integer() else round(checked, 2)


def normalize_time_node_by_range(start: float, end: float) -> str:
    return f"{format_second(start)}-{format_second(end)}秒"


def ensure_result_schema(result: dict[str, Any]) -> dict[str, Any]:
    final_result = {
        "原视频内容": result.get("原视频内容", "无"),
        "原视频分类": result.get("原视频分类", "无"),
        "原视频受众人群": result.get("原视频受众人群", "无"),
        "拆镜内容剖析": result.get("拆镜内容剖析", []),
    }
    for key in ("原视频内容", "原视频分类", "原视频受众人群"):
        if not isinstance(final_result[key], str):
            final_result[key] = json.dumps(final_result[key], ensure_ascii=False)
    if not isinstance(final_result["拆镜内容剖析"], list):
        final_result["拆镜内容剖析"] = []
    return final_result


def filter_and_normalize_breakdown(result: dict[str, Any], *, mode: AnalysisMode = "full_breakdown") -> dict[str, Any]:
    if mode == "summary_only":
        result["拆镜内容剖析"] = []
        return result

    breakdown = result.get("拆镜内容剖析")
    if not isinstance(breakdown, list):
        result["拆镜内容剖析"] = []
        return result

    normalized_items: list[dict[str, Any]] = []
    for item in breakdown:
        if not isinstance(item, dict):
            continue
        raw_start = first_present(item, ["拆镜开始秒", "开始秒", "起始秒", "开始时间", "起始时间", "start", "start_second", "start_seconds"])
        raw_end = first_present(item, ["拆镜结束秒", "结束秒", "结束时间", "end", "end_second", "end_seconds"])
        start = parse_number(raw_start)
        end = parse_number(raw_end)
        time_node = str(first_present(item, ["拆镜时间节点", "时间节点", "时间段", "镜头时间", "time_node", "time_range"]) or "").strip()
        if start is None or end is None:
            parsed_range = parse_time_node_to_range(time_node)
            if parsed_range is None:
                continue
            start, end = parsed_range
        if start is None or end is None or start < 0 or end < 0 or end <= start:
            continue
        duration = end - start
        if duration < _split_min_seconds() or duration > _split_max_seconds():
            continue
        content = first_present(item, ["对应时间节点内的内容", "内容", "画面内容", "片段内容", "镜头内容", "content"]) or "无"
        category = first_present(item, ["分类", "片段分类", "内容分类", "镜头分类", "category"]) or "无"
        audience = first_present(item, ["受众人群", "目标受众", "片段受众", "镜头受众", "audience"]) or "无"
        normalized_items.append({
            "拆镜开始秒": format_second(start),
            "拆镜结束秒": format_second(end),
            "拆镜时间节点": normalize_time_node_by_range(start, end),
            "对应时间节点内的内容": str(content or "无"),
            "分类": str(category or "无"),
            "受众人群": str(audience or "无"),
        })
    normalized_items.sort(key=lambda x: float(x.get("拆镜开始秒", 0)))
    result["拆镜内容剖析"] = normalized_items
    return result


async def _select_model_config(db: AsyncSession) -> ModelConfig | None:
    result = await db.execute(
        select(ModelConfig)
        .where(ModelConfig.is_active == True)
        .order_by(ModelConfig.priority.desc())
        .limit(1)
    )
    return result.scalar_one_or_none()


def _int_usage(value: Any) -> int:
    try:
        return int(value or 0)
    except Exception:
        return 0

async def analyze_video_for_shot_split(
    db: AsyncSession,
    video_url: str,
    *,
    user_id: str | None = None,
    mode: AnalysisMode = "full_breakdown",
) -> ShotVideoAnalysisResult:
    """调用模型完成拆镜/片段分析。

    模型配置统一从 model_configs 表选择当前启用且 priority 最高的配置；
    不再读取 SHOT_ANALYSIS_API_BASE / SHOT_ANALYSIS_API_KEY / SHOT_ANALYSIS_MODEL_NAME，
    也不再 fallback 到 SEEDANCE_*，避免拆镜分析走错通道。
    """
    config = await _select_model_config(db)
    if not config:
        raise RuntimeError("拆镜分析模型未配置：请先在 model_configs 表启用可用模型")
    if not str(config.api_key or "").strip():
        raise RuntimeError(f"拆镜分析模型 API Key 为空: model_config_id={config.id}")
    if not str(config.api_base or "").strip():
        raise RuntimeError(f"拆镜分析模型 API Base 为空: model_config_id={config.id}")
    if not str(config.model_name or "").strip():
        raise RuntimeError(f"拆镜分析模型名称为空: model_config_id={config.id}")

    system_prompt = build_video_analysis_system_prompt(mode=mode)
    user_text = build_video_analysis_user_text(mode=mode)
    user_message, log_user_message = build_user_message(user_text, video_url)

    request_data: dict[str, Any] = {
        "model": config.model_name,
        "messages": [
            {"role": "system", "content": system_prompt},
            user_message,
        ],
        "max_tokens": int(getattr(settings, "SHOT_ANALYSIS_MAX_TOKENS", 5000) or getattr(config, "max_tokens", 5000) or 5000),
        "temperature": float(getattr(settings, "SHOT_ANALYSIS_TEMPERATURE", 0.1) or getattr(config, "temperature", 0.1) or 0.1),
        "response_format": {"type": "json_object"},
    }
    log_request_data: dict[str, Any] = {
        **request_data,
        "messages": [
            {"role": "system", "content": system_prompt},
            log_user_message,
        ],
        "model_config_id": config.id,
        "model_config_name": config.name,
        "provider": config.provider,
        "analysis_mode": mode,
    }

    url = f"{str(config.api_base).rstrip('/')}/chat/completions"
    async with httpx.AsyncClient(timeout=_timeout_seconds()) as client:
        response = await client.post(
            url,
            headers={"Authorization": f"Bearer {config.api_key}", "Content-Type": "application/json"},
            json=request_data,
        )
    if response.status_code >= 400:
        raise RuntimeError(f"视频拆镜分析 API 请求失败: HTTP {response.status_code}: {response.text}")

    raw = response.json()
    content = get_message_content_or_raise(raw)
    result = parse_model_json(content)
    result = fill_none_with_wu(result)
    result = ensure_result_schema(result)
    result = filter_and_normalize_breakdown(result, mode=mode)

    usage = raw.get("usage") or {}
    token_usage = {
        "input_tokens": _int_usage(usage.get("prompt_tokens") or usage.get("input_tokens")),
        "output_tokens": _int_usage(usage.get("completion_tokens") or usage.get("output_tokens")),
        "total_tokens": _int_usage(usage.get("total_tokens")),
        "finish_reason": ((raw.get("choices") or [{}])[0] or {}).get("finish_reason"),
        "model_config_id": config.id,
        "model_config_name": config.name,
        "model": config.model_name,
        "provider": config.provider,
        "video_fps": _video_fps(),
        "split_min_seconds": _split_min_seconds(),
        "split_max_seconds": _split_max_seconds(),
        "analysis_mode": mode,
        "log_request": log_request_data,
    }
    if not token_usage["total_tokens"]:
        token_usage["total_tokens"] = token_usage["input_tokens"] + token_usage["output_tokens"]

    token_usage_id = generate_id()
    db.add(
        TokenUsage(
            id=token_usage_id,
            model_config_id=config.id,
            user_id=user_id,
            input_tokens=token_usage["input_tokens"],
            output_tokens=token_usage["output_tokens"],
            total_tokens=token_usage["total_tokens"],
        )
    )
    await db.flush()
    token_usage.update({
        "token_usage_id": token_usage_id,
        "model_config_id": config.id,
        "model_config_name": config.name,
        "model_provider": config.provider,
        "model_name": config.model_name,
    })

    return ShotVideoAnalysisResult(result=result, raw_response=raw, usage=token_usage)

