SupplyAgent/agents/find_agent/support/douyin_user_videos.py

"""查询抖音作者作品，输出 find_agent 统一候选格式。"""
from __future__ import annotations

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

import httpx

from agents.find_agent.support.search_persistence import persist_search_payload

logger = logging.getLogger(__name__)

DOUYIN_USER_VIDEOS_API = "http://crawapi.piaoquantv.com/crawler/dou_yin/blogger"
DEFAULT_TIMEOUT = 60.0
_MIN_REQUEST_INTERVAL_SECONDS = 10.1
_rate_limit_lock = asyncio.Lock()
_last_request_monotonic = 0.0
_SORT_TYPES = {"最新", "最热"}


def _safe_int(value: Any, default: int = 0) -> int:
    if isinstance(value, bool) or value is None:
        return default
    try:
        return int(float(str(value).strip()))
    except (TypeError, ValueError):
        return default


def _extract_topics(item: dict[str, Any]) -> list[str]:
    topics: list[str] = []
    for topic in item.get("topic_list") or []:
        if isinstance(topic, str):
            topics.append(topic.strip())
        elif isinstance(topic, dict):
            name = (
                topic.get("topic_name")
                or topic.get("cha_name")
                or topic.get("hashtag_name")
                or topic.get("name")
            )
            if name:
                topics.append(str(name).strip())
    for topic in item.get("text_extra") or []:
        if isinstance(topic, dict) and topic.get("hashtag_name"):
            topics.append(str(topic["hashtag_name"]).strip())
    for topic in item.get("cha_list") or []:
        if isinstance(topic, dict) and topic.get("cha_name"):
            topics.append(str(topic["cha_name"]).strip())
    return list(dict.fromkeys(topic for topic in topics if topic))


def _normalize_video(item: dict[str, Any]) -> dict[str, Any] | None:
    aweme_id = str(item.get("aweme_id") or "").strip()
    if not aweme_id:
        return None
    author = item.get("author") if isinstance(item.get("author"), dict) else {}
    stats = (
        item.get("statistics")
        if isinstance(item.get("statistics"), dict)
        else {}
    )
    video = item.get("video") if isinstance(item.get("video"), dict) else {}
    duration_ms = _safe_int(video.get("duration") or item.get("duration"))
    return {
        "aweme_id": aweme_id,
        "desc": str(item.get("desc") or item.get("item_title") or "无标题")[:200],
        "url": f"https://www.douyin.com/video/{aweme_id}",
        "author": {
            "nickname": str(author.get("nickname") or "未知作者"),
            "sec_uid": str(author.get("sec_uid") or ""),
        },
        "statistics": {
            "digg_count": _safe_int(stats.get("digg_count")),
            "comment_count": _safe_int(stats.get("comment_count")),
            "share_count": _safe_int(stats.get("share_count")),
            "collect_count": _safe_int(stats.get("collect_count")),
            "play_count": _safe_int(stats.get("play_count")),
        },
        "duration_ms": duration_ms,
        "topics": _extract_topics(item),
    }


def _summary(
    account_id: str,
    results: list[dict[str, Any]],
    *,
    filtered_count: int,
    has_more: bool,
    next_cursor: str,
) -> str:
    lines = [
        f"账号 {account_id} 的作品列表",
        (
            f"保留 {len(results)} 条"
            + (f"，过滤短视频 {filtered_count} 条" if filtered_count else "")
            + (f"，还有更多（cursor={next_cursor}）" if has_more else "")
        ),
        "",
    ]
    for index, item in enumerate(results, 1):
        stats = item["statistics"]
        lines.extend(
            [
                f"{index}. {item['desc'][:50]}",
                f"   ID: {item['aweme_id']}",
                f"   链接: {item['url']}",
                (
                    f"   数据: 点赞 {stats['digg_count']:,} | "
                    f"评论 {stats['comment_count']:,} | "
                    f"分享 {stats['share_count']:,} | "
                    f"收藏 {stats['collect_count']:,}"
                ),
                f"   标签: {'、'.join(item['topics']) or '无'}",
                "",
            ]
        )
    return "\n".join(lines).rstrip()


def _error_result(error: str) -> str:
    return json.dumps(
        {"error": error, "title": "抖音作者作品获取失败"},
        ensure_ascii=False,
    )


async def _wait_rate_limit() -> None:
    global _last_request_monotonic
    async with _rate_limit_lock:
        elapsed = time.monotonic() - _last_request_monotonic
        if elapsed < _MIN_REQUEST_INTERVAL_SECONDS:
            await asyncio.sleep(_MIN_REQUEST_INTERVAL_SECONDS - elapsed)
        _last_request_monotonic = time.monotonic()


async def _douyin_user_videos_raw(
    account_id: str,
    sort_type: str = "最热",
    cursor: str = "",
    min_duration_seconds: int = 0,
    timeout: float | None = None,
) -> str:
    """
    获取指定抖音作者的作品列表，支持最热/最新排序和游标翻页。

    当候选作者的粉丝画像偏老或某条视频表现优秀时，可用该工具扩展同作者内容。
    返回结构与 douyin_search.search_results 一致，可直接保存为搜索轨迹和候选。

    Args:
        account_id: author.sec_uid，必须使用完整值。
        sort_type: 最热 / 最新，默认最热。
        cursor: 首次为空；翻页使用上次返回的 next_cursor。
        min_duration_seconds: 最短时长过滤，默认 0 表示不过滤。
        timeout: 请求超时秒数，默认 60。

    Returns:
        JSON 字符串，包含 user_videos、search_results、has_more 和 next_cursor。
        user_videos 与 search_results 是同一个统一结构化列表。
    """
    account_text = str(account_id).strip()
    if not account_text:
        return _error_result("account_id 不能为空")
    if sort_type not in _SORT_TYPES:
        return _error_result(f"sort_type 必须是: {sorted(_SORT_TYPES)}")

    start_time = time.time()
    request_timeout = timeout if timeout is not None else DEFAULT_TIMEOUT
    payload = {
        "account_id": account_text,
        "sort_type": sort_type,
        "cursor": str(cursor or ""),
    }

    try:
        await _wait_rate_limit()
        async with httpx.AsyncClient(
            timeout=request_timeout,
            trust_env=False,
            headers={"Content-Type": "application/json"},
        ) as client:
            response = await client.post(DOUYIN_USER_VIDEOS_API, json=payload)
            response.raise_for_status()
            body = response.json()

        data = body.get("data") if isinstance(body.get("data"), dict) else {}
        items = data.get("data") if isinstance(data.get("data"), list) else []
        minimum_ms = max(0, int(min_duration_seconds)) * 1000
        filtered_count = 0
        seen: set[str] = set()
        results: list[dict[str, Any]] = []
        for raw_item in items:
            if not isinstance(raw_item, dict):
                continue
            normalized = _normalize_video(raw_item)
            if normalized is None or normalized["aweme_id"] in seen:
                continue
            if (
                minimum_ms
                and normalized["duration_ms"]
                and normalized["duration_ms"] < minimum_ms
            ):
                filtered_count += 1
                continue
            seen.add(normalized["aweme_id"])
            results.append(normalized)

        has_more = bool(data.get("has_more") in (1, True, "1"))
        next_cursor = str(data.get("next_cursor") or "")
        duration_ms = int((time.time() - start_time) * 1000)
        result = {
            "title": f"抖音作者作品: {account_text}",
            "output": _summary(
                account_text,
                results,
                filtered_count=filtered_count,
                has_more=has_more,
                next_cursor=next_cursor,
            ),
            "provider": "internal_blogger",
            "account_id": account_text,
            "sort_type": sort_type,
            "cursor": str(cursor or ""),
            "results_count": len(results),
            "filtered_count": filtered_count,
            "has_more": has_more,
            "next_cursor": next_cursor,
            "user_videos": results,
            "search_results": results,
            "duration_ms": duration_ms,
        }
        logger.info(
            "douyin_user_videos completed: account_id=%s results=%d has_more=%s duration_ms=%d",
            account_text,
            len(results),
            has_more,
            duration_ms,
        )
        return json.dumps(result, ensure_ascii=False)
    except httpx.HTTPStatusError as exc:
        text = exc.response.text[:1000]
        logger.error(
            "douyin_user_videos HTTP error: account_id=%s status=%d",
            account_text,
            exc.response.status_code,
        )
        return _error_result(f"HTTP {exc.response.status_code}: {text}")
    except httpx.TimeoutException:
        return _error_result(f"请求超时（{request_timeout}秒）")
    except httpx.RequestError as exc:
        return _error_result(f"网络错误: {exc}")
    except Exception as exc:
        logger.error(
            "douyin_user_videos unexpected error: account_id=%s error=%s",
            account_text,
            exc,
            exc_info=True,
        )
        return _error_result(f"未知错误: {exc}")


async def douyin_user_videos(
    run_id: str,
    account_id: str,
    query_reason: str,
    source_value: str | None = None,
    parent_search_id: int | None = None,
    page_no: int = 1,
    sort_type: str = "最热",
    cursor: str = "",
    min_duration_seconds: int = 0,
    timeout: float | None = None,
) -> str:
    """
    获取一页作者作品，创建搜索记录和候选记录并返回对应数据库 ID。
    """
    result = await _douyin_user_videos_raw(
        account_id=account_id,
        sort_type=sort_type,
        cursor=cursor,
        min_duration_seconds=min_duration_seconds,
        timeout=timeout,
    )
    return await asyncio.to_thread(
        persist_search_payload,
        result,
        run_id=run_id,
        keyword=f"author:{account_id}",
        query_reason=query_reason,
        source_type="author",
        source_value=source_value or account_id,
        parent_search_id=parent_search_id,
        cursor=cursor,
        page_no=page_no,
        provider="internal_blogger",
        sort_type=sort_type,
    )