HelperUtils

async def prepare(self, file_paths: Sequence[str | Path]) -> None:
    """准备上传,获取或复用有效的临时凭证(按文件 SHA1 去重)."""
    paths = tuple(Path(p) for p in file_paths)
    if not paths:
        raise ValueError("至少需要提供一个文件路径.")

    file_info_map: dict[int, InitUploadFileDict] = {}
    async with anyio.create_task_group() as tg:

        async def _get_info(idx: int, p: anyio.Path):
            file_info_map[idx] = await self._get_file_info(p)

        for i, p in enumerate(paths):
            tg.start_soon(_get_info, i, anyio.Path(p))

    file_infos = [file_info_map[i] for i in range(len(paths))]
    current_shas = tuple(info["FileSha1"] for info in file_infos)

    if self._last_file_shas != current_shas:
        self._init_data = None
        self._last_file_shas = current_shas

    if self._init_data is not None:
        now = int(time.time())
        # 留 10 分钟余量防止临界过期
        if now < self._init_data.auth_info.expired_time - 600:
            return

    self._init_data = await self.api.init_upload(
        bus_id=self.bus_id,
        files=list(file_infos),
        credential=self.credential,
    )

async def upload(
    self,
    file_paths: str | Path | Sequence[str | Path],
) -> list[UploadObjectInfo]:
    """执行多文件的完整上传流程.

    Args:
        file_paths: 单个或多个文件路径.

    Returns:
        list[UploadObjectInfo]: 所有文件上传成功后的信息列表.
    """
    if isinstance(file_paths, str | Path):
        file_paths = [file_paths]

    await self.prepare(file_paths)

    if self._init_data is None:
        raise ApiDataError("获取上传凭证失败: 服务器未返回凭证信息")

    auth_info = self._init_data.auth_info
    files_info = self._init_data.files

    if not files_info or len(files_info) != len(file_paths):
        raise ValueError("InitUpload 返回的文件目标数量不匹配.")

    secret_id = auth_info.secret_id
    secret_key = auth_info.secret_key
    token = auth_info.token

    # 并发直传 COS
    semaphore = anyio.Semaphore(self.max_concurrency)

    async def _upload_worker(
        path: Path,
        region: str,
        secret_id: str,
        secret_key: str,
        token: str,
        bucket_name: str,
        object_key: str,
        file_size: int = 0,
    ) -> None:
        async with semaphore:
            await to_thread.run_sync(
                self._upload_to_cos,
                path,
                region,
                secret_id,
                secret_key,
                token,
                bucket_name,
                object_key,
                file_size,
            )

    finish_results: list[FinishUploadResultDict] = []

    stat_map: dict[int, os.stat_result] = {}
    async with anyio.create_task_group() as tg:

        async def _get_stat(idx: int, p: str | Path):
            stat_map[idx] = await anyio.Path(p).stat()

        for i, p in enumerate(file_paths):
            tg.start_soon(_get_stat, i, p)

    async with anyio.create_task_group() as tg:
        for i, file_info in enumerate(files_info):
            buckets = file_info.buckets
            if not buckets:
                raise ApiDataError(f"获取上传凭证失败: 文件 {file_paths[i]} 未返回目标存储桶信息.")

            target_bucket = buckets[0]
            bucket_info = target_bucket.bucket
            bucket_name = bucket_info.name
            region = bucket_info.region
            object_key = file_info.object_key
            upload_status = target_bucket.upload_status
            stat = stat_map.get(i)
            fsize = stat.st_size if stat is not None else 0

            if not all([secret_id, secret_key, token, object_key, bucket_name, region]):
                raise ApiDataError(f"获取上传凭证失败: 文件 {file_paths[i]} 上传凭证信息不完整.")

            if upload_status != 1:
                tg.start_soon(
                    _upload_worker,
                    Path(file_paths[i]),
                    region,
                    secret_id,
                    secret_key,
                    token,
                    bucket_name,
                    object_key,
                    fsize,
                )

            finish_results.append(
                {
                    "Storage": {"Bucket": {"Name": bucket_name, "Region": region}, "ObjectKey": object_key},
                    "UploadResult": 0,
                }
            )
    finish_data = await self.api.finish_upload(
        bus_id=self.bus_id,
        results=finish_results,
        credential=self.credential,
    )
    objects = finish_data.objects
    if not objects:
        raise ApiDataError("FinishUpload 未返回上传成功的文件对象.")
    return objects