#!/usr/bin/env python3 """Submit a PDF URL, poll BeaglePDF, and download its result. No dependencies.""" import argparse import json import os from pathlib import Path import sys import time import urllib.error import urllib.parse import urllib.request import uuid class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): raise RuntimeError("Unexpected API redirect; check --api-base.") def api_request(base, key, path, payload=None, idempotency=None): headers = {"Authorization": "Bearer " + key, "Accept": "application/json"} data = None if payload is not None: headers["Content-Type"] = "application/json" headers["Idempotency-Key"] = idempotency data = json.dumps(payload).encode() request = urllib.request.Request(base.rstrip("/") + path, data=data, headers=headers) try: with urllib.request.build_opener(NoRedirect).open(request, timeout=45) as response: return json.load(response) except urllib.error.HTTPError as exc: try: message = json.loads(exc.read()).get("error", {}).get("message", "Request rejected") except (ValueError, AttributeError): message = "Request rejected" raise RuntimeError("API HTTP {}: {}".format(exc.code, message)) from None def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("source_url", help="Public HTTPS URL of the source PDF") parser.add_argument("-o", "--output", type=Path, help="Destination filename (must not already exist)") parser.add_argument("--level", choices=("basic", "standard", "maximum"), default="standard", help="basic: Maximum Compression; standard: Balanced (default); maximum: High Fidelity") parser.add_argument("--callback-url", help="Public HTTPS endpoint to receive the signed completion webhook") parser.add_argument("--poll-interval", type=float, default=5) parser.add_argument("--timeout", type=float, default=1800, help="Maximum polling seconds") parser.add_argument("--api-base", default="https://api.beaglepdf.com/v1") parser.add_argument("--credentials", type=Path, default=Path(__file__).resolve().with_name(".beaglepdf-test-credentials.json")) parser.add_argument("--idempotency-key", default=None, help="Reuse with the same URL and level to resume a submission") args = parser.parse_args() if args.poll_interval <= 0 or args.timeout <= 0: parser.error("Polling interval and timeout must be positive") for url in (args.source_url, args.api_base): if urllib.parse.urlsplit(url).scheme != "https": parser.error("Source URL and API base must use HTTPS") key = os.environ.get("BEAGLEPDF_API_KEY", "").strip() if not key and args.credentials.exists(): key = json.loads(args.credentials.read_text())["api_key"] if not key: parser.error("Set BEAGLEPDF_API_KEY or provide --credentials") name = Path(urllib.parse.unquote(urllib.parse.urlsplit(args.source_url).path)).stem or "document" output = args.output or Path(name + "-compressed.pdf") if output.exists(): parser.error("Output already exists: {}".format(output)) idem = args.idempotency_key or str(uuid.uuid4()) print("Idempotency key: " + idem, flush=True) payload = { "source_url": args.source_url, "compression_level": args.level, } if args.callback_url: payload["callback_url"] = args.callback_url print("Callback requested; webhook delivery runs separately from polling.", flush=True) job = api_request(args.api_base, key, "/compressions", payload, idem) tag = job["tag"] path = "/compressions/" + urllib.parse.quote(tag, safe="") print("Job tag: " + tag, flush=True) start = time.monotonic() while True: job = api_request(args.api_base, key, path) status = job.get("status", "unknown") label = status if status == "processing": label = { "downloading_source": "Downloading source PDF on BeaglePDF...", "compressing": "Compressing PDF", "saving_result": "Saving compressed PDF...", }.get(job.get("stage"), "Processing PDF...") if job.get("stage") == "compressing" and job.get("percent") is not None: label += " — {}%".format(job["percent"]) print("[{:.0f}s] {} (status_code={})".format(time.monotonic() - start, label, job.get("status_code")), flush=True) if job.get("status_code") == 3: break if status not in ("queued", "processing"): raise RuntimeError(job.get("error", {}).get("message", "Compression failed: " + status)) if time.monotonic() - start >= args.timeout: raise RuntimeError("Polling timed out. Job may still run; rerun with --idempotency-key " + idem) time.sleep(args.poll_interval) print("Preparing download...", flush=True) download = api_request(args.api_base, key, path + "/download") url = download["download_url"] if urllib.parse.urlsplit(url).scheme != "https": raise RuntimeError("Download URL must use HTTPS") # Deliberately do not forward the API Authorization header to storage. created = False print("Downloading now...", flush=True) try: with urllib.request.urlopen(url, timeout=60) as response, output.open("xb") as target: created = True length = response.headers.get("Content-Length", "") total = int(length) if length.isdigit() and int(length) > 0 else None downloaded = 0 last_report = time.monotonic() def report_download(): amount = downloaded / (1024 * 1024) if total: print("Downloading: {:.1f}% ({:.2f} / {:.2f} MiB)".format( min(100, downloaded * 100 / total), amount, total / (1024 * 1024)), flush=True) else: print("Downloading: {:.2f} MiB received (total size unknown)".format(amount), flush=True) report_download() first = response.read(1024) if b"%PDF-" not in first: raise RuntimeError("Downloaded response is not a PDF") target.write(first) downloaded += len(first) while True: chunk = response.read(1024 * 1024) if not chunk: break target.write(chunk) downloaded += len(chunk) if time.monotonic() - last_report >= 1: report_download() last_report = time.monotonic() if total is not None and downloaded != total: raise RuntimeError("Download was incomplete; retry with the printed idempotency key") report_download() except Exception: if created: output.unlink(missing_ok=True) raise print("Saved: {} ({} bytes)".format(output.resolve(), output.stat().st_size)) if "reduction_percent" in job: print("Compression reduction: {}%".format(job["reduction_percent"])) if __name__ == "__main__": try: main() except KeyboardInterrupt: print("\nStopped polling; the server job may still complete.", file=sys.stderr) sys.exit(130) except urllib.error.URLError: print("Network request failed. Check connectivity and retry using the printed idempotency key.", file=sys.stderr) sys.exit(1) except (RuntimeError, OSError, ValueError, KeyError) as exc: print("Error: " + str(exc), file=sys.stderr) sys.exit(1)