"""查询抖音作者作品，输出 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, )