63093990d2
- Empty state UI: first-time prompt with file selector - Mapping review step in CSVImport with confidence scores and override dropdowns - Confirmation step in ConfirmVisitModal (one last look before save) - Next-priority toast after visit confirmation - Active patient banner with autosave and one-click return for interruptions - Simplified cascade to 2-step (what is missing, device shipment at risk) - Copy rewrites: status labels, doc checklist text, action messages, cascade language - Pilot readiness updated to 85% (11/13) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
523 lines
20 KiB
Python
523 lines
20 KiB
Python
"""Signal API — FastAPI backend for CSV ingestion, scoring, and export."""
|
|
|
|
import csv
|
|
import hashlib
|
|
import io
|
|
import os
|
|
import sys
|
|
from datetime import date, date as date_type
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import jwt
|
|
from jwt import PyJWKClient, ExpiredSignatureError, InvalidTokenError
|
|
from fastapi import Depends, FastAPI, File, Header, HTTPException, UploadFile
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import StreamingResponse
|
|
from pydantic import BaseModel
|
|
|
|
# Ensure python-backend root is on path (works both locally and in Docker)
|
|
_backend_root = Path(__file__).parent.parent
|
|
if str(_backend_root) not in sys.path:
|
|
sys.path.insert(0, str(_backend_root))
|
|
|
|
from core.coverage_calculator import ShipmentRecord, calculate_batch
|
|
from core.audit_logger import AuditAction, log_event
|
|
from core.doc_state_machine import compute_doc_state, DocState
|
|
from core.persistence import (
|
|
persist_export, persist_upload,
|
|
upsert_confirmed_visit, load_confirmed_visits_for_org, get_or_create_org,
|
|
)
|
|
from api.normalizer import normalize_csv
|
|
|
|
app = FastAPI(title="Signal API", version="1.0.0", docs_url="/docs")
|
|
|
|
# CORS — locked to Vercel frontend and localhost for dev.
|
|
# Set ALLOWED_ORIGINS in Railway as a comma-separated list for production.
|
|
_origins_env = os.getenv("ALLOWED_ORIGINS", "")
|
|
_allowed_origins: list[str] = (
|
|
[o.strip() for o in _origins_env.split(",") if o.strip()]
|
|
if _origins_env
|
|
else [
|
|
"http://localhost:5173",
|
|
"http://localhost:5174",
|
|
"http://127.0.0.1:5173",
|
|
]
|
|
)
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=_allowed_origins,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# API key auth — enforced when SIGNAL_API_KEY env var is set.
|
|
# In dev (no env var), all requests pass. In production, X-API-Key header is required.
|
|
_api_key = os.getenv("SIGNAL_API_KEY", "")
|
|
_clerk_jwks_url = os.getenv("CLERK_JWKS_URL", "")
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def _get_jwks_client() -> PyJWKClient:
|
|
if not _clerk_jwks_url:
|
|
raise RuntimeError("CLERK_JWKS_URL is not set")
|
|
return PyJWKClient(_clerk_jwks_url, cache_keys=True, cache_jwk_set=True, lifespan=300)
|
|
|
|
|
|
def verify_clerk_token(authorization: str) -> dict:
|
|
"""Verify a Clerk Bearer JWT. Returns decoded claims or raises HTTP 401."""
|
|
if not authorization.startswith("Bearer "):
|
|
raise HTTPException(status_code=401, detail="Missing or malformed Authorization header")
|
|
token = authorization.removeprefix("Bearer ").strip()
|
|
try:
|
|
client = _get_jwks_client()
|
|
signing_key = client.get_signing_key_from_jwt(token)
|
|
return jwt.decode(token, signing_key.key, algorithms=["RS256"],
|
|
options={"verify_exp": True, "verify_nbf": True})
|
|
except ExpiredSignatureError:
|
|
raise HTTPException(status_code=401, detail="Token has expired")
|
|
except InvalidTokenError as exc:
|
|
raise HTTPException(status_code=401, detail=f"Invalid token: {exc}")
|
|
except RuntimeError:
|
|
raise HTTPException(status_code=503, detail="Auth service not configured")
|
|
except Exception:
|
|
raise HTTPException(status_code=401, detail="Token verification failed")
|
|
|
|
|
|
def require_auth(
|
|
authorization: str = Header(default=""),
|
|
x_api_key: str = Header(default=""),
|
|
) -> dict:
|
|
"""
|
|
Accept either a Clerk JWT (Authorization: Bearer) or a direct API key (X-Api-Key).
|
|
Returns a claims dict. API-key path returns a synthetic service-account dict.
|
|
"""
|
|
# Fast path: API key (for direct access / testing)
|
|
if _api_key and x_api_key == _api_key:
|
|
return {"sub": "service-account", "via": "api_key"}
|
|
# Clerk JWT path (frontend / production)
|
|
if authorization.startswith("Bearer "):
|
|
return verify_clerk_token(authorization)
|
|
# Dev mode: no keys set — allow all
|
|
if not _api_key and not _clerk_jwks_url:
|
|
return {"sub": "dev", "via": "open"}
|
|
raise HTTPException(status_code=401, detail="Authentication required")
|
|
|
|
DEVICE_DISPLAY = {
|
|
"dexcom_g7": "Dexcom G7",
|
|
"dexcom_g6": "Dexcom G6",
|
|
"freestyle_libre_2": "FreeStyle Libre 2",
|
|
"freestyle_libre_3": "FreeStyle Libre 3",
|
|
"omnipod_5": "Omnipod 5",
|
|
}
|
|
|
|
FLAG_LABELS = {
|
|
"SUPPLY_LAPSED": "Docs Required",
|
|
"VISIT_REQUIRED": "Escalate",
|
|
"TRANSFER_PENDING": "New Patient",
|
|
"RENEWAL_CRITICAL": "Escalate",
|
|
"RENEWAL_ELEVATED": "Confirm Appointment",
|
|
"RENEWAL_SOON": "Begin Outreach",
|
|
"RESUPPLY_READY": "Clear to Ship",
|
|
"ACTIVE": "On Track",
|
|
"NO_RECENT_SHIPMENT": "No Recent Shipment",
|
|
}
|
|
|
|
FLAG_ACTIONS = {
|
|
"SUPPLY_LAPSED": "Pull patient file — verify documentation is current before initiating any new order",
|
|
"VISIT_REQUIRED": "Contact prescriber's office and patient directly — ask: 'Was there a visit in the past 6 months to evaluate this patient's CGM dosing needs?' No shipment until visit is confirmed and documented",
|
|
"TRANSFER_PENDING": "Obtain full documentation set (SWO, PECOS, PA, diagnosis) — Signal cannot assess status until records are on file",
|
|
"RENEWAL_CRITICAL": "Contact prescriber's office and patient directly — ask: 'Was there a visit in the past 6 months to evaluate CGM dosing needs?' Appointment must be on calendar within 45 days",
|
|
"RENEWAL_ELEVATED": "Follow up with prescriber's office — ask: 'Was there a visit in the past 6 months to evaluate CGM dosing needs?' Confirm appointment is scheduled",
|
|
"RENEWAL_SOON": "Contact prescriber's office — ask: 'Was there a visit in the past 6 months to evaluate this patient's CGM dosing needs?' Schedule evaluation now if not yet completed",
|
|
"RESUPPLY_READY": "Initiate resupply order — patient is within the 30-day refill window",
|
|
"ACTIVE": "No action needed — documentation current, visit more than 90 days out",
|
|
"NO_RECENT_SHIPMENT": "Verify status with prescriber office — confirm patient is still active",
|
|
}
|
|
|
|
|
|
class DocStateOut(BaseModel):
|
|
swo: str
|
|
visit: str
|
|
pecos: str
|
|
pa: str
|
|
diagnosis: str
|
|
|
|
|
|
class RecordOut(BaseModel):
|
|
patient_id: str
|
|
device_type: str
|
|
device_display: str
|
|
payer: str
|
|
component: str
|
|
days_until_coverage_end: int
|
|
days_until_visit_due: Optional[int] = None
|
|
flag: str
|
|
priority_score: int
|
|
coverage_end_date: str
|
|
next_visit_due_date: Optional[str] = None
|
|
action: str
|
|
status_label: str
|
|
reason: str
|
|
rule_version: str
|
|
visit_date_confidence: str = "estimated"
|
|
is_transfer: bool = False
|
|
doc_state: Optional[DocStateOut] = None
|
|
cascade: list[str] = []
|
|
|
|
|
|
class UploadResponse(BaseModel):
|
|
records: list[RecordOut]
|
|
total: int
|
|
skipped: int
|
|
skipped_reasons: list[str]
|
|
stats: dict
|
|
mapping_summary: dict
|
|
batch_id: Optional[str] = None
|
|
|
|
|
|
def _build_reason(flag_val: str, days_until_end: int, days_until_visit: Optional[int]) -> str:
|
|
if flag_val == "SUPPLY_LAPSED":
|
|
ago = abs(days_until_end)
|
|
unit = "day" if ago == 1 else "days"
|
|
return f"Supply lapsed {ago} {unit} ago. Prescriber contact required before next shipment."
|
|
if flag_val == "VISIT_REQUIRED":
|
|
if days_until_visit is not None and days_until_visit <= 0:
|
|
overdue = abs(days_until_visit)
|
|
unit = "day" if overdue == 1 else "days"
|
|
return f"Qualifying visit overdue by {overdue} {unit}. Confirm documentation immediately."
|
|
return "Qualifying visit renewal required. Confirm documentation before resupply."
|
|
if flag_val == "RENEWAL_CRITICAL":
|
|
if days_until_visit is not None:
|
|
unit = "day" if days_until_visit == 1 else "days"
|
|
return f"Qualifying visit due in {days_until_visit} {unit}. Contact prescriber immediately."
|
|
return "Qualifying visit renewal critical. Confirm visit documentation before resupply."
|
|
if flag_val == "RENEWAL_ELEVATED":
|
|
if days_until_visit is not None:
|
|
unit = "day" if days_until_visit == 1 else "days"
|
|
return f"Qualifying visit due in {days_until_visit} {unit}. Schedule prescriber appointment soon."
|
|
return "Qualifying visit renewal elevated. Schedule prescriber appointment."
|
|
if flag_val == "RENEWAL_SOON":
|
|
if days_until_visit is not None:
|
|
unit = "day" if days_until_visit == 1 else "days"
|
|
return f"Qualifying visit due in {days_until_visit} {unit}. Monitor and plan ahead."
|
|
return "Qualifying visit renewal approaching. Monitor status."
|
|
if flag_val == "RESUPPLY_READY":
|
|
unit = "day" if days_until_end == 1 else "days"
|
|
return f"Coverage ends in {days_until_end} {unit}. Patient is within resupply window — initiate shipment now."
|
|
if flag_val == "TRANSFER_PENDING":
|
|
return "Patient transferred from prior supplier. Verify all documentation before proceeding."
|
|
if flag_val == "NO_RECENT_SHIPMENT":
|
|
return "No shipment on file in 12+ months. Contact the prescriber office to confirm the patient is still active and has not transferred care."
|
|
unit = "day" if days_until_end == 1 else "days"
|
|
return f"Coverage on track. Resupply window opens in approximately {days_until_end} {unit}."
|
|
|
|
|
|
def _normalize_payer_type(payer: str) -> str:
|
|
"""Map raw payer string to doc_state_machine payer_type."""
|
|
from core.coverage_calculator import _normalize_payer
|
|
normalized = _normalize_payer(payer)
|
|
# Ensure medicare_advantage is not collapsed to medicare
|
|
if "medicare advantage" in payer.lower() or "medicare_advantage" in payer.lower():
|
|
return "medicare_advantage"
|
|
return normalized
|
|
|
|
|
|
def _to_record_out(r, record=None) -> RecordOut:
|
|
flag_val = r.flag.value if hasattr(r.flag, "value") else str(r.flag)
|
|
|
|
# Compute doc state if we have the original record
|
|
doc_state_out = None
|
|
cascade = []
|
|
if record is not None:
|
|
normalized_payer = _normalize_payer_type(r.payer)
|
|
doc = compute_doc_state(
|
|
payer_type=normalized_payer,
|
|
csv_swo_status=record.csv_swo_status,
|
|
csv_visit_date=record.csv_visit_date,
|
|
confirmed_visit_date=None, # already factored into r.next_visit_due_date
|
|
csv_pecos_verified=record.csv_pecos_verified,
|
|
csv_pa_status=record.csv_pa_status,
|
|
csv_diagnosis_on_file=record.csv_diagnosis_on_file,
|
|
is_transfer=bool(record.csv_transfer_from),
|
|
)
|
|
doc_state_out = DocStateOut(
|
|
swo=doc.swo, visit=doc.visit, pecos=doc.pecos, pa=doc.pa, diagnosis=doc.diagnosis
|
|
)
|
|
cascade = doc.cascade
|
|
|
|
return RecordOut(
|
|
patient_id=r.patient_id,
|
|
device_type=r.device_type,
|
|
device_display=DEVICE_DISPLAY.get(r.device_type, r.device_type),
|
|
payer=r.payer,
|
|
component=r.component,
|
|
days_until_coverage_end=r.days_until_coverage_end,
|
|
days_until_visit_due=r.days_until_visit_due,
|
|
flag=flag_val,
|
|
priority_score=r.priority_score,
|
|
coverage_end_date=r.coverage_end_date.isoformat(),
|
|
next_visit_due_date=r.next_visit_due_date.isoformat() if r.next_visit_due_date else None,
|
|
action=FLAG_ACTIONS.get(flag_val, "Review"),
|
|
status_label=FLAG_LABELS.get(flag_val, flag_val),
|
|
reason=_build_reason(flag_val, r.days_until_coverage_end, r.days_until_visit_due),
|
|
rule_version=r.rule_version,
|
|
visit_date_confidence=getattr(r, "visit_date_confidence", "estimated"),
|
|
is_transfer=getattr(r, "is_transfer", False),
|
|
doc_state=doc_state_out,
|
|
cascade=cascade,
|
|
)
|
|
|
|
|
|
def _compute_stats(records: list[RecordOut]) -> dict:
|
|
flags = [r.flag for r in records]
|
|
return {
|
|
"total": len(records),
|
|
"supply_lapsed": flags.count("SUPPLY_LAPSED"),
|
|
"visit_required": flags.count("VISIT_REQUIRED"),
|
|
"transfer_pending": flags.count("TRANSFER_PENDING"),
|
|
"renewal_critical": flags.count("RENEWAL_CRITICAL"),
|
|
"renewal_elevated": flags.count("RENEWAL_ELEVATED"),
|
|
"renewal_soon": flags.count("RENEWAL_SOON"),
|
|
"resupply_ready": flags.count("RESUPPLY_READY"),
|
|
"active": flags.count("ACTIVE"),
|
|
"no_recent_shipment": flags.count("NO_RECENT_SHIPMENT"),
|
|
"prescriber_action": (
|
|
flags.count("SUPPLY_LAPSED") + flags.count("VISIT_REQUIRED") +
|
|
flags.count("RENEWAL_CRITICAL") + flags.count("RENEWAL_ELEVATED")
|
|
),
|
|
}
|
|
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {"status": "ok", "service": "signal-api", "version": "1.0.0"}
|
|
|
|
|
|
@app.get("/health/db")
|
|
def health_db():
|
|
from core.supabase_client import get_client
|
|
import os
|
|
client = get_client()
|
|
if not client:
|
|
return {
|
|
"status": "unavailable",
|
|
"supabase_url_set": bool(os.getenv("SUPABASE_URL")),
|
|
"service_key_set": bool(os.getenv("SUPABASE_SERVICE_KEY")),
|
|
}
|
|
try:
|
|
result = client.table("organizations").select("id").limit(1).execute()
|
|
return {"status": "ok", "org_count": len(result.data)}
|
|
except Exception as e:
|
|
return {"status": "error", "detail": str(e)}
|
|
|
|
|
|
@app.post("/api/upload", response_model=UploadResponse)
|
|
async def upload_csv(
|
|
file: UploadFile = File(...),
|
|
claims: dict = Depends(require_auth),
|
|
):
|
|
if not (file.filename or "").endswith(".csv"):
|
|
raise HTTPException(status_code=400, detail="File must be a .csv")
|
|
|
|
content = await file.read()
|
|
try:
|
|
text = content.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
text = content.decode("latin-1")
|
|
|
|
records, skipped_reasons, mapping_summary = normalize_csv(text)
|
|
|
|
if not records:
|
|
log_event(AuditAction.CSV_INGEST, file.filename or "unknown", "demo_user",
|
|
"failure", "0.0.0.0", detail="No processable rows")
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail={
|
|
"message": "No processable rows found in the uploaded file.",
|
|
"skipped": skipped_reasons[:10],
|
|
"mapping_summary": mapping_summary,
|
|
},
|
|
)
|
|
|
|
# Load confirmed visit dates from Supabase for this org
|
|
clerk_org_id = claims.get("o", {}).get("id") if isinstance(claims.get("o"), dict) else None
|
|
org_id = get_or_create_org(clerk_org_id=clerk_org_id)
|
|
confirmed_visits = load_confirmed_visits_for_org(org_id) if org_id else {}
|
|
|
|
results = calculate_batch(records, as_of=date.today(), confirmed_visits=confirmed_visits)
|
|
|
|
# Build a lookup from patient_id to original record for doc state computation
|
|
record_lookup = {r.patient_id: r for r in records}
|
|
out = [_to_record_out(r, record=record_lookup.get(r.patient_id)) for r in results]
|
|
|
|
log_event(AuditAction.CSV_INGEST, file.filename or "unknown", "demo_user",
|
|
"success", "0.0.0.0", detail=f"{len(out)} records scored")
|
|
|
|
batch_id = persist_upload(
|
|
filename=file.filename or "unknown",
|
|
content_bytes=content,
|
|
shipment_records=records,
|
|
coverage_results=results,
|
|
skipped_count=len(skipped_reasons),
|
|
mapping_summary=mapping_summary,
|
|
clerk_org_id=clerk_org_id,
|
|
)
|
|
|
|
return UploadResponse(
|
|
records=out,
|
|
total=len(out),
|
|
skipped=len(skipped_reasons),
|
|
skipped_reasons=skipped_reasons[:20],
|
|
stats=_compute_stats(out),
|
|
mapping_summary=mapping_summary,
|
|
batch_id=batch_id,
|
|
)
|
|
|
|
|
|
class ExportRequest(BaseModel):
|
|
records: list[RecordOut]
|
|
batch_id: Optional[str] = None
|
|
|
|
|
|
@app.post("/api/export")
|
|
async def export_work_queue(
|
|
body: ExportRequest,
|
|
claims: dict = Depends(require_auth),
|
|
):
|
|
"""Generate a downloadable work-queue CSV from a list of scored records."""
|
|
records = body.records
|
|
output = io.StringIO()
|
|
writer = csv.writer(output)
|
|
writer.writerow([
|
|
"Patient ID",
|
|
"Device",
|
|
"Payer",
|
|
"Status",
|
|
"Priority Score",
|
|
"Days Until Resupply End",
|
|
"Next Visit Due",
|
|
"Recommended Action",
|
|
"Resupply End Date",
|
|
"Reason",
|
|
"Visit Confidence",
|
|
"SWO Status",
|
|
"Visit Status",
|
|
"PA Status",
|
|
])
|
|
for r in records:
|
|
doc = r.doc_state
|
|
writer.writerow([
|
|
r.patient_id,
|
|
r.device_display,
|
|
r.payer,
|
|
r.status_label,
|
|
r.priority_score,
|
|
r.days_until_coverage_end,
|
|
r.next_visit_due_date or "",
|
|
r.action,
|
|
r.coverage_end_date,
|
|
r.reason,
|
|
r.visit_date_confidence,
|
|
doc.swo if doc else "",
|
|
doc.visit if doc else "",
|
|
doc.pa if doc else "",
|
|
])
|
|
|
|
output.seek(0)
|
|
today = date.today().isoformat()
|
|
export_filename = f"signal-work-queue-{today}.csv"
|
|
log_event(AuditAction.WORKLIST_EXPORT, export_filename, "demo_user",
|
|
"success", "0.0.0.0", detail=f"{len(records)} records exported")
|
|
persist_export(batch_id=body.batch_id, filename=export_filename, row_count=len(records))
|
|
return StreamingResponse(
|
|
io.BytesIO(output.getvalue().encode("utf-8")),
|
|
media_type="text/csv",
|
|
headers={
|
|
"Content-Disposition": f"attachment; filename=signal-work-queue-{today}.csv"
|
|
},
|
|
)
|
|
|
|
|
|
class ConfirmVisitRequest(BaseModel):
|
|
patient_id: str
|
|
confirmed_date: str # ISO date string YYYY-MM-DD
|
|
# Fields needed to recompute coverage after confirmation
|
|
shipment_date: str
|
|
payer: str
|
|
device_type: str
|
|
quantity: int = 1
|
|
component: str = "sensor"
|
|
|
|
|
|
@app.post("/api/confirm-visit")
|
|
async def confirm_visit(
|
|
body: ConfirmVisitRequest,
|
|
claims: dict = Depends(require_auth),
|
|
):
|
|
"""
|
|
Store a staff-confirmed qualifying visit date and return the updated coverage record.
|
|
The confirmed date persists across all future CSV imports for this patient_id.
|
|
"""
|
|
from core.coverage_calculator import ShipmentRecord, calculate_coverage
|
|
|
|
# Validate dates
|
|
try:
|
|
confirmed = date_type.fromisoformat(body.confirmed_date)
|
|
shipment = date_type.fromisoformat(body.shipment_date)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=f"Invalid date: {e}. Use YYYY-MM-DD.")
|
|
|
|
if confirmed > date_type.today():
|
|
raise HTTPException(status_code=400, detail="Confirmed date cannot be in the future.")
|
|
|
|
# Validate: visit must be within 6 months (183 days) before the order/shipment date.
|
|
# 183 days = ~6 months, consistent with CMS CGM qualifying visit window.
|
|
from datetime import timedelta as _td
|
|
six_months_before = shipment - _td(days=183)
|
|
if confirmed < six_months_before:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=(
|
|
f"Visit date {confirmed.isoformat()} is more than 6 months before "
|
|
f"the order date {shipment.isoformat()}. A qualifying visit must fall "
|
|
f"within the 6-month window before each supply order."
|
|
),
|
|
)
|
|
|
|
# Get org
|
|
clerk_org_id = claims.get("o", {}).get("id") if isinstance(claims.get("o"), dict) else None
|
|
org_id = get_or_create_org(clerk_org_id=clerk_org_id)
|
|
if not org_id:
|
|
raise HTTPException(status_code=503, detail="Organization not found.")
|
|
|
|
# Hash and store
|
|
patient_hash = hashlib.sha256(body.patient_id.encode()).hexdigest()
|
|
success = upsert_confirmed_visit(org_id, patient_hash, confirmed, confirmed_by="staff")
|
|
if not success:
|
|
raise HTTPException(status_code=503, detail="Failed to save confirmed visit.")
|
|
|
|
# Recompute coverage with confirmed visit date
|
|
record = ShipmentRecord(
|
|
patient_id=body.patient_id,
|
|
device_type=body.device_type,
|
|
shipment_date=shipment,
|
|
quantity=body.quantity,
|
|
payer=body.payer,
|
|
component=body.component,
|
|
)
|
|
result = calculate_coverage(record, confirmed_visit_date=confirmed)
|
|
|
|
log_event(
|
|
AuditAction.CSV_INGEST,
|
|
f"confirm-visit:{body.patient_id[:8]}",
|
|
"staff",
|
|
"success",
|
|
"0.0.0.0",
|
|
detail=f"Visit confirmed {confirmed.isoformat()}",
|
|
)
|
|
|
|
return _to_record_out(result, record=record)
|