"""Small, dependency-free client for https://pdf-central.com.""" from __future__ import annotations import json import argparse import os import sys import time import uuid from pathlib import Path from typing import Any, Dict, Iterable, Optional from urllib.error import HTTPError, URLError from urllib.parse import quote, urlencode from urllib.request import Request, urlopen class PDFCentralError(RuntimeError): def __init__(self, message: str, *, status: int = 0, code: str = "CLIENT_ERROR", request_id: Optional[str] = None, retryable: bool = False, details: Any = None): super().__init__(message) self.status = status self.code = code self.request_id = request_id self.retryable = retryable self.details = details class PDFCentralClient: def __init__(self, api_key: Optional[str] = None, *, base_url: str = "https://pdf-central.com", client_name: str = "pdf-central-python", client_version: str = "0.7.0", acquisition_channel: Optional[str] = None, retries: int = 2, timeout: float = 30.0): self.api_key = api_key self.base_url = base_url.rstrip("/") self.client_name = client_name self.client_version = client_version self.acquisition_channel = acquisition_channel self.retries = max(0, int(retries)) self.timeout = timeout def request(self, path: str, *, method: str = "GET", body: Any = None, headers: Optional[Dict[str, str]] = None, token: Any = ..., idempotency_key: Optional[str] = None, response_type: str = "json", retry: bool = True) -> Any: target = path if path.startswith("http") else self.base_url + path request_headers = {"Accept": "application/json" if response_type == "json" else "*/*", "X-PDF-Central-Client": f"{self.client_name}/{self.client_version}"} request_headers.update(headers or {}) if self.acquisition_channel: request_headers["X-PDF-Central-Channel"] = self.acquisition_channel actual_token = self.api_key if token is ... else token if actual_token: request_headers["Authorization"] = f"Bearer {actual_token}" if idempotency_key: request_headers["Idempotency-Key"] = idempotency_key data = body if body is not None and not isinstance(body, (bytes, bytearray, memoryview)): request_headers["Content-Type"] = "application/json" data = json.dumps(body, separators=(",", ":")).encode("utf-8") elif isinstance(body, (bytearray, memoryview)): data = bytes(body) last_error: Optional[HTTPError] = None last_network_error: Optional[BaseException] = None attempts = self.retries if retry else 0 for attempt in range(attempts + 1): try: with urlopen(Request(target, data=data, headers=request_headers, method=method), timeout=self.timeout) as response: payload = response.read() if response_type == "bytes": return payload try: return json.loads(payload or b"{}") except (ValueError, TypeError) as error: raise PDFCentralError("PDF Central returned an invalid JSON response.", status=getattr(response, "status", 200), code="INVALID_RESPONSE", request_id=response.headers.get("X-Request-ID")) from error except HTTPError as error: last_error = error last_network_error = None can_retry = error.code in (429, 502, 503, 504) and attempt < attempts and (method == "GET" or bool(idempotency_key)) if can_retry: try: retry_after = float(error.headers.get("Retry-After", "0") or 0) except (TypeError, ValueError): retry_after = 0.0 time.sleep(min(5.0, retry_after if retry_after > 0 else 0.25 * (2**attempt))) continue break except (URLError, TimeoutError, OSError) as error: last_error = None last_network_error = error can_retry = attempt < attempts and (method == "GET" or bool(idempotency_key)) if can_retry: time.sleep(min(5.0, 0.25 * (2**attempt))) continue break if last_error is None: cause = str(getattr(last_network_error, "reason", last_network_error)) if last_network_error else None raise PDFCentralError("Could not reach PDF Central.", code="NETWORK_ERROR", retryable=True, details={"cause": cause} if cause else None) from last_network_error raw = last_error.read() try: parsed = json.loads(raw or b"{}") except (ValueError, TypeError): parsed = {} if not isinstance(parsed, dict): parsed = {} error_data = parsed.get("error") if isinstance(parsed.get("error"), dict) else {} raise PDFCentralError(error_data.get("message") or f"PDF Central request failed with HTTP {last_error.code}.", status=last_error.code, code=error_data.get("code", "HTTP_ERROR"), request_id=parsed.get("request_id") or last_error.headers.get("X-Request-ID"), retryable=bool(error_data.get("retryable")), details=error_data.get("details")) def capabilities(self, *, available_only: bool = True, summary: bool = True) -> Dict[str, Any]: query = urlencode({**({"view": "summary"} if summary else {}), **({"status": "available"} if available_only else {})}) return self.request(f"/api/v1/capabilities?{query}") def create_guest_session(self, *, acquisition_channel: Optional[str] = None) -> Dict[str, Any]: previous = self.acquisition_channel if acquisition_channel: self.acquisition_channel = acquisition_channel try: result = self.request("/api/v1/guest-sessions", method="POST", body={}, token=None, retry=False)["result"] self.api_key = result["token"] return result finally: self.acquisition_channel = previous or acquisition_channel def limits(self) -> Dict[str, Any]: return self.request("/api/v1/limits") def status(self) -> Dict[str, Any]: return self.request("/api/v1/status", token=None) def recipes(self) -> Dict[str, Any]: return self.request("/api/v1/recipes", token=None) def doctor(self) -> Dict[str, Any]: return self.request(f"/api/v1/doctor{'?authenticated=true' if self.api_key else ''}") def list_api_keys(self) -> Dict[str, Any]: return self.request("/api/v1/api-keys")["result"] def create_api_key(self, scopes: Iterable[str], *, expires_at: Optional[str] = None, idempotency_key: Optional[str] = None) -> Dict[str, Any]: key = idempotency_key or str(uuid.uuid4()) body: Dict[str, Any] = {"scopes": list(scopes)} if expires_at: body["expires_at"] = expires_at return self.request("/api/v1/api-keys", method="POST", idempotency_key=key, body=body)["result"] def rotate_api_key(self, credential_id: str, *, scopes: Optional[Iterable[str]] = None, expires_at: Optional[str] = None, idempotency_key: Optional[str] = None) -> Dict[str, Any]: key = idempotency_key or str(uuid.uuid4()) body: Dict[str, Any] = {} if scopes is not None: body["scopes"] = list(scopes) if expires_at: body["expires_at"] = expires_at return self.request(f"/api/v1/api-keys/{quote(str(credential_id), safe='')}/rotate", method="POST", idempotency_key=key, body=body)["result"] def revoke_api_key(self, credential_id: str) -> Dict[str, Any]: return self.request(f"/api/v1/api-keys/{quote(str(credential_id), safe='')}/revoke", method="POST", body={})["result"] def upload(self, file: Any, *, filename: Optional[str] = None, mime_type: str = "application/pdf", idempotency_key: Optional[str] = None) -> Dict[str, Any]: if isinstance(file, (str, Path)): path = Path(file) content = path.read_bytes() filename = filename or path.name else: content = bytes(file) filename = filename or "input.pdf" key = idempotency_key or str(uuid.uuid4()) created = self.request("/api/v1/files/uploads", method="POST", idempotency_key=key, body={"filename": filename, "mime_type": mime_type, "size_bytes": len(content)})["result"] staged = self.request(created["upload_url"], method="PUT", headers={"Content-Type": mime_type}, body=content, retry=False)["result"] return self.request(f"/api/v1/files/{quote(str(created['file_id']), safe='')}/complete", method="POST", idempotency_key=f"{key}-complete", body={"upload_etag": staged["upload_etag"]})["result"] def import_url(self, source_url: str, *, filename: Optional[str] = None, mime_type: Optional[str] = None, expected_sha256: Optional[str] = None, idempotency_key: Optional[str] = None) -> Dict[str, Any]: key = idempotency_key or str(uuid.uuid4()) body = {"source_url": source_url} if filename: body["filename"] = filename if mime_type: body["mime_type"] = mime_type if expected_sha256: body["expected_sha256"] = expected_sha256 return self.request("/api/v1/files/imports", method="POST", idempotency_key=key, body=body)["result"] def process(self, *, recipe: str = "read_document.v1", file_ids: Iterable[str] = (), source_url: Optional[str] = None, parameters: Optional[Dict[str, Any]] = None, max_cost_microusd: int = 100000, allow_ocr: bool = False, idempotency_key: Optional[str] = None) -> Dict[str, Any]: key = idempotency_key or str(uuid.uuid4()) body = {"recipe": recipe, "file_ids": list(file_ids), "parameters": parameters or {}, "execution": {"max_cost_microusd": max_cost_microusd, "allow_ocr": allow_ocr}, "idempotency_key": key} if source_url: body["source_url"] = source_url return self.request("/api/v1/process", method="POST", idempotency_key=key, body=body)["result"] def create_batch(self, tasks: Iterable[Dict[str, Any]], *, max_cost_microusd: int = 500000, idempotency_key: Optional[str] = None) -> Dict[str, Any]: key = idempotency_key or str(uuid.uuid4()) return self.request("/api/v1/batches", method="POST", idempotency_key=key, body={"tasks": list(tasks), "execution": {"max_cost_microusd": max_cost_microusd}, "idempotency_key": key})["result"] def get_batch(self, batch_id: str) -> Dict[str, Any]: return self.request(f"/api/v1/batches/{quote(str(batch_id), safe='')}")["result"] def list_collections(self) -> Dict[str, Any]: return self.request("/api/v1/collections")["result"] def create_collection(self, name: str, *, idempotency_key: Optional[str] = None) -> Dict[str, Any]: key = idempotency_key or str(uuid.uuid4()) return self.request("/api/v1/collections", method="POST", idempotency_key=key, body={"name": name})["result"] def add_collection_document(self, collection_id: str, file_id: str, *, document_key: Optional[str] = None, idempotency_key: Optional[str] = None) -> Dict[str, Any]: key = idempotency_key or str(uuid.uuid4()) body = {"file_id": file_id} if document_key: body["document_key"] = document_key return self.request(f"/api/v1/collections/{quote(str(collection_id), safe='')}/documents", method="POST", idempotency_key=key, body=body)["result"] def get_collection_changes(self, collection_id: str, *, after: Optional[str] = None, limit: int = 50) -> Dict[str, Any]: query_string = urlencode({"limit": limit, **({"after": after} if after else {})}) return self.request(f"/api/v1/collections/{quote(str(collection_id), safe='')}/changes?{query_string}")["result"] def search_collection(self, collection_id: str, query: str, *, limit: int = 10, idempotency_key: Optional[str] = None) -> Dict[str, Any]: key = idempotency_key or str(uuid.uuid4()) return self.request(f"/api/v1/collections/{quote(str(collection_id), safe='')}/search", method="POST", idempotency_key=key, body={"query": query, "limit": limit})["result"] def delete_collection(self, collection_id: str) -> Dict[str, Any]: return self.request(f"/api/v1/collections/{quote(str(collection_id), safe='')}", method="DELETE")["result"] def compare_documents(self, before_file_id: str, after_file_id: str, *, idempotency_key: Optional[str] = None) -> Dict[str, Any]: key = idempotency_key or str(uuid.uuid4()) return self.request("/api/v1/documents/compare", method="POST", idempotency_key=key, body={"before_file_id": before_file_id, "after_file_id": after_file_id})["result"] def classify_bundle(self, file_id: str, *, create_files: bool = False, idempotency_key: Optional[str] = None) -> Dict[str, Any]: key = idempotency_key or str(uuid.uuid4()) return self.request("/api/v1/document-bundles/classify", method="POST", idempotency_key=key, body={"file_id": file_id, "create_files": create_files})["result"] def create_review(self, artifact_id: str, corrections: Iterable[Dict[str, Any]], *, reason: Optional[str] = None, parent_review_id: Optional[str] = None, authorization_ref: Optional[str] = None, save_rule: bool = False, idempotency_key: Optional[str] = None) -> Dict[str, Any]: key = idempotency_key or str(uuid.uuid4()) body = {"artifact_id": artifact_id, "corrections": list(corrections), "save_rule": save_rule} if reason: body["reason"] = reason if parent_review_id: body["parent_review_id"] = parent_review_id if authorization_ref: body["authorization_ref"] = authorization_ref return self.request("/api/v1/reviews", method="POST", idempotency_key=key, body=body)["result"] def list_reviews(self, artifact_id: str) -> Dict[str, Any]: return self.request(f"/api/v1/reviews?artifact_id={quote(str(artifact_id), safe='')}")["result"] def get_review(self, review_id: str) -> Dict[str, Any]: return self.request(f"/api/v1/reviews/{quote(str(review_id), safe='')}")["result"] def change_review_state(self, review_id: str, action: str, *, reason: Optional[str] = None, idempotency_key: Optional[str] = None) -> Dict[str, Any]: key = idempotency_key or str(uuid.uuid4()) body = {"action": action} if reason: body["reason"] = reason return self.request(f"/api/v1/reviews/{quote(str(review_id), safe='')}/state", method="POST", idempotency_key=key, body=body)["result"] def agent_analyze(self, endpoint: str, body: Dict[str, Any], *, idempotency_key: Optional[str] = None) -> Dict[str, Any]: options: Dict[str, Any] = {"method": "POST", "body": body} if idempotency_key: options["idempotency_key"] = idempotency_key return self.request(f"/api/v1/agent/{endpoint}", **options)["result"] def prepare_pdf_for_agent(self, file_id: str) -> Dict[str, Any]: return self.agent_analyze("prepare", {"file_id": file_id}) def detect_pdf_agent_threats(self, file_id: str) -> Dict[str, Any]: return self.agent_analyze("threats", {"file_id": file_id}) def extract_pdf_to_schema(self, file_id: str, field_schema: Dict[str, Any], *, options: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: return self.agent_analyze("extract-schema", {"file_id": file_id, "field_schema": field_schema, "options": options or {}}) def retrieve_pdf_evidence(self, file_id: str, query: Any, *, max_tokens: Optional[int] = None, limit: Optional[int] = None) -> Dict[str, Any]: body: Dict[str, Any] = {"file_id": file_id, "queries" if isinstance(query, list) else "query": query} if max_tokens is not None: body["max_tokens"] = max_tokens if limit is not None: body["limit"] = limit return self.agent_analyze("evidence", body) def verify_pdf_claims(self, file_id: str, claims: Iterable[Dict[str, Any]], *, minimum_coverage: Optional[float] = None) -> Dict[str, Any]: body: Dict[str, Any] = {"file_id": file_id, "claims": list(claims)} if minimum_coverage is not None: body["minimum_coverage"] = minimum_coverage return self.agent_analyze("verify-claims", body) def answer_document(self, file_id: str, question: Any, *, max_tokens: Optional[int] = None, limit: Optional[int] = None) -> Dict[str, Any]: body: Dict[str, Any] = {"file_id": file_id, "questions" if isinstance(question, list) else "question": question} if max_tokens is not None: body["max_tokens"] = max_tokens if limit is not None: body["limit"] = limit return self.agent_analyze("answer", body) def reconcile_document_set(self, file_ids: Iterable[str]) -> Dict[str, Any]: return self.agent_analyze("reconcile", {"file_ids": list(file_ids)}) def check_document_requirements(self, file_id: str, requirements: Iterable[Dict[str, Any]]) -> Dict[str, Any]: return self.agent_analyze("check-requirements", {"file_id": file_id, "requirements": list(requirements)}) def extract_actions_and_deadlines(self, file_id: str) -> Dict[str, Any]: return self.agent_analyze("actions-deadlines", {"file_id": file_id}) def extract_pdf_visual_candidates(self, file_id: str) -> Dict[str, Any]: return self.agent_analyze("visuals", {"file_id": file_id}) def audit_pdf_accessibility(self, file_id: str) -> Dict[str, Any]: return self.agent_analyze("accessibility-audit", {"file_id": file_id}) def repair_pdf_accessibility_metadata(self, file_id: str, *, language: Optional[str] = None, title: Optional[str] = None, execute: bool = False, idempotency_key: Optional[str] = None) -> Dict[str, Any]: body: Dict[str, Any] = {"file_id": file_id, "execute": execute} if language: body["language"] = language if title: body["title"] = title key = idempotency_key or (str(uuid.uuid4()) if execute else None) return self.agent_analyze("accessibility-repair", body, idempotency_key=key) def plan_pdf_redactions(self, file_id: str, policy: Dict[str, Any]) -> Dict[str, Any]: return self.agent_analyze("redaction-plan", {"file_id": file_id, "policy": policy}) def make_pdf_model_ready(self, file_id: str, *, execute: bool = False, dpi: Optional[int] = None, languages: Optional[Iterable[str]] = None, linearize: Optional[bool] = None, max_output_bytes: Optional[int] = None, idempotency_key: Optional[str] = None) -> Dict[str, Any]: body: Dict[str, Any] = {"file_id": file_id, "execute": execute} if dpi is not None: body["dpi"] = dpi if languages is not None: body["languages"] = list(languages) if linearize is not None: body["linearize"] = linearize if max_output_bytes is not None: body["max_output_bytes"] = max_output_bytes key = idempotency_key or (str(uuid.uuid4()) if execute else None) return self.agent_analyze("model-ready", body, idempotency_key=key) def create_source_watch(self, source_url: str, *, name: Optional[str] = None, interval_seconds: Optional[int] = None, max_bytes: Optional[int] = None, idempotency_key: Optional[str] = None) -> Dict[str, Any]: body: Dict[str, Any] = {"source_url": source_url} if name: body["name"] = name if interval_seconds is not None: body["interval_seconds"] = interval_seconds if max_bytes is not None: body["max_bytes"] = max_bytes return self.request("/api/v1/source-watches", method="POST", idempotency_key=idempotency_key or str(uuid.uuid4()), body=body)["result"] def list_source_watches(self) -> Dict[str, Any]: return self.request("/api/v1/source-watches")["result"] def get_source_watch(self, watch_id: str) -> Dict[str, Any]: return self.request(f"/api/v1/source-watches/{quote(str(watch_id), safe='')}")["result"] def delete_source_watch(self, watch_id: str) -> Dict[str, Any]: return self.request(f"/api/v1/source-watches/{quote(str(watch_id), safe='')}", method="DELETE")["result"] def poll_source_watch(self, watch_id: str) -> Dict[str, Any]: return self.request(f"/api/v1/source-watches/{quote(str(watch_id), safe='')}/poll", method="POST", body={})["result"] def get_source_watch_events(self, watch_id: str, *, limit: int = 50) -> Dict[str, Any]: return self.request(f"/api/v1/source-watches/{quote(str(watch_id), safe='')}/events?{urlencode({'limit': limit})}")["result"] def run(self, operation: str, *, file_ids: Iterable[str] = (), parameters: Optional[Dict[str, Any]] = None, max_cost_microusd: int = 100000, deadline_seconds: Optional[int] = None, allow_partial: bool = False, idempotency_key: Optional[str] = None) -> Dict[str, Any]: key = idempotency_key or str(uuid.uuid4()) execution = {"max_cost_microusd": max_cost_microusd, "allow_partial": allow_partial} if deadline_seconds: execution["deadline_seconds"] = deadline_seconds return self.request(f"/api/v1/pdf/{quote(str(operation), safe='')}", method="POST", idempotency_key=key, body={"file_ids": list(file_ids), "parameters": parameters or {}, "execution": execution, "idempotency_key": key})["result"] def get_job(self, job_id: str) -> Dict[str, Any]: return self.request(f"/api/v1/jobs/{quote(str(job_id), safe='')}")["result"] def wait_for_job(self, job_id: str, *, interval: float = 0.5, timeout: float = 120.0) -> Dict[str, Any]: deadline = time.monotonic() + timeout while True: job = self.get_job(job_id) if job["status"] in ("completed", "failed", "cancelled"): return job if time.monotonic() >= deadline: raise PDFCentralError(f"Timed out waiting for job {job_id}.", code="CLIENT_TIMEOUT", retryable=True) time.sleep(interval) def artifact_records(self, artifact_id: str) -> Dict[str, Any]: return self.request(f"/api/v1/artifacts/{quote(str(artifact_id), safe='')}/records")["result"] def get_artifact(self, artifact_id: str) -> Dict[str, Any]: return self.request(f"/api/v1/artifacts/{quote(str(artifact_id), safe='')}")["result"] def download_file(self, file_id: str) -> bytes: return self.request(f"/api/v1/files/{quote(str(file_id), safe='')}/content", response_type="bytes") def extract_text(self, file: Any, **options: Any) -> Dict[str, Any]: acquisition_channel = options.pop("acquisition_channel", None) filename = options.pop("filename", None) mime_type = options.pop("mime_type", "application/pdf") upload_key = options.pop("idempotency_key", None) job_key = options.pop("job_idempotency_key", None) parameters = options.pop("parameters", {}) max_cost = options.pop("max_cost_microusd", 100000) interval = options.pop("interval", 0.5) timeout = options.pop("timeout", 120.0) if options: raise TypeError(f"Unsupported extract_text options: {', '.join(sorted(options))}") if not self.api_key: self.create_guest_session(acquisition_channel=acquisition_channel) uploaded = self.upload(file, filename=filename, mime_type=mime_type, idempotency_key=upload_key) job = self.run("extract_pdf_text", file_ids=[uploaded["file_id"]], parameters=parameters, max_cost_microusd=max_cost, idempotency_key=job_key) if job["status"] not in ("completed", "failed", "cancelled"): job = self.wait_for_job(job["job_id"], interval=interval, timeout=timeout) if job["status"] != "completed": failure = job.get("error") or {} raise PDFCentralError(failure.get("message", "Text extraction failed."), code=failure.get("code", "JOB_FAILED"), retryable=bool(failure.get("retryable")), details=failure) records = [] for artifact in job.get("artifacts", []): records.extend(self.artifact_records(artifact["artifact_id"]).get("items", [])) return {"file": uploaded, "job": job, "records": records} def read_document(self, file_or_url: Any, **options: Any) -> Dict[str, Any]: acquisition_channel = options.pop("acquisition_channel", None) filename = options.pop("filename", None) mime_type = options.pop("mime_type", None) expected_sha256 = options.pop("expected_sha256", None) upload_key = options.pop("idempotency_key", None) job_key = options.pop("job_idempotency_key", None) parameters = options.pop("parameters", {}) max_cost = options.pop("max_cost_microusd", 100000) allow_ocr = options.pop("allow_ocr", False) interval = options.pop("interval", 0.5) timeout = options.pop("timeout", 120.0) if options: raise TypeError(f"Unsupported read_document options: {', '.join(sorted(options))}") if not self.api_key: self.create_guest_session(acquisition_channel=acquisition_channel) if isinstance(file_or_url, str) and file_or_url.lower().startswith("https://"): uploaded = self.import_url(file_or_url, filename=filename, mime_type=mime_type, expected_sha256=expected_sha256, idempotency_key=upload_key) else: uploaded = self.upload(file_or_url, filename=filename, mime_type=mime_type or "application/pdf", idempotency_key=upload_key) job = self.process(recipe="read_document.v1", file_ids=[uploaded["file_id"]], parameters=parameters, max_cost_microusd=max_cost, allow_ocr=allow_ocr, idempotency_key=job_key) if job["status"] not in ("completed", "failed", "cancelled"): job = self.wait_for_job(job["job_id"], interval=interval, timeout=timeout) if job["status"] != "completed": failure = job.get("error") or {} raise PDFCentralError(failure.get("message", "Document processing failed."), code=failure.get("code", "JOB_FAILED"), retryable=bool(failure.get("retryable")), details=failure) records = [] for artifact in job.get("artifacts", []): records.extend(self.artifact_records(artifact["artifact_id"]).get("items", [])) return {"file": uploaded, "job": job, "records": records} __all__ = ["PDFCentralClient", "PDFCentralError"] def _read_json(path: str) -> Any: return json.loads(Path(path).read_text(encoding="utf-8")) def _ensure_credential(client: PDFCentralClient) -> None: if not client.api_key: client.create_guest_session(acquisition_channel="cli") def _ingest_cli_source(client: PDFCentralClient, source: str, filename: Optional[str] = None) -> Dict[str, Any]: _ensure_credential(client) if source.lower().startswith("https://"): return client.import_url(source, filename=filename) return client.upload(source, filename=filename) def _finish_cli_job(client: PDFCentralClient, job: Dict[str, Any]) -> Dict[str, Any]: return job if job["status"] in ("completed", "failed", "cancelled") else client.wait_for_job(job["job_id"]) def _build_cli_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="pdf-central", description="PDF Central agent API client") parser.add_argument("--base-url", default=os.environ.get("PDF_CENTRAL_BASE_URL", "https://pdf-central.com")) parser.add_argument("--api-key", default=os.environ.get("PDF_CENTRAL_API_KEY")) subparsers = parser.add_subparsers(dest="command", required=True) for name in ("status", "capabilities", "recipes", "doctor", "collection-list"): simple = subparsers.add_parser(name) simple.add_argument("--out") read = subparsers.add_parser("read") read.add_argument("source") read.add_argument("--filename") read.add_argument("--out") read.add_argument("--max-cost", type=int, default=100000) read.add_argument("--allow-ocr", action="store_true") read.add_argument("--idempotency-key") imported = subparsers.add_parser("import-url") imported.add_argument("url") imported.add_argument("--filename") imported.add_argument("--out") imported.add_argument("--idempotency-key") process = subparsers.add_parser("process") process.add_argument("recipe") process.add_argument("source") process.add_argument("--parameters") process.add_argument("--out") process.add_argument("--max-cost", type=int, default=100000) process.add_argument("--allow-ocr", action="store_true") process.add_argument("--idempotency-key") batch = subparsers.add_parser("batch") batch.add_argument("tasks") batch.add_argument("--out") batch.add_argument("--max-cost", type=int, default=500000) batch.add_argument("--idempotency-key") collection_create = subparsers.add_parser("collection-create") collection_create.add_argument("name") collection_create.add_argument("--out") collection_create.add_argument("--idempotency-key") collection_add = subparsers.add_parser("collection-add") collection_add.add_argument("collection_id") collection_add.add_argument("source") collection_add.add_argument("--out") collection_add.add_argument("--idempotency-key") collection_search = subparsers.add_parser("collection-search") collection_search.add_argument("collection_id") collection_search.add_argument("query") collection_search.add_argument("--limit", type=int, default=10) collection_search.add_argument("--out") collection_search.add_argument("--idempotency-key") collection_delete = subparsers.add_parser("collection-delete") collection_delete.add_argument("collection_id") collection_delete.add_argument("--out") compare = subparsers.add_parser("compare") compare.add_argument("before") compare.add_argument("after") compare.add_argument("--out") compare.add_argument("--idempotency-key") classify = subparsers.add_parser("classify") classify.add_argument("source") classify.add_argument("--create-files", action="store_true") classify.add_argument("--out") classify.add_argument("--idempotency-key") review = subparsers.add_parser("review") review.add_argument("artifact_id") review.add_argument("corrections") review.add_argument("--reason") review.add_argument("--out") review.add_argument("--idempotency-key") review_list = subparsers.add_parser("review-list") review_list.add_argument("artifact_id") review_list.add_argument("--out") review_get = subparsers.add_parser("review-get") review_get.add_argument("review_id") review_get.add_argument("--out") review_state = subparsers.add_parser("review-state") review_state.add_argument("review_id") review_state.add_argument("action", choices=("revoke", "restore")) review_state.add_argument("--reason") review_state.add_argument("--out") review_state.add_argument("--idempotency-key") extract = subparsers.add_parser("extract-text") extract.add_argument("file") extract.add_argument("--out") extract.add_argument("--max-cost", type=int, default=100000) extract.add_argument("--idempotency-key") run = subparsers.add_parser("run") run.add_argument("operation") run.add_argument("file") run.add_argument("--out") run.add_argument("--max-cost", type=int, default=100000) run.add_argument("--idempotency-key") return parser def _run_cli(args: argparse.Namespace) -> Any: client = PDFCentralClient(args.api_key, base_url=args.base_url, client_name="pdf-central-cli", acquisition_channel="cli") command = args.command if command == "status": return client.status() if command == "capabilities": return client.capabilities() if command == "recipes": return client.recipes() if command == "doctor": return client.doctor() if command == "read": return client.read_document(args.source, filename=args.filename, max_cost_microusd=args.max_cost, allow_ocr=args.allow_ocr, job_idempotency_key=args.idempotency_key) if command == "import-url": _ensure_credential(client) return client.import_url(args.url, filename=args.filename, idempotency_key=args.idempotency_key) if command == "process": uploaded = _ingest_cli_source(client, args.source) parameters = _read_json(args.parameters) if args.parameters else {} job = client.process(recipe=args.recipe, file_ids=[uploaded["file_id"]], parameters=parameters, max_cost_microusd=args.max_cost, allow_ocr=args.allow_ocr, idempotency_key=args.idempotency_key) return {"file": uploaded, "job": _finish_cli_job(client, job)} if command == "batch": _ensure_credential(client) spec = _read_json(args.tasks) tasks = spec if isinstance(spec, list) else spec.get("tasks") if not isinstance(tasks, list): raise ValueError("The batch JSON must be an array or contain a tasks array.") return client.create_batch(tasks, max_cost_microusd=args.max_cost, idempotency_key=args.idempotency_key) if command == "collection-list": _ensure_credential(client) return client.list_collections() if command == "collection-create": _ensure_credential(client) return client.create_collection(args.name, idempotency_key=args.idempotency_key) if command == "collection-add": uploaded = _ingest_cli_source(client, args.source) return {"file": uploaded, "index": client.add_collection_document(args.collection_id, uploaded["file_id"], idempotency_key=args.idempotency_key)} if command == "collection-search": _ensure_credential(client) return client.search_collection(args.collection_id, args.query, limit=args.limit, idempotency_key=args.idempotency_key) if command == "collection-delete": _ensure_credential(client) return client.delete_collection(args.collection_id) if command == "compare": before = _ingest_cli_source(client, args.before) after = _ingest_cli_source(client, args.after) return client.compare_documents(before["file_id"], after["file_id"], idempotency_key=args.idempotency_key) if command == "classify": uploaded = _ingest_cli_source(client, args.source) return {"file": uploaded, "classification": client.classify_bundle(uploaded["file_id"], create_files=args.create_files, idempotency_key=args.idempotency_key)} if command == "review": _ensure_credential(client) spec = _read_json(args.corrections) corrections = spec if isinstance(spec, list) else spec.get("corrections") if not isinstance(corrections, list): raise ValueError("The corrections JSON must be an array or contain a corrections array.") return client.create_review(args.artifact_id, corrections, reason=args.reason or (spec.get("reason") if isinstance(spec, dict) else None), parent_review_id=spec.get("parent_review_id") if isinstance(spec, dict) else None, authorization_ref=spec.get("authorization_ref") if isinstance(spec, dict) else None, save_rule=spec.get("save_rule") is True if isinstance(spec, dict) else False, idempotency_key=args.idempotency_key) if command == "review-list": _ensure_credential(client) return client.list_reviews(args.artifact_id) if command == "review-get": _ensure_credential(client) return client.get_review(args.review_id) if command == "review-state": _ensure_credential(client) return client.change_review_state(args.review_id, args.action, reason=args.reason, idempotency_key=args.idempotency_key) if command == "extract-text": return client.extract_text(args.file, max_cost_microusd=args.max_cost, job_idempotency_key=args.idempotency_key) if command == "run": uploaded = _ingest_cli_source(client, args.file) job = client.run(args.operation, file_ids=[uploaded["file_id"]], max_cost_microusd=args.max_cost, idempotency_key=args.idempotency_key) return {"file": uploaded, "job": _finish_cli_job(client, job)} raise ValueError(f"Unknown command: {command}") def main() -> int: parser = _build_cli_parser() args = parser.parse_args() if hasattr(args, "max_cost") and args.max_cost < 1: parser.error("--max-cost must be a positive integer.") if hasattr(args, "limit") and not 1 <= args.limit <= 25: parser.error("--limit must be from 1 through 25.") try: value = _run_cli(args) rendered = json.dumps(value, ensure_ascii=False, indent=2) + "\n" output = getattr(args, "out", None) if output: Path(output).write_text(rendered, encoding="utf-8") else: print(rendered, end="") job = value.get("job") if isinstance(value, dict) else None return 1 if isinstance(job, dict) and job.get("status") != "completed" else 0 except PDFCentralError as error: payload = {"success": False, "error": {"code": error.code, "message": str(error), "status": error.status, "request_id": error.request_id, "retryable": error.retryable, "details": error.details}} except (OSError, ValueError, TypeError, KeyError) as error: payload = {"success": False, "error": {"code": "CLI_ERROR", "message": str(error)}} print(json.dumps(payload, ensure_ascii=False), file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())