Method merge_chunks

core/storage.py:419–492

合并 S3 上的分片文件 使用 S3 的 multipart upload API 实现流式合并，避免内存问题

(self, upload_id: str, chunk_info: UploadChunk, save_path: str)

            )

    async def merge_chunks(self, upload_id: str, chunk_info: UploadChunk, save_path: str) -> tuple[str, str]:
        """
        合并 S3 上的分片文件
        使用 S3 的 multipart upload API 实现流式合并，避免内存问题
        """
        file_sha256 = hashlib.sha256()
        chunk_dir = str(Path(save_path).parent / "chunks" / upload_id)

        async with self._client() as s3:
            # 创建 multipart upload
            mpu = await s3.create_multipart_upload(
                Bucket=self.bucket_name,
                Key=save_path,
                ContentType='application/octet-stream'
            )
            mpu_id = mpu['UploadId']
            parts = []

            try:
                # 按顺序读取、验证并上传每个分片
                for i in range(chunk_info.total_chunks):
                    chunk_key = f"{chunk_dir}/{i}.part"
                    chunk_record = await UploadChunk.filter(upload_id=upload_id, chunk_index=i).first()
                    if not chunk_record:
                        raise ValueError(f"分片{i}记录不存在")

                    try:
                        response = await s3.get_object(
                            Bucket=self.bucket_name,
                            Key=chunk_key
                        )
                        chunk_data = await response['Body'].read()
                    except Exception as e:
                        raise ValueError(f"分片{i}文件不存在: {e}")

                    current_hash = hashlib.sha256(chunk_data).hexdigest()
                    if current_hash != chunk_record.chunk_hash:
                        raise ValueError(f"分片{i}哈希不匹配: 期望 {chunk_record.chunk_hash}, 实际 {current_hash}")

                    file_sha256.update(chunk_data)

                    # 上传分片到 multipart upload
                    part_response = await s3.upload_part(
                        Bucket=self.bucket_name,
                        Key=save_path,
                        UploadId=mpu_id,
                        PartNumber=i + 1,  # S3 part numbers start at 1
                        Body=chunk_data
                    )
                    parts.append({
                        'PartNumber': i + 1,
                        'ETag': part_response['ETag']
                    })

                    # 释放内存
                    del chunk_data

                # 完成 multipart upload