"""
抖音视频详情工具

根据 content_id（aweme_id）调用内部爬虫服务获取视频详情与真实播放链接。
支持单个或批量查询。
"""
from __future__ import annotations

import asyncio
import json
import logging
import time
from typing import Any, Optional

import httpx

from agents.find_agent.support.rate_limit import make_async_interval_limiter
from supply_infra.video_discovery_gates import parse_datetime_value
from supply_infra.services.video_discovery_service import (
    RunNotFoundError,
    build_evidence_values,
    get_video_discovery_service,
)

logger = logging.getLogger(__name__)

_MIN_REQUEST_INTERVAL_SECONDS = 10.1
_wait_rate_limit = make_async_interval_limiter(_MIN_REQUEST_INTERVAL_SECONDS)

DOUYIN_DETAIL_API = "http://8.217.190.241:8888/crawler/dou_yin/detail"
DEFAULT_TIMEOUT = 60.0
MAX_DETAIL_ITEMS = 8

_PLAY_URL_MARKER = "douyin.com/aweme/v1/play/"

# 详情接口中保留的有效字段（去掉长期无意义的空壳字段）
_KEEP_FIELDS = (
    "channel",
    "channel_content_id",
    "content_link",
    "title",
    "content_type",
    "body_text",
    "location",
    "source_url",
    "topic_list",
    "image_url_list",
    "video_url_list",
    "multi_bitrate",
    "bgm_data",
    "is_original",
    "channel_account_id",
    "channel_account_name",
    "channel_account_avatar",
    "view_count",
    "play_count",
    "like_count",
    "collect_count",
    "comment_count",
    "share_count",
    "looking_count",
    "create_time",
    "create_timestamp",
    "publish_at",
    "publish_time",
    "publish_timestamp",
    "modify_timestamp",
    "update_timestamp",
)


def _is_play_url(url: str) -> bool:
    return bool(url) and _PLAY_URL_MARKER in url


def _pick_url(*candidates: str) -> str:
    """优先选 aweme/v1/play 链接，否则回退第一个非空 URL。"""
    urls = [u for u in candidates if u]
    for url in urls:
        if _is_play_url(url):
            return url
    return urls[0] if urls else ""


def _extract_video_url(detail_data: dict[str, Any]) -> str:
    """优先取 video_url_list[0]，并偏好 aweme/v1/play 可播放链接。"""
    candidates: list[str] = []

    video_url_list = detail_data.get("video_url_list")
    if isinstance(video_url_list, list):
        for item in video_url_list:
            if isinstance(item, dict):
                url = item.get("video_url") or ""
                if url:
                    candidates.append(url)

    multi_bitrate = detail_data.get("multi_bitrate")
    if isinstance(multi_bitrate, dict):
        for ratio in ("1080p", "720p", "540p", "default"):
            bit_info = multi_bitrate.get(ratio)
            if isinstance(bit_info, dict):
                url = bit_info.get("video_url") or ""
                if url:
                    candidates.append(url)

    return _pick_url(*candidates)


def _extract_cover_url(detail_data: dict[str, Any]) -> str:
    image_url_list = detail_data.get("image_url_list")
    if isinstance(image_url_list, list) and image_url_list:
        first = image_url_list[0]
        if isinstance(first, dict):
            return first.get("image_url") or ""
    return ""


def _extract_video_duration(detail_data: dict[str, Any]) -> int:
    video_url_list = detail_data.get("video_url_list")
    if isinstance(video_url_list, list) and video_url_list:
        first = video_url_list[0]
        if isinstance(first, dict):
            try:
                return int(first.get("video_duration") or 0)
            except (TypeError, ValueError):
                return 0
    return 0


def _normalize_content_ids(content_ids: list[str]) -> list[str]:
    """去重且保序，过滤空值。"""
    seen: set[str] = set()
    result: list[str] = []
    for item in content_ids:
        cid = str(item).strip()
        if not cid or cid in seen:
            continue
        seen.add(cid)
        result.append(cid)
    return result


def _build_detail_result(detail: dict[str, Any], content_id: str) -> dict[str, Any]:
    """保留接口有效字段，并补充常用便捷字段。"""
    channel_content_id = str(detail.get("channel_content_id") or content_id)
    result: dict[str, Any] = {
        "content_id": content_id,
        "video_url": _extract_video_url(detail),
        "video_duration": _extract_video_duration(detail),
        "duration_seconds": _extract_video_duration(detail),
        "cover_url": _extract_cover_url(detail),
    }
    for key in _KEEP_FIELDS:
        if key not in detail:
            continue
        value = detail.get(key)
        if key == "channel_content_id":
            result[key] = channel_content_id
        elif key == "content_link":
            result[key] = value or (
                f"https://www.douyin.com/video/{channel_content_id}" if channel_content_id else ""
            )
        else:
            result[key] = value

    publish_at = parse_datetime_value(
        detail.get("publish_at")
        or detail.get("publish_time")
        or detail.get("publish_timestamp")
        or detail.get("create_time")
        or detail.get("create_timestamp")
    )
    result["publish_at"] = (
        publish_at.isoformat(timespec="seconds") if publish_at is not None else None
    )
    return result


def _build_item_summary(index: int, result: dict[str, Any]) -> str:
    lines = [
        f"{index}. {result.get('title') or result.get('body_text') or '无标题'}",
        f"   content_id: {result.get('content_id', '')}",
        f"   页面链接: {result.get('content_link', '')}",
        f"   视频链接: {result.get('video_url', '') or '未获取到'}",
        f"   时长: {result.get('video_duration', 0)} 秒",
        f"   作者: {result.get('channel_account_name', '')}",
        f"   sec_uid: {result.get('channel_account_id', '')}",
        (
            f"   数据: 点赞 {result.get('like_count') or 0:,} | "
            f"评论 {result.get('comment_count') or 0:,} | "
            f"分享 {result.get('share_count') or 0:,} | "
            f"收藏 {result.get('collect_count') or 0:,}"
        ),
    ]
    return "\n".join(lines)


def _build_output_summary(
    details: list[dict[str, Any]],
    errors: list[dict[str, str]],
) -> str:
    lines = [
        f"抖音视频详情：成功 {len(details)} 条"
        + (f"，失败 {len(errors)} 条" if errors else "")
    ]
    lines.append("")

    for i, item in enumerate(details, 1):
        lines.append(_build_item_summary(i, item))
        lines.append("")

    if errors:
        lines.append("失败列表:")
        for err in errors:
            lines.append(f"- {err.get('content_id', '')}: {err.get('error', '')}")

    return "\n".join(lines).rstrip()


def _error_result(
    error: str,
    *,
    title: str = "抖音详情获取失败",
    input_error: bool = False,
) -> str:
    return json.dumps(
        {"error": error, "title": title, "input_error": input_error},
        ensure_ascii=False,
    )


async def _fetch_one_detail(
    client: httpx.AsyncClient,
    content_id: str,
) -> tuple[dict[str, Any], dict[str, Any]]:
    """拉取单条详情。成功返回 detail 字典；失败抛出 Exception。"""
    await _wait_rate_limit()
    response = await client.post(
        DOUYIN_DETAIL_API,
        json={"content_id": content_id},
        headers={"Content-Type": "application/json"},
    )
    response.raise_for_status()
    body = response.json()

    if body.get("code") not in (0, None):
        raise RuntimeError(f"接口返回错误: code={body.get('code')} msg={body.get('msg')}")

    data_block = body.get("data", {}) if isinstance(body.get("data"), dict) else {}
    detail_raw = data_block.get("data", {}) if isinstance(data_block.get("data"), dict) else {}
    if not detail_raw:
        raise RuntimeError(f"未查到视频详情: content_id={content_id}")

    return _build_detail_result(detail_raw, content_id), body


def _detail_candidate_values(detail: dict[str, Any]) -> dict[str, Any]:
    topics = detail.get("topic_list")
    tags = topics if isinstance(topics, list) else []
    publish_at = parse_datetime_value(detail.get("publish_at"))
    return {
        "title": detail.get("title") or detail.get("body_text"),
        "content_link": detail.get("content_link"),
        "author_name": detail.get("channel_account_name"),
        "author_sec_uid": detail.get("channel_account_id"),
        "tags_json": json.dumps(tags, ensure_ascii=False) if tags else None,
        "publish_at": (
            publish_at.replace(tzinfo=None) if publish_at is not None else None
        ),
        "duration_seconds": detail.get("duration_seconds"),
        "play_count": (
            detail.get("play_count")
            if detail.get("play_count") is not None
            else detail.get("view_count")
        ),
        "like_count": detail.get("like_count"),
        "comment_count": detail.get("comment_count"),
        "collect_count": detail.get("collect_count"),
        "share_count": detail.get("share_count"),
    }


async def douyin_detail(
    content_ids: list[str] | None = None,
    timeout: Optional[float] = None,
    run_id: str | None = None,
    candidate_ids: list[int] | None = None,
) -> str:
    """
    抖音视频详情（支持批量）

    根据 content_id（搜索结果中的 aweme_id）获取视频详情与真实播放链接。
    用于在 douyin_search 选中目标视频后，再拉取可播放的 video_url。

    Args:
        content_ids: 兼容旧调用的视频 ID 列表；新流程不要使用。
        timeout: 单次请求超时时间（秒），默认 60
        run_id: 视频发现运行 id；与 candidate_ids 一起使用时自动落库并重算门禁。
        candidate_ids: 搜索工具返回的数据库候选 id。工具会从数据库读取 aweme_id。

    Returns:
        JSON 字符串，包含：
        - output: 文本摘要
        - details: 详情列表（含 video_url、作者、互动、BGM、多码率等有效字段）
        - errors: 失败项列表
        - success_count / failed_count / results_count
    """
    start_time = time.time()
    request_timeout = timeout if timeout is not None else DEFAULT_TIMEOUT
    candidate_by_aweme: dict[str, int] = {}
    if candidate_ids:
        if not run_id:
            return _error_result("candidate_ids 模式必须提供 run_id", input_error=True)
        try:
            candidates = await asyncio.to_thread(
                get_video_discovery_service().get_candidates_by_ids,
                str(run_id),
                [int(value) for value in candidate_ids],
            )
        except (RunNotFoundError, ValueError) as exc:
            return _error_result(str(exc), input_error=True)
        for candidate in candidates:
            aweme_id = str(candidate.get("aweme_id") or "")
            if aweme_id:
                candidate_by_aweme.setdefault(aweme_id, int(candidate["candidate_id"]))
        ids = _normalize_content_ids(list(candidate_by_aweme))
    else:
        ids = _normalize_content_ids(content_ids or [])

    if not ids:
        return _error_result("content_ids 不能为空")
    if len(ids) > MAX_DETAIL_ITEMS:
        return _error_result(
            f"content_ids 最多 {MAX_DETAIL_ITEMS} 条，请先按相关性、分享价值和备选潜力筛选",
            input_error=True,
        )

    details: list[dict[str, Any]] = []
    errors: list[dict[str, str]] = []

    try:
        async with httpx.AsyncClient(
            timeout=request_timeout,
            trust_env=False,
            headers={"User-Agent": "curl/8.6.0", "Accept": "*/*"},
        ) as client:
            for content_id in ids:
                try:
                    detail, raw_response = await _fetch_one_detail(client, content_id)
                    details.append(detail)
                    if run_id and content_id in candidate_by_aweme:
                        normalized = _detail_candidate_values(detail)
                        evidence_values = build_evidence_values(
                            run_id=str(run_id),
                            evidence_type="detail",
                            provider="crawler",
                            subject_key=content_id,
                            request={"content_id": content_id},
                            raw_response=raw_response,
                            normalized=normalized,
                            fetch_status="success",
                            trigger_candidate_id=candidate_by_aweme[content_id],
                            aweme_id=content_id,
                        )
                        await asyncio.to_thread(
                            get_video_discovery_service().record_candidate_evidence,
                            run_id=str(run_id),
                            evidence_values=evidence_values,
                            aweme_id=content_id,
                            normalized=normalized,
                        )
                except httpx.HTTPStatusError as e:
                    msg = f"HTTP {e.response.status_code}: {e.response.text}"
                    logger.error("douyin_detail HTTP error: content_id=%s status=%d", content_id, e.response.status_code)
                    errors.append({"content_id": content_id, "error": msg})
                except httpx.TimeoutException:
                    msg = f"请求超时（{request_timeout}秒）"
                    logger.error("douyin_detail timeout: content_id=%s", content_id)
                    errors.append({"content_id": content_id, "error": msg})
                except httpx.RequestError as e:
                    msg = f"网络错误: {e}"
                    logger.error("douyin_detail network error: content_id=%s error=%s", content_id, e)
                    errors.append({"content_id": content_id, "error": msg})
                except Exception as e:
                    msg = str(e)
                    logger.warning("douyin_detail item failed: content_id=%s error=%s", content_id, e)
                    errors.append({"content_id": content_id, "error": msg})

                if errors and errors[-1].get("content_id") == content_id and run_id and content_id in candidate_by_aweme:
                    error_message = str(errors[-1].get("error") or "详情获取失败")
                    evidence_values = build_evidence_values(
                        run_id=str(run_id),
                        evidence_type="detail",
                        provider="crawler",
                        subject_key=content_id,
                        request={"content_id": content_id},
                        raw_response={"error": error_message},
                        fetch_status="failed",
                        trigger_candidate_id=candidate_by_aweme[content_id],
                        aweme_id=content_id,
                        error_message=error_message,
                    )
                    await asyncio.to_thread(
                        get_video_discovery_service().record_candidate_evidence,
                        run_id=str(run_id),
                        evidence_values=evidence_values,
                        aweme_id=content_id,
                        normalized={},
                    )

        duration_ms = int((time.time() - start_time) * 1000)
        logger.info(
            "douyin_detail completed: requested=%d success=%d failed=%d duration_ms=%d",
            len(ids),
            len(details),
            len(errors),
            duration_ms,
        )

        if not details and errors and not (run_id and candidate_ids):
            return _error_result(
                f"全部失败（{len(errors)} 条）: {errors[0].get('error', '')}"
            )

        payload = {
            "title": f"抖音详情: {len(details)}/{len(ids)}",
            "output": _build_output_summary(details, errors),
            "results_count": len(ids),
            "success_count": len(details),
            "failed_count": len(errors),
            "details": details,
            "errors": errors,
            "duration_ms": duration_ms,
        }
        if run_id and candidate_ids:
            refreshed = await asyncio.to_thread(
                get_video_discovery_service().get_candidates_by_ids,
                str(run_id),
                [int(value) for value in candidate_ids],
            )
            payload["candidates"] = [
                {
                    "candidate_id": item["candidate_id"],
                    "aweme_id": item["aweme_id"],
                    "gate_status": item["gate_status"],
                    "failed_reason_codes": (
                        item.get("gate_results", {}).get("failed_reason_codes") or []
                    ),
                    "evidence_version": item.get("evidence_version"),
                }
                for item in refreshed
            ]
            payload.pop("details", None)
            payload["errors"] = [
                {
                    "content_id": item.get("content_id"),
                    "error": str(item.get("error") or "")[:500],
                }
                for item in errors
            ]
        # 单条时额外提供 detail，方便旧逻辑取值
        if len(details) == 1 and not (run_id and candidate_ids):
            payload["detail"] = details[0]
        return json.dumps(payload, ensure_ascii=False)

    except Exception as e:
        logger.error("douyin_detail unexpected error: error=%s", e, exc_info=True)
        return _error_result(f"未知错误: {e}")


async def main() -> None:
    result_json = await douyin_detail(
        content_ids=["7641118685977614586", "7307654921879358747"]
    )
    result = json.loads(result_json)
    if "error" in result and "details" not in result:
        print(f"获取失败: {result['error']}")
    else:
        print(result["output"])
        print(f"\nsuccess={result.get('success_count')} failed={result.get('failed_count')}")
        for item in result.get("details", []):
            print(f"- {item.get('content_id')}: {item.get('video_url')}")


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