content-find-agent-new/content_agent/integrations/oss_upload.py

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"}