Spaces:
Sleeping
Sleeping
File size: 5,378 Bytes
8de5584 | 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 | from sqlalchemy.orm import Session
from app.models.database_models import (
Document,
PathologyReport,
DocumentEmbedding,
User,
)
from datetime import datetime
import json
class DatabaseService:
"""Service for database operations"""
@staticmethod
def save_document(
db: Session, file_id: str, filename: str, file_size: int, file_type: str,
file_data: bytes = None
) -> Document:
"""Save document metadata and optionally the raw file bytes (PostgreSQL vault)"""
doc = Document(
file_id=file_id, filename=filename, file_size=file_size, file_type=file_type,
file_data=file_data
)
db.add(doc)
db.commit()
db.refresh(doc)
return doc
@staticmethod
def save_raw_text(
db: Session, file_id: str, raw_text: str, status: str = "completed"
):
"""Save OCR extracted text"""
doc = db.query(Document).filter(Document.file_id == file_id).first()
if doc:
doc.raw_text = raw_text
doc.ocr_status = status
db.commit()
db.refresh(doc)
return doc
return None
@staticmethod
def save_cleaned_text(db: Session, file_id: str, cleaned_text: str):
"""Save cleaned text"""
doc = db.query(Document).filter(Document.file_id == file_id).first()
if doc:
doc.cleaned_text = cleaned_text
db.commit()
db.refresh(doc)
return doc
return None
@staticmethod
def save_pathology_report(
db: Session, file_id: str, extraction_data: dict
) -> PathologyReport:
"""Save extracted pathology report data"""
test_date_value = extraction_data.get("test_date")
parsed_test_date = None
if isinstance(test_date_value, datetime):
parsed_test_date = test_date_value
elif isinstance(test_date_value, str) and test_date_value.strip():
# Accept ISO date or datetime strings (e.g., 2026-03-28 or 2026-03-28T12:34:56)
try:
parsed_test_date = datetime.fromisoformat(test_date_value.strip())
except Exception:
parsed_test_date = None
report = PathologyReport(
document_id=file_id,
patient_id=extraction_data.get("patient_id"),
patient_name=extraction_data.get("patient_name"),
test_type=extraction_data.get("test_type", ""),
test_date=parsed_test_date,
findings=json.dumps(extraction_data.get("findings", [])),
diagnosis=extraction_data.get("diagnosis"),
recommendations=extraction_data.get("recommendations"),
summary=extraction_data.get("summary", ""),
)
db.add(report)
# Mark extraction status on the parent document (if present)
doc = db.query(Document).filter(Document.file_id == file_id).first()
if doc:
doc.extraction_status = "completed"
db.commit()
db.refresh(report)
return report
@staticmethod
def save_embedding(db: Session, file_id: str, embedding: list, text_chunk: str):
"""Save document embedding"""
emb = (
db.query(DocumentEmbedding)
.filter(DocumentEmbedding.document_id == file_id)
.first()
)
if emb:
emb.embedding = embedding
emb.text_chunk = text_chunk
else:
emb = DocumentEmbedding(
document_id=file_id, embedding=embedding, text_chunk=text_chunk
)
db.add(emb)
db.commit()
db.refresh(emb)
return emb
@staticmethod
def get_document(db: Session, file_id: str):
"""Get document by file ID"""
return db.query(Document).filter(Document.file_id == file_id).first()
@staticmethod
def list_documents(db: Session, limit: int = 10):
"""List all documents"""
return (
db.query(Document).order_by(Document.upload_date.desc()).limit(limit).all()
)
@staticmethod
def get_pathology_report(db: Session, file_id: str):
"""Get pathology report by file ID"""
return (
db.query(PathologyReport)
.filter(PathologyReport.document_id == file_id)
.first()
)
@staticmethod
def get_embedding(db: Session, file_id: str):
"""Get embedding by file ID"""
return (
db.query(DocumentEmbedding)
.filter(DocumentEmbedding.document_id == file_id)
.first()
)
@staticmethod
def save_user(
db: Session, username: str, email: str, password_hash: str, role: str = "doctor"
) -> User:
"""Save user to database"""
user = User(
username=username, email=email, password_hash=password_hash, role=role
)
db.add(user)
db.commit()
db.refresh(user)
return user
@staticmethod
def get_user_by_username(db: Session, username: str):
"""Get user by username"""
return db.query(User).filter(User.username == username).first()
@staticmethod
def get_user_by_email(db: Session, email: str):
"""Get user by email"""
return db.query(User).filter(User.email == email).first()
|