""" database.py ----------- SQLite database for Radiology AI Report System. Stores all generated reports with patient info, findings, and metadata. Tables: patients — patient demographics reports — generated radiology reports images — uploaded image metadata """ import json import logging import sqlite3 from datetime import datetime from pathlib import Path from typing import Optional logger = logging.getLogger(__name__) DB_PATH = Path("data/radiology.db") def get_connection() -> sqlite3.Connection: DB_PATH.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(str(DB_PATH), check_same_thread=False) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA foreign_keys=ON") return conn def init_db(): """Create all tables if they don't exist.""" conn = get_connection() try: conn.executescript(""" CREATE TABLE IF NOT EXISTS patients ( id INTEGER PRIMARY KEY AUTOINCREMENT, patient_name TEXT NOT NULL, age INTEGER NOT NULL, gender TEXT NOT NULL, created_at TEXT DEFAULT (datetime('now')), updated_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS reports ( id INTEGER PRIMARY KEY AUTOINCREMENT, report_uuid TEXT NOT NULL UNIQUE, patient_id INTEGER NOT NULL REFERENCES patients(id), patient_name TEXT NOT NULL, age INTEGER NOT NULL, gender TEXT NOT NULL, symptoms TEXT, clinical_history TEXT, modality TEXT, body_region TEXT, view TEXT, examination_details TEXT, clinical_information TEXT, findings TEXT, impression TEXT, recommendations TEXT, confidence_score REAL, generation_time_sec REAL, image_caption TEXT, validation_json TEXT, medical_coding_json TEXT, created_at TEXT DEFAULT (datetime('now')) ); CREATE INDEX IF NOT EXISTS idx_reports_uuid ON reports(report_uuid); CREATE INDEX IF NOT EXISTS idx_reports_patient ON reports(patient_id); CREATE INDEX IF NOT EXISTS idx_reports_created ON reports(created_at); CREATE INDEX IF NOT EXISTS idx_patients_name ON patients(patient_name); """) conn.commit() logger.info(f"Database initialised at {DB_PATH}") finally: conn.close() # ═══════════════════════════════════════════════════════════════════════════ # Patient operations # ═══════════════════════════════════════════════════════════════════════════ def create_or_get_patient(name: str, age: int, gender: str) -> int: """ Creates a new patient record or returns existing ID if same name+age+gender exists. """ conn = get_connection() try: row = conn.execute( "SELECT id FROM patients WHERE patient_name=? AND age=? AND gender=?", (name.strip(), age, gender) ).fetchone() if row: return row["id"] cur = conn.execute( "INSERT INTO patients (patient_name, age, gender) VALUES (?,?,?)", (name.strip(), age, gender) ) conn.commit() return cur.lastrowid finally: conn.close() def search_patients(query: str) -> list: """Search patients by name.""" conn = get_connection() try: rows = conn.execute( "SELECT DISTINCT patient_name, age, gender, COUNT(*) as report_count " "FROM reports WHERE patient_name LIKE ? " "GROUP BY patient_name, age, gender " "ORDER BY created_at DESC LIMIT 20", (f"%{query}%",) ).fetchall() return [dict(r) for r in rows] finally: conn.close() # ═══════════════════════════════════════════════════════════════════════════ # Report operations # ═══════════════════════════════════════════════════════════════════════════ def save_report(payload: dict, patient_name: str) -> int: """Save a complete report payload to the database. Returns row ID.""" patient = payload.get("patient", {}) imaging = payload.get("imaging", {}) report = payload.get("report", {}) valid = payload.get("validation", {}) coding = payload.get("medical_coding", {}) age = patient.get("age", 0) gender = patient.get("gender", "Unknown") patient_id = create_or_get_patient(patient_name, age, gender) conn = get_connection() try: cur = conn.execute(""" INSERT INTO reports ( report_uuid, patient_id, patient_name, age, gender, symptoms, clinical_history, modality, body_region, view, examination_details, clinical_information, findings, impression, recommendations, confidence_score, generation_time_sec, image_caption, validation_json, medical_coding_json ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) """, ( payload.get("report_id", ""), patient_id, patient_name.strip(), age, gender, patient.get("symptoms", ""), patient.get("clinical_history", ""), imaging.get("modality", ""), imaging.get("body_region", ""), imaging.get("view", ""), report.get("examination_details", ""), report.get("clinical_information", ""), report.get("findings", ""), report.get("impression", ""), report.get("recommendations", ""), report.get("confidence_score", 0.0), report.get("generation_time_seconds", 0.0), payload.get("image_caption", ""), json.dumps(valid), json.dumps(coding), )) conn.commit() logger.info(f"Report saved — DB row: {cur.lastrowid} | patient: {patient_name}") return cur.lastrowid finally: conn.close() def get_all_reports(limit: int = 50, offset: int = 0) -> list: """Get all reports ordered by newest first.""" conn = get_connection() try: rows = conn.execute(""" SELECT report_uuid, patient_name, age, gender, modality, body_region, confidence_score, created_at FROM reports ORDER BY created_at DESC LIMIT ? OFFSET ? """, (limit, offset)).fetchall() return [dict(r) for r in rows] finally: conn.close() def get_report_by_uuid(uuid: str) -> Optional[dict]: """Get full report by UUID.""" conn = get_connection() try: row = conn.execute( "SELECT * FROM reports WHERE report_uuid=?", (uuid,) ).fetchone() if not row: return None result = dict(row) # Parse JSON fields for field in ("validation_json", "medical_coding_json"): try: result[field] = json.loads(result[field] or "{}") except Exception: result[field] = {} return result finally: conn.close() def get_reports_by_patient(patient_name: str) -> list: """Get all reports for a specific patient.""" conn = get_connection() try: rows = conn.execute(""" SELECT report_uuid, patient_name, age, gender, modality, body_region, confidence_score, findings, impression, created_at FROM reports WHERE patient_name LIKE ? ORDER BY created_at DESC """, (f"%{patient_name}%",)).fetchall() return [dict(r) for r in rows] finally: conn.close() def get_db_stats() -> dict: """Get database statistics.""" conn = get_connection() try: total_reports = conn.execute("SELECT COUNT(*) FROM reports").fetchone()[0] total_patients = conn.execute("SELECT COUNT(DISTINCT patient_name) FROM reports").fetchone()[0] avg_confidence = conn.execute("SELECT AVG(confidence_score) FROM reports").fetchone()[0] modality_counts = conn.execute( "SELECT modality, COUNT(*) as cnt FROM reports GROUP BY modality ORDER BY cnt DESC" ).fetchall() return { "total_reports": total_reports, "total_patients": total_patients, "avg_confidence": round(avg_confidence or 0, 2), "modality_counts": {r["modality"]: r["cnt"] for r in modality_counts}, } finally: conn.close() def delete_report(uuid: str) -> bool: """Delete a report by UUID.""" conn = get_connection() try: conn.execute("DELETE FROM reports WHERE report_uuid=?", (uuid,)) conn.commit() return True except Exception as e: logger.error(f"Delete failed: {e}") return False finally: conn.close() # Auto-initialise on import init_db()