fix: 重试临时外部 API 错误

This commit is contained in:
meijiali
2026-07-06 16:09:08 +08:00
parent 918dd62e21
commit 44d459fc50
2 changed files with 81 additions and 5 deletions
+26 -5
View File
@@ -94,15 +94,36 @@ class TikHubClient:
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,
)
if attempt >= self.max_retries:
raise PlatformAPIError(
self._format_api_error(response),
error_type="api_error",
status_code=response.status_code,
)
time.sleep(2**attempt)
continue
return response.json()
raise PlatformAPIError("External API request failed", error_type="api_error", status_code=last_status)
def _format_api_error(self, response: httpx.Response) -> str:
message = f"External API returned HTTP {response.status_code}"
detail = self._response_error_detail(response)
return f"{message}: {detail}" if detail else message
def _response_error_detail(self, response: httpx.Response) -> str | None:
try:
payload = response.json()
except ValueError:
text = response.text.strip()
return text[:200] if text else None
if isinstance(payload, dict):
for key in ("message_zh", "message", "error", "detail"):
value = payload.get(key)
if value:
return str(value)[:200]
return None
def parse_timestamp(value: Any) -> datetime | None:
if value in (None, ""):