Create-knowledge-find-decode/acquisition/oss.py

"""OSS 转存：把帖子媒体直链转存到公网 CDN，返回 cdn_url（绕过防盗链 / 临时签名链时效）。

接口（实测 2026-06-26）：
    POST settings.oss_upload_url  body{src_url, src_type:"video"|"image"}
      → {status:0, oss_object:{cdn_url:"https://res.cybertogether.net/...", oss_object_key, ...}}

纯 parser（parse_oss_response）可离线测；upload_stream 负责 HTTP（失败抛 OssError）；
to_oss 是兜底包装——任何失败返回原 src_url、永不抛，供 decompose 降级不阻断。
"""
from __future__ import annotations

from typing import Any, Callable, Optional

import httpx

from core.config import Settings


class OssError(RuntimeError):
    pass


def parse_oss_response(response: Any) -> str:
    """{status:0, oss_object:{cdn_url}} → cdn_url。status!=0 或缺 cdn_url 抛 OssError。"""
    if not isinstance(response, dict):
        raise OssError("bad_response: not a dict")
    status = response.get("status")
    if status not in (0, "0"):
        raise OssError(f"business_error: status={status} msg={response.get('msg')}")
    obj = response.get("oss_object") or {}
    cdn = obj.get("cdn_url")
    if not cdn:
        raise OssError("empty_cdn_url: oss_object.cdn_url missing")
    return cdn


def upload_stream(
    src_url: str,
    *,
    src_type: str,
    settings: Optional[Settings] = None,
    http_post: Callable[..., Any] = httpx.post,
    env_file: str = ".env",
) -> str:
    """转存一条直链到 OSS，返回 cdn_url；HTTP/解析失败抛 OssError。"""
    settings = settings or Settings.from_env(env_file)
    if src_type not in ("video", "image"):
        raise OssError(f"bad_src_type: {src_type}")
    try:
        resp = http_post(
            settings.oss_upload_url,
            json={"src_url": src_url, "src_type": src_type},
            headers={"Content-Type": "application/json"},
            timeout=settings.oss_upload_timeout_seconds,
        )
        resp.raise_for_status()
        data = resp.json()
    except httpx.HTTPError as exc:
        raise OssError(f"http_error: {exc}") from exc
    except ValueError as exc:
        raise OssError("bad_json") from exc
    return parse_oss_response(data)


def to_oss(
    src_url: str,
    src_type: str,
    *,
    settings: Optional[Settings] = None,
    http_post: Callable[..., Any] = httpx.post,
    env_file: str = ".env",
) -> str:
    """兜底包装：成功返回 cdn_url，任何失败（含空入参）返回原 src_url，永不抛。"""
    if not src_url:
        return src_url
    try:
        return upload_stream(src_url, src_type=src_type, settings=settings,
                             http_post=http_post, env_file=env_file)
    except Exception:
        return src_url