85 lines
3.6 KiB
Python
85 lines
3.6 KiB
Python
from typing import Any
|
|
|
|
from app.platforms.base import CommentData, ContentItemData, HotspotData, TikHubClient, first_present, parse_timestamp
|
|
|
|
|
|
class XiaohongshuPlatform:
|
|
def __init__(self, client: TikHubClient | None) -> None:
|
|
self.client = client
|
|
|
|
def map_hotspots(self, payload: dict[str, Any], *, limit: int) -> list[HotspotData]:
|
|
items = payload.get("data", {}).get("data", {}).get("items", [])
|
|
hotspots = []
|
|
for index, item in enumerate(items[:limit], start=1):
|
|
hotspots.append(
|
|
HotspotData(
|
|
source_hot_id=str(item.get("id")) if item.get("id") is not None else None,
|
|
title=str(item.get("title") or ""),
|
|
rank=index,
|
|
heat_value=str(item.get("score")) if item.get("score") is not None else None,
|
|
raw_data=item,
|
|
)
|
|
)
|
|
return hotspots
|
|
|
|
def map_items(self, payload: dict[str, Any], *, limit: int) -> list[ContentItemData]:
|
|
raw_items = payload.get("data", {}).get("data", {}).get("items", [])
|
|
notes = [item.get("note", item) for item in raw_items]
|
|
notes.sort(key=lambda note: 0 if int(note.get("comments_count") or 0) > 0 else 1)
|
|
result = []
|
|
for note in notes[:limit]:
|
|
note_id = note.get("id")
|
|
if not note_id:
|
|
continue
|
|
result.append(
|
|
ContentItemData(
|
|
source_item_id=str(note_id),
|
|
item_type="note",
|
|
title=note.get("title"),
|
|
summary=note.get("desc"),
|
|
url=note.get("url"),
|
|
raw_data=note,
|
|
)
|
|
)
|
|
return result
|
|
|
|
def map_comments(self, payload: dict[str, Any], *, limit: int) -> list[CommentData]:
|
|
comments = payload.get("data", {}).get("data", {}).get("comments", [])
|
|
result = []
|
|
for comment in comments[:limit]:
|
|
content = first_present(comment, "content", "text")
|
|
if not content:
|
|
continue
|
|
result.append(
|
|
CommentData(
|
|
source_comment_id=first_present(comment, "comment_id", "id"),
|
|
content=str(content),
|
|
author=self._author_name(comment),
|
|
like_count=comment.get("like_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]:
|
|
return self.map_hotspots(self.client.get("/api/v1/xiaohongshu/web_v2/fetch_hot_list"), limit=limit)
|
|
|
|
def search_items_by_hotspot(self, keyword: str, *, limit: int) -> list[ContentItemData]:
|
|
payload = self.client.get(
|
|
"/api/v1/xiaohongshu/app_v2/search_notes",
|
|
params={"keyword": keyword, "page": 1, "sort": "general", "note_type": 0},
|
|
)
|
|
return self.map_items(payload, limit=limit)
|
|
|
|
def fetch_comments(self, source_item_id: str, *, limit: int) -> list[CommentData]:
|
|
payload = self.client.get(
|
|
"/api/v1/xiaohongshu/app_v2/get_note_comments",
|
|
params={"note_id": source_item_id, "cursor": "", "index": 0, "pageArea": "UNFOLDED", "sort_strategy": "latest_v2"},
|
|
)
|
|
return self.map_comments(payload, limit=limit)
|
|
|
|
def _author_name(self, comment: dict[str, Any]) -> str | None:
|
|
user = comment.get("user_info") or comment.get("user") or {}
|
|
return user.get("nickname") or user.get("name") if isinstance(user, dict) else None
|