112 lines
4.7 KiB
Python
112 lines
4.7 KiB
Python
from typing import Any
|
|
|
|
from app.platforms.base import CommentData, ContentItemData, HotspotData, TikHubClient, first_present, parse_timestamp
|
|
|
|
|
|
class DouyinPlatform:
|
|
def __init__(self, client: TikHubClient | None) -> None:
|
|
self.client = client
|
|
|
|
def map_hotspots(self, payload: dict[str, Any], *, limit: int) -> list[HotspotData]:
|
|
data = payload.get("data", {})
|
|
items = data.get("word_list") or data.get("list") or data.get("item_list") or []
|
|
result = []
|
|
for index, item in enumerate(items[:limit], start=1):
|
|
result.append(
|
|
HotspotData(
|
|
source_hot_id=str(item.get("query_id")) if item.get("query_id") is not None else None,
|
|
title=str(first_present(item, "title", "sentence", "word") or ""),
|
|
rank=item.get("rank") or index,
|
|
heat_value=str(item.get("hot_score")) if item.get("hot_score") is not None else None,
|
|
raw_data=item,
|
|
)
|
|
)
|
|
return result
|
|
|
|
def map_items(self, payload: dict[str, Any], *, limit: int) -> list[ContentItemData]:
|
|
data = payload.get("data", [])
|
|
raw_items = data.get("business_data") or data.get("data") or data.get("items") or [] if isinstance(data, dict) else data
|
|
result = []
|
|
for item in raw_items[:limit]:
|
|
nested = item.get("data", item) if isinstance(item, dict) else {}
|
|
aweme = nested.get("aweme_info", nested) if isinstance(nested, dict) else {}
|
|
aweme_id = aweme.get("aweme_id")
|
|
if not aweme_id:
|
|
continue
|
|
result.append(
|
|
ContentItemData(
|
|
source_item_id=str(aweme_id),
|
|
item_type="video",
|
|
title=aweme.get("desc"),
|
|
summary=aweme.get("desc"),
|
|
url=aweme.get("share_url"),
|
|
raw_data=aweme,
|
|
)
|
|
)
|
|
return result
|
|
|
|
def map_comments(self, payload: dict[str, Any], *, limit: int) -> list[CommentData]:
|
|
comments = payload.get("comments") or payload.get("data", {}).get("comments") or []
|
|
result = []
|
|
for comment in comments[:limit]:
|
|
content = comment.get("text")
|
|
if not content:
|
|
continue
|
|
result.append(
|
|
CommentData(
|
|
source_comment_id=first_present(comment, "comment_id", "cid"),
|
|
content=str(content),
|
|
author=self._author_name(comment),
|
|
like_count=comment.get("digg_count"),
|
|
comment_time=parse_timestamp(first_present(comment, "create_time", "create_time_str")),
|
|
raw_data=comment,
|
|
)
|
|
)
|
|
return result
|
|
|
|
def fetch_hotspots(self, *, limit: int) -> list[HotspotData]:
|
|
payload = self.client.get(
|
|
"/api/v1/douyin/creator/fetch_creator_hot_spot_billboard",
|
|
params={"billboard_tag": 0, "hot_search_type": 1},
|
|
)
|
|
return self.map_hotspots(payload, limit=limit)
|
|
|
|
def search_items_by_hotspot(self, keyword: str, *, limit: int) -> list[ContentItemData]:
|
|
payload = self.client.post(
|
|
"/api/v1/douyin/search/fetch_video_search_v2",
|
|
json={
|
|
"keyword": keyword,
|
|
"cursor": 0,
|
|
"sort_type": "0",
|
|
"publish_time": "0",
|
|
"filter_duration": "0",
|
|
"content_type": "1",
|
|
"search_id": "",
|
|
"backtrace": "",
|
|
},
|
|
)
|
|
return self.map_items(payload, limit=limit)
|
|
|
|
def fetch_comments(self, source_item_id: str, *, limit: int) -> list[CommentData]:
|
|
comments: list[CommentData] = []
|
|
cursor = 0
|
|
page_size = min(20, limit)
|
|
while len(comments) < limit:
|
|
payload = self.client.get(
|
|
"/api/v1/douyin/app/v3/fetch_video_comments",
|
|
params={"aweme_id": source_item_id, "cursor": cursor, "count": page_size},
|
|
)
|
|
page_comments = self.map_comments(payload, limit=limit - len(comments))
|
|
comments.extend(page_comments)
|
|
data = payload.get("data", {})
|
|
next_cursor = data.get("cursor") or payload.get("cursor")
|
|
has_more = data.get("has_more", payload.get("has_more", 0))
|
|
if not page_comments or not has_more or next_cursor in (None, cursor):
|
|
break
|
|
cursor = next_cursor
|
|
return comments[:limit]
|
|
|
|
def _author_name(self, comment: dict[str, Any]) -> str | None:
|
|
user = comment.get("user") or {}
|
|
return user.get("nickname") or user.get("name") if isinstance(user, dict) else None
|