Export data

from io import BytesIO

import pandas as pd
import requests

API_BASE = "https://api.lendpathway.com/api"
TOKEN = "pat_your_token_here"
BOOK_ID = "99cc93e6-1f3f-42b1-9fe4-ba5a95be9c78"
HEADERS = {"Authorization": f"Bearer {TOKEN}"}

response = requests.get(
    f"{API_BASE}/books/{BOOK_ID}/csv-export",
    headers=HEADERS,
    params={"table_format": "month_as_row"},
    timeout=60,
)
response.raise_for_status()

underwriting = pd.read_csv(BytesIO(response.content))

# Keep statement/account rows for analysis. The final two rows are rollups.
detail = underwriting[~underwriting["Period"].isin(["NET", "Average"])]
print(detail[["Period", "Account", "Deposits", "True Revenue", "DTI %"]])

from pathlib import Path

import requests


def get_template(template_id: str) -> dict:
    response = requests.get(
        f"{API_BASE}/parse/spreadsheet-templates",
        headers=HEADERS,
        timeout=30,
    )
    response.raise_for_status()

    templates = response.json()["templates"]
    return next(template for template in templates if template["id"] == template_id)


def save_gallery_export(book_id: str, template_id: str, destination: Path) -> Path:
    template = get_template(template_id)
    sheet_type = "vlad" if template["type"] == "credit_report" else template["type"]

    response = requests.get(
        f"{API_BASE}/books/{book_id}/spreadsheet-export",
        headers=HEADERS,
        params={"sheet_type": sheet_type, "template_id": template_id},
        timeout=120,
    )
    response.raise_for_status()

    expected_type = "application/pdf" if template["format"] == "pdf" else "spreadsheet"
    content_type = response.headers.get("content-type", "").lower()
    if expected_type not in content_type:
        raise RuntimeError(f"Unexpected export content type: {content_type}")

    suffix = ".pdf" if template["format"] == "pdf" else ".xlsx"
    output = destination.with_suffix(suffix)
    output.write_bytes(response.content)
    return output


path = save_gallery_export(
    BOOK_ID,
    template_id="mca_stacking_pro",
    destination=Path("acme-coffee-underwriting"),
)
print(path)

import json
from datetime import UTC, datetime
from pathlib import Path

import requests


def get_json(path: str) -> dict | list:
    response = requests.get(
        f"{API_BASE}{path}",
        headers=HEADERS,
        timeout=60,
    )
    response.raise_for_status()
    return response.json()


output = Path("acme-coffee")
output.mkdir(exist_ok=True)

book = get_json(f"/books/{BOOK_ID}")
analytics = get_json(f"/books/{BOOK_ID}/analytics")
documents = get_json(f"/documents/?book_id={BOOK_ID}")

(output / "book.json").write_text(json.dumps(book, indent=2), encoding="utf-8")
(output / "analytics.json").write_text(json.dumps(analytics, indent=2), encoding="utf-8")
(output / "documents.json").write_text(json.dumps(documents, indent=2), encoding="utf-8")
(output / "manifest.json").write_text(
    json.dumps(
        {
            "book_id": BOOK_ID,
            "exported_at": datetime.now(UTC).isoformat(),
            "parse_status": book["parse_status"],
            "last_parsed_at": book.get("last_parsed_at"),
        },
        indent=2,
    ),
    encoding="utf-8",
)

from pathlib import Path

import requests


def safe_filename(name: str) -> str:
    return Path(name).name.replace("\x00", "")


documents = requests.get(
    f"{API_BASE}/documents/",
    headers=HEADERS,
    params={"book_id": BOOK_ID},
    timeout=30,
)
documents.raise_for_status()

source_dir = Path("acme-coffee/source-documents")
source_dir.mkdir(parents=True, exist_ok=True)

for document in documents.json():
    response = requests.get(
        f"{API_BASE}/documents/{document['id']}/download",
        headers=HEADERS,
        timeout=30,
    )
    response.raise_for_status()
    download_url = response.json()["download_url"]

    file_response = requests.get(download_url, timeout=120)
    file_response.raise_for_status()
    destination = source_dir / safe_filename(document["original_filename"])
    destination.write_bytes(file_response.content)