91a37b2f18
Treat success=false responses from HDHive search/unlock endpoints as explicit service errors so missing API keys or auth issues are surfaced instead of silently returning empty resources. Co-authored-by: Cursor <cursoragent@cursor.com>
94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
from config import Config
|
|
from http_client import request_json
|
|
from error_handling import AppServiceError, classify_http_error
|
|
|
|
|
|
def _headers():
|
|
headers = {
|
|
"Accept": "application/json",
|
|
"X-API-Key": Config.HDHIVE_API_KEY,
|
|
}
|
|
if Config.HDHIVE_ACCESS_TOKEN:
|
|
headers["Authorization"] = f"Bearer {Config.HDHIVE_ACCESS_TOKEN}"
|
|
return headers
|
|
|
|
|
|
def _ensure_hdhive_success(response_data):
|
|
if not isinstance(response_data, dict):
|
|
return
|
|
success = response_data.get("success")
|
|
if success is not False:
|
|
return
|
|
|
|
code = str(response_data.get("code") or "HDHIVE_API_ERROR")
|
|
message = response_data.get("message") or response_data.get("description") or "HDHIVE request failed"
|
|
category = classify_http_error(400, code)
|
|
raise AppServiceError(
|
|
message,
|
|
category=category,
|
|
code=code,
|
|
status=400,
|
|
provider="hdhive",
|
|
detail=response_data,
|
|
)
|
|
|
|
|
|
def search_resource(media_type, tmdb_id):
|
|
url = f"{Config.HDHIVE_BASE_URL}/api/open/resources/{media_type}/{tmdb_id}"
|
|
result = request_json(
|
|
url,
|
|
headers=_headers(),
|
|
max_retry=Config.MAX_RETRY,
|
|
retry_delay_ms=Config.RETRY_DELAY_MS,
|
|
provider="hdhive",
|
|
)
|
|
_ensure_hdhive_success(result.get("data"))
|
|
return result
|
|
|
|
|
|
def unlock_link(slug):
|
|
url = f"{Config.HDHIVE_BASE_URL}/api/open/resources/unlock"
|
|
result = request_json(
|
|
url,
|
|
method="POST",
|
|
payload={"slug": slug},
|
|
headers={**_headers(), "Content-Type": "application/json"},
|
|
max_retry=Config.MAX_RETRY,
|
|
retry_delay_ms=Config.RETRY_DELAY_MS,
|
|
provider="hdhive",
|
|
)
|
|
_ensure_hdhive_success(result.get("data"))
|
|
return result
|
|
|
|
|
|
def normalize_resource(search_data, unlock_data):
|
|
resolution = (search_data or {}).get("video_resolution")
|
|
source = (search_data or {}).get("source")
|
|
subtitle_language = (search_data or {}).get("subtitle_language")
|
|
unlock_url = (unlock_data or {}).get("full_url") or (unlock_data or {}).get("url") or ""
|
|
media_url = (search_data or {}).get("media_url") or ""
|
|
detail_url = media_url
|
|
if not detail_url and (search_data or {}).get("media_slug"):
|
|
detail_url = f"{Config.HDHIVE_BASE_URL}/movie/{(search_data or {}).get('media_slug')}"
|
|
|
|
validate_status = (search_data or {}).get("validate_status")
|
|
validate_message = (search_data or {}).get("validate_message")
|
|
|
|
return {
|
|
"resourceTitle": (search_data or {}).get("title", ""),
|
|
"quality": ", ".join(resolution) if isinstance(resolution, list) else "",
|
|
"size": (search_data or {}).get("share_size", ""),
|
|
"diskType": (search_data or {}).get("pan_type", ""),
|
|
"source": ", ".join(source) if isinstance(source, list) else "",
|
|
"subtitleLanguage": ", ".join(subtitle_language)
|
|
if isinstance(subtitle_language, list)
|
|
else "",
|
|
"slug": (search_data or {}).get("slug", ""),
|
|
"unlockUrl": unlock_url,
|
|
"detailUrl": detail_url,
|
|
"availability": "available" if unlock_url else ("index_only" if detail_url else "unknown"),
|
|
"validateStatus": validate_status,
|
|
"validateMessage": validate_message,
|
|
"raw": {"searchData": search_data, "unlockData": unlock_data},
|
|
}
|