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

1
import requests
2
import math
3
from typing import List
4
from tqdm import tqdm  # For progress bar
5
6
def upload_file_in_chunks(file_path: str, upload_urls: list[str], content_type: str | None = None, chunk_size: int | None = None) -> bool:
7
    """
8
    Upload a file in chunks using presigned URLs.
9
    """
10
    try:
11
        # Auto-detect content type based on file extension
12
        if content_type is None:
13
            detected_content_type, _ = mimetypes.guess_type(file_path)
14
            content_type = detected_content_type # Default fallback
15
16
        print(f"Detected content type: {content_type}")
17
18
        # Get file size
19
        with open(file_path, 'rb') as f:
20
            f.seek(0, 2)  # Seek to end of file
21
            file_size = f.tell()
22
23
        # Calculate chunk size if not provided
24
        if chunk_size is None:
25
            chunk_size = math.ceil(file_size / len(upload_urls))
26
27
        print(f"File size: {file_size} bytes")
28
        print(f"Chunk size: {chunk_size} bytes")
29
        print(f"Number of chunks: {len(upload_urls)}")
30
31
        # Upload each chunk
32
        with open(file_path, 'rb') as f:
33
            with tqdm(total=len(upload_urls), desc="Uploading chunks") as pbar:
34
                for i, url in enumerate(upload_urls):
35
                    start_byte = i * chunk_size
36
                    end_byte = min(start_byte + chunk_size, file_size)
37
38
                    # Read chunk from file
39
                    f.seek(start_byte)
40
                    chunk = f.read(end_byte - start_byte)
41
42
                    print(f"Uploading chunk {i+1}: {len(chunk)} bytes")
43
44
                    # Upload chunk with minimal headers matching the signature
45
                    response = requests.put(
46
                        url,
47
                        data=chunk,
48
                        headers={
49
                            'content-type': content_type,  
50
                            'x-amz-acl': 'private'       
51
                        }
52
                    )
53
54
                    if response.status_code != 200:
55
                        print(f"Failed to upload chunk {i+1}. Status code: {response.status_code}")
56
                        print(f"Response text: {response.text}")
57
                        print(f"Response headers: {dict(response.headers)}")
58
                        return False
59
                    else:
60
                        print(f"Chunk {i+1} uploaded successfully!")
61
62
                    pbar.update(1)
63
64
        return True
65
66
    except Exception as e:
67
        print(f"Error during upload: {str(e)}")
68
        return False
69
70
# Example usage
71
if __name__ == "__main__":
72
    # Replace these with your actual values
73
    file_path = "/Users/MyComputer/local_upload/sample.jpg"  # Path to your file
74
    upload_urls = [
75
        "https://frameio-uploads-development.s3-accelerate.amazonaws.com/parts/fa18ba7b-b3ee-4dd6-9b31-bd07e554241d/part_1?...", 
76
        "https://frameio-uploads-development.s3-accelerate.amazonaws.com/parts/fa18ba7b-b3ee-4dd6-9b31-bd07e554241d/part_2?...", 
77
        "https://frameio-uploads-development.s3-accelerate.amazonaws.com/parts/fa18ba7b-b3ee-4dd6-9b31-bd07e554241d/part_3?..."
78
    ]
79
    content_type = "image/jpeg"
80
81
    print("Starting file upload...")
82
    success = upload_file_in_chunks(file_path, upload_urls, content_type)
83
84
    if success:
85
        print("File upload completed successfully!")
86
    else:
87
        print("File upload failed!")