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

from __future__ import annotations

import os
from typing import Any, Callable, Mapping

import httpx


DEFAULT_OSS_UPLOAD_URL = "http://crawler-upload-v2.aiddit.com/crawler/oss/upload_stream"
DEFAULT_OSS_TIMEOUT_SECONDS = 60 * 60.0


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
    try:
        response = http_post(endpoint, json=payload, timeout=timeout_seconds)
        response.raise_for_status()
        body = response.json()
    except httpx.HTTPError as exc:
        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,
        )
    except Exception as exc:
        return _failure(
            "oss_upload_failed",
            exception_type=type(exc).__name__,
            error_message=str(exc),
            oss_payload_mode=payload_mode,
        )

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