examples/quickstart.py

#!/usr/bin/env python3
"""
VelaOps Data Estate — quickstart.

Usage:
    python quickstart.py [ROOT]

ROOT defaults to the parent folder of this script (so it works in-place inside the dataset).
It loads the structured tables, joins across the CRM/ERP/claims "systems", then shows how to
reach a scanned form, an email thread, and a call transcript for the SAME claim.
"""
import os, sys, json
import pandas as pd

ROOT = sys.argv[1] if len(sys.argv) > 1 else os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

def path(*a): return os.path.join(ROOT, *a)

print(f"== VelaOps Data Estate :: root = {ROOT} ==\n")

# 1) structured tables -------------------------------------------------------
cust   = pd.read_csv(path("structured", "crm_customers.csv"))
pol    = pd.read_csv(path("structured", "erp_policies.csv"))
claims = pd.read_csv(path("structured", "claims.csv"))
print(f"customers={len(cust):>6}  policies={len(pol):>6}  claims={len(claims):>6}")

# 2) join across the three "systems" (note the column-name mismatch) ---------
full = (claims
        .merge(cust, on="customer_id", how="left")
        .merge(pol, left_on="policy_no", right_on="PolicyNo", how="left"))
print(f"joined claims+customer+policy -> {full.shape[0]} rows x {full.shape[1]} cols")

# 3) how many customers look like near-duplicates ----------------------------
dup_keys = cust["customer_id"].astype(str).str.startswith("CUST9").sum()
print(f"near-duplicate customer rows (CUST9*): {dup_keys}")

# 4) find a customer that appears across ALL modalities, then trace them -----
docs   = json.load(open(path("documents_index.json")))
emails = json.load(open(path("emails_index.json")))
calls  = json.load(open(path("calls_index.json")))

doc_custs  = {d["customer_id"] for d in docs}
mail_custs = {e["customer_id"] for e in emails}
call_custs = {c["customer_id"] for c in calls}
fully_linked = doc_custs & mail_custs & call_custs
cust_id = sorted(fully_linked)[0] if fully_linked else docs[0]["customer_id"]

sample = next(d for d in docs if d["customer_id"] == cust_id)
e = next(x for x in emails if x["customer_id"] == cust_id)
c = next(x for x in calls  if x["customer_id"] == cust_id)
claim_no = sample["claim_no"]

print(f"\n-- tracing customer {cust_id} across ALL sources --")
print("  claims       :", list(claims[claims.customer_id == cust_id].claim_no))
print("  scanned form :", sample["scan"], "(open with PIL / feed to OCR)")
print("  pdf          :", sample["pdf"])
print("  email thread :", e["file"])
print("  transcript   :", c["file"])

row = claims[claims["claim_no"] == claim_no].iloc[0]
print("  claim detail :", dict(row[["policy_no", "customer_id", "line", "status", "amount_claimed"]]))

# 5) verify the scan opens ---------------------------------------------------
try:
    from PIL import Image
    im = Image.open(path(sample["scan"]))
    print(f"\n  scan opens OK -> {im.size} {im.mode}")
except Exception as ex:
    print("  (PIL not available or scan missing:", ex, ")")

print("\nDone. See README.md for the full data dictionary and per-question hints.")