Viên ngọc phương Đông

import os
from typing import Any, Dict, Optional

import requests
from loguru import logger

from app.errors.error_codes import ErrorCode
from app.errors.exceptions import BartError
from app.utils.job_activity_logger import JobActivityLogger

activity_logger: Optional[JobActivityLogger] = None

def _set_logger(cbs_job_id, is_cli: bool = False) -> None:
    global activity_logger
    activity_logger = JobActivityLogger(cbs_job_id, echo_to_stdout=is_cli)

def upload_bulk(request_body: Dict[str, Any], is_cli: bool = False) -> Dict[str, Any]:
    cbs_job_id = request_body.get("cbs_job_id")
    if cbs_job_id is None:
        raise BartError(ErrorCode.VALIDATION_FAILED, "cbs_job_id is required")
    _set_logger(cbs_job_id, is_cli)
    assert activity_logger is not None

upload_url = (request_body.get("upload_url") or "").rstrip("/")
    if not upload_url:
        raise BartError(ErrorCode.VALIDATION_FAILED, "upload_url is required")

upload_requests = request_body.get("upload_requests") or {}
    if not isinstance(upload_requests, dict) or not upload_requests:
        raise BartError(ErrorCode.VALIDATION_FAILED, "upload_requests must be a non-empty object")

object_prefix = (request_body.get("object_prefix") or "").strip("/")
    if not object_prefix:
        raise BartError(ErrorCode.VALIDATION_FAILED, "object_prefix is required")

action_data: Dict[str, Any] = {}

for req_id, request_info in upload_requests.items():
        req_id_s = str(req_id)
        if not isinstance(request_info, dict):
            raise BartError(ErrorCode.VALIDATION_FAILED, f"upload_requests[{req_id!r}] must be an object")

files_path_prefix = (request_info.get("files_path_prefix") or "").strip("/")
        if not files_path_prefix:
            raise BartError(
                ErrorCode.VALIDATION_FAILED,
                f"upload_requests[{req_id!r}].files_path_prefix is required",
            )

files_map = request_info.get("files")
        if not isinstance(files_map, dict) or not files_map:
            raise BartError(
                ErrorCode.VALIDATION_FAILED,
                f"upload_requests[{req_id!r}].files must be a non-empty object",
            )

action_data[req_id_s] = {}

for file_key, local_path in files_map.items():
            file_key_s = str(file_key)
            try:
                if not local_path or not isinstance(local_path, str):
                    raise BartError(ErrorCode.VALIDATION_FAILED, f"Invalid local path for {req_id_s}/{file_key!r}")
                local_path = os.path.normpath(local_path)
                if not os.path.isfile(local_path):
                    raise BartError(ErrorCode.FILE_NOT_FOUND, f"File not found: {local_path}")

local_filename = os.path.basename(local_path)
                put_url = f"{upload_url}/{object_prefix}/{files_path_prefix}/{local_filename}"

activity_logger.note(f"Uploading {local_path} -> {put_url}")
                try:
                    with open(local_path, "rb") as f:
                        response = requests.put(put_url, data=f, timeout=3600)
                        response.raise_for_status()
                except requests.RequestException as e:
                    raise BartError(ErrorCode.COMMAND_FAILED, f"Upload failed for {req_id_s}/{file_key_s}: {e}") from e

logger.info(f"Upload successful: {local_path} -> {put_url}")
                action_data[req_id_s][f"{file_key_s}_upload_status"] = "success"
            except BartError:
                action_data[req_id_s][f"{file_key_s}_upload_status"] = "failed"

activity_logger.success("upload_bulk completed")
    return action_data

Registrieren