SupplyAgent/supply_infra/oss/client.py

"""Aliyun OSS client for uploading agent run artifacts."""

from __future__ import annotations

from pathlib import Path

import oss2

from supply_infra.config import InfraSettings, get_infra_settings


class OssClient:
    """Thin wrapper around oss2 Bucket."""

    def __init__(
        self,
        settings: InfraSettings | None = None,
        *,
        root_prefix: str | None = None,
    ) -> None:
        self.settings = settings or get_infra_settings()
        if not self.settings.aliyun_oss_access_key_id:
            raise ValueError("ALIYUN_OSS_ACCESS_KEY_ID is not configured")
        if not self.settings.aliyun_oss_access_key_secret:
            raise ValueError("ALIYUN_OSS_ACCESS_KEY_SECRET is not configured")

        endpoint = f"https://oss-{self.settings.aliyun_oss_region}.aliyuncs.com"
        auth = oss2.Auth(
            self.settings.aliyun_oss_access_key_id,
            self.settings.aliyun_oss_access_key_secret,
        )
        self._bucket = oss2.Bucket(
            auth,
            endpoint,
            self.settings.aliyun_oss_bucket,
            connect_timeout=self.settings.aliyun_oss_connect_timeout_seconds,
        )
        prefix = root_prefix if root_prefix is not None else self.settings.aliyun_oss_root_prefix
        self._root = prefix.strip("/")
        self._public_base = self.settings.aliyun_oss_public_base_url.rstrip("/")

    def object_key(self, *parts: str) -> str:
        cleaned = [self._root] if self._root else []
        for part in parts:
            p = str(part).strip("/")
            if p:
                cleaned.append(p)
        return "/".join(cleaned)

    def public_url(self, object_key: str) -> str:
        key = object_key.lstrip("/")
        return f"{self._public_base}/{key}"

    def upload_file(self, local_path: Path | str, object_key: str) -> str:
        """Upload a local file and return its public CDN URL."""
        path = Path(local_path)
        if not path.is_file():
            raise FileNotFoundError(f"File not found: {path}")
        self._bucket.put_object_from_file(object_key, str(path))
        return self.public_url(object_key)