"""Full-element Partner API client. Both test and live keys run real conversions.""" import argparse import os from pathlib import Path import re import tempfile import time from urllib.parse import quote import zipfile import httpx BASE_URL = "https://2pptx.com/api/v1/partner" SUBMIT_OPTIONS = {"mode": "full_element_editable"} def api_json(client, api_key, method, path, *, body=None, idempotency_key=None): """Retry only the API's idempotent operations, retaining the same payload.""" headers = {"Authorization": f"Bearer {api_key}"} if idempotency_key: headers["Idempotency-Key"] = idempotency_key for attempt in range(5): try: response = client.request(method, BASE_URL + path, headers=headers, json=body) except httpx.TransportError: if attempt == 4: raise RuntimeError("API network failure. Resume with the same idempotency key or job ID.") from None time.sleep(2 ** attempt) continue if response.status_code == 429 or response.status_code >= 500: if attempt < 4: try: delay = float(response.headers.get("Retry-After", 2 ** attempt)) except ValueError: delay = 2 ** attempt time.sleep(max(0, min(60, delay))) continue try: data = response.json() except ValueError: raise RuntimeError(f"API returned non-JSON HTTP {response.status_code}. Resume the same request later.") from None if not response.is_success: error = data.get("error", {}) raise RuntimeError(f"API HTTP {response.status_code}: {error.get('code', 'unknown_error')}: {error.get('message', 'Request failed')}") return data def upload_source(client, upload, source): # Signed storage URLs authorize the transfer themselves. Never send the API key. with source.open("rb") as stream: response = client.put(upload["url"], content=stream, headers={"Content-Type": "application/octet-stream", "Content-Length": str(source.stat().st_size)}, timeout=600) if not response.is_success: raise RuntimeError(f"Upload HTTP {response.status_code}. Resume with the same file and idempotency key before upload expiry.") def wait_for_result(client, api_key, job_id, max_wait): path = "/jobs/" + quote(job_id, safe="") deadline = time.monotonic() + max_wait previous_state = None while True: job = api_json(client, api_key, "GET", path) state = (job["status"], job.get("phase")) if state != previous_state: print(f"{job_id}: {state[0]} / {state[1] or '-'}", flush=True) previous_state = state if job["status"] == "completed": if job.get("mode") != "full_element_editable": raise RuntimeError("This job is not a full-element conversion.") return job if job["status"] in {"failed", "cancelled", "expired"}: raise RuntimeError(f"Job {job_id} is {job['status']}. Contact support before creating a replacement order.") if job["status"] not in {"queued", "processing"}: raise RuntimeError(f"Job {job_id} is {job['status']}; resume upload using the original file and idempotency key.") if time.monotonic() >= deadline: raise RuntimeError(f"Polling stopped; the order is still active. Resume with --job-id {job_id}.") time.sleep(5) def download_result(client, download_url, output, page_count): output.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile(dir=output.parent, suffix=".pptx.part", delete=False) as stream: temporary = Path(stream.name) try: with client.stream("GET", download_url, timeout=600) as response: if response.status_code != 200: raise RuntimeError(f"Download HTTP {response.status_code}. Resume the same job to request a fresh link.") with temporary.open("wb") as stream: for chunk in response.iter_bytes(): stream.write(chunk) expected_size = response.headers.get("Content-Length") if expected_size and temporary.stat().st_size != int(expected_size): raise RuntimeError("Incomplete download; resume the same job.") with zipfile.ZipFile(temporary) as pptx: if pptx.testzip() is not None or "ppt/presentation.xml" not in pptx.namelist(): raise RuntimeError("Downloaded file is not a valid PPTX.") slides = [name for name in pptx.namelist() if re.fullmatch(r"ppt/slides/slide\d+\.xml", name)] if len(slides) != page_count: raise RuntimeError("PPTX page count does not match the completed job. Contact support.") if output.exists(): raise RuntimeError("Output already exists; select another --output path.") temporary.replace(output) finally: temporary.unlink(missing_ok=True) def convert(client, api_key, *, source, idempotency_key, job_id, output, max_wait): if output.exists(): raise RuntimeError("Output already exists; select another --output path.") if source is not None: file_type = source.suffix.lower().lstrip(".") file_size = source.stat().st_size if file_type not in {"pdf", "png", "jpg", "jpeg"} or not 0 < file_size <= 35 * 1024 * 1024: raise RuntimeError("Use a nonempty PDF, PNG, JPG or JPEG file no larger than 35 MiB.") job = api_json(client, api_key, "POST", "/jobs/init-upload", idempotency_key=idempotency_key, body={ "file_name": source.name, "file_type": file_type, "file_size": file_size, "external_job_id": idempotency_key, }) job_id = job["partner_job_id"] print(f"Partner job ID: {job_id}", flush=True) if job["status"] == "awaiting_upload": upload_source(client, job["upload"], source) if job["status"] in {"awaiting_upload", "queued", "processing", "completed"}: # Replays verify the original mode and never create another order. api_json(client, api_key, "POST", "/jobs/" + quote(job_id, safe="") + "/submit", body=SUBMIT_OPTIONS) completed = wait_for_result(client, api_key, job_id, max_wait) download = api_json(client, api_key, "GET", "/jobs/" + quote(job_id, safe="") + "/download") download_result(client, download["download_url"], output, completed["page_count"]) print(f"Saved {completed['page_count']} page(s) to {output}") return job_id def main(): parser = argparse.ArgumentParser(description=__doc__) source = parser.add_mutually_exclusive_group(required=True) source.add_argument("--file", type=Path) source.add_argument("--job-id", help="Resume polling/downloading an already submitted job") parser.add_argument("--idempotency-key", help="Keep this value and the file unchanged on retries") parser.add_argument("--output", type=Path, required=True) parser.add_argument("--max-wait", type=int, default=7200, help="Polling limit in seconds; does not cancel an order") args = parser.parse_args() api_key = os.environ.get("PARTNER_API_KEY", "").strip() if not api_key: parser.error("Set PARTNER_API_KEY in your environment.") if args.file and not args.idempotency_key: parser.error("--file requires --idempotency-key.") if args.max_wait <= 0: parser.error("--max-wait must be positive.") try: # Authentication is added only by api_json, never as a client-wide header. with httpx.Client(timeout=60, follow_redirects=False) as client: convert(client, api_key, source=args.file, idempotency_key=args.idempotency_key, job_id=args.job_id, output=args.output, max_wait=args.max_wait) except httpx.HTTPError as exc: parser.exit(1, f"Transfer failed ({type(exc).__name__}); retry with the same file/key or job ID.\n") except (RuntimeError, OSError, zipfile.BadZipFile) as exc: parser.exit(1, f"{exc}\n") if __name__ == "__main__": main()