mporwoll/dak.c2s
1
0
Fork 0
mirror of https://github.com/complexcaresolutions/dak.c2s.git synced 2026-03-17 16:03:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions
dak.c2s/backend/app/services/report_service.py
CCS Admin 2f0a556371 fix: use same KW range for year-over-year KPI comparison 
The previous implementation compared a partial current year against a
full previous year, producing misleading percentages. Now determines the
max KW with data in the selected year and filters the previous year to
the same KW range for a fair comparison.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-28 13:23:03 +00:00

843 lines
27 KiB
Python
Raw Blame History

"""Report service — all 5 sheet calculations for the DAK Berichtswesen.
Sheet 1: Auswertung KW gesamt — weekly totals + year summary
Sheet 2: Auswertung nach Fachgebieten — per-KW per-Fallgruppe breakdown
Sheet 3: Auswertung Gutachten — per-KW gutachten / alternative / bestaetigung
Sheet 4: Auswertung Therapieaenderungen — per-KW therapy-change metrics
Sheet 5: Auswertung ICD onko — ICD code frequency for onko cases
All queries use SQLAlchemy (not pandas) against the cases / case_icd_codes tables.
"""
from __future__ import annotations
import logging
from typing import Any
from sqlalchemy import Integer, and_, func
from sqlalchemy.orm import Session
from app.config import get_settings
from app.models.case import Case, CaseICDCode
settings = get_settings()
logger = logging.getLogger(__name__)
# Canonical Fallgruppen in display order
FALLGRUPPEN = ("onko", "kardio", "intensiv", "galle", "sd")
# Report type definitions: each maps to a subset of Fallgruppen
REPORT_TYPES: dict[str, tuple[str, ...]] = {
"gesamt": FALLGRUPPEN,
"onko_intensiv": ("onko", "intensiv"),
"galle_schild": ("galle", "sd"),
}
REPORT_TYPE_LABELS: dict[str, str] = {
"gesamt": "Gesamt",
"onko_intensiv": "Onko-Intensiv",
"galle_schild": "Galle-Schild",
}
# Number of calendar weeks to include (ISO weeks 1..52; 53 is rare)
MAX_KW = 52
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _int(val: Any) -> int:
"""Safely coerce a query result to int (None -> 0)."""
if val is None:
return 0
return int(val)
def _pct(part: int, total: int) -> float | None:
"""Return part/total as a float, or None if total==0."""
if total == 0:
return None
return round(part / total, 4)
def _empty_weekly_row(kw: int) -> dict:
"""Return a zeroed-out weekly row template for Sheet 1."""
return {
"kw": kw,
"erstberatungen": 0,
"unterlagen": 0,
"ablehnungen": 0,
"keine_rm": 0,
"gutachten": 0,
}
def _empty_fg_weekly_row(kw: int, fallgruppen: tuple[str, ...] = FALLGRUPPEN) -> dict:
"""Return a zeroed-out weekly row template for Sheet 2."""
row: dict[str, Any] = {"kw": kw}
for fg in fallgruppen:
row[fg] = {"anzahl": 0, "gutachten": 0, "keine_rm": 0}
return row
def _empty_gutachten_weekly_row(kw: int, fallgruppen: tuple[str, ...] = FALLGRUPPEN) -> dict:
"""Return a zeroed-out weekly row template for Sheet 3."""
row: dict[str, Any] = {"kw": kw}
for group in ("gesamt",) + fallgruppen:
row[group] = {"gutachten": 0, "alternative": 0, "bestaetigung": 0}
return row
def _empty_ta_weekly_row(kw: int) -> dict:
"""Return a zeroed-out weekly row template for Sheet 4."""
return {
"kw": kw,
"gutachten": 0,
"ta_ja": 0,
"ta_nein": 0,
"diagnosekorrektur": 0,
"unterversorgung": 0,
"uebertherapie": 0,
}
# ---------------------------------------------------------------------------
# Sheet 1: Auswertung KW gesamt
# ---------------------------------------------------------------------------
def calculate_sheet1_data(db: Session, jahr: int, max_kw: int | None = None, fallgruppen: tuple[str, ...] | None = None) -> dict:
"""Calculate *Auswertung KW gesamt*.
Returns::
{
"summary": {
"erstberatungen": int,
"ablehnungen": int,
"unterlagen": int,
"keine_rueckmeldung": int,
"gutachten": int,
},
"weekly": [
{"kw": 1, "erstberatungen": X, "unterlagen": X,
"ablehnungen": X, "keine_rm": X, "gutachten": X},
... # kw 1..52
]
}
Business rules (matching the Excel formulas):
* Erstberatungen = total cases for the KW
* Unterlagen = cases where unterlagen == True
* Ablehnungen = cases where ablehnung == True
* Gutachten = cases where gutachten == True
* Keine RM = Unterlagen - Gutachten (derived, per KW row)
If *max_kw* is given, only data up to and including that KW is included.
"""
# One query: group by kw, count the four flags
filters = [Case.versicherung == settings.VERSICHERUNG_FILTER, Case.jahr == jahr]
if fallgruppen is not None:
filters.append(Case.fallgruppe.in_(fallgruppen))
if max_kw is not None:
filters.append(Case.kw <= max_kw)
rows = (
db.query(
Case.kw,
func.count(Case.id).label("erstberatungen"),
func.sum(Case.unterlagen.cast(Integer)).label("unterlagen"),
func.sum(Case.ablehnung.cast(Integer)).label("ablehnungen"),
func.sum(Case.gutachten.cast(Integer)).label("gutachten"),
)
.filter(*filters)
.group_by(Case.kw)
.all()
)
# Build a lookup kw -> values
kw_map: dict[int, dict] = {}
for row in rows:
kw = _int(row.kw)
unterlagen = _int(row.unterlagen)
gutachten = _int(row.gutachten)
kw_map[kw] = {
"kw": kw,
"erstberatungen": _int(row.erstberatungen),
"unterlagen": unterlagen,
"ablehnungen": _int(row.ablehnungen),
"keine_rm": unterlagen - gutachten,
"gutachten": gutachten,
}
# Build full 1..52 list (filling gaps with zeros)
weekly = []
for kw in range(1, MAX_KW + 1):
weekly.append(kw_map.get(kw, _empty_weekly_row(kw)))
# Summary (sums across all weeks)
total_erst = sum(w["erstberatungen"] for w in weekly)
total_abl = sum(w["ablehnungen"] for w in weekly)
total_unt = sum(w["unterlagen"] for w in weekly)
total_keine = sum(w["keine_rm"] for w in weekly)
total_gut = sum(w["gutachten"] for w in weekly)
summary = {
"erstberatungen": total_erst,
"ablehnungen": total_abl,
"unterlagen": total_unt,
"keine_rueckmeldung": total_keine,
"gutachten": total_gut,
}
return {"summary": summary, "weekly": weekly}
# ---------------------------------------------------------------------------
# Sheet 2: Auswertung nach Fachgebieten
# ---------------------------------------------------------------------------
def calculate_sheet2_data(db: Session, jahr: int, max_kw: int | None = None, fallgruppen: tuple[str, ...] | None = None) -> dict:
"""Calculate *Auswertung nach Fachgebieten*.
Per KW, per Fallgruppe: Anzahl, Gutachten, Keine RM/Ablehnung.
Returns::
{
"weekly": [
{
"kw": 1,
"onko": {"anzahl": X, "gutachten": X, "keine_rm": X},
"kardio": {...},
"intensiv": {...},
"galle": {...},
"sd": {...},
},
...
]
}
Keine RM/Ablehnung = Anzahl - Gutachten (per the Excel formula).
If *max_kw* is given, only data up to and including that KW is included.
If *fallgruppen* is given, only those Fallgruppen are included.
"""
fgs = fallgruppen or FALLGRUPPEN
filters = [Case.versicherung == settings.VERSICHERUNG_FILTER, Case.jahr == jahr]
if fallgruppen is not None:
filters.append(Case.fallgruppe.in_(fallgruppen))
if max_kw is not None:
filters.append(Case.kw <= max_kw)
rows = (
db.query(
Case.kw,
Case.fallgruppe,
func.count(Case.id).label("anzahl"),
func.sum(Case.gutachten.cast(Integer)).label("gutachten"),
)
.filter(*filters)
.group_by(Case.kw, Case.fallgruppe)
.all()
)
# Build kw -> fg -> values
kw_map: dict[int, dict] = {}
for row in rows:
kw = _int(row.kw)
fg = row.fallgruppe
if fg not in fgs:
logger.warning("Unknown fallgruppe '%s' in case data, skipping", fg)
continue
if kw not in kw_map:
kw_map[kw] = _empty_fg_weekly_row(kw, fgs)
anzahl = _int(row.anzahl)
gutachten = _int(row.gutachten)
kw_map[kw][fg] = {
"anzahl": anzahl,
"gutachten": gutachten,
"keine_rm": anzahl - gutachten,
}
weekly = []
for kw in range(1, MAX_KW + 1):
weekly.append(kw_map.get(kw, _empty_fg_weekly_row(kw, fgs)))
return {"weekly": weekly}
# ---------------------------------------------------------------------------
# Sheet 3: Auswertung Gutachten
# ---------------------------------------------------------------------------
def calculate_sheet3_data(db: Session, jahr: int, max_kw: int | None = None, fallgruppen: tuple[str, ...] | None = None) -> dict:
"""Calculate *Auswertung Gutachten*.
Per KW, per group (gesamt + 5 Fallgruppen):
Gutachten count, Alternative, Bestaetigung.
Returns::
{
"weekly": [
{
"kw": 1,
"gesamt": {"gutachten": X, "alternative": X, "bestaetigung": X},
"onko": {...},
"kardio": {...},
"intensiv": {...},
"galle": {...},
"sd": {...},
},
...
]
}
In the Excel:
- Per Fallgruppe: Gutachten = count, Alternative = count where typ='Alternative',
Bestaetigung = Gutachten - Alternative
- Gesamt = sum across all Fallgruppen
If *max_kw* is given, only data up to and including that KW is included.
If *fallgruppen* is given, only those Fallgruppen are included.
"""
fgs = fallgruppen or FALLGRUPPEN
filters = [
Case.versicherung == settings.VERSICHERUNG_FILTER,
Case.jahr == jahr,
Case.gutachten == True, # noqa: E712
]
if fallgruppen is not None:
filters.append(Case.fallgruppe.in_(fallgruppen))
if max_kw is not None:
filters.append(Case.kw <= max_kw)
rows = (
db.query(
Case.kw,
Case.fallgruppe,
func.count(Case.id).label("gutachten"),
func.sum(
(Case.gutachten_typ == "Alternative").cast(Integer)
).label("alternative"),
)
.filter(*filters)
.group_by(Case.kw, Case.fallgruppe)
.all()
)
kw_map: dict[int, dict] = {}
for row in rows:
kw = _int(row.kw)
fg = row.fallgruppe
if fg not in fgs:
continue
if kw not in kw_map:
kw_map[kw] = _empty_gutachten_weekly_row(kw, fgs)
gutachten = _int(row.gutachten)
alternative = _int(row.alternative)
kw_map[kw][fg] = {
"gutachten": gutachten,
"alternative": alternative,
"bestaetigung": gutachten - alternative,
}
# Compute gesamt (sum of included Fallgruppen per KW)
for kw_data in kw_map.values():
total_g = sum(kw_data[fg]["gutachten"] for fg in fgs)
total_a = sum(kw_data[fg]["alternative"] for fg in fgs)
kw_data["gesamt"] = {
"gutachten": total_g,
"alternative": total_a,
"bestaetigung": total_g - total_a,
}
weekly = []
for kw in range(1, MAX_KW + 1):
weekly.append(kw_map.get(kw, _empty_gutachten_weekly_row(kw, fgs)))
return {"weekly": weekly}
# ---------------------------------------------------------------------------
# Sheet 4: Auswertung Therapieaenderungen
# ---------------------------------------------------------------------------
def calculate_sheet4_data(db: Session, jahr: int, max_kw: int | None = None, fallgruppen: tuple[str, ...] | None = None) -> dict:
"""Calculate *Auswertung Therapieaenderungen*.
Per KW: Gutachten count, TA Ja, TA Nein, Diagnosekorrektur,
Unterversorgung, Uebertherapie.
Returns::
{
"weekly": [
{
"kw": 1,
"gutachten": X,
"ta_ja": X,
"ta_nein": X,
"diagnosekorrektur": X,
"unterversorgung": X,
"uebertherapie": X,
},
...
]
}
If *max_kw* is given, only data up to and including that KW is included.
If *fallgruppen* is given, only those Fallgruppen are included.
"""
filters = [
Case.versicherung == settings.VERSICHERUNG_FILTER,
Case.jahr == jahr,
Case.gutachten == True, # noqa: E712
]
if fallgruppen is not None:
filters.append(Case.fallgruppe.in_(fallgruppen))
if max_kw is not None:
filters.append(Case.kw <= max_kw)
rows = (
db.query(
Case.kw,
func.count(Case.id).label("gutachten"),
func.sum(
(Case.therapieaenderung == "Ja").cast(Integer)
).label("ta_ja"),
func.sum(
(Case.therapieaenderung == "Nein").cast(Integer)
).label("ta_nein"),
func.sum(Case.ta_diagnosekorrektur.cast(Integer)).label("diagnosekorrektur"),
func.sum(Case.ta_unterversorgung.cast(Integer)).label("unterversorgung"),
func.sum(Case.ta_uebertherapie.cast(Integer)).label("uebertherapie"),
)
.filter(*filters)
.group_by(Case.kw)
.all()
)
kw_map: dict[int, dict] = {}
for row in rows:
kw = _int(row.kw)
kw_map[kw] = {
"kw": kw,
"gutachten": _int(row.gutachten),
"ta_ja": _int(row.ta_ja),
"ta_nein": _int(row.ta_nein),
"diagnosekorrektur": _int(row.diagnosekorrektur),
"unterversorgung": _int(row.unterversorgung),
"uebertherapie": _int(row.uebertherapie),
}
weekly = []
for kw in range(1, MAX_KW + 1):
weekly.append(kw_map.get(kw, _empty_ta_weekly_row(kw)))
return {"weekly": weekly}
# ---------------------------------------------------------------------------
# Sheet 5: Auswertung ICD onko
# ---------------------------------------------------------------------------
def calculate_sheet5_data(db: Session, jahr: int, max_kw: int | None = None, fallgruppen: tuple[str, ...] | None = None) -> dict:
"""Calculate *Auswertung ICD onko*.
Returns sorted list of ICD codes from onko cases with counts.
Query: case_icd_codes JOIN cases
WHERE cases.fallgruppe = 'onko' AND cases.jahr = jahr
GROUP BY UPPER(icd_code)
ORDER BY count DESC, icd_code ASC
Returns::
{
"icd_codes": [
{"icd": "C18", "count": 17},
{"icd": "C50", "count": 12},
...
]
}
If *max_kw* is given, only data up to and including that KW is included.
If *fallgruppen* is given and ``'onko'`` is not in the set, returns an
empty list (ICD codes are only relevant for onko cases).
"""
fgs = fallgruppen or FALLGRUPPEN
if "onko" not in fgs:
return {"icd_codes": []}
filter_conditions = [
Case.versicherung == settings.VERSICHERUNG_FILTER,
Case.fallgruppe == "onko",
Case.jahr == jahr,
]
if max_kw is not None:
filter_conditions.append(Case.kw <= max_kw)
rows = (
db.query(
func.upper(CaseICDCode.icd_code).label("icd"),
func.count(CaseICDCode.id).label("cnt"),
)
.join(Case, CaseICDCode.case_id == Case.id)
.filter(and_(*filter_conditions))
.group_by(func.upper(CaseICDCode.icd_code))
.order_by(func.count(CaseICDCode.id).desc(), func.upper(CaseICDCode.icd_code))
.all()
)
icd_codes = [{"icd": row.icd, "count": _int(row.cnt)} for row in rows]
return {"icd_codes": icd_codes}
# ---------------------------------------------------------------------------
# Dashboard KPIs
# ---------------------------------------------------------------------------
def calculate_dashboard_kpis(db: Session, jahr: int, max_kw: int | None = None) -> dict:
"""Calculate live KPIs for the dashboard.
If *max_kw* is given, only cases up to and including that calendar week
are counted. This is used for fair year-over-year comparisons.
Returns::
{
"total_cases": int,
"pending_icd": int,
"pending_coding": int,
"total_gutachten": int,
"total_ablehnungen": int,
"total_unterlagen": int,
"fallgruppen": {"onko": X, "kardio": X, "intensiv": X, "galle": X, "sd": X},
"gutachten_typen": {"alternative": X, "bestaetigung": X, "uncodiert": X},
}
"""
# Base filters for this portal's insurance + year (+ optional KW cutoff)
v_filter = Case.versicherung == settings.VERSICHERUNG_FILTER
base_filters = [v_filter, Case.jahr == jahr]
if max_kw is not None:
base_filters.append(Case.kw <= max_kw)
# Total cases for the year
total_cases = (
db.query(func.count(Case.id)).filter(*base_filters).scalar() or 0
)
# Cases without ICD codes entered
pending_icd = (
db.query(func.count(Case.id))
.filter(*base_filters, Case.icd == None) # noqa: E711
.scalar()
or 0
)
# Gutachten without gutachten_typ (need coding)
pending_coding = (
db.query(func.count(Case.id))
.filter(
*base_filters,
Case.gutachten == True, # noqa: E712
Case.gutachten_typ == None, # noqa: E711
)
.scalar()
or 0
)
# Gutachten totals
total_gutachten = (
db.query(func.count(Case.id))
.filter(*base_filters, Case.gutachten == True) # noqa: E712
.scalar()
or 0
)
# Ablehnungen
total_ablehnungen = (
db.query(func.count(Case.id))
.filter(*base_filters, Case.ablehnung == True) # noqa: E712
.scalar()
or 0
)
# Unterlagen
total_unterlagen = (
db.query(func.count(Case.id))
.filter(*base_filters, Case.unterlagen == True) # noqa: E712
.scalar()
or 0
)
# Per-Fallgruppe counts
fg_rows = (
db.query(Case.fallgruppe, func.count(Case.id).label("cnt"))
.filter(*base_filters)
.group_by(Case.fallgruppe)
.all()
)
fallgruppen = {fg: 0 for fg in FALLGRUPPEN}
for row in fg_rows:
if row.fallgruppe in fallgruppen:
fallgruppen[row.fallgruppe] = _int(row.cnt)
# Gutachten type breakdown
typ_rows = (
db.query(Case.gutachten_typ, func.count(Case.id).label("cnt"))
.filter(*base_filters, Case.gutachten == True) # noqa: E712
.group_by(Case.gutachten_typ)
.all()
)
gutachten_typen = {"alternative": 0, "bestaetigung": 0, "uncodiert": 0}
for row in typ_rows:
if row.gutachten_typ == "Alternative":
gutachten_typen["alternative"] = _int(row.cnt)
elif row.gutachten_typ == "Bestätigung":
gutachten_typen["bestaetigung"] = _int(row.cnt)
else:
gutachten_typen["uncodiert"] = _int(row.cnt)
return {
"total_cases": total_cases,
"pending_icd": pending_icd,
"pending_coding": pending_coding,
"total_gutachten": total_gutachten,
"total_ablehnungen": total_ablehnungen,
"total_unterlagen": total_unterlagen,
"fallgruppen": fallgruppen,
"gutachten_typen": gutachten_typen,
}
# ---------------------------------------------------------------------------
# Dashboard: Yearly KW comparison (multi-year)
# ---------------------------------------------------------------------------
def calculate_yearly_kw_comparison(db: Session) -> list[dict]:
"""Return per-KW total case counts for each year from 2022 onwards.
Returns a list of dicts like::
[
{"kw": 1, "2022": 5, "2023": 8, "2024": 12, ...},
{"kw": 2, "2022": 3, "2023": 7, "2024": 9, ...},
...
]
"""
v_filter = Case.versicherung == settings.VERSICHERUNG_FILTER
rows = (
db.query(
Case.jahr,
Case.kw,
func.count(Case.id).label("cnt"),
)
.filter(v_filter, Case.jahr >= 2022)
.group_by(Case.jahr, Case.kw)
.all()
)
# Build lookup: {kw: {year: count}}
kw_data: dict[int, dict[int, int]] = {}
all_years: set[int] = set()
for row in rows:
yr = int(row.jahr)
kw = int(row.kw)
all_years.add(yr)
if kw not in kw_data:
kw_data[kw] = {}
kw_data[kw][yr] = int(row.cnt)
sorted_years = sorted(all_years)
result = []
for kw in range(1, 53):
entry: dict[str, Any] = {"kw": kw}
for yr in sorted_years:
entry[str(yr)] = kw_data.get(kw, {}).get(yr, 0)
result.append(entry)
return result
# ---------------------------------------------------------------------------
# Dashboard: Top ICD codes
# ---------------------------------------------------------------------------
def calculate_top_icd(db: Session, jahr: int, limit: int = 10) -> list[dict]:
"""Return the most common ICD codes across all cases for the given year.
Uses the ``icd`` text field on cases (not the junction table) so it
covers all Fallgruppen, not just onko.
Returns::
[{"icd": "C50.9", "count": 42}, {"icd": "I25.1", "count": 31}, ...]
"""
v_filter = Case.versicherung == settings.VERSICHERUNG_FILTER
rows = (
db.query(
func.upper(func.trim(Case.icd)).label("icd_code"),
func.count(Case.id).label("cnt"),
)
.filter(
v_filter,
Case.jahr == jahr,
Case.icd != None, # noqa: E711
Case.icd != "",
)
.group_by(func.upper(func.trim(Case.icd)))
.order_by(func.count(Case.id).desc())
.limit(limit)
.all()
)
return [{"icd": row.icd_code, "count": int(row.cnt)} for row in rows]
# ---------------------------------------------------------------------------
# Dashboard: Gutachten-Statistik (combined sheet3 + sheet4 + KPIs)
# ---------------------------------------------------------------------------
def calculate_gutachten_statistik(db: Session, jahr: int) -> dict:
"""Return combined gutachten statistics for the statistics page.
Combines:
- KPI aggregates (total, bestätigung, alternative, uncodiert, therapy changes)
- Sheet 3 weekly data (gutachten types per KW)
- Sheet 4 weekly data (therapy changes per KW)
Returns::
{
"kpis": {
"total_gutachten": int,
"bestaetigung": int,
"alternative": int,
"uncodiert": int,
"ta_ja": int,
"ta_nein": int,
"diagnosekorrektur": int,
"unterversorgung": int,
"uebertherapie": int,
},
"gutachten_weekly": [{"kw": 1, "bestaetigung": X, "alternative": X}, ...],
"therapie_weekly": [{"kw": 1, "ta_ja": X, "ta_nein": X, ...}, ...],
}
"""
v_filter = Case.versicherung == settings.VERSICHERUNG_FILTER
# KPI aggregates
total_gutachten = (
db.query(func.count(Case.id))
.filter(v_filter, Case.jahr == jahr, Case.gutachten == True) # noqa: E712
.scalar() or 0
)
typ_rows = (
db.query(Case.gutachten_typ, func.count(Case.id).label("cnt"))
.filter(v_filter, Case.jahr == jahr, Case.gutachten == True) # noqa: E712
.group_by(Case.gutachten_typ)
.all()
)
bestaetigung = 0
alternative = 0
uncodiert = 0
for row in typ_rows:
cnt = _int(row.cnt)
if row.gutachten_typ == "Bestätigung":
bestaetigung = cnt
elif row.gutachten_typ == "Alternative":
alternative = cnt
else:
uncodiert = cnt
# Therapy change aggregates
ta_row = (
db.query(
func.sum((Case.therapieaenderung == "Ja").cast(Integer)).label("ta_ja"),
func.sum((Case.therapieaenderung == "Nein").cast(Integer)).label("ta_nein"),
func.sum(Case.ta_diagnosekorrektur.cast(Integer)).label("dk"),
func.sum(Case.ta_unterversorgung.cast(Integer)).label("uv"),
func.sum(Case.ta_uebertherapie.cast(Integer)).label("ut"),
)
.filter(v_filter, Case.jahr == jahr, Case.gutachten == True) # noqa: E712
.first()
)
# Sheet 3 weekly (simplified: gesamt only)
sheet3 = calculate_sheet3_data(db, jahr)
gutachten_weekly = [
{
"kw": w["kw"],
"bestaetigung": w["gesamt"]["bestaetigung"],
"alternative": w["gesamt"]["alternative"],
}
for w in sheet3["weekly"]
]
# Sheet 4 weekly
sheet4 = calculate_sheet4_data(db, jahr)
therapie_weekly = sheet4["weekly"]
return {
"kpis": {
"total_gutachten": total_gutachten,
"bestaetigung": bestaetigung,
"alternative": alternative,
"uncodiert": uncodiert,
"ta_ja": _int(ta_row.ta_ja) if ta_row else 0,
"ta_nein": _int(ta_row.ta_nein) if ta_row else 0,
"diagnosekorrektur": _int(ta_row.dk) if ta_row else 0,
"unterversorgung": _int(ta_row.uv) if ta_row else 0,
"uebertherapie": _int(ta_row.ut) if ta_row else 0,
},
"gutachten_weekly": gutachten_weekly,
"therapie_weekly": therapie_weekly,
}
# ---------------------------------------------------------------------------
# Full report generation (all 5 sheets)
# ---------------------------------------------------------------------------
def generate_full_report(db: Session, jahr: int, kw: int | None = None, report_type: str = "gesamt") -> dict:
"""Generate complete report data for all 5 sheets.
If *kw* is given, only data up to and including that calendar week is
included in the report. This allows generating historical reports
that reflect the state at a specific point in the year.
If *report_type* is given, only the Fallgruppen belonging to that type
are included (e.g. ``"onko_intensiv"`` → only onko + intensiv).
Returns::
{
"jahr": int,
"kw": int | None,
"report_type": str,
"fallgruppen": list[str],
"sheet1": {...},
"sheet2": {...},
"sheet3": {...},
"sheet4": {...},
"sheet5": {...},
}
"""
fallgruppen = REPORT_TYPES.get(report_type, FALLGRUPPEN)
fg_arg = fallgruppen if report_type != "gesamt" else None
logger.info("Generating full report for jahr=%d, kw=%s, report_type=%s", jahr, kw, report_type)
return {
"jahr": jahr,
"kw": kw,
"report_type": report_type,
"fallgruppen": list(fallgruppen),
"sheet1": calculate_sheet1_data(db, jahr, max_kw=kw, fallgruppen=fg_arg),
"sheet2": calculate_sheet2_data(db, jahr, max_kw=kw, fallgruppen=fg_arg),
"sheet3": calculate_sheet3_data(db, jahr, max_kw=kw, fallgruppen=fg_arg),
"sheet4": calculate_sheet4_data(db, jahr, max_kw=kw, fallgruppen=fg_arg),
"sheet5": calculate_sheet5_data(db, jahr, max_kw=kw, fallgruppen=fg_arg),
}
Reference in a new issue View git blame Copy permalink