#!/usr/bin/env python3 """Run one billed PDF conversion, or resume its saved task. Python 3.9+, no dependencies.""" import argparse from email.utils import parsedate_to_datetime import io import zipfile import json import os from pathlib import Path import sys import tempfile import time from urllib.error import HTTPError, URLError from urllib.parse import quote, urlparse from urllib.request import HTTPRedirectHandler, Request, build_opener API_BASE = "https://api.pdfcraft.ai/v1/pcc-api" SAMPLE_URL = "https://pdfcraft.ai/examples/api-quickstart.pdf" EXPECTED_TEXT = "Paper to structured text." REQUEST_TIMEOUT = 30 POLL_INITIAL = 2 POLL_MAX = 30 MAX_RESPONSE_BYTES = 10 * 1024 * 1024 DEFAULT_DEADLINE = 600 RETRYABLE_STATUS = {429, 502, 503, 504} class NoRedirect(HTTPRedirectHandler): """Never forward credentials or change a POST through an HTTP redirect.""" def redirect_request(self, req, fp, code, msg, headers, newurl): return None class ApiError(Exception): def __init__(self, status, retry_after=None): super().__init__(f"HTTP {status}") self.status = status self.retry_after = retry_after def fetch(url, key=None, payload=None, timeout=REQUEST_TIMEOUT, method=None, headers=None): headers = {"User-Agent": "PDFCraft-Quickstart/2.0", "Accept": "application/json" if key else "*/*", **(headers or {})} if key: headers["Authorization"] = f"Bearer {key}" data = None if isinstance(payload, bytes): data = payload elif payload is not None: headers["Content-Type"] = "application/json" data = json.dumps(payload).encode() request = Request(url, data=data, headers=headers, method=method) try: with build_opener(NoRedirect()).open(request, timeout=timeout) as response: body = response.read(MAX_RESPONSE_BYTES + 1) if len(body) > MAX_RESPONSE_BYTES: raise ValueError("Response exceeds the example's size limit") if key: if "application/json" not in response.headers.get("Content-Type", "").lower(): raise ValueError("API response is not JSON") result = json.loads(body) if not isinstance(result, dict): raise ValueError("API response must be a JSON object") return result if "text/html" in response.headers.get("Content-Type", "").lower(): raise ValueError("Download returned an HTML page instead of Markdown") return body except HTTPError as error: retry = error.headers.get("Retry-After") error.close() raise ApiError(error.code, retry) from None def remaining(deadline): seconds = deadline - time.monotonic() if seconds <= 0: raise TimeoutError("Deadline reached; resume the saved task later") return min(REQUEST_TIMEOUT, seconds) def pause(seconds, deadline): available = deadline - time.monotonic() if seconds >= available: raise TimeoutError("Next poll would exceed the deadline; resume the saved task later") time.sleep(seconds) def save_accepted_task(path, task_id): temporary = None try: with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.", delete=False) as state_file: temporary = Path(state_file.name) json.dump({"uuid": task_id, "sampleURL": SAMPLE_URL}, state_file) state_file.flush() os.fsync(state_file.fileno()) os.replace(temporary, path) finally: if temporary is not None: temporary.unlink(missing_ok=True) def https_url(value): parsed = urlparse(value) if isinstance(value, str) else None if not parsed or parsed.scheme != "https" or not parsed.netloc or parsed.username or parsed.password: raise ValueError("Invalid HTTPS storage URL") return value def run(args): key = os.environ.get("PDF_CRAFT_API_KEY", "").strip() if not key: raise ValueError("Set PDF_CRAFT_API_KEY securely in your environment") if args.timeout <= 0: raise ValueError("--timeout must be positive") if args.output.exists(): raise FileExistsError("Output already exists; choose another --output") deadline = time.monotonic() + args.timeout record = args.resume or args.state_file if args.resume: task_id = json.loads(record.read_text()).get("uuid") if not isinstance(task_id, str) or not task_id: raise ValueError("No saved uuid. Investigate the previous creation before creating another task") else: with record.open("x", encoding="utf-8") as state_file: json.dump({"state": "creation_unknown"}, state_file) state_file.flush() os.fsync(state_file.fileno()) created = fetch(f"{API_BASE}/transforms", key, {"input_format": "pdf"}, remaining(deadline)) task_id = created.get("uuid") if not isinstance(task_id, str) or not task_id: raise ValueError("Creation did not return uuid; do not automatically recreate") save_accepted_task(record, task_id) task_url = f"{API_BASE}/transforms/{quote(task_id, safe='')}" delay = POLL_INITIAL while True: try: result = fetch(task_url, key, timeout=remaining(deadline)) except ApiError as error: if error.status not in RETRYABLE_STATUS: raise retry_delay = delay if error.retry_after: try: retry_delay = max(delay, float(error.retry_after)) except ValueError: retry_delay = max(delay, parsedate_to_datetime(error.retry_after).timestamp() - time.time()) pause(retry_delay, deadline) except URLError: pause(delay, deadline) else: status = result.get("status") if status == "waiting_for_upload": upload = fetch(f"{task_url}/upload-urls", key, timeout=remaining(deadline)) if "put_url" in upload: headers = upload.get("headers") if not isinstance(headers, dict) or any(k.lower() == "authorization" for k in headers): raise ValueError("Invalid signed upload headers") source = fetch(SAMPLE_URL, timeout=remaining(deadline)) fetch(https_url(upload["put_url"]), payload=source, method="PUT", headers=headers, timeout=remaining(deadline)) # Same parameters on every resume. A terminal task is never relaunched. fetch(f"{task_url}/launch", key, {"output_format": "markdown"}, remaining(deadline)) elif status == "succeeded": urls = fetch(f"{task_url}/download-urls", key, timeout=remaining(deadline)) data = fetch(https_url(urls.get("get_url")), timeout=remaining(deadline)) with zipfile.ZipFile(io.BytesIO(data)) as archive: texts = [archive.read(info).decode("utf-8") for info in archive.infolist() if info.filename.endswith(".md") and info.file_size <= MAX_RESPONSE_BYTES] if not any(EXPECTED_TEXT in text for text in texts): raise ValueError("Output does not contain the sample phrase") with args.output.open("xb") as output: output.write(data) print(f"Verified output saved to {args.output}") return elif status in {"failed", "cancelled"}: raise ValueError("Conversion ended without success; inspect the saved uuid") elif status != "running": raise ValueError(f"Unknown task state: {status!r}") pause(delay, deadline) delay = min(POLL_MAX, delay * 2) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--resume", type=Path, help="Poll an existing state file without creating another task") parser.add_argument("--state-file", type=Path, default=Path("pdf-craft-task.json")) parser.add_argument("--output", type=Path, default=Path("pdf-craft-result.zip")) parser.add_argument("--timeout", type=int, default=DEFAULT_DEADLINE, help="Overall deadline in seconds") try: run(parser.parse_args()) except (ApiError, OSError, ValueError, TypeError) as error: print(f"Conversion stopped: {error}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": sys.exit(main())