from __future__ import annotations import os from typing import Any, Callable, Mapping from urllib.parse import urlparse import httpx from content_agent.integrations import timeout_config DEFAULT_OSS_UPLOAD_URL = "http://crawler-upload-v2.aiddit.com/crawler/oss/upload_stream" DEFAULT_OSS_TIMEOUT_SECONDS = 300.0 # 5min(原 3600);read 相与单次尝试同为 300s def upload_video_from_url( src_url: str, *, project: str | None = None, referer: Mapping[str, str] | None = None, use_proxy: bool = True, endpoint: str = DEFAULT_OSS_UPLOAD_URL, timeout_seconds: float = DEFAULT_OSS_TIMEOUT_SECONDS, http_post: Callable[..., Any] = httpx.post, ) -> dict[str, Any]: if not src_url: return _failure("missing_src_url") payload: dict[str, Any] = { "src_url": src_url, "src_type": "video", "use_proxy": use_proxy, } payload_mode = "no_referer" if project: payload["project"] = project timeout = timeout_config.as_httpx_timeout(timeout_seconds, read=timeout_config.read_timeout("oss")) try: response = http_post( endpoint, json=payload, timeout=timeout, ) response.raise_for_status() body = response.json() except httpx.HTTPError as exc: response_absent = getattr(exc, "response", None) is None return _failure( "oss_upload_http_error", exception_type=type(exc).__name__, error_message=str(exc), http_status_code=_http_status(exc), oss_payload_mode=payload_mode, endpoint_host=urlparse(endpoint).netloc, read_timeout_seconds=timeout_config.read_timeout("oss"), attempt_timeout_seconds=timeout_seconds, response_absent=response_absent, ) except Exception as exc: return _failure( "oss_upload_failed", exception_type=type(exc).__name__, error_message=str(exc), oss_payload_mode=payload_mode, endpoint_host=urlparse(endpoint).netloc, read_timeout_seconds=timeout_config.read_timeout("oss"), attempt_timeout_seconds=timeout_seconds, response_absent=True, ) oss_object = body.get("oss_object") if isinstance(body, dict) else None if not isinstance(oss_object, dict) or not oss_object.get("cdn_url"): return _failure( "oss_upload_response_invalid", oss_payload_mode=payload_mode, oss_response_summary=_response_summary(body), ) return { "status": "ok", "oss_url": oss_object.get("cdn_url"), "oss_object_key": oss_object.get("oss_object_key"), "save_oss_timestamp": oss_object.get("save_oss_timestamp"), "oss_payload_mode": payload_mode, "raw_payload": body, } def upload_video_from_env( src_url: str, *, referer: Mapping[str, str] | None = None, timeout_seconds: float | None = None, http_post: Callable[..., Any] = httpx.post, ) -> dict[str, Any]: return upload_video_from_url( src_url, project=os.environ.get("CONTENT_AGENT_OSS_PROJECT") or None, referer=referer, use_proxy=_env_bool("CONTENT_AGENT_OSS_USE_PROXY", default=True), endpoint=os.environ.get("CONTENT_AGENT_OSS_UPLOAD_URL") or DEFAULT_OSS_UPLOAD_URL, timeout_seconds=( timeout_seconds if timeout_seconds is not None else float(os.environ.get("CONTENT_AGENT_OSS_TIMEOUT_SECONDS") or DEFAULT_OSS_TIMEOUT_SECONDS) ), http_post=http_post, ) def _failure( failure_type: str, *, exception_type: str | None = None, error_message: str | None = None, http_status_code: int | None = None, oss_payload_mode: str | None = None, oss_response_summary: dict[str, Any] | None = None, endpoint_host: str | None = None, read_timeout_seconds: float | None = None, attempt_timeout_seconds: float | None = None, response_absent: bool | None = None, ) -> dict[str, Any]: result: dict[str, Any] = {"status": "failed", "failure_type": failure_type} if exception_type: result["exception_type"] = exception_type if error_message: result["error_message"] = error_message if http_status_code is not None: result["http_status_code"] = http_status_code if oss_payload_mode: result["oss_payload_mode"] = oss_payload_mode if oss_response_summary: result["oss_response_summary"] = oss_response_summary if endpoint_host: result["endpoint_host"] = endpoint_host if read_timeout_seconds is not None: result["read_timeout_seconds"] = read_timeout_seconds if attempt_timeout_seconds is not None: result["attempt_timeout_seconds"] = attempt_timeout_seconds if response_absent is not None: result["response_absent"] = response_absent return result def _response_summary(body: Any) -> dict[str, Any]: if not isinstance(body, dict): return {"body_type": type(body).__name__} oss_object = body.get("oss_object") return { "status": body.get("status"), "msg": body.get("msg"), "oss_object_present": isinstance(oss_object, dict), "oss_object_has_cdn_url": isinstance(oss_object, dict) and bool(oss_object.get("cdn_url")), } def _http_status(exc: httpx.HTTPError) -> int | None: response = getattr(exc, "response", None) status_code = getattr(response, "status_code", None) return int(status_code) if isinstance(status_code, int) else None def _env_bool(key: str, *, default: bool) -> bool: value = os.environ.get(key) if value is None or value == "": return default return value.strip().lower() not in {"0", "false", "no", "off"}