SupplyAgent/agents/find_agent/support/douyin_search.py

"""
抖音关键词搜索工具

调用内部爬虫服务进行抖音关键词搜索。
"""
from __future__ import annotations

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

import httpx

from agents.find_agent.support.search_persistence import persist_search_payload

logger = logging.getLogger(__name__)

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

# API 基础配置
DOUYIN_SEARCH_API = "http://crawapi.piaoquantv.com/crawler/dou_yin/keyword"
DEFAULT_TIMEOUT = 60.0
DOUYIN_ACCOUNT_ID = "771431222"
DEFAULT_MIN_DURATION_SECONDS = 30


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_duration_ms(item: dict[str, Any]) -> int:
    video = item.get("video") if isinstance(item.get("video"), dict) else {}
    return _safe_int(
        item.get("duration_ms")
        or video.get("duration")
        or item.get("duration")
    )


def _extract_publish_at(item: dict[str, Any]) -> Any:
    return (
        item.get("publish_at")
        or item.get("create_time")
        or item.get("create_timestamp")
        or item.get("publish_timestamp")
    )


def _build_search_results(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """将 API 原始条目转换为结构化搜索结果。"""
    results = []
    for item in items:
        author = item.get("author", {}) if isinstance(item.get("author"), dict) else {}
        stats = item.get("statistics", {}) if isinstance(item.get("statistics"), dict) else {}
        aweme_id = item.get("aweme_id", "")
        results.append(
            {
                "aweme_id": aweme_id,
                "desc": (item.get("desc") or item.get("item_title") or "无标题")[:100],
                "url": f"https://www.douyin.com/video/{aweme_id}" if aweme_id else "",
                "author": {
                    "nickname": author.get("nickname", "未知作者"),
                    "sec_uid": author.get("sec_uid", ""),
                },
                "statistics": {
                    "digg_count": stats.get("digg_count", 0),
                    "comment_count": stats.get("comment_count", 0),
                    "share_count": stats.get("share_count", 0),
                },
                "duration_ms": _extract_duration_ms(item),
                "publish_at": _extract_publish_at(item),
            }
        )
    return results


def _build_output_summary(
    keyword: str,
    items: list[dict[str, Any]],
    has_more: bool,
    cursor_value: str,
) -> str:
    """生成给 LLM 阅读的文本摘要。"""
    lines = [f"搜索关键词「{keyword}」"]
    lines.append(
        f"找到 {len(items)} 条结果"
        + (f"，还有更多（cursor={cursor_value}）" if has_more else "")
    )
    lines.append("")

    for i, item in enumerate(items, 1):
        aweme_id = item.get("aweme_id", "unknown")
        desc = (item.get("desc") or item.get("item_title") or "无标题")[:50]

        author = item.get("author", {}) if isinstance(item.get("author"), dict) else {}
        author_name = author.get("nickname", "未知作者")
        author_id = author.get("sec_uid", "")

        stats = item.get("statistics", {}) if isinstance(item.get("statistics"), dict) else {}
        digg_count = stats.get("digg_count", 0)
        comment_count = stats.get("comment_count", 0)
        share_count = stats.get("share_count", 0)

        lines.append(f"{i}. {desc}")
        lines.append(f"   ID: {aweme_id}")
        lines.append(f"   链接: https://www.douyin.com/video/{aweme_id}")
        lines.append(f"   作者: {author_name}")
        lines.append(f"   sec_uid: {author_id}")
        lines.append(f"   数据: 点赞 {digg_count:,} | 评论 {comment_count:,} | 分享 {share_count:,}")
        lines.append("")

    return "\n".join(lines)


def _success_result(
    keyword: str,
    data: dict[str, Any],
    items: list[dict[str, Any]],
    has_more: bool,
    cursor_value: str,
    duration_ms: int,
    filtered_count: int = 0,
) -> str:
    """构建成功时的 JSON 字符串返回值。"""
    search_results = _build_search_results(items)
    payload = {
        "title": f"抖音搜索: {keyword}",
        "output": _build_output_summary(keyword, items, has_more, cursor_value),
        "keyword": keyword,
        "results_count": len(items),
        "filtered_count": filtered_count,
        "has_more": has_more,
        "next_cursor": cursor_value,
        "search_results": search_results,
        "duration_ms": duration_ms,
    }
    return json.dumps(payload, ensure_ascii=False)


def _error_result(error: str, *, title: str = "抖音搜索失败") -> str:
    """构建失败时的 JSON 字符串返回值。"""
    return json.dumps({"error": error, "title": title}, ensure_ascii=False)


async def _douyin_search_raw(
    keyword: str,
    content_type: str = "视频",
    sort_type: str = "综合排序",
    publish_time: str = "不限",
    cursor: str = "0",
    account_id: str = DOUYIN_ACCOUNT_ID,
    min_duration_seconds: int = DEFAULT_MIN_DURATION_SECONDS,
    timeout: Optional[float] = None,
) -> str:
    """
    抖音关键词搜索

    通过关键词搜索抖音平台的视频内容，支持多种排序和筛选方式。

    Args:
        keyword: 搜索关键词
        content_type: 内容类型（可选：视频/图文, 默认 "视频"）
        sort_type: 排序方式（可选：综合排序/最新发布/最多点赞, 默认 "综合排序"）
        publish_time: 发布时间范围（可选：不限/一天内/一周内/半年内, 默认 "不限"）
        cursor: 分页游标，用于获取下一页结果，默认 "0"
        account_id: 账号ID（可选）
        min_duration_seconds: 客户端最短时长过滤，固定不低于 30 秒。
        timeout: 超时时间（秒），默认 60

    Returns:
        JSON 字符串，包含 output（文本摘要）和 search_results（结构化列表）。
        search_results 中每项含 aweme_id、desc、author、statistics。
        使用 next_cursor 可获取下一页。
    """
    start_time = time.time()
    request_timeout = timeout if timeout is not None else DEFAULT_TIMEOUT

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

        payload = {
            "keyword": keyword,
            "content_type": content_type,
            "sort_type": sort_type,
            "publish_time": publish_time,
            "cursor": cursor,
            "account_id": account_id,
        }

        async with httpx.AsyncClient(timeout=request_timeout) as client:
            response = await client.post(
                DOUYIN_SEARCH_API,
                json=payload,
                headers={"Content-Type": "application/json"},
            )
            response.raise_for_status()
            data = response.json()

        data_block = data.get("data", {}) if isinstance(data.get("data"), dict) else {}
        items = data_block.get("data", []) if isinstance(data_block.get("data"), list) else []
        minimum_ms = max(
            DEFAULT_MIN_DURATION_SECONDS,
            int(min_duration_seconds or DEFAULT_MIN_DURATION_SECONDS),
        ) * 1000
        filtered_count = 0
        filtered_items: list[dict[str, Any]] = []
        for item in items:
            duration = _extract_duration_ms(item)
            if duration and duration < minimum_ms:
                filtered_count += 1
                continue
            filtered_items.append(item)
        items = filtered_items
        has_more = bool(data_block.get("has_more", False))
        cursor_value = str(data_block.get("next_cursor", ""))

        duration_ms = int((time.time() - start_time) * 1000)
        logger.info(
            "douyin_search completed: keyword=%s results=%d has_more=%s duration_ms=%d",
            keyword,
            len(items),
            has_more,
            duration_ms,
        )

        return _success_result(
            keyword,
            data,
            items,
            has_more,
            cursor_value,
            duration_ms,
            filtered_count,
        )

    except httpx.HTTPStatusError as e:
        logger.error(
            "douyin_search HTTP error: keyword=%s status=%d",
            keyword,
            e.response.status_code,
        )
        return _error_result(f"HTTP {e.response.status_code}: {e.response.text}")
    except httpx.TimeoutException:
        logger.error("douyin_search timeout: keyword=%s timeout=%s", keyword, request_timeout)
        return _error_result(f"请求超时（{request_timeout}秒）")
    except httpx.RequestError as e:
        logger.error("douyin_search network error: keyword=%s error=%s", keyword, e)
        return _error_result(f"网络错误: {e}")
    except Exception as e:
        logger.error("douyin_search unexpected error: keyword=%s error=%s", keyword, e, exc_info=True)
        return _error_result(f"未知错误: {e}")


async def douyin_search(
    run_id: str,
    keyword: str,
    query_reason: str,
    source_type: str,
    source_value: str | None = None,
    parent_search_id: int | None = None,
    page_no: int = 1,
    content_type: str = "视频",
    sort_type: str = "综合排序",
    publish_time: str = "不限",
    cursor: str = "0",
    account_id: str = DOUYIN_ACCOUNT_ID,
    min_duration_seconds: int = DEFAULT_MIN_DURATION_SECONDS,
    timeout: Optional[float] = None,
) -> str:
    """
    搜索一页抖音视频，创建搜索记录和候选记录，返回视频基础信息及数据库 ID。
    """
    result = await _douyin_search_raw(
        keyword=keyword,
        content_type=content_type,
        sort_type=sort_type,
        publish_time=publish_time,
        cursor=cursor,
        account_id=account_id,
        min_duration_seconds=min_duration_seconds,
        timeout=timeout,
    )
    return await asyncio.to_thread(
        persist_search_payload,
        result,
        run_id=run_id,
        keyword=keyword,
        query_reason=query_reason,
        source_type=source_type,
        source_value=source_value,
        parent_search_id=parent_search_id,
        cursor=cursor,
        page_no=page_no,
        provider="internal_keyword",
        content_type=content_type,
        sort_type=sort_type,
        publish_time=publish_time,
    )


async def main() -> None:
    result_json = await _douyin_search_raw(
        keyword="养老政策",
        account_id=DOUYIN_ACCOUNT_ID,
    )
    result = json.loads(result_json)
    if "error" in result:
        print(f"搜索失败: {result['error']}")
    else:
        print(result["output"])
        print(f"\n共 {result['results_count']} 条结果")


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