mirror of
https://github.com/complexcaresolutions/dak.c2s.git
synced 2026-03-17 19:33:41 +00:00
CCS Admin
32127c30c3
Add VERSICHERUNG_FILTER="DAK" to config and apply it to all case queries: list, detail, pending-icd, pending-coding, coding queue, dashboard KPIs, all 5 report sheets, and excel sync export. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
113 lines
3.3 KiB
Python
113 lines
3.3 KiB
Python
"""Coding queue and batch coding operations for Gutachten classification."""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import get_settings
|
|
from app.core.exceptions import CaseNotFoundError
|
|
from app.models.case import Case
|
|
|
|
settings = get_settings()
|
|
|
|
|
|
def get_coding_queue(
|
|
db: Session,
|
|
fallgruppe: str | None = None,
|
|
page: int = 1,
|
|
per_page: int = 50,
|
|
) -> tuple[list[Case], int]:
|
|
"""Get cases with gutachten=True but no gutachten_typ yet.
|
|
|
|
These cases have received a Gutachten but still need classification
|
|
(Bestaetigung/Alternative) and therapy-change coding.
|
|
|
|
Returns:
|
|
Tuple of (cases, total_count) for pagination.
|
|
"""
|
|
query = db.query(Case).filter(
|
|
Case.versicherung == settings.VERSICHERUNG_FILTER,
|
|
Case.gutachten == True, # noqa: E712
|
|
Case.gutachten_typ == None, # noqa: E711
|
|
)
|
|
if fallgruppe:
|
|
query = query.filter(Case.fallgruppe == fallgruppe)
|
|
|
|
total = query.count()
|
|
cases = (
|
|
query.order_by(Case.datum.desc())
|
|
.offset((page - 1) * per_page)
|
|
.limit(per_page)
|
|
.all()
|
|
)
|
|
return cases, total
|
|
|
|
|
|
def update_coding(
|
|
db: Session,
|
|
case_id: int,
|
|
gutachten_typ: str,
|
|
therapieaenderung: str,
|
|
ta_diagnosekorrektur: bool = False,
|
|
ta_unterversorgung: bool = False,
|
|
ta_uebertherapie: bool = False,
|
|
user_id: int | None = None,
|
|
) -> Case:
|
|
"""Set coding for a single case.
|
|
|
|
Validates gutachten_typ and therapieaenderung values, then persists the
|
|
coding classification along with the user and timestamp.
|
|
|
|
Raises:
|
|
CaseNotFoundError: If no case with the given ID exists.
|
|
ValueError: If gutachten_typ or therapieaenderung have invalid values.
|
|
"""
|
|
case = db.query(Case).filter(Case.id == case_id).first()
|
|
if not case:
|
|
raise CaseNotFoundError()
|
|
|
|
if gutachten_typ not in ("Bestätigung", "Alternative"):
|
|
raise ValueError(f"Invalid gutachten_typ: {gutachten_typ}")
|
|
if therapieaenderung not in ("Ja", "Nein"):
|
|
raise ValueError(f"Invalid therapieaenderung: {therapieaenderung}")
|
|
|
|
case.gutachten_typ = gutachten_typ
|
|
case.therapieaenderung = therapieaenderung
|
|
case.ta_diagnosekorrektur = ta_diagnosekorrektur
|
|
case.ta_unterversorgung = ta_unterversorgung
|
|
case.ta_uebertherapie = ta_uebertherapie
|
|
case.coding_completed_by = user_id
|
|
case.coding_completed_at = datetime.now(timezone.utc)
|
|
|
|
db.commit()
|
|
db.refresh(case)
|
|
return case
|
|
|
|
|
|
def batch_update_coding(
|
|
db: Session,
|
|
updates: list[dict],
|
|
user_id: int | None = None,
|
|
) -> dict:
|
|
"""Batch update coding for multiple cases.
|
|
|
|
Each dict in *updates* must contain at minimum:
|
|
- ``case_id`` (int)
|
|
- ``gutachten_typ`` (str)
|
|
- ``therapieaenderung`` (str)
|
|
and optionally the ``ta_*`` boolean flags.
|
|
|
|
Returns:
|
|
Dict with ``updated`` count and ``errors`` list of error messages.
|
|
"""
|
|
updated = 0
|
|
errors: list[str] = []
|
|
for item in updates:
|
|
case_id = item.get("case_id")
|
|
try:
|
|
params = {k: v for k, v in item.items() if k != "case_id"}
|
|
update_coding(db, case_id, user_id=user_id, **params)
|
|
updated += 1
|
|
except Exception as exc:
|
|
errors.append(f"Case {case_id}: {exc}")
|
|
return {"updated": updated, "errors": errors}
|