ローカルおよびリモートアップロードの仕組み

import requests
import math
from typing import List
from tqdm import tqdm  # For progress bar
def upload_file_in_chunks(file_path: str, upload_urls: list[str], content_type: str | None = None, chunk_size: int | None = None) -> bool:
    """
    Upload a file in chunks using presigned URLs.
    """
    try:
        # Auto-detect content type based on file extension
        if content_type is None:
            detected_content_type, _ = mimetypes.guess_type(file_path)
            content_type = detected_content_type # Default fallback
        print(f"Detected content type: {content_type}")
        # Get file size
        with open(file_path, 'rb') as f:
            f.seek(0, 2)  # Seek to end of file
            file_size = f.tell()
        # Calculate chunk size if not provided
        if chunk_size is None:
            chunk_size = math.ceil(file_size / len(upload_urls))
        print(f"File size: {file_size} bytes")
        print(f"Chunk size: {chunk_size} bytes")
        print(f"Number of chunks: {len(upload_urls)}")
        # Upload each chunk
        with open(file_path, 'rb') as f:
            with tqdm(total=len(upload_urls), desc="Uploading chunks") as pbar:
                for i, url in enumerate(upload_urls):
                    start_byte = i * chunk_size
                    end_byte = min(start_byte + chunk_size, file_size)
                    # Read chunk from file
                    f.seek(start_byte)
                    chunk = f.read(end_byte - start_byte)
                    print(f"Uploading chunk {i+1}: {len(chunk)} bytes")
                    # Upload chunk with minimal headers matching the signature
                    response = requests.put(
                        url,
                        data=chunk,
                        headers={
                            'content-type': content_type,  
                            'x-amz-acl': 'private'       
                        }
                    )
                    if response.status_code != 200:
                        print(f"Failed to upload chunk {i+1}. Status code: {response.status_code}")
                        print(f"Response text: {response.text}")
                        print(f"Response headers: {dict(response.headers)}")
                        return False
                    else:
                        print(f"Chunk {i+1} uploaded successfully!")
                    pbar.update(1)
        return True
    except Exception as e:
        print(f"Error during upload: {str(e)}")
        return False
# Example usage
if __name__ == "__main__":
    # Replace these with your actual values
    file_path = "/Users/MyComputer/local_upload/sample.jpg"  # Path to your file
    upload_urls = [
        "https://frameio-uploads-development.s3-accelerate.amazonaws.com/parts/fa18ba7b-b3ee-4dd6-9b31-bd07e554241d/part_1?...", 
        "https://frameio-uploads-development.s3-accelerate.amazonaws.com/parts/fa18ba7b-b3ee-4dd6-9b31-bd07e554241d/part_2?...", 
        "https://frameio-uploads-development.s3-accelerate.amazonaws.com/parts/fa18ba7b-b3ee-4dd6-9b31-bd07e554241d/part_3?..."
    ]
    content_type = "image/jpeg"
    print("Starting file upload...")
    success = upload_file_in_chunks(file_path, upload_urls, content_type)
    if success:
        print("File upload completed successfully!")
    else:
        print("File upload failed!")