""" DermaConnect backend (FastAPI). A deliberately small, readable API that ties together patients, dermatologists, upload schedules, and the severity model. No real auth, no encryption, synthetic data only. This is a prototype to demonstrate the workflow, NOT a HIPAA-ready clinical system. See README before pointing it at anything real. """ from __future__ import annotations import json import os import uuid from datetime import datetime, timedelta, timezone from fastapi import FastAPI, File, Form, HTTPException, UploadFile from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles import database as db from severity_model import assess_severity BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) UPLOAD_DIR = os.path.join(BASE, "uploads") FRONTEND_DIR = os.path.join(BASE, "frontend") app = FastAPI(title="DermaConnect", version="0.1.0") def _now() -> datetime: return datetime.now(timezone.utc) @app.on_event("startup") def _startup() -> None: os.makedirs(UPLOAD_DIR, exist_ok=True) db.init_db() # On a fresh deploy the DB is empty; seed demo data so the dashboards aren't # blank. Safe to run every boot — seed.run() no-ops when users already exist. if os.environ.get("DERMA_AUTOSEED", "1") == "1": try: import seed seed.run() except Exception as e: # never let seeding crash the app print(f"[startup] auto-seed skipped: {e}") # ------------------------------------------------------------------ users ----- @app.post("/api/users") def create_user(payload: dict): role = payload.get("role") name = payload.get("name") email = payload.get("email") if role not in ("patient", "dermatologist") or not name or not email: raise HTTPException(400, "role (patient|dermatologist), name, email required") try: uid = db.create_user(role, name, email) except Exception as e: raise HTTPException(400, f"could not create user: {e}") return {"id": uid, "role": role, "name": name, "email": email} @app.get("/api/users") def list_users(role: str | None = None): conn = db.get_conn() try: if role: rows = conn.execute("SELECT * FROM users WHERE role = ? ORDER BY name", (role,)).fetchall() else: rows = conn.execute("SELECT * FROM users ORDER BY name").fetchall() return [dict(r) for r in rows] finally: conn.close() # ------------------------------------------------------------- care links ----- @app.post("/api/care-links") def create_care_link(payload: dict): patient_id = payload.get("patient_id") derm_id = payload.get("dermatologist_id") body_site = payload.get("body_site", "general") cadence = int(payload.get("cadence_days", 7)) if not patient_id or not derm_id: raise HTTPException(400, "patient_id and dermatologist_id required") next_due = (_now() + timedelta(days=cadence)).isoformat() conn = db.get_conn() try: cur = conn.execute( """INSERT INTO care_links (patient_id, dermatologist_id, body_site, cadence_days, next_due, created_at) VALUES (?, ?, ?, ?, ?, ?)""", (patient_id, derm_id, body_site, cadence, next_due, _now().isoformat()), ) conn.commit() return {"id": cur.lastrowid, "next_due": next_due, "cadence_days": cadence} except Exception as e: raise HTTPException(400, f"could not create care link: {e}") finally: conn.close() # ---------------------------------------------------------------- uploads ----- @app.post("/api/upload") async def upload_photo(care_link_id: int = Form(...), file: UploadFile = File(...)): conn = db.get_conn() try: link = conn.execute("SELECT * FROM care_links WHERE id = ?", (care_link_id,)).fetchone() if not link: raise HTTPException(404, "care_link not found") ext = os.path.splitext(file.filename or "")[1].lower() or ".jpg" fname = f"{uuid.uuid4().hex}{ext}" fpath = os.path.join(UPLOAD_DIR, fname) with open(fpath, "wb") as out: out.write(await file.read()) result = assess_severity(fpath) conn.execute( """INSERT INTO readings (care_link_id, image_path, score, category, confidence, signals, model, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", ( care_link_id, fname, result["score"], result["category"], result["confidence"], json.dumps(result["signals"]), result["model"], _now().isoformat(), ), ) # advance the schedule next_due = (_now() + timedelta(days=link["cadence_days"])).isoformat() conn.execute("UPDATE care_links SET next_due = ? WHERE id = ?", (next_due, care_link_id)) conn.commit() return {"reading": result, "image": fname, "next_due": next_due} finally: conn.close() # -------------------------------------------------------------- dashboards ---- def _triage(category: str) -> int: return {"Severe": 0, "Moderate": 1, "Mild": 2, "Clear": 3}.get(category, 4) @app.get("/api/dashboard/dermatologist/{derm_id}") def derm_dashboard(derm_id: int): """Every patient under this derm, sorted worst-first by latest reading.""" conn = db.get_conn() try: links = conn.execute( """SELECT cl.*, u.name AS patient_name, u.email AS patient_email FROM care_links cl JOIN users u ON u.id = cl.patient_id WHERE cl.dermatologist_id = ?""", (derm_id,), ).fetchall() cards = [] for link in links: readings = conn.execute( "SELECT * FROM readings WHERE care_link_id = ? ORDER BY created_at DESC", (link["id"],), ).fetchall() latest = readings[0] if readings else None prev = readings[1] if len(readings) > 1 else None trend = None if latest and prev: delta = latest["score"] - prev["score"] trend = "worse" if delta > 3 else "better" if delta < -3 else "stable" overdue = bool(link["next_due"] and link["next_due"] < _now().isoformat()) cards.append({ "care_link_id": link["id"], "patient_name": link["patient_name"], "patient_email": link["patient_email"], "body_site": link["body_site"], "cadence_days": link["cadence_days"], "next_due": link["next_due"], "overdue": overdue, "latest": dict(latest) if latest else None, "trend": trend, "reading_count": len(readings), }) cards.sort(key=lambda c: ( _triage(c["latest"]["category"]) if c["latest"] else 5, -(c["latest"]["score"] if c["latest"] else 0), )) return {"dermatologist_id": derm_id, "patients": cards} finally: conn.close() @app.get("/api/dashboard/patient/{patient_id}") def patient_dashboard(patient_id: int): conn = db.get_conn() try: links = conn.execute( """SELECT cl.*, u.name AS derm_name FROM care_links cl JOIN users u ON u.id = cl.dermatologist_id WHERE cl.patient_id = ?""", (patient_id,), ).fetchall() out = [] for link in links: readings = conn.execute( "SELECT * FROM readings WHERE care_link_id = ? ORDER BY created_at DESC LIMIT 20", (link["id"],), ).fetchall() out.append({ "care_link_id": link["id"], "derm_name": link["derm_name"], "body_site": link["body_site"], "cadence_days": link["cadence_days"], "next_due": link["next_due"], "overdue": bool(link["next_due"] and link["next_due"] < _now().isoformat()), "readings": [dict(r) for r in readings], }) return {"patient_id": patient_id, "care_links": out} finally: conn.close() @app.get("/api/readings/{care_link_id}") def readings(care_link_id: int): conn = db.get_conn() try: rows = conn.execute( "SELECT * FROM readings WHERE care_link_id = ? ORDER BY created_at", (care_link_id,), ).fetchall() return [dict(r) for r in rows] finally: conn.close() # ---------------------------------------------------------------- static ------ app.mount("/uploads", StaticFiles(directory=UPLOAD_DIR), name="uploads") @app.get("/") def index(): return FileResponse(os.path.join(FRONTEND_DIR, "index.html")) app.mount("/app", StaticFiles(directory=FRONTEND_DIR, html=True), name="frontend")