Files
kbcrm/App_Server/root/crm/routes.py
T
gg c96ed6e97d feat(workflow): Tages-Workflow-UI für Telefon-Akquise (Liste + Maske)
Server: neuer Endpunkt /xhr/workflow/list aggregiert pro Kontaktadresse
die letzte Gesprächsnotiz, das effektiv gültige Wiedervorlage-Datum
(jüngste Notiz mit Wert), den Erledigt-Status und eine priorisierte
Sortierung (überfällig → ohne Notizen → Rest). Filter: kunden_id, status
(offen/erledigt), wiedervorlage (heute/3tage/ueberfaellig/woche).
Sichtbarkeit wie kontaktadressen über projekt_users.

CallLog erfasst den Ersteller serverseitig (user_id, Migration _db_V01_14,
kein FK wegen separater Auth-DB); to_dict ergänzt mitarbeiter_name.
calllog_list unterstützt kontaktadresse_id als Filter.

Client: TagesWorkflowListe (Route /tagesworkflow, Filterleiste) und
TagesWorkflowMaske (Route /tagesworkflowkontakt, lesende Historie,
Notiz-Formular mit "Speichern und weiter"), Routes + Menüpunkt erster
Eintrag. berechneWiedervorlageDatum aus der Call_Log-Maske nach
models/tag.ts ausgelagert (gemeinsame Nutzung).

Generated by Mistral Vibe.
Co-Authored-By: Mistral Vibe <vibe@mistral.ai>
2026-09-24 18:56:23 +02:00

475 lines
18 KiB
Python

"""
REST-Endpunkte für die CRM-Domain.
Die Endpunkte orientieren sich 1:1 an den clientseitig definierten Pfaden
(siehe Client/models/*.ts). Pro Entität gibt es einen zusammenhängenden Block
mit den Aktionen list/get/create/update/delete.
Die Autorisierung erfolgt über den Decorator authorisation_required; AppAdmin
und SysAdmin sind darin implizit immer erlaubt. Die Rollen Mitarbeiter und
KundenManager werden aus crm.model importiert.
:author: Mistral Vibe <vibe@mistral.ai>
:date: 2026-09-02
:version: 1.0
"""
from flask import current_app as app
from flask import jsonify, request
from flask_login import current_user
from auth.decorators import authorisation_required
from auth.model import APP_ADMIN
from auth.model.errors import ApplicationError, HtmlStatusCode
from crm.model import KUNDEN_MANAGER, MITARBEITER
from crm.model import kunde as kunden
from crm.model import kontakt as kontakte
from crm.model import projekt as projekte
from crm.model import projekt_user as projekt_users
from crm.model import kontaktadresse as kontaktadressen
from crm.model import call_log as call_logs
from crm.model import tages_workflow
from crm.model import tag as tags
print("crm-routes")
def _get_visible_user_id():
"""
Liefert die user_id für die Sichtbarkeits-Filterung.
SysAdmin und AppAdmin sehen alle Projekte und abhängige Entitäten —
für sie wird None zurückgegeben, sodass die Modell-Funktionen keinen
Filter anwenden. Alle anderen Nutzer (Mitarbeiter, KundenManager)
erhalten ihre current_user.id, sodass nur die ihnen zugeordneten
Projekte bzw. abhängige Einträge geliefert werden.
"""
if current_user.is_SysAdmin() or current_user.has_permission(
app.config["APPID"], [APP_ADMIN]
):
return None
return current_user.id
def _require_projekt_access(projekt_id: int):
"""
Prüft, ob der aktuelle Nutzer für das angegebene Projekt schreibberechtigt ist.
SysAdmin und AppAdmin haben immer Zugriff. Alle anderen Nutzer müssen
dem Projekt über projekt_users zugeordnet sein. Ist der Nutzer nicht
zugeordnet, wird ein ApplicationError (404) geworfen — 404 statt 403,
damit nicht zugeordnete Nutzer keine Rückschlüsse auf die Existenz
des Projekts ziehen können.
Muss in allen Schreib-Endpunkten (create/update/delete) aufgerufen
werden, die ein konkretes Projekt betreffen, bevor die Änderung
durchgeführt wird.
- projekt_id: int
- Die ID des Projekts, auf das zugegriffen werden soll.
"""
user_id = _get_visible_user_id()
if user_id is not None:
if not projekt_users.is_user_assigned(projekt_id, user_id):
raise ApplicationError(
"Projekt mit der ID {} wurde nicht gefunden.".format(projekt_id),
htmlstatus=HtmlStatusCode.Not_Found,
)
# --------------------------------------------------------------------------- #
# Kunde #
# --------------------------------------------------------------------------- #
@app.get("/xhr/kunde/list")
@authorisation_required([MITARBEITER, KUNDEN_MANAGER])
def kunde_list():
"""Liefert alle Kunden als Liste."""
result = [k.to_dict() for k in kunden.get_kunden()]
return jsonify(result), 200
@app.get("/xhr/kunde/get")
@authorisation_required([MITARBEITER, KUNDEN_MANAGER])
def kunde_get():
"""Liefert einen einzelnen Kunden anhand seiner ID (Query-Param id)."""
kunde = kunden.get_kunde(int(request.args.get("id")))
return jsonify(kunde.to_dict()), 200
@app.post("/xhr/kunde/create")
@authorisation_required([KUNDEN_MANAGER])
def kunde_create():
"""Legt einen neuen Kunden an. ID wird serverseitig vergeben."""
kunde = kunden.create_kunde(request.json)
return jsonify(kunde.to_dict()), 200
@app.put("/xhr/kunde/update")
@authorisation_required([KUNDEN_MANAGER])
def kunde_update():
"""Aktualisiert einen bestehenden Kunden."""
kunde = kunden.update_kunde(request.json)
return jsonify(kunde.to_dict()), 200
@app.delete("/xhr/kunde/delete")
@authorisation_required([KUNDEN_MANAGER])
def kunde_delete():
"""Löscht einen Kunden anhand seiner ID (Query-Param id)."""
kunden.delete_kunde(int(request.args.get("id")))
return jsonify({}), 200
# --------------------------------------------------------------------------- #
# Kontakt #
# --------------------------------------------------------------------------- #
@app.get("/xhr/kontakt/list")
@authorisation_required([MITARBEITER, KUNDEN_MANAGER])
def kontakt_list():
"""Liefert Kontakte als Liste. Optional nach kunden_id gefiltert."""
kunden_id = request.args.get("kunden_id", type=int)
result = [k.to_dict() for k in kontakte.get_kontakte(kunden_id)]
return jsonify(result), 200
@app.get("/xhr/kontakt/get")
@authorisation_required([MITARBEITER, KUNDEN_MANAGER])
def kontakt_get():
"""Liefert einen einzelnen Kontakt anhand seiner ID (Query-Param id)."""
kontakt = kontakte.get_kontakt(int(request.args.get("id")))
return jsonify(kontakt.to_dict()), 200
@app.post("/xhr/kontakt/create")
@authorisation_required([KUNDEN_MANAGER])
def kontakt_create():
"""Legt einen neuen Kontakt an. ID wird serverseitig vergeben."""
kontakt = kontakte.create_kontakt(request.json)
return jsonify(kontakt.to_dict()), 200
@app.put("/xhr/kontakt/update")
@authorisation_required([KUNDEN_MANAGER])
def kontakt_update():
"""Aktualisiert einen bestehenden Kontakt."""
kontakt = kontakte.update_kontakt(request.json)
return jsonify(kontakt.to_dict()), 200
@app.delete("/xhr/kontakt/delete")
@authorisation_required([KUNDEN_MANAGER])
def kontakt_delete():
"""Löscht einen Kontakt anhand seiner ID (Query-Param id)."""
kontakte.delete_kontakt(int(request.args.get("id")))
return jsonify({}), 200
# --------------------------------------------------------------------------- #
# Projekt #
# --------------------------------------------------------------------------- #
@app.get("/xhr/projekt/list")
@authorisation_required([MITARBEITER, KUNDEN_MANAGER])
def projekt_list():
"""Liefert Projekte als Liste. Optional nach kunden_id gefiltert.
Für Nicht-SysAdmin/Nicht-AppAdmin werden nur die Projekte geliefert,
denen der Nutzer über projekt_users zugeordnet ist."""
kunden_id = request.args.get("kunden_id", type=int)
user_id = _get_visible_user_id()
result = [p.to_dict() for p in projekte.get_projekte(kunden_id, user_id)]
return jsonify(result), 200
@app.get("/xhr/projekt/get")
@authorisation_required([MITARBEITER, KUNDEN_MANAGER])
def projekt_get():
"""Liefert ein einzelnes Projekt anhand seiner ID (Query-Param id).
Für Nicht-SysAdmin/Nicht-AppAdmin wird geprüft, ob der Nutzer dem
Projekt zugeordnet ist; sonst wird 404 geliefert."""
user_id = _get_visible_user_id()
projekt = projekte.get_projekt(int(request.args.get("id")), user_id)
return jsonify(projekt.to_dict()), 200
@app.post("/xhr/projekt/create")
@authorisation_required([KUNDEN_MANAGER])
def projekt_create():
"""Legt ein neues Projekt an. ID wird serverseitig vergeben."""
projekt = projekte.create_projekt(request.json)
return jsonify(projekt.to_dict()), 200
@app.put("/xhr/projekt/update")
@authorisation_required([KUNDEN_MANAGER])
def projekt_update():
"""Aktualisiert ein bestehendes Projekt."""
_require_projekt_access(request.json["id"])
projekt = projekte.update_projekt(request.json)
return jsonify(projekt.to_dict()), 200
@app.delete("/xhr/projekt/delete")
@authorisation_required([KUNDEN_MANAGER])
def projekt_delete():
"""Löscht ein Projekt anhand seiner ID (Query-Param id)."""
_require_projekt_access(int(request.args.get("id")))
projekte.delete_projekt(int(request.args.get("id")))
return jsonify({}), 200
# --------------------------------------------------------------------------- #
# ProjektUser (Nutzer-Zuordnung / Sichtbarkeit) #
# --------------------------------------------------------------------------- #
@app.get("/xhr/projektuser/list")
@authorisation_required([MITARBEITER, KUNDEN_MANAGER])
def projektuser_list():
"""Liefert alle einem Projekt zugeordneten Nutzer (Query-Param projekt_id)."""
projekt_id = request.args.get("projekt_id", type=int)
result = projekt_users.get_projekt_users(projekt_id)
return jsonify(result), 200
@app.get("/xhr/projektuser/listall")
# Leeres allowed_roles-Array als Erweiterungspunkt (nur AppAdmin/SysAdmin).
@authorisation_required([])
def projektuser_listall():
"""Liefert alle Projekt-User-Zuordnungen inkl. projektname und Nutzerdaten."""
result = projekt_users.get_all_projekt_users()
return jsonify(result), 200
@app.post("/xhr/projektuser/add")
# Leeres allowed_roles-Array als Erweiterungspunkt: Aktuell passieren nur
# AppAdmin und SysAdmin (vom Decorator implizit ergänzt). Sobald später eine
# weitere Rolle diese Endpunkte nutzen soll, wird sie einfach hier eingetragen,
# z.B. @authorisation_required([KUNDEN_MANAGER]).
@authorisation_required([])
def projektuser_add():
"""Ordnet einen Nutzer einem Projekt zu (body: {projekt_id, user_id})."""
data = request.json
projekt_users.add_projekt_user(data["projekt_id"], data["user_id"])
return jsonify({}), 200
@app.delete("/xhr/projektuser/remove")
# Leeres allowed_roles-Array als Erweiterungspunkt (siehe projektuser_add).
@authorisation_required([])
def projektuser_remove():
"""Entfernt die Zuordnung eines Nutzers zu einem Projekt (params: projekt_id, user_id)."""
projekt_id = request.args.get("projekt_id", type=int)
user_id = request.args.get("user_id", type=int)
projekt_users.remove_projekt_user(projekt_id, user_id)
return jsonify({}), 200
# --------------------------------------------------------------------------- #
# Kontaktadresse #
# --------------------------------------------------------------------------- #
@app.get("/xhr/kontaktadresse/list")
@authorisation_required([MITARBEITER, KUNDEN_MANAGER])
def kontaktadresse_list():
"""Liefert Kontaktadressen als Liste. Optional nach projekt_id gefiltert.
Für Nicht-SysAdmin/Nicht-AppAdmin nur Kontaktadressen sichtbarer Projekte."""
projekt_id = request.args.get("projekt_id", type=int)
user_id = _get_visible_user_id()
result = [k.to_dict() for k in kontaktadressen.get_kontaktadressen(projekt_id, user_id)]
return jsonify(result), 200
@app.get("/xhr/kontaktadresse/get")
@authorisation_required([MITARBEITER, KUNDEN_MANAGER])
def kontaktadresse_get():
"""Liefert eine einzelne Kontaktadresse anhand ihrer ID (Query-Param id).
Für Nicht-SysAdmin/Nicht-AppAdmin wird geprüft, ob der Nutzer dem
Projekt der Kontaktadresse zugeordnet ist; sonst 404."""
user_id = _get_visible_user_id()
kontaktadresse = kontaktadressen.get_kontaktadresse(int(request.args.get("id")), user_id)
return jsonify(kontaktadresse.to_dict()), 200
@app.post("/xhr/kontaktadresse/create")
@authorisation_required([KUNDEN_MANAGER])
def kontaktadresse_create():
"""Legt eine neue Kontaktadresse an. ID wird serverseitig vergeben."""
_require_projekt_access(int(request.json["projekt_id"]))
kontaktadresse = kontaktadressen.create_kontaktadresse(request.json)
return jsonify(kontaktadresse.to_dict()), 200
@app.put("/xhr/kontaktadresse/update")
@authorisation_required([KUNDEN_MANAGER])
def kontaktadresse_update():
"""Aktualisiert eine bestehende Kontaktadresse."""
_require_projekt_access(int(request.json["projekt_id"]))
kontaktadresse = kontaktadressen.update_kontaktadresse(request.json)
return jsonify(kontaktadresse.to_dict()), 200
@app.delete("/xhr/kontaktadresse/delete")
@authorisation_required([KUNDEN_MANAGER])
def kontaktadresse_delete():
"""Löscht eine Kontaktadresse anhand ihrer ID (Query-Param id)."""
# projekt_id der Kontaktadresse ermitteln, um die Projekt-Zuordnung
# des Nutzers zu prüfen, bevor gelöscht wird.
user_id = _get_visible_user_id()
kontaktadresse = kontaktadressen.get_kontaktadresse(int(request.args.get("id")), user_id)
kontaktadressen.delete_kontaktadresse(int(request.args.get("id")))
return jsonify({}), 200
# --------------------------------------------------------------------------- #
# CallLog (Gesprächsnotiz) #
# --------------------------------------------------------------------------- #
@app.get("/xhr/calllog/list")
@authorisation_required([MITARBEITER, KUNDEN_MANAGER])
def calllog_list():
"""Liefert Gesprächsnotizen als Liste. Optional nach projekt_id und/oder
kontaktadresse_id gefiltert. Für Nicht-SysAdmin/Nicht-AppAdmin nur
Notizen sichtbarer Projekte."""
projekt_id = request.args.get("projekt_id", type=int)
kontaktadresse_id = request.args.get("kontaktadresse_id", type=int)
user_id = _get_visible_user_id()
result = [l.to_dict() for l in call_logs.get_call_logs(
projekt_id, user_id, kontaktadresse_id
)]
return jsonify(result), 200
@app.get("/xhr/calllog/get")
@authorisation_required([MITARBEITER, KUNDEN_MANAGER])
def calllog_get():
"""Liefert eine einzelne Gesprächsnotiz anhand ihrer ID (Query-Param id).
Für Nicht-SysAdmin/Nicht-AppAdmin wird geprüft, ob der Nutzer dem
Projekt der Notiz zugeordnet ist; sonst 404."""
user_id = _get_visible_user_id()
call_log = call_logs.get_call_log(int(request.args.get("id")), user_id)
return jsonify(call_log.to_dict()), 200
@app.post("/xhr/calllog/create")
@authorisation_required([KUNDEN_MANAGER])
def calllog_create():
"""Legt eine neue Gesprächsnotiz an. ID wird serverseitig vergeben.
current_user.id wird als Ersteller (user_id) erfasst."""
_require_projekt_access(int(request.json["projekt_id"]))
call_log = call_logs.create_call_log(request.json, current_user.id)
return jsonify(call_log.to_dict()), 200
@app.put("/xhr/calllog/update")
@authorisation_required([KUNDEN_MANAGER])
def calllog_update():
"""Aktualisiert eine bestehende Gesprächsnotiz."""
_require_projekt_access(int(request.json["projekt_id"]))
call_log = call_logs.update_call_log(request.json)
return jsonify(call_log.to_dict()), 200
@app.delete("/xhr/calllog/delete")
@authorisation_required([KUNDEN_MANAGER])
def calllog_delete():
"""Löscht eine Gesprächsnotiz anhand ihrer ID (Query-Param id)."""
# projekt_id der Notiz ermitteln, um die Projekt-Zuordnung
# des Nutzers zu prüfen, bevor gelöscht wird.
user_id = _get_visible_user_id()
call_log = call_logs.get_call_log(int(request.args.get("id")), user_id)
call_logs.delete_call_log(int(request.args.get("id")))
return jsonify({}), 200
# --------------------------------------------------------------------------- #
# Tages-Workflow #
# --------------------------------------------------------------------------- #
@app.get("/xhr/workflow/list")
@authorisation_required([MITARBEITER, KUNDEN_MANAGER])
def workflow_list():
"""Tages-Workflow-Liste: Kontaktadressen mit letzter Notiz, Wiedervorlage
und dynamisch berechnetem Status. Optional gefiltert nach kunden_id,
status ('offen'/'erledigt') und wiedervorlage ('heute'/'3tage'/
'ueberfaellig'/'woche'). Für Nicht-SysAdmin/Nicht-AppAdmin nur
Kontaktadressen sichtbarer Projekte. Die Sortierung ist priorisiert:
überfällige Wiedervorlagen, dann Kontakte ohne Notizen, dann der Rest."""
user_id = _get_visible_user_id()
result = tages_workflow.get_tages_workflow_list(
user_id,
request.args.get("kunden_id", type=int),
request.args.get("status"),
request.args.get("wiedervorlage"),
)
return jsonify(result), 200
# --------------------------------------------------------------------------- #
# Tag (hierarchische Taxonomie) #
# --------------------------------------------------------------------------- #
# Tags sind nicht projektbezogen — die projektbezogene Sichtbarkeits-
# Filterung entfaellt. Lesen: alle Mitarbeiter. Schreiben: nur AppAdmin/SysAdmin.
# --------------------------------------------------------------------------- #
@app.get("/xhr/tag/list")
@authorisation_required([MITARBEITER, KUNDEN_MANAGER])
def tag_list():
"""Liefert alle Tags als flache Liste. Optional nach parent_id gefiltert."""
parent_id = request.args.get("parent_id", type=int)
result = [t.to_dict() for t in tags.get_tags(parent_id)]
return jsonify(result), 200
@app.get("/xhr/tag/hierarchy")
@authorisation_required([MITARBEITER, KUNDEN_MANAGER])
def tag_hierarchy():
"""Liefert die vollstaendige Tag-Hierarchie als verschachtelte Struktur."""
result = tags.get_tag_hierarchy()
return jsonify(result), 200
@app.get("/xhr/tag/get")
@authorisation_required([MITARBEITER, KUNDEN_MANAGER])
def tag_get():
"""Liefert einen einzelnen Tag anhand seiner ID (Query-Param id)."""
tag = tags.get_tag(int(request.args.get("id")))
return jsonify(tag.to_dict()), 200
@app.post("/xhr/tag/create")
# Leeres allowed_roles-Array: Nur AppAdmin/SysAdmin (vom Decorator
# implizit ergaenzt). Tags sind eine globale Taxonomie — die Anlage
# und Aenderung ist ausschliesslich den Administratoren vorbehalten.
@authorisation_required([])
def tag_create():
"""Legt einen neuen Tag an. ID wird serverseitig vergeben."""
tag = tags.create_tag(request.json)
return jsonify(tag.to_dict()), 200
@app.put("/xhr/tag/update")
@authorisation_required([])
def tag_update():
"""Aktualisiert einen bestehenden Tag."""
tag = tags.update_tag(request.json)
return jsonify(tag.to_dict()), 200
@app.delete("/xhr/tag/delete")
@authorisation_required([])
def tag_delete():
"""Loescht einen Tag inkl. aller Nachfahren (kaskadierend) anhand seiner ID."""
tags.delete_tag(int(request.args.get("id")))
return jsonify({}), 200
print("########## crm.routes importiert")