Agent/examples/content_finder/tools/douyin_user_videos.py

"""
抖音账号历史作品工具（示例）

调用内部爬虫服务获取指定账号的历史作品列表。
"""
import json
from typing import Optional

import asyncio
import httpx

from agent.tools import tool, ToolResult


# API 基础配置
DOUYIN_BLOGGER_API = "http://crawapi.piaoquantv.com/crawler/dou_yin/blogger"
DEFAULT_TIMEOUT = 60.0


@tool(description="根据账号ID获取抖音历史作品，支持排序与游标")
async def douyin_user_videos(
    account_id: str,
    sort_type: str = "最新",
    cursor: str = "",
    timeout: Optional[float] = None,
) -> ToolResult:
    """
    抖音账号历史作品查询

    Args:
        account_id: 抖音账号ID（account_id）
        sort_type: 排序方式（默认 "最新"）
        cursor: 分页游标（默认 ""）
        timeout: 超时时间（秒），默认 60

    Returns:
        ToolResult: 作品列表 JSON
    """
    try:
        payload = {
            "account_id": account_id,
            "sort_type": sort_type,
            "cursor": cursor,
        }

        request_timeout = timeout if timeout is not None else DEFAULT_TIMEOUT

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

        return ToolResult(
            title=f"账号作品: {account_id}",
            output=json.dumps(data, ensure_ascii=False, indent=2),
            long_term_memory=f"Fetched Douyin blogger videos for '{account_id}'",
        )
    except httpx.HTTPStatusError as e:
        return ToolResult(
            title="账号作品获取失败",
            output="",
            error=f"HTTP error {e.response.status_code}: {e.response.text}",
        )
    except Exception as e:
        return ToolResult(
            title="账号作品获取失败",
            output="",
            error=str(e),
        )
# async def main():
#     result = await douyin_user_videos(
#         account_id="MS4wLjABAAAAPRCMGPAFM1VGcJrxRuvTXgJp0Sk95EW1DynNmbKSPg8",
#         sort_type="最新",
#         cursor=""
#     )
#     print(result.output)
#
# if __name__ == "__main__":
#     asyncio.run(main())