""" Service for generating clinical PDFs with ReportLab and HTML previews with Jinja2. """ import base64 from dataclasses import dataclass from io import BytesIO from pathlib import Path from typing import Any from jinja2 import Environment, FileSystemLoader, select_autoescape from app.models.lab import Lab from app.models.order import VisitOrder from app.models.patient import Patient from app.models.report import Report, ReportStatus from app.services.clinical_engine import ClinicalEngine from app.services.report_pdf_renderer import render_diagnostic_pdf from app.services.signature_service import read_signature_bytes def build_qr_b64(url: str) -> str | None: """Render a QR code PNG for a URL as a data URI (free, no external service).""" try: import qrcode except ImportError: # pragma: no cover - optional dependency guard return None img = qrcode.make(url, box_size=6, border=2) buf = BytesIO() img.save(buf, format="PNG") return "data:image/png;base64," + base64.b64encode(buf.getvalue()).decode("ascii") @dataclass class ReportParameterView: """Single analyte parameter view for report presentation.""" name: str value: str unit: str | None reference_interval: str flag: str | None is_abnormal: bool notes: str | None @dataclass class ReportTestGroupView: """Group of parameters under a clinical test heading.""" test_name: str test_code: str parameters: list[ReportParameterView] class PDFReportService: """Renders diagnostic reports to HTML and binary PDF.""" def __init__(self) -> None: templates_dir = Path(__file__).resolve().parent.parent / "templates" self.jinja_env = Environment( loader=FileSystemLoader(str(templates_dir)), autoescape=select_autoescape(["html", "xml"]), ) self.template = self.jinja_env.get_template("reports/diagnostic_report.html") self.engine = ClinicalEngine() @staticmethod def _trim_decimal(raw: str | None) -> str | None: """Display trim: '13.0000' → '13', '4.5000' → '4.5'.""" if raw is None: return None try: num = float(str(raw).replace(",", "")) except (TypeError, ValueError): return str(raw) return str(int(num)) if num.is_integer() else str(num) @staticmethod def _format_reference_interval( ref_low: str | None, ref_high: str | None, ref_text: str | None, ) -> str: ref_low = PDFReportService._trim_decimal(ref_low) ref_high = PDFReportService._trim_decimal(ref_high) if ref_low is not None and ref_high is not None: return f"{ref_low} – {ref_high}" if ref_low is not None: return f"> {ref_low}" if ref_high is not None: return f"< {ref_high}" if ref_text: return ref_text return "—" @staticmethod def _test_on_side(test: object, modality: str) -> bool: """Keep tests of the report's own side; unclassifiable tests stay visible.""" category = getattr(test, "category", None) if category is None: return True value = getattr(category.modality, "value", category.modality) return str(value) == modality def _build_context(self, report: Report, modality: str = "mixed") -> dict[str, Any]: order: VisitOrder | None = report.order patient: Patient | None = order.patient if order else None lab: Lab | None = order.lab if order else None # Patient demographics patient_age = ( self.engine.calculate_patient_age(patient.date_of_birth, patient.age) if patient else None ) gender_str = patient.gender.value.capitalize() if patient and patient.gender else "Unknown" age_str = f"{patient_age} Y" if patient_age is not None else "Age Unknown" patient_age_gender = f"{age_str} / {gender_str}" # Specimen display specimen_list: list[str] = [] if order and order.samples: for s in order.samples: if s.sample_type: specimen_list.append(s.sample_type.title()) if specimen_list: specimen_display = ", ".join(dict.fromkeys(specimen_list)) elif modality == "imaging": specimen_display = "Not applicable (imaging)" else: specimen_display = "Clinical Specimen" # Dates order_date = order.created_at.strftime("%Y-%m-%d") if order and order.created_at else "—" reported_date = "Pending" if report.released_at: reported_date = report.released_at.strftime("%Y-%m-%d %H:%M UTC") elif report.verified_at: reported_date = report.verified_at.strftime("%Y-%m-%d %H:%M UTC") elif report.updated_at: reported_date = report.updated_at.strftime("%Y-%m-%d %H:%M UTC") # Watermark watermark_text = None if report.status in (ReportStatus.DRAFT, ReportStatus.PENDING_VERIFICATION): watermark_text = "DRAFT — NOT FOR CLINICAL USE" elif report.status == ReportStatus.REJECTED: watermark_text = "REJECTED — INVALID" # Stored observations from JSONB result_data stored_params: dict[str, dict[str, Any]] = {} if report.result_data and isinstance(report.result_data, dict): raw_params = report.result_data.get("parameters") if isinstance(raw_params, dict): stored_params = raw_params # Build test groups test_groups: list[ReportTestGroupView] = [] rendered_param_ids: set[str] = set() if order and order.order_tests: for ot in order.order_tests: test = ot.test if not test or not self._test_on_side(test, modality): continue # Honor analyte scopes: partial-panel orders print only picked analytes. scope = {str(x) for x in ot.selected_parameter_ids} if ot.selected_parameter_ids else None param_views: list[ReportParameterView] = [] for param in sorted(test.parameters, key=lambda p: p.sort_order): param_id_str = str(param.id) if scope is not None and param_id_str not in scope: continue rendered_param_ids.add(param_id_str) if param_id_str in stored_params: data = stored_params[param_id_str] val_str = str(data.get("value", "—")) ref_low = data.get("reference_low") ref_high = data.get("reference_high") ref_text = data.get("reference_text") flag = data.get("flag") is_abnormal = bool(data.get("is_abnormal", False)) notes = data.get("notes") else: val_str = "—" flag = None is_abnormal = False notes = None ref_range = self.engine.select_reference_range( param.ranges, patient.gender if patient else None, patient_age, ) ref_low = ( str(ref_range.numeric_low) if ref_range and ref_range.numeric_low is not None else None ) ref_high = ( str(ref_range.numeric_high) if ref_range and ref_range.numeric_high is not None else None ) ref_text = ref_range.text_normal if ref_range else None interval_str = self._format_reference_interval(ref_low, ref_high, ref_text) param_views.append( ReportParameterView( name=param.name, value=val_str, unit=param.unit, reference_interval=interval_str, flag=flag, is_abnormal=is_abnormal, notes=notes, ) ) if param_views: test_groups.append( ReportTestGroupView( test_name=test.name, test_code=test.code, parameters=param_views, ) ) # Package members with no order_test row (legacy + package-only orders): # render as proper panel sections, not the generic fallback bucket. if order: covered_codes = set() for ot in order.order_tests: if ot.test: covered_codes.add(ot.test_id) for op in order.order_packages: pkg = op.package if not pkg: continue for item in sorted(pkg.items, key=lambda i: i.sort_order): test = item.test if not test or test.id in covered_codes or not self._test_on_side(test, modality): continue covered_codes.add(test.id) pkg_views: list[ReportParameterView] = [] for param in sorted(test.parameters, key=lambda p: p.sort_order): if item.parameter_id and param.id != item.parameter_id: continue param_id_str = str(param.id) rendered_param_ids.add(param_id_str) if param_id_str in stored_params: data = stored_params[param_id_str] pkg_views.append( ReportParameterView( name=param.name, value=str(data.get("value", "—")), unit=param.unit, reference_interval=self._format_reference_interval( data.get("reference_low"), data.get("reference_high"), data.get("reference_text"), ), flag=data.get("flag"), is_abnormal=bool(data.get("is_abnormal", False)), notes=data.get("notes"), ) ) else: ref_range = self.engine.select_reference_range( param.ranges, patient.gender if patient else None, patient_age, ) pkg_views.append( ReportParameterView( name=param.name, value="—", unit=param.unit, reference_interval=self._format_reference_interval( str(ref_range.numeric_low) if ref_range and ref_range.numeric_low is not None else None, str(ref_range.numeric_high) if ref_range and ref_range.numeric_high is not None else None, ref_range.text_normal if ref_range else None, ), flag=None, is_abnormal=False, notes=None, ) ) if pkg_views: test_groups.append( ReportTestGroupView( test_name=test.name, test_code=test.code, parameters=pkg_views, ) ) # Fallback for any stored parameters not belonging to order_tests unassigned_views: list[ReportParameterView] = [] for pid, data in stored_params.items(): if pid not in rendered_param_ids: unassigned_views.append( ReportParameterView( name=str(data.get("parameter_name", f"Parameter {pid[:6]}")), value=str(data.get("value", "—")), unit=data.get("unit"), reference_interval=self._format_reference_interval( data.get("reference_low"), data.get("reference_high"), data.get("reference_text"), ), flag=data.get("flag"), is_abnormal=bool(data.get("is_abnormal", False)), notes=data.get("notes"), ) ) if unassigned_views: test_groups.append( ReportTestGroupView( test_name="Diagnostic Observations", test_code="", parameters=unassigned_views, ) ) # Per-study imaging narrative (professional radiology sections). imaging_sections: list[dict[str, object]] = [] if modality == "imaging": raw_sections: dict[str, object] = {} if report.result_data and isinstance(report.result_data, dict): maybe_sections = report.result_data.get("imaging_sections") if isinstance(maybe_sections, dict): raw_sections = maybe_sections for key in sorted(raw_sections): section = raw_sections[key] if isinstance(section, dict): imaging_sections.append(section) # Verifier signature overlay (transparent PNG at the exact sign-off block) verifier_signature_b64: str | None = None verifier = getattr(report, "verifier", None) signature_path = getattr(verifier, "signature_image_path", None) if verifier else None if signature_path: raw_sig = read_signature_bytes(signature_path) if raw_sig: verifier_signature_b64 = "data:image/png;base64," + base64.b64encode(raw_sig).decode("ascii") # QR code for the printed report: durable no-login capability URL. report_qr_b64: str | None = None report_qr_url: str | None = None public_token = getattr(report, "public_token", None) if report.status == ReportStatus.RELEASED and public_token: from app.config import settings report_qr_url = f"{settings.CLIENT_APP_BASE_URL.rstrip('/')}/client/reports/{public_token}" report_qr_b64 = build_qr_b64(report_qr_url) return { "report": report, "order": order, "patient": patient, "lab": lab, "report_modality": modality, "imaging_sections": imaging_sections, "test_groups": test_groups, "watermark_text": watermark_text, "order_date": order_date, "reported_date": reported_date, "patient_age_gender": patient_age_gender, "specimen_display": specimen_display, "verifier_signature_b64": verifier_signature_b64, "report_qr_b64": report_qr_b64, "report_qr_url": report_qr_url, } def render_html(self, report: Report, modality: str = "mixed") -> str: """Render diagnostic report to HTML string.""" context = self._build_context(report, modality) return self.template.render(context) def render_pdf(self, report: Report, modality: str = "mixed") -> bytes: """Render diagnostic report dynamically to binary PDF bytes.""" return render_diagnostic_pdf(self._build_context(report, modality))