API Documentation

Python Example

import os
import time
import requests
from pathlib import Path
import json

# Configuration
folder_path = r"user folder"  # Folder with images
base_url = "https://pro.convertcom.online/apiv2"
signature_url = f"{base_url}/signature"
upload_url = f"{base_url}/upload-chunk"
status_url = f"{base_url}/conversion/status"
result_url = f"{base_url}/conversion/result"
formats_url = f"{base_url}/formats"
output_folder = r"user folder"  # Output folder
email = "[email protected]"  # Replace with your real email
target_format = "jpg"  # Adjust to desired format

# Setup session
session = requests.Session()

# Create output folder if it doesn't exist
Path(output_folder).mkdir(parents=True, exist_ok=True)

# Validate formats
formats_response = session.get(formats_url)
if formats_response.status_code == 200:
    formats_data = formats_response.json()
    categories = formats_data.get("categories", {})
    all_formats = []
    
    for category, formats in categories.items():
        all_formats.extend(formats.keys())
    
    if all_formats and target_format.lower() not in [fmt.lower() for fmt in all_formats]:
        print(f"Warning: Format {target_format} not found in available formats.")
        print(f"Available formats: {all_formats}")
    else:
        print(f"Format {target_format} validated successfully.")

# List files
image_files = [f for f in os.listdir(folder_path) if f.lower().endswith(('.jpg', '.png', '.jpeg'))]
total_files = len(image_files)
print(f"Total images: {total_files}")

# Get initial signature (valid for 5 minutes)
signature_response = session.post(signature_url, json={"email": email})
if signature_response.status_code != 200:
    raise Exception(f"Error getting initial signature: {signature_response.text}")
signature_data = signature_response.json()
signature = signature_data["signature"]
timestamp = signature_data["timestamp"]
expires_in = signature_data.get("expires_in", 300)
print(f"Signature obtained, valid for {expires_in} seconds")

# Process each file individually
for idx, filename in enumerate(image_files, 1):
    file_path = os.path.join(folder_path, filename)
    job_id = None

    # 1. Upload image
    with open(file_path, 'rb') as file_obj:
        upload_headers = {
            "X-Signature": signature,
            "X-Timestamp": str(timestamp),
            "X-User-Email": email
        }
        
        upload_files = {"file": (filename, file_obj, "application/octet-stream")}
        upload_data = {
            "chunk_index": "0",
            "total_chunks": "1", 
            "file_name": filename,
            "target_format": target_format.lower()
        }
        
        print(f"Sending file {idx}/{total_files}: {filename}")
        upload_response = session.post(upload_url, files=upload_files, data=upload_data, headers=upload_headers)
        
        if upload_response.status_code != 200:
            raise Exception(f"Upload failed for {filename}: {upload_response.text}")
        
        upload_result = upload_response.json()
        if upload_result.get("status") == "completed" or upload_result.get("file_complete"):
            job_id = upload_result.get("job_id")
            print(f"File {filename} uploaded, job_id: {job_id}")
        else:
            raise Exception(f"Incomplete upload for {filename}: {upload_response.text}")

    # 2. Conversion starts automatically by server
    print(f"File uploaded successfully. Server will start conversion automatically for job_id: {job_id}")
    
    # 3. Status polling
    status_headers = {
        "X-Signature": signature,
        "X-Timestamp": str(timestamp),
        "X-User-Email": email
    }
    
    while True:
        status_response = session.get(f"{status_url}/{job_id}", headers=status_headers)
        if status_response.status_code != 200:
            raise Exception(f"Status error for {filename}: {status_response.text}")
        status_data = status_response.json()
        print(f"Status for {filename}: {status_data['status']}, Progress: {status_data['progress']}%")
        if status_data["status"] == "completed":
            break
        time.sleep(5)  # Polling every 5s

    # 4. Get result
    result_response = session.get(f"{result_url}/{job_id}", headers=status_headers)
    if result_response.status_code == 200:
        with open(os.path.join(output_folder, f"converted_{filename}"), 'wb') as f:
            f.write(result_response.content)
        print(f"File {filename} converted and downloaded!")
    else:
        print(f"Download error for {filename}: {result_response.text}")

    # Rate limit: 3s per file
    time.sleep(3)

    # Renew signature if close to expiration
    if time.time() > timestamp + expires_in - 30:
        signature_response = session.post(signature_url, json={"email": email})
        if signature_response.status_code != 200:
            raise Exception(f"Error renewing signature: {signature_response.text}")
        signature_data = signature_response.json()
        signature = signature_data["signature"]
        timestamp = signature_data["timestamp"]
        expires_in = signature_data.get("expires_in", 300)
        print(f"Signature renewed, valid for {expires_in} seconds")

print("Completed!")