#!/usr/bin/env python3 """Generate a landscape, wrapped physician discussion report without medical risk scoring. Output: - reports/arztbericht_aktuell.pdf - reports/arztbericht_drive_link.json """ from __future__ import annotations import json import os import re import sqlite3 import subprocess from datetime import datetime from pathlib import Path from statistics import mean from typing import Any from reportlab.graphics.charts.barcharts import VerticalBarChart from reportlab.graphics.shapes import Drawing, String from reportlab.lib import colors from reportlab.lib.colors import HexColor from reportlab.lib.enums import TA_CENTER from reportlab.lib.pagesizes import A4, landscape from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet from reportlab.lib.units import cm from reportlab.platypus import Paragraph, SimpleDocTemplate, Spacer, Table, TableStyle, PageBreak import sys BASE = Path.home() / ".hermes" / "assets" / "Gesundheit" sys.path.insert(0, str(BASE / "scripts")) from apple_health_analytics import daily_series as canonical_apple_daily, unit_for as canonical_apple_unit DB = BASE / "health_data.db" REPORTS = BASE / "reports" OUT = REPORTS / "arztbericht_aktuell.pdf" META = REPORTS / "arztbericht_drive_link.json" DRIVE_REPORT_FOLDER_ID = "1oVpcbpelbt2IwHL9oGw9QTi2haQL63-v" ACCOUNT = "friday.uplink@gmail.com" GOG_SECRET_ENV = Path.home() / ".hermes" / "secrets" / "gog_keyring.env" def load_gog_keyring_password() -> str: value = os.environ.get("GOG_KEYRING_PASSWORD", "").strip() if value: return value if not GOG_SECRET_ENV.is_file(): raise RuntimeError(f"Missing gog keyring secret environment file: {GOG_SECRET_ENV}") if GOG_SECRET_ENV.stat().st_mode & 0o077: raise RuntimeError(f"Unsafe permissions on gog keyring secret file: {GOG_SECRET_ENV}") for raw_line in GOG_SECRET_ENV.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if line.startswith("export "): line = line[7:].strip() key, sep, raw_value = line.partition("=") if sep and key.strip() == "GOG_KEYRING_PASSWORD": secret = raw_value.strip().strip('"').strip("'") if secret: return secret raise RuntimeError(f"GOG_KEYRING_PASSWORD missing in {GOG_SECRET_ENV}") LAB_PARAMS = [ "C-Reaktives Protein (CRP)", "Blutsenkungsreaktion miniiSED", "Leukozyten", "Neutrophile", "Lymphozyten", "Thrombozyten", "D-Dimer", "Faktor VIII", "Fibrinogen", "Cholesterin gesamt", "LDL-Cholesterin", "HDL-Cholesterin", "Triglyceride", "Kreatinin", "eGFR (Niere)", "Albumin/Kreatinin", "ALAT (GPT)", "ASAT (GOT)", "GGT", "Vitamin D (25-OH)", ] APPLE_METRICS = { "step_count": ("Schritte", "sum"), "walking_running_distance": ("Geh-/Laufdistanz", "sum"), "resting_heart_rate": ("Ruhepuls", "avg"), "heart_rate_variability": ("HRV", "avg"), "sleep_analysis": ("Schlaf", "avg"), "blood_oxygen_saturation": ("SpO₂", "avg"), "respiratory_rate": ("Atemfrequenz", "avg"), "active_energy": ("Aktive Energie", "sum"), "physical_effort": ("Physical Effort", "avg"), "weight_body_mass": ("Gewicht", "avg"), } BLUE = HexColor("#1f4e78") PURPLE = HexColor("#7c3aed") LIGHT = HexColor("#f6f8fb") def c() -> sqlite3.Connection: con = sqlite3.connect(DB) con.row_factory = sqlite3.Row return con def fnum(x: Any) -> float | None: try: if x is None: return None return float(str(x).replace(",", ".").replace("<", "").replace(">", "").strip()) except Exception: return None def fmt(x: Any) -> str: if x is None: return "" if isinstance(x, float): return f"{x:.2f}".rstrip("0").rstrip(".") return str(x) def styles(): st = getSampleStyleSheet() st.add(ParagraphStyle(name="Small", parent=st["BodyText"], fontSize=8, leading=10)) st.add(ParagraphStyle(name="Tiny", parent=st["BodyText"], fontSize=7, leading=8)) st.add(ParagraphStyle(name="Cell", parent=st["BodyText"], fontSize=7, leading=8, wordWrap="CJK")) st.add(ParagraphStyle(name="CellCenter", parent=st["Cell"], alignment=TA_CENTER)) return st ST = styles() def P(text: Any, style="Cell") -> Paragraph: safe = str(text or "").replace("&", "&").replace("<", "<").replace(">", ">") return Paragraph(safe, ST[style]) def lab_date_expr() -> str: # ermittlung_datum is an import/extraction timestamp and must not be used as medical sampling date. return "COALESCE(abnahme_datum,befund_datum)" def parse_reference_boundary(value: Any) -> float | None: if value is None: return None text = str(value).strip().replace(",", ".") if not re.fullmatch(r"[+-]?(?:\d+(?:\.\d*)?|\.\d+)", text): return None try: return float(text) except ValueError: return None def latest_labs(con: sqlite3.Connection) -> list[list[Any]]: rows: list[list[Any]] = [] for p in LAB_PARAMS: r = con.execute(f""" SELECT parameter_name, {lab_date_expr()} datum, wert, einheit, reference_min, reference_max, validierungsstatus, quelle FROM laborwerte WHERE parameter_name=? AND wert IS NOT NULL AND lower(COALESCE(validierungsstatus,'')) = 'validiert' AND verified_against_original = 1 AND lower(COALESCE(reference_range_source,'')) = 'scanned_original' AND NULLIF(TRIM(COALESCE(einheit,'')), '') IS NOT NULL AND ( NULLIF(TRIM(COALESCE(reference_min,'')), '') IS NOT NULL OR NULLIF(TRIM(COALESCE(reference_max,'')), '') IS NOT NULL ) AND {lab_date_expr()} IS NOT NULL ORDER BY date({lab_date_expr()}) DESC, id DESC LIMIT 1 """, (p,)).fetchone() if not r: continue ref_min = parse_reference_boundary(r["reference_min"]) ref_max = parse_reference_boundary(r["reference_max"]) if ref_min is None and ref_max is None: continue ref = f"{fmt(ref_min)}–{fmt(ref_max)}" if ref_min is not None and ref_max is not None else (f"> {fmt(ref_min)}" if ref_min is not None else f"< {fmt(ref_max)}") rows.append([r["parameter_name"], str(r["datum"])[:10], fmt(r["wert"]), r["einheit"] or "", ref, r["validierungsstatus"] or "", r["quelle"] or ""]) return rows def apple_daily(con: sqlite3.Connection, metric: str, mode: str) -> dict[str, float]: # Canonical aggregation avoids double-counting overlapping daily/weekly exports # and converts weekly grouped sum rows to daily-equivalent values. return canonical_apple_daily(metric, mode) def apple_summary(con: sqlite3.Connection) -> list[list[Any]]: rows = [] for metric, (label, mode) in APPLE_METRICS.items(): series = apple_daily(con, metric, mode) if not series: continue unit = canonical_apple_unit(metric) labels = sorted(series) data = [series[d] for d in labels] rows.append([label, f"{labels[0]} bis {labels[-1]}", len(data), f"{fmt(data[-1])} {unit}", fmt(mean(data)), fmt(min(data)), fmt(max(data))]) return rows def chart_bar(labels: list[str], values: list[float], title: str, width=12.5*cm, height=6.5*cm, color=BLUE) -> Drawing: d = Drawing(width, height) d.add(String(0.2*cm, height-0.4*cm, title, fontName="Helvetica-Bold", fontSize=9, fillColor=BLUE)) bc = VerticalBarChart() bc.x = 0.9*cm; bc.y = 0.9*cm; bc.width = width - 1.5*cm; bc.height = height - 1.6*cm bc.data = [values] bc.categoryAxis.categoryNames = labels bc.categoryAxis.labels.angle = 45; bc.categoryAxis.labels.fontSize = 5 mn = min(values) if values else 0; mx = max(values) if values else 100 if mn >= 0: bc.valueAxis.valueMin = 0 bc.valueAxis.valueMax = max(mx * 1.2, 1) bc.bars[0].fillColor = color d.add(bc) return d def make_table(title: str, headers: list[str], rows: list[list[Any]], widths: list[float] | None = None, repeat=1, extra_style=None) -> list[Any]: story: list[Any] = [Paragraph(title, ST["Heading2"])] if not rows: story += [Paragraph("Keine Daten vorhanden.", ST["BodyText"]), Spacer(1, .25*cm)] return story data = [[P(h, "CellCenter") for h in headers]] + [[P(cell) for cell in row] for row in rows] tbl = Table(data, colWidths=widths, repeatRows=repeat) style = TableStyle([ ("BACKGROUND", (0,0), (-1,0), BLUE), ("TEXTCOLOR", (0,0), (-1,0), colors.white), ("FONTNAME", (0,0), (-1,0), "Helvetica-Bold"), ("GRID", (0,0), (-1,-1), 0.25, colors.lightgrey), ("VALIGN", (0,0), (-1,-1), "TOP"), ("ROWBACKGROUNDS", (0,1), (-1,-1), [colors.white, LIGHT]), ("LEFTPADDING", (0,0), (-1,-1), 3), ("RIGHTPADDING", (0,0), (-1,-1), 3), ]) if extra_style: for cmd in extra_style: style.add(*cmd) tbl.setStyle(style) story += [tbl, Spacer(1, .35*cm)] return story def recent_events(con): return [[str(r["date"])[:10], r["category"] or "", r["parameter"] or "", (r["value"] or r["notes"] or "")[:220]] for r in con.execute(""" SELECT date,category,parameter,value,notes FROM health_events WHERE date(date) IS NOT NULL AND upper(coalesce(category,'')) NOT IN ('LABOR','PROFIL','BEFUNDE') ORDER BY date DESC LIMIT 12 """)] def recent_symptoms(con): return [[str(r["datum"])[:10], r["symptom"] or "", r["schwergrad"] or "", (r["notizen"] or "")[:220]] for r in con.execute("SELECT datum,symptom,schwergrad,notizen FROM symptome WHERE date(datum) IS NOT NULL ORDER BY datum DESC LIMIT 12")] def nutrition(con): return [[r["datum"], fmt(r["plan_adherence_score"]), r["negative_hits"] or "", r["positive_hits"] or ""] for r in con.execute("SELECT datum,plan_adherence_score,negative_hits,positive_hits FROM nutrition_daily_features ORDER BY datum DESC LIMIT 10")] def upload(path: Path) -> dict: env = os.environ.copy(); env["GOG_KEYRING_PASSWORD"] = load_gog_keyring_password() cmd = ["gog", "-a", ACCOUNT, "drive", "upload", str(path), "--parent", DRIVE_REPORT_FOLDER_ID, "--json"] p = subprocess.run(cmd, text=True, capture_output=True, env=env) if p.returncode != 0: return {"status": "upload_failed", "stderr": p.stderr, "stdout": p.stdout} try: return json.loads(p.stdout) except Exception: return {"status": "uploaded_unparsed", "stdout": p.stdout} def main() -> None: REPORTS.mkdir(parents=True, exist_ok=True) con = c() doc = SimpleDocTemplate(str(OUT), pagesize=landscape(A4), rightMargin=.9*cm, leftMargin=.9*cm, topMargin=.9*cm, bottomMargin=.9*cm) story: list[Any] = [] story.append(Paragraph("JARVIS Gesundheitsübersicht für Arztgespräch", ST["Title"])) story.append(Paragraph(f"Erstellt: {datetime.now():%Y-%m-%d %H:%M} · Querformat · Tabellen mit automatischem Zeilenumbruch · Datenbasis: validierte Laborwerte + Apple Health + Events/Symptome", ST["Small"])) story.append(Spacer(1, .3*cm)) story.append(Paragraph("Hinweis: Strukturierter Gesprächsbericht; ersetzt keine ärztliche Beurteilung. Es erfolgt keine automatische medizinische Risiko-, Ampel- oder Entwarnungsbewertung. Angezeigte Laborwerte sind validiert, am Original geprüft und verwenden den dokumentbezogenen Referenzbereich; die ärztliche Interpretation bleibt erforderlich.", ST["Small"])) # Apple mini charts hrv = apple_daily(con, "heart_rate_variability", "avg") rhr = apple_daily(con, "resting_heart_rate", "avg") sleep = apple_daily(con, "sleep_analysis", "avg") steps = apple_daily(con, "step_count", "sum") mini = [] for title, series, color in [("HRV", hrv, PURPLE), ("Ruhepuls", rhr, BLUE), ("Schlaf", sleep, BLUE), ("Schritte", steps, PURPLE)]: if series: labs = sorted(series)[-12:] mini.append(chart_bar(labs, [series[x] for x in labs], title, color=color)) if mini: # Two charts per row using nested table. rows = [] for i in range(0, len(mini), 2): rows.append(mini[i:i+2]) story.append(Paragraph("1. Verlaufsgrafiken Apple Health", ST["Heading2"])) story.append(Table(rows, colWidths=[13.4*cm, 13.4*cm])) story.append(Spacer(1, .35*cm)) story += make_table("2. Aktuelle relevante Laborwerte", ["Parameter", "Datum", "Wert", "Einheit", "Referenz", "Status", "Quelle"], latest_labs(con), [5.2*cm, 2.2*cm, 1.8*cm, 1.9*cm, 2.5*cm, 2.2*cm, 10.5*cm]) story.append(PageBreak()) story += make_table("3. Apple Health Überblick", ["Metrik", "Zeitraum", "Tage", "letzter Wert", "Ø", "Min", "Max"], apple_summary(con), [4.0*cm, 4.8*cm, 1.4*cm, 3.0*cm, 2.0*cm, 2.0*cm, 2.0*cm]) story += make_table("4. Klinische Events", ["Datum", "Kategorie", "Ereignis", "Wert/Notiz"], recent_events(con), [2.2*cm, 3.0*cm, 5.2*cm, 16.3*cm]) story += make_table("5. Symptome", ["Datum", "Symptom", "Schweregrad", "Notiz"], recent_symptoms(con), [2.2*cm, 5.0*cm, 2.6*cm, 16.9*cm]) story += make_table("6. Ernährung / Plan-Treue", ["Datum", "Score", "Risiken", "Positive Treffer"], nutrition(con), [2.2*cm, 1.8*cm, 12.0*cm, 10.7*cm]) story.append(Paragraph("Besprechungspunkte: Behçet-/Vaskulitis-Aktivität, Thromboserisiko, Augen-/Haut-/Schleimhaut-Symptome, Schlaf/HRV/Ruhepuls als Belastungsmarker, Medikamentensicherheit und Labortrend-Kontrollen.", ST["Small"])) doc.build(story) con.close() up = upload(OUT) meta = {"generated_at": datetime.now().isoformat(), "pdf_path": str(OUT), "drive_upload": up, "drive_folder_id": DRIVE_REPORT_FOLDER_ID, "layout": "landscape_wrapped_v2"} META.write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps(meta, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()