ai-flow-python/app/util/aliyun_util.py

import logging
import os

from app.config import settings
from app.util.file_util import generate_unique_filename

logger = logging.getLogger(__name__)


def upload_file(
    local_path: str,
    oss_key: str,
    endpoint: str | None = None,
    access_key_id: str | None = None,
    access_key_secret: str | None = None,
) -> str | None:
    if not local_path or not oss_key:
        logger.error("Local file path or OSS key is blank.")
        return None

    if not os.path.isfile(local_path):
        logger.error("Local file does not exist: %s", local_path)
        return None

    ep = endpoint or settings.aliyun_endpoint
    aki = access_key_id or settings.aliyun_access_key_id
    aks = access_key_secret or settings.aliyun_access_key_secret
    bucket_name = settings.aliyun_bucket_name

    try:
        import oss2

        auth = oss2.Auth(aki, aks)
        bucket = oss2.Bucket(auth, ep, bucket_name)

        logger.debug("Uploading to OSS: %s -> %s", local_path, oss_key)
        bucket.put_object_from_file(oss_key, local_path)

        oss_url = f"https://{bucket_name}.{ep}/{oss_key}"
        logger.debug("Successfully uploaded: %s", oss_url)
        return oss_url
    except Exception as e:
        logger.error("Failed to upload file to OSS: %s -> %s, %s", local_path, oss_key, e)
        return None


def upload_files(
    local_paths: list[str],
    oss_prefix: str,
    endpoint: str | None = None,
    access_key_id: str | None = None,
    access_key_secret: str | None = None,
) -> list[str]:
    uploaded: list[str] = []
    if not local_paths:
        return uploaded

    if not oss_prefix:
        oss_prefix = "uploads/"
    if not oss_prefix.endswith("/"):
        oss_prefix += "/"

    for path in local_paths:
        try:
            name = generate_unique_filename(os.path.basename(path))
            key = oss_prefix + name
            url = upload_file(path, key, endpoint, access_key_id, access_key_secret)
            if url:
                uploaded.append(url)
        except Exception as e:
            logger.error("Failed to upload file: %s, %s", path, e)

    return uploaded