124 lines
3.9 KiB
Python
124 lines
3.9 KiB
Python
import time
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
|
|
@dataclass
|
|
class HotspotData:
|
|
source_hot_id: str | None
|
|
title: str
|
|
rank: int | None = None
|
|
heat_value: str | None = None
|
|
raw_data: dict[str, Any] | None = None
|
|
|
|
|
|
@dataclass
|
|
class ContentItemData:
|
|
source_item_id: str
|
|
item_type: str
|
|
title: str | None = None
|
|
summary: str | None = None
|
|
url: str | None = None
|
|
raw_data: dict[str, Any] | None = None
|
|
|
|
|
|
@dataclass
|
|
class CommentData:
|
|
source_comment_id: str | None
|
|
content: str
|
|
author: str | None = None
|
|
like_count: int | None = None
|
|
comment_time: datetime | None = None
|
|
raw_data: dict[str, Any] | None = None
|
|
|
|
|
|
class PlatformAPIError(Exception):
|
|
def __init__(self, message: str, *, error_type: str, status_code: int | None = None):
|
|
super().__init__(message)
|
|
self.error_type = error_type
|
|
self.status_code = status_code
|
|
|
|
|
|
class TikHubClient:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
base_url: str,
|
|
api_key: str,
|
|
timeout_seconds: int = 20,
|
|
max_retries: int = 3,
|
|
http_client: httpx.Client | None = None,
|
|
) -> None:
|
|
self.base_url = base_url.rstrip("/")
|
|
self.api_key = api_key
|
|
self.timeout_seconds = timeout_seconds
|
|
self.max_retries = max_retries
|
|
self._http_client = http_client or httpx.Client(timeout=timeout_seconds)
|
|
|
|
def get(self, path: str, *, params: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
return self._request("GET", path, params=params)
|
|
|
|
def post(self, path: str, *, json: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
return self._request("POST", path, json=json)
|
|
|
|
def _request(self, method: str, path: str, **kwargs) -> dict[str, Any]:
|
|
url = f"{self.base_url}{path}"
|
|
headers = {"Authorization": f"Bearer {self.api_key}"} if self.api_key else {}
|
|
last_status: int | None = None
|
|
for attempt in range(self.max_retries + 1):
|
|
try:
|
|
response = self._http_client.request(method, url, headers=headers, **kwargs)
|
|
except httpx.RequestError as exc:
|
|
if attempt >= self.max_retries:
|
|
raise PlatformAPIError("External API request failed", error_type="network_error") from exc
|
|
time.sleep(2**attempt)
|
|
continue
|
|
|
|
last_status = response.status_code
|
|
if response.status_code == 429:
|
|
if attempt >= self.max_retries:
|
|
raise PlatformAPIError(
|
|
"External API rate limited",
|
|
error_type="rate_limited",
|
|
status_code=response.status_code,
|
|
)
|
|
time.sleep(2**attempt)
|
|
continue
|
|
if response.status_code == 401:
|
|
raise PlatformAPIError(
|
|
"TikHub 鉴权失败，请检查 TIKHUB_API_KEY 是否有效",
|
|
error_type="auth_error",
|
|
status_code=response.status_code,
|
|
)
|
|
if response.is_error:
|
|
raise PlatformAPIError(
|
|
f"External API returned HTTP {response.status_code}",
|
|
error_type="api_error",
|
|
status_code=response.status_code,
|
|
)
|
|
return response.json()
|
|
|
|
raise PlatformAPIError("External API request failed", error_type="api_error", status_code=last_status)
|
|
|
|
|
|
def parse_timestamp(value: Any) -> datetime | None:
|
|
if value in (None, ""):
|
|
return None
|
|
try:
|
|
number = int(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if number > 10_000_000_000:
|
|
number = number // 1000
|
|
return datetime.fromtimestamp(number, tz=UTC)
|
|
|
|
|
|
def first_present(data: dict[str, Any], *keys: str) -> Any:
|
|
for key in keys:
|
|
if key in data and data[key] not in (None, ""):
|
|
return data[key]
|
|
return None
|