SupplyAgent/agents/find_agent/tools/douyin_search_tikhub.py

"""通过 TikHub 搜索一页抖音视频并自动保存。"""
from __future__ import annotations

import asyncio
import os as os

import httpx as httpx

from agents.find_agent.support.douyin_search_tikhub import (
    _douyin_search_tikhub_raw,
    _ensure_env_loaded as _ensure_env_loaded,
    _wait_rate_limit as _wait_rate_limit,
)
from agents.find_agent.support.search_persistence import persist_search_payload
from supply_agent.tools import tool


@tool
async def douyin_search_tikhub(
    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: int = 0,
    filter_duration: str = "不限",
    search_id: str = "",
    backtrace: str = "",
    min_duration_seconds: int = 0,
    timeout: float | None = None,
) -> str:
    """使用 TikHub 搜索一页抖音视频，返回基础信息、数据库 ID 和分页状态。"""
    result = await _douyin_search_tikhub_raw(
        keyword=keyword,
        content_type=content_type,
        sort_type=sort_type,
        publish_time=publish_time,
        cursor=cursor,
        filter_duration=filter_duration,
        search_id=search_id,
        backtrace=backtrace,
        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=str(cursor),
        page_no=page_no,
        provider="tikhub",
        content_type=content_type,
        sort_type=sort_type,
        publish_time=publish_time,
        provider_state={"search_id": search_id, "backtrace": backtrace},
    )