SupplyAgent/agents/find_agent/tools/douyin_search.py

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

import asyncio
from typing import Optional

import httpx as httpx

from agents.find_agent.support.douyin_search import (
    DOUYIN_ACCOUNT_ID,
    _build_search_results as _build_search_results,
    _douyin_search_raw,
    _error_result as _error_result,
    _success_result as _success_result,
)
from agents.find_agent.support.search_persistence import persist_search_payload
from supply_agent.tools import tool


@tool
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,
    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,
        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,
    )