import hashlib
import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
API_KEY = 'YOUR_API_KEY' # Get from dashboard settings
BASE_URL = 'https://api.beecargo.net'
MULTIPART_THRESHOLD = 4 * 1024 * 1024 # 4MB
def get_optimal_parallelism(file_size):
"""Determine optimal number of parallel uploads based on file size"""
if file_size > 50 * 1024**3: # > 50GB
return 3
elif file_size > 10 * 1024**3: # > 10GB
return 4
elif file_size > 1 * 1024**3: # > 1GB
return 5
else:
return 6
def upload_small_file(file_path, folder_id=None):
"""Upload small files (< 4MB) directly"""
url = f'{BASE_URL}/files/upload'
file_size = os.path.getsize(file_path)
file_name = os.path.basename(file_path)
print(f"Uploading {file_name} ({file_size / (1024**2):.2f} MB)...")
headers = {}
if API_KEY:
headers['Authorization'] = f'Bearer {API_KEY}'
with open(file_path, 'rb') as f:
files = {'file': (file_name, f)}
data = {}
if folder_id:
data['folderId'] = folder_id
response = requests.post(url, headers=headers, files=files, data=data)
response.raise_for_status()
result = response.json()
if not result.get('success'):
raise Exception(result.get('error', 'Upload failed'))
print("Upload completed!")
return result['data']
def upload_large_file_parallel(file_path, folder_id=None):
"""Upload large files (>= 4MB) with parallel multipart parts"""
file_size = os.path.getsize(file_path)
file_name = os.path.basename(file_path)
headers = {'Content-Type': 'application/json'}
if API_KEY:
headers['Authorization'] = f'Bearer {API_KEY}'
print(f"Initializing upload for {file_name} ({file_size / (1024**3):.2f} GB)...")
init_res = requests.post(
f'{BASE_URL}/files/multipart/init',
headers=headers,
json={
'fileName': file_name,
'fileSize': file_size,
'fileType': 'application/octet-stream',
'folderId': folder_id
}
)
init_res.raise_for_status()
init_data = init_res.json()
upload_id = init_data['uploadId']
key = init_data['key']
chunk_size = init_data['chunkSize']
total_parts = init_data['totalParts']
upload_session_token = init_data.get('uploadSessionToken')
print(f"Upload initialized: {total_parts} parts x {chunk_size / (1024**2):.1f} MB")
part_digests = {}
with open(file_path, 'rb') as f:
for part_number in range(1, total_parts + 1):
chunk = f.read(chunk_size)
part_digests[str(part_number)] = hashlib.sha256(chunk).hexdigest()
batch_body = {
'key': key,
'uploadId': upload_id,
'totalParts': total_parts,
'partDigests': part_digests,
}
if upload_session_token:
batch_body['uploadSessionToken'] = upload_session_token
print(f"Getting presigned URLs for {total_parts} parts...")
urls_res = requests.post(
f'{BASE_URL}/files/multipart/batch-urls',
headers=headers,
json=batch_body,
)
urls_res.raise_for_status()
urls_data = urls_res.json()
if not urls_data.get('success'):
raise Exception(urls_data.get('error', 'Failed to get upload URLs'))
urls_dict = urls_data['urls']
print("Got all presigned URLs")
parallelism = get_optimal_parallelism(file_size)
print(f"Uploading {total_parts} parts with {parallelism}x parallelism...")
start_time = time.time()
uploaded_parts = []
completed_parts = 0
def upload_part(part_number):
url = urls_dict[str(part_number)]
with open(file_path, 'rb') as f:
f.seek((part_number - 1) * chunk_size)
chunk = f.read(chunk_size)
max_retries = 3
for attempt in range(max_retries):
try:
res = requests.put(url, data=chunk, timeout=300)
res.raise_for_status()
etag = res.headers.get('ETag', '').strip('"')
return {'partNumber': part_number, 'etag': etag}
except Exception:
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt)
with ThreadPoolExecutor(max_workers=parallelism) as executor:
futures = {
executor.submit(upload_part, i): i for i in range(1, total_parts + 1)
}
for future in as_completed(futures):
part_number = futures[future]
try:
result = future.result()
uploaded_parts.append(result)
completed_parts += 1
elapsed = time.time() - start_time
uploaded_bytes = completed_parts * chunk_size
speed = uploaded_bytes / elapsed / (1024**2) if elapsed > 0 else 0
progress = (completed_parts / total_parts) * 100
eta = (
(total_parts - completed_parts)
* chunk_size
/ (uploaded_bytes / elapsed)
if uploaded_bytes > 0
else 0
)
bar_length = 30
filled = int(bar_length * completed_parts / total_parts)
bar = '#' * filled + '-' * (bar_length - filled)
print(
f"\r[{bar}] {progress:.1f}% | {completed_parts}/{total_parts} parts | "
f"{speed:.1f} MB/s | ETA: {eta:.0f}s",
end='',
flush=True,
)
except Exception as e:
print(f"\nFailed to upload part {part_number}: {e}")
raise
elapsed_total = time.time() - start_time
avg_speed = file_size / elapsed_total / (1024**2)
print(f"\nUpload completed in {elapsed_total:.1f}s (avg: {avg_speed:.1f} MB/s)")
uploaded_parts.sort(key=lambda x: x['partNumber'])
print("Finalizing upload...")
complete_body = {
'key': key,
'uploadId': upload_id,
'parts': uploaded_parts,
'fileName': file_name,
'fileSize': file_size,
'contentType': 'application/octet-stream',
'folderId': folder_id,
}
if upload_session_token:
complete_body['uploadSessionToken'] = upload_session_token
complete_res = requests.post(
f'{BASE_URL}/files/multipart/complete',
headers=headers,
json=complete_body,
)
complete_res.raise_for_status()
result = complete_res.json()
if not result.get('success'):
raise Exception(result.get('error'))
print("File saved")
return result['file']
def upload_file(file_path, folder_id=None):
"""Main upload function - direct under 4MB, parallel multipart otherwise"""
file_size = os.path.getsize(file_path)
if file_size < MULTIPART_THRESHOLD:
return upload_small_file(file_path, folder_id)
return upload_large_file_parallel(file_path, folder_id)
if __name__ == '__main__':
try:
result = upload_file('./large_file.zip')
print(f"\nSuccess! Share: https://beecargo.net/d/{result['shortId']}")
except Exception as e:
print(f'Upload failed: {e}')
