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!")
