suro/aza
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
suro
2026-05-06 22:43:22 +02:00
parent 2363564013
commit 01f05b23e9
5221 changed files with 872060 additions and 734 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
AzA march 2026/empfang_routes.py
View File

@@ -13,6 +13,7 @@ import os
import re
import secrets
import time
import unicodedata
import uuid
from pathlib import Path
from typing import Optional, Tuple
@@ -119,6 +120,14 @@ def _save_practices(data: dict):
_save_json(_PRACTICES_FILE, data)
def _invite_code_key(raw: str) -> str:
"""Vergleicht Einladungscodes unabhaengig von Leerzeichen und Gedankenstrich-Varianten."""
s = (raw or "").strip().upper().replace(" ", "")
for ch in ("\u2011", "\u2013", "\u2014", "\u2212"):
s = s.replace(ch, "-")
return s
def _generate_chat_invite_code() -> str:
"""Lesbarer Chat-Einladungscode im Format CHAT-XXXX-XXXX."""
import random
@@ -969,9 +978,10 @@ async def auth_provision(request: Request):
# Geraet ohne gespeicherte practice_id eine eigene Praxis an (Realbefund).
if invite_code_in:
practices = _load_practices()
want = _invite_code_key(invite_code_in)
target_pid = None
for pida, pdata in practices.items():
if (pdata.get("invite_code") or "").strip() == invite_code_in:
if _invite_code_key(pdata.get("invite_code")) == want:
target_pid = pida
break
if not target_pid:
@@ -1685,6 +1695,48 @@ async def empfang_register_user(request: Request):
if a["display_name"] == name and a.get("practice_id") == pid:
a["display_name"] = new_name
_save_accounts(accounts)
elif action == "add_secure":
s = _session_from_request(request)
if not s:
raise HTTPException(status_code=401, detail="Nicht angemeldet")
sess_role = str(s.get("role") or "").strip().lower()
if sess_role not in ("admin", "empfang"):
raise HTTPException(status_code=403, detail="Keine Berechtigung Benutzer anzulegen")
pid = str(s.get("practice_id") or "").strip()
if not pid:
return JSONResponse(content={"success": False, "detail": "Keine Praxis"}, status_code=400)
pw = (body.get("password") or "").strip()
pw2 = (body.get("password_repeat") or body.get("password2") or "").strip()
if pw != pw2:
return JSONResponse(content={"success": False, "detail": "Passwoerter stimmen nicht ueberein"}, status_code=400)
if len(pw) < 4:
return JSONResponse(content={"success": False, "detail": "Passwort mindestens 4 Zeichen"}, status_code=400)
allowed_roles = {"mpa", "arzt"}
if sess_role == "admin":
allowed_roles.update({"admin", "empfang"})
role_new = str(body.get("role") or "mpa").strip().lower()
if role_new not in allowed_roles:
role_new = "mpa"
exists = any(a["display_name"] == name and a.get("practice_id") == pid
for a in accounts.values())
if not exists:
uid = uuid.uuid4().hex[:12]
pw_hash, pw_salt = _hash_password(pw)
accounts[uid] = {
"user_id": uid,
"practice_id": pid,
"display_name": name,
"role": role_new,
"pw_hash": pw_hash,
"pw_salt": pw_salt,
"created": time.strftime("%Y-%m-%d %H:%M:%S"),
"status": "active",
"last_login": "",
"email": "",
}
_save_accounts(accounts)
else:
return JSONResponse(content={"success": False, "detail": "Name bereits vergeben"}, status_code=409)
else:
exists = any(a["display_name"] == name and a.get("practice_id") == pid
for a in accounts.values())
@@ -1834,7 +1886,10 @@ def _pulse_get(practice_id: str) -> dict:
def _norm_name(s: str) -> str:
return (s or "").strip().lower()
"""Vergleichts-String fuer Namen: lower, trim, Akzente ueber NFKD entfernen."""
t = (s or "").strip().lower()
t = unicodedata.normalize("NFKD", t)
return "".join(ch for ch in t if unicodedata.combining(ch) == "")
def _sender_core(absender: str) -> str:
@@ -2055,6 +2110,7 @@ async def empfang_tasks_list(request: Request):
@router.post("/tasks")
async def empfang_tasks_create(request: Request):
pid = _require_practice_id(request)
s = _session_from_request(request)
try:
body = await request.json()
except Exception:
@@ -2062,14 +2118,23 @@ async def empfang_tasks_create(request: Request):
text = (body.get("text") or "").strip()
if not text:
raise HTTPException(status_code=400, detail="Text erforderlich")
title_opt = (body.get("title") or "").strip()
meta_opt = (body.get("source_meta") or "").strip()
peer_opt = (body.get("source_peer") or "").strip()
stid_opt = (body.get("source_thread_id") or "").strip()
task = {
"task_id": uuid.uuid4().hex[:12],
"practice_id": pid,
"text": text,
"title": title_opt or "",
"source_meta": meta_opt or "",
"source_peer": peer_opt or "",
"source_thread_id": stid_opt or "",
"done": False,
"assignee": (body.get("assignee") or "").strip(),
"created_by": s["display_name"] if s else "",
"created_by": (s or {}).get("display_name", "") if s else "",
"created": time.strftime("%Y-%m-%d %H:%M:%S"),
"source_msg_id": (body.get("source_msg_id") or "").strip(),
}
tasks = _load_tasks()
tasks.insert(0, task)
@@ -2091,8 +2156,12 @@ async def empfang_tasks_update(task_id: str, request: Request):
target["done"] = bool(body["done"])
if "text" in body:
target["text"] = (body["text"] or "").strip() or target["text"]
if "title" in body:
target["title"] = (body.get("title") or "").strip()
if "assignee" in body:
target["assignee"] = (body["assignee"] or "").strip()
target["assignee"] = (body.get("assignee") or "").strip()
if "source_meta" in body:
target["source_meta"] = (body.get("source_meta") or "").strip()
_save_tasks(tasks)
return JSONResponse(content={"success": True, "task": target})