pip install requests

export HTPBE_API_KEY="your_api_key_here"

import os
import time
import requests

API_KEY = os.environ["HTPBE_API_KEY"]
BASE_URL = "https://api.htpbe.tech/v1"
HEADERS = {
 "Authorization": f"Bearer {API_KEY}",
 "Content-Type": "application/json",
}


def check_pdf(pdf_url: str) -> dict:
 """Submit a PDF URL for forensic analysis and return the result."""
 # Step 1: submit the URL
 submit = requests.post(
 f"{BASE_URL}/analyze",
 headers=HEADERS,
 json={"url": pdf_url},
 timeout=30,
 )
 submit.raise_for_status()
 check_id = submit.json()["id"]

 # Step 2: retrieve the verdict
 result = requests.get(
 f"{BASE_URL}/result/{check_id}",
 headers=HEADERS,
 timeout=30,
 )
 result.raise_for_status()
 return result.json()


if __name__ == "__main__":
 report = check_pdf("https://your-storage.example.com/documents/statement.pdf")
 print(report["status"]) # intact | modified | inconclusive

def route_document(pdf_url: str, document_type: str = "generic") -> dict:
 """
 Analyze a PDF and return a routing decision.

 Returns a dict with keys:
 - action: "accept" | "reject" | "review"
 - check_id: str (store this for audit purposes)
 - reason: str
 """
 report = check_pdf(pdf_url)
 status = report["status"]
 check_id = report["id"]
 markers = report.get("modification_markers", [])
 producer = report.get("producer") or "unknown"
 xref_count = report.get("xref_count", 1)

 if status == "intact":
 return {
 "action": "accept",
 "check_id": check_id,
 "reason": "Document structure is consistent with original issuance.",
 }

 if status == "modified":
 reason_parts = [f"Forensic markers: {', '.join(markers)}."]
 if xref_count > 1:
 reason_parts.append(f"File was saved {xref_count} times after creation.")
 if producer:
 reason_parts.append(f"Last processed by: {producer}.")
 return {
 "action": "reject",
 "check_id": check_id,
 "reason": "".join(reason_parts),
 }

 # status == "inconclusive"
 # The document was created with consumer software and lacks institutional metadata.
 # For documents that should come from banks, tax authorities, or payroll systems,
 # inconclusive is a strong signal of fraud — treat it accordingly.
 institutional_types = {"bank_statement", "tax_document", "payslip", "insurance_policy"}
 if document_type in institutional_types:
 return {
 "action": "reject",
 "check_id": check_id,
 "reason": (
 f"Document origin ({producer}) is inconsistent with a {document_type}. "
 "Expected institutional software; consumer software detected."
 ),
 }

 return {
 "action": "review",
 "check_id": check_id,
 "reason": "Document created with consumer software. Route to manual review.",
 }

from concurrent.futures import ThreadPoolExecutor, as_completed


def batch_check(
 pdf_urls: list[str],
 document_type: str = "generic",
 max_workers: int = 5,
) -> list[dict]:
 """
 Run forensic checks on multiple PDFs in parallel.

 Returns a list of result dicts sorted by the original URL order.
 Each dict includes the URL, routing decision, and raw report fields.
 """
 results: dict[str, dict] = {}

 def check_one(url: str) -> tuple[str, dict]:
 try:
 decision = route_document(url, document_type)
 report = check_pdf(url)
 return url, {
 "url": url,
 "action": decision["action"],
 "check_id": decision["check_id"],
 "reason": decision["reason"],
 "status": report["status"],
 "markers": report.get("modification_markers", []),
 "producer": report.get("producer"),
 "xref_count": report.get("xref_count"),
 "error": None,
 }
 except requests.HTTPError as exc:
 return url, {
 "url": url,
 "action": "error",
 "check_id": None,
 "reason": f"HTTP {exc.response.status_code}",
 "status": None,
 "markers": [],
 "producer": None,
 "xref_count": None,
 "error": str(exc),
 }
 except Exception as exc:
 return url, {
 "url": url,
 "action": "error",
 "check_id": None,
 "reason": str(exc),
 "status": None,
 "markers": [],
 "producer": None,
 "xref_count": None,
 "error": str(exc),
 }

 with ThreadPoolExecutor(max_workers=max_workers) as pool:
 futures = {pool.submit(check_one, url): url for url in pdf_urls}
 for future in as_completed(futures):
 url, result = future.result()
 results[url] = result

 # preserve original order
 ordered = [results[url] for url in pdf_urls]
 return ordered


def filter_flagged(results: list[dict]) -> list[dict]:
 """Return only rejected documents for downstream handling."""
 return [r for r in results if r["action"] == "reject"]


# Example usage
if __name__ == "__main__":
 urls = [
 "https://your-storage.example.com/docs/statement-jan.pdf",
 "https://your-storage.example.com/docs/statement-feb.pdf",
 "https://your-storage.example.com/docs/statement-mar.pdf",
 ]

 all_results = batch_check(urls, document_type="bank_statement")
 flagged = filter_flagged(all_results)

 print(f"Checked: {len(all_results)}, Flagged: {len(flagged)}")
 for doc in flagged:
 print(f" REJECT {doc['url']}")
 print(f" Markers: {doc['markers']}")
 print(f" Producer: {doc['producer']}")

import os
import time
import requests
from dataclasses import dataclass


API_KEY = os.environ["HTPBE_API_KEY"]
BASE_URL = "https://api.htpbe.tech/v1"
HEADERS = {
 "Authorization": f"Bearer {API_KEY}",
 "Content-Type": "application/json",
}


@dataclass
class PendingCheck:
 check_id: str
 pdf_url: str
 submitted_at: float


def submit_for_analysis(pdf_url: str) -> PendingCheck:
 """Submit a PDF URL and return a handle for later retrieval."""
 response = requests.post(
 f"{BASE_URL}/analyze",
 headers=HEADERS,
 json={"url": pdf_url},
 timeout=30,
 )
 response.raise_for_status()
 check_id = response.json()["id"]
 return PendingCheck(
 check_id=check_id,
 pdf_url=pdf_url,
 submitted_at=time.time(),
 )


def poll_result(
 pending: PendingCheck,
 poll_interval: float = 1.0,
 timeout: float = 30.0,
) -> dict:
 """
 Poll for a result until it is ready or the timeout expires.

 In practice the API returns synchronously, so the first poll
 almost always succeeds. This loop handles edge cases and network
 retries for long-running queues.
 """
 deadline = time.time() + timeout
 while time.time() < deadline:
 response = requests.get(
 f"{BASE_URL}/result/{pending.check_id}",
 headers=HEADERS,
 timeout=30,
 )
 if response.status_code == 200:
 return response.json()
 if response.status_code == 404:
 # Not ready yet — wait and retry
 time.sleep(poll_interval)
 continue
 response.raise_for_status()

 raise TimeoutError(
 f"Result for {pending.check_id} not ready after {timeout}s"
 )


# Pipeline pattern: submit a batch, then retrieve when convenient
def intake_pipeline(urls: list[str]) -> list[dict]:
 # Phase 1: submit all documents
 pending_checks = []
 for url in urls:
 try:
 pending = submit_for_analysis(url)
 pending_checks.append(pending)
 print(f"Submitted {url} → {pending.check_id}")
 except requests.HTTPError as exc:
 print(f"Submit failed for {url}: {exc}")

 # Phase 2: retrieve results (could be in a separate job/worker)
 results = []
 for pending in pending_checks:
 try:
 result = poll_result(pending)
 results.append(result)
 print(f"{pending.check_id}: {result['status']}")
 except (TimeoutError, requests.HTTPError) as exc:
 print(f"Retrieval failed for {pending.check_id}: {exc}")

 return results

{"id":"ck_9f4a2e1b-3d7c-4a8e-b1f2-9e0d3c5a7b8f","status":"modified","modification_confidence":"high","modification_markers":["HTPBE_EDITING_TOOL_FINGERPRINT","HTPBE_MULTIPLE_REVISION_LAYERS"],"xref_count":3,"has_digital_signature":false,"modifications_after_signature":false,"signature_removed":false,"creator":"HSBC Document Service","producer":"Smallpdf","creation_date":1764547200,"modification_date":1742688000}

import os
import requests

TEST_KEY = os.environ["HTPBE_TEST_KEY"]
BASE_URL = "https://api.htpbe.tech/v1"
HEADERS = {"Authorization": f"Bearer {TEST_KEY}", "Content-Type": "application/json"}

# Predictable test fixtures
TEST_URLS = {
 "intact": "https://api.htpbe.tech/v1/test/clean.pdf",
 "modified_high": "https://api.htpbe.tech/v1/test/modified-high.pdf",
 "inconclusive": "https://api.htpbe.tech/v1/test/inconclusive.pdf",
 "sig_removed": "https://api.htpbe.tech/v1/test/signature-removed.pdf",
 "modified_after_sig":"https://api.htpbe.tech/v1/test/modified-medium.pdf",
}


def test_all_verdict_branches():
 for scenario, url in TEST_URLS.items():
 submit = requests.post(f"{BASE_URL}/analyze", headers=HEADERS, json={"url": url}, timeout=30)
 submit.raise_for_status()
 check_id = submit.json()["id"]

 result = requests.get(f"{BASE_URL}/result/{check_id}", headers=HEADERS, timeout=30)
 result.raise_for_status()
 data = result.json()

 print(f"{scenario:20s} → status={data['status']}, confidence={data['modification_confidence']}")


if __name__ == "__main__":
 test_all_verdict_branches()