#!/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 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://fusion-api.oomol.com/v1" 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): headers = {"User-Agent": "PDFCraft-Quickstart/1.0", "Accept": "application/json" if key else "text/markdown"} if key: headers["Authorization"] = f"Bearer {key}" data = None if payload is not None: headers["Content-Type"] = "application/json" data = json.dumps(payload).encode() request = Request(url, data=data, headers=headers) 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({"sessionID": 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 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 if args.resume: state = json.loads(args.resume.read_text()) task_id = state.get("sessionID") if not isinstance(task_id, str) or not task_id: raise ValueError("No saved sessionID. Investigate the previous submission before creating another task") else: # Reserve the state file before a potentially billed submit. Never auto-resubmit. with args.state_file.open("x", encoding="utf-8") as state_file: json.dump({"state": "submission_unknown", "sampleURL": SAMPLE_URL}, state_file) state_file.flush() os.fsync(state_file.fileno()) print("Submitting one conversion; this may consume credits.", file=sys.stderr) submitted = fetch(f"{API_BASE}/pdf-transform-markdown/submit", key, {"pdfURL": SAMPLE_URL, "model": "gundam"}, remaining(deadline)) task_id = submitted.get("sessionID") if not isinstance(task_id, str) or not task_id: raise ValueError("Submit did not return sessionID; do not automatically resubmit") print(f"sessionID: {task_id}", file=sys.stderr) save_accepted_task(args.state_file, task_id) delay = POLL_INITIAL while True: try: result = fetch(f"{API_BASE}/pdf-transform-markdown/result/{quote(task_id, safe='')}", 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: state = result.get("state") if state == "completed": data = result.get("data") if not isinstance(data, dict): raise ValueError("Completed result does not contain a data object") download = data.get("downloadURL") if not isinstance(download, str): raise ValueError("Completed result does not contain a download URL") parsed = urlparse(download) if parsed.scheme != "https" or not parsed.netloc or parsed.username or parsed.password: raise ValueError("Result does not contain an HTTPS download URL") body = fetch(download, timeout=remaining(deadline)) # No Authorization on downloads. if EXPECTED_TEXT not in body.decode("utf-8"): raise ValueError("Output does not contain the sample phrase; inspect the task result") with args.output.open("xb") as output: output.write(body) print(f"Verified output saved to {args.output}") return if state == "failed": raise ValueError("Conversion failed; inspect the saved task in your account") if state != "processing": raise ValueError(f"Unknown task state: {state!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.md")) 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())