SupplyAgent/agents/find_agent/support/douyin_detail.py

"""
抖音视频详情工具

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

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

import httpx

from supply_infra.video_discovery_gates import parse_datetime_value

logger = logging.getLogger(__name__)

_MIN_REQUEST_INTERVAL_SECONDS = 10.1
_rate_limit_lock = asyncio.Lock()
_last_request_monotonic: float = 0.0

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 _wait_rate_limit() -> None:
    global _last_request_monotonic
    async with _rate_limit_lock:
        now_mono = time.monotonic()
        wait_seconds = _MIN_REQUEST_INTERVAL_SECONDS - (now_mono - _last_request_monotonic)
        if wait_seconds > 0:
            await asyncio.sleep(wait_seconds)
        _last_request_monotonic = time.monotonic()


async def _fetch_one_detail(
    client: httpx.AsyncClient,
    content_id: str,
) -> 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)


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

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

    Args:
        content_ids: 视频 ID 列表，对应搜索结果中的 aweme_id。
            单个传 ["123"]，多个传 ["123", "456"]
        timeout: 单次请求超时时间（秒），默认 60

    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
    ids = _normalize_content_ids(content_ids)

    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 = await _fetch_one_detail(client, content_id)
                    details.append(detail)
                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})

        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:
            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,
        }
        # 单条时额外提供 detail，方便旧逻辑取值
        if len(details) == 1:
            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())