Spaces:
Sleeping
Sleeping
File size: 8,748 Bytes
60d4850 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 | import uuid
from datetime import datetime
from sqlalchemy import Boolean, Column, DateTime, Integer, String, Text
from app.db.database import Base
class IntakeSession(Base):
__tablename__ = "intake_sessions"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
created_at = Column(DateTime, default=datetime.utcnow, index=True)
patient_name = Column(String, nullable=True) # Optional for now
# Original Data
transcript = Column(Text, nullable=False)
detected_language = Column(String, default="en")
# Extracted Data
chief_complaint = Column(Text, nullable=True)
# SOAP Sections
soap_subjective = Column(Text, nullable=True)
soap_objective = Column(Text, nullable=True)
soap_assessment = Column(Text, nullable=True)
soap_plan = Column(Text, nullable=True)
# Encryption tracking
is_encrypted = Column(Boolean, default=False, nullable=False)
class User(Base):
__tablename__ = "users"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
username = Column(String, unique=True, index=True, nullable=False)
full_name = Column(String, nullable=True)
role = Column(String, index=True, nullable=False)
hashed_password = Column(Text, nullable=False)
is_active = Column(Boolean, default=True, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow, index=True)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Phase 4: MFA
totp_secret = Column(Text, nullable=True)
mfa_enrolled_at = Column(DateTime, nullable=True)
class AuditLog(Base):
__tablename__ = "audit_logs"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
timestamp = Column(DateTime, default=datetime.utcnow, index=True)
user_id = Column(String, nullable=True, index=True)
username = Column(String, nullable=True, index=True)
role = Column(String, nullable=True, index=True)
action = Column(String, nullable=False)
resource = Column(String, nullable=False)
resource_id = Column(String, nullable=True)
endpoint = Column(String, nullable=False, index=True)
http_method = Column(String, nullable=False)
status_code = Column(Integer, nullable=False)
ip_address = Column(String, nullable=True)
user_agent = Column(Text, nullable=True)
details = Column(Text, nullable=True)
# Enhanced HIPAA fields
data_access_type = Column(String, nullable=True, index=True) # read, write, export, delete, purge
phi_accessed = Column(Boolean, default=False, nullable=False)
correlation_id = Column(String, nullable=True, index=True)
class ConversationSession(Base):
"""Stores AI Voice Assistant conversation sessions."""
__tablename__ = "conversation_sessions"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
created_at = Column(DateTime, default=datetime.utcnow, index=True)
ended_at = Column(DateTime, nullable=True)
intake_session_id = Column(String, nullable=True, index=True)
mode = Column(String, nullable=False, default="patient") # "patient" or "clinician"
state = Column(String, default="greeting")
turns_json = Column(Text, nullable=True) # JSON array of conversation turns
accumulated_transcript = Column(Text, nullable=True)
entities_json = Column(Text, nullable=True) # JSON of extracted entities
is_encrypted = Column(Boolean, default=False, nullable=False)
class RefreshToken(Base):
"""Stores hashed refresh tokens for JWT auth."""
__tablename__ = "refresh_tokens"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
user_id = Column(String, nullable=False, index=True)
token_hash = Column(String, nullable=False, unique=True, index=True)
expires_at = Column(DateTime, nullable=False)
revoked = Column(Boolean, default=False, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
ip_address = Column(String, nullable=True)
user_agent = Column(Text, nullable=True)
class ConsentRecord(Base):
"""Records patient verbal/written consent before intake processing."""
__tablename__ = "consent_records"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
session_id = Column(String, nullable=False, index=True)
patient_identifier = Column(String, nullable=True)
consent_type = Column(String, nullable=False, default="verbal") # verbal, written, electronic
consented_at = Column(DateTime, default=datetime.utcnow)
recorded_by_user_id = Column(String, nullable=True)
recorded_by_username = Column(String, nullable=True)
details = Column(Text, nullable=True)
revoked = Column(Boolean, default=False, nullable=False)
revoked_at = Column(DateTime, nullable=True)
class DataExportLog(Base):
"""Tracks all data exports for HIPAA compliance — who exported what, when, and where."""
__tablename__ = "data_export_logs"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
timestamp = Column(DateTime, default=datetime.utcnow, index=True)
user_id = Column(String, nullable=True, index=True)
username = Column(String, nullable=True, index=True)
export_type = Column(String, nullable=False) # fhir, csv, json, pdf
resource_type = Column(String, nullable=False) # session, audit_log, report
resource_ids = Column(Text, nullable=True) # JSON list of exported resource IDs
record_count = Column(Integer, default=0)
destination = Column(String, nullable=True) # e.g., "ehr_endpoint", "download", "api"
ip_address = Column(String, nullable=True)
status = Column(String, default="success") # success, failed, partial
details = Column(Text, nullable=True)
# =====================================================================
# Phase 2: Document Versioning & Collaborative Annotations
# =====================================================================
class DocumentVersion(Base):
"""Versioned SOAP document with diff tracking.
Each edit to a SOAP note creates a new version, preserving full history
for audit compliance and collaborative editing.
"""
__tablename__ = "document_versions"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
session_id = Column(String, nullable=False, index=True)
version_number = Column(Integer, nullable=False, default=1)
created_at = Column(DateTime, default=datetime.utcnow, index=True)
# Author
author_id = Column(String, nullable=True, index=True)
author_username = Column(String, nullable=True)
author_role = Column(String, nullable=True) # clinician, system, ai
# SOAP content (JSON blob with all sections)
content_json = Column(Text, nullable=False)
# Change tracking
diff_json = Column(Text, nullable=True) # JSON diff from previous version
change_summary = Column(Text, nullable=True) # Human-readable summary
change_type = Column(String, nullable=False, default="edit") # initial, edit, ai_generated, review, correction
# Metadata
confidence_json = Column(Text, nullable=True) # Confidence scores per section
is_encrypted = Column(Boolean, default=False, nullable=False)
class DocumentAnnotation(Base):
"""Inline annotation on a specific field of a SOAP document version.
Supports corrections, additions, questions, and approval markers
from collaborating clinicians.
"""
__tablename__ = "document_annotations"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
document_version_id = Column(String, nullable=False, index=True)
session_id = Column(String, nullable=False, index=True)
created_at = Column(DateTime, default=datetime.utcnow, index=True)
# Author
author_id = Column(String, nullable=True, index=True)
author_username = Column(String, nullable=True)
# Location within the SOAP note
soap_section = Column(String, nullable=False) # subjective, objective, assessment, plan
field_path = Column(String, nullable=True) # e.g., "symptom_details.onset"
text_offset_start = Column(Integer, nullable=True) # Character offset in section text
text_offset_end = Column(Integer, nullable=True)
# Annotation content
annotation_type = Column(String, nullable=False) # correction, addition, question, approval, flag
content = Column(Text, nullable=False)
suggested_replacement = Column(Text, nullable=True) # For corrections
# Status
status = Column(String, default="open") # open, resolved, rejected
resolved_by_id = Column(String, nullable=True)
resolved_at = Column(DateTime, nullable=True)
|