from __future__ import annotations import copy import datetime from pprint import pformat from typing import TYPE_CHECKING, Any, cast import polars as pl import sqlalchemy as sa from dopt_basics.result_pattern import wrap_result from sqlalchemy.dialects.sqlite import insert as sqlite_insert from wce_crm import db from wce_crm.constants import TIMEZONE_CEST from wce_crm.data_models import ( FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN, FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN, Beratungsgespraech_Einzelgespraech, Beratungsgespraech_Vorgang, InitRec, Page_InitRec_Form_Data, ) from wce_crm.logging import logger_back as logger from wce_crm.types import ( CompanyInfo, CompanyProfileConsultationEntry, CompanyProfileConsultations, ConsultingType, ContactPersonInfo, EntityIds, EntityType, LinkingConsultationsEntryCompany, LinkingConsultationsEntryPerson, MainPageEntry, RecordingType, ) if TYPE_CHECKING: from dopt_basics.datastructures import DualDict from wce_crm.types import ConsId, ExtAnId, ExtMaId, RecId def _transform_for_gui_output( data: pl.DataFrame, ) -> pl.DataFrame: q = ( data.lazy() .with_columns( pl.col(pl.Datetime).dt.to_string("%d.%m.%Y"), pl.col(pl.Date).dt.to_string("%d.%m.%Y"), pl.when(pl.col(pl.Boolean)) .then(pl.lit("Ja")) .otherwise(pl.lit("Nein")) .name.keep(), ) .with_columns(pl.all().cast(pl.String)) ) return q.collect() def initrec_comp_search_choices() -> tuple[tuple[str, int], ...]: # TODO no reload functionality logger.debug("[Call backend] comp_search_choices") q = db.DF_CRM_MASTER.lazy() counter = pl.int_range(0, pl.len()).over(pl.col.ma_unternehmensname) q = q.with_columns( dedupl=pl.when(counter == 0) .then(pl.col.ma_unternehmensname) .otherwise(pl.format("{} ({})", pl.col.ma_unternehmensname, counter)) ) df = q.collect() return tuple(zip(df["dedupl"], df["ma_id"])) # TODO change to direct SQL interaction def external_company_get_info( ma_id: ExtMaId, ) -> CompanyInfo: logger.debug("[Call backend] comp_search_get_info") df = db.DF_CRM_MASTER.filter(pl.col.ma_id == ma_id) if df.height > 1 or df.height == 0: raise ValueError(f"Größe des zurückgelieferten Datenpakets ungültig: {df.height}") df = _transform_for_gui_output(df) return cast(CompanyInfo, df.row(0, named=True)) def initrec_comp_contact_person_search_choices( ma_id: ExtMaId | None, use_both_names: bool, ) -> tuple[tuple[str, int], ...]: # TODO no reload functionality logger.debug("[Call backend] contact_person_search_choices") q = db.DF_CONTACT_PERSON.lazy() if ma_id is not None: q = q.filter(pl.col.ma_id == ma_id) dedupl_col = pl.col.an_nachname if use_both_names: q = q.with_columns( name_search=(pl.format("{}, {}", pl.col.an_nachname, pl.col.an_vorname)) ) dedupl_col = pl.col.name_search counter = pl.int_range(0, pl.len()).over(dedupl_col) q = q.with_columns( dedupl=pl.when(counter == 0) .then(dedupl_col) .otherwise(pl.format("{} ({})", dedupl_col, counter)) ) df = q.collect() return tuple(zip(df["dedupl"], df["an_id"])) def initrec_comp_contact_person_search_get_info( an_id: ExtAnId, ) -> ContactPersonInfo: logger.debug("[Call backend] contact_person_search_get_info") df = db.DF_CONTACT_PERSON.filter(pl.col.an_id == an_id) if df.height > 1 or df.height == 0: raise ValueError(f"Größe des zurückgelieferten Datenpakets ungültig: {df.height}") df = _transform_for_gui_output(df) return cast(ContactPersonInfo, df.row(0, named=True)) # // internals # TODO check removal # def initrec_company_insert_initial_recording( # data: dict[str, Any], # ) -> RecId: # logger.debug("[Call backend] insert_initial_recording") # stmt = db.grunderfassung_unternehmen.insert() # with db.ENGINE.begin() as conn: # ret = conn.execute(stmt, data) # if ret.rowcount == 0: # raise IOError("Entry was not inserted correctly") # prim_keys = ret.inserted_primary_key # assert prim_keys # return prim_keys[0] # def initrec_company_update_initial_recording( # id_: RecId, # data: dict[str, Any], # ) -> None: # logger.debug("[Call backend] update_initial_recording") # stmt = db.grunderfassung_unternehmen.update().where( # db.grunderfassung_unternehmen.c.un_id == id_ # ) # with db.ENGINE.begin() as conn: # conn.execute(stmt, data) # def initrec_company_to_db( # auto_form_data: InitRec, # ) -> InitRec: # logger.debug("[AutoForm -- backend] Call database saving routine...") # dump_data = copy.deepcopy(auto_form_data.db_data) # dump_data["geloescht"] = auto_form_data.geloescht # with db.ENGINE.begin() as conn: # if auto_form_data.ids.un_id is None: # logger.debug("[AutoForm -- backend] Insert...") # stmt = db.grunderfassung_unternehmen.insert().returning( # db.grunderfassung_unternehmen.c.un_id, # db.grunderfassung_unternehmen.c.Metadaten_aktualisierung, # db.grunderfassung_unternehmen.c.geloescht, # ) # ret = conn.execute(stmt, dump_data) # from_db = ret.mappings().fetchall() # assert len(from_db) == 1, "expected excactly one returned row" # data_from_db = from_db[0] # auto_form_data.ids.un_id = data_from_db["un_id"] # auto_form_data.geloescht = data_from_db["geloescht"] # auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[ # "Metadaten_aktualisierung" # ] # logger.debug("[AutoForm -- backend] Inserted InitRec Company successfully") # else: # logger.debug("[AutoForm -- backend] Update...") # stmt = ( # db.grunderfassung_unternehmen.update() # .where(db.grunderfassung_unternehmen.c.un_id == auto_form_data.ids.un_id) # .returning( # db.grunderfassung_unternehmen.c.Metadaten_aktualisierung, # db.grunderfassung_unternehmen.c.geloescht, # ) # ) # ret = conn.execute(stmt, dump_data) # from_db = ret.mappings().fetchall() # assert len(from_db) == 1, "expected excatly one returned row" # data_from_db = from_db[0] # auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[ # "Metadaten_aktualisierung" # ] # logger.debug( # "[AutoForm -- backend] Updated InitRec Company with ID %d successfully", # auto_form_data.ids.un_id, # ) # return auto_form_data # def initrec_company_from_db( # id_: RecId, # ) -> InitRec: # logger.debug("[AutoForm -- backend] Call database loading routine...") # stmt = db.grunderfassung_unternehmen.select().where( # db.grunderfassung_unternehmen.c.un_id == id_ # ) # with db.ENGINE.connect() as conn: # ret = conn.execute(stmt) # from_db = ret.mappings().all() # assert len(from_db) == 1, "not excactly one company initial recording obtained" # data_from_db = dict(from_db[0]) # geloescht = data_from_db["geloescht"] # del data_from_db["geloescht"] # return InitRec( # ids=EntityIds(un_id=id_), # # rec_id=id_, # TODO remove # geloescht=geloescht, # db_data=data_from_db, # ) # def initrec_person_to_db( # auto_form_data: InitRec, # ) -> InitRec: # logger.debug("[AutoForm -- backend] Call database saving routine...") # dump_data = copy.deepcopy(auto_form_data.db_data) # dump_data["geloescht"] = auto_form_data.geloescht # with db.ENGINE.begin() as conn: # if auto_form_data.ids.pers_id is None: # logger.debug("[AutoForm -- backend] Insert...") # stmt = sa.insert(db.t_grunderfassung_personen).returning( # db.t_grunderfassung_personen.c.pers_id, # db.t_grunderfassung_personen.c.Metadaten_aktualisierung, # db.t_grunderfassung_personen.c.geloescht, # ) # stmt_compiled = str(stmt.compile(db.ENGINE)) # logger.debug( # "[AutoForm -- backend] Data to insert:\nStatement: %s\n%s", # stmt_compiled, # pformat(dump_data), # ) # ret = conn.execute(stmt, dump_data) # from_db = ret.mappings().fetchall() # assert len(from_db) == 1, "expected excactly one returned row" # data_from_db = from_db[0] # auto_form_data.ids.pers_id = data_from_db["pers_id"] # auto_form_data.geloescht = data_from_db["geloescht"] # auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[ # "Metadaten_aktualisierung" # ] # logger.debug("[AutoForm -- backend] Inserted InitRec Person successfully") # else: # logger.debug("[AutoForm -- backend] Update...") # stmt = ( # db.t_grunderfassung_personen.update() # .where(db.t_grunderfassung_personen.c.pers_id == auto_form_data.ids.pers_id) # .returning( # db.t_grunderfassung_personen.c.Metadaten_aktualisierung, # db.t_grunderfassung_personen.c.geloescht, # ) # ) # ret = conn.execute(stmt, dump_data) # from_db = ret.mappings().fetchall() # assert len(from_db) == 1, "expected excatly one returned row" # data_from_db = from_db[0] # auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[ # "Metadaten_aktualisierung" # ] # logger.debug( # "[AutoForm -- backend] Updated InitRec Person with ID %d successfully", # auto_form_data.ids.pers_id, # ) # return auto_form_data # def initrec_person_from_db( # id_: RecId, # ) -> InitRec: # logger.debug("[AutoForm -- backend] Call database loading routine...") # stmt = db.t_grunderfassung_personen.select().where( # db.t_grunderfassung_personen.c.pers_id == id_ # ) # with db.ENGINE.connect() as conn: # ret = conn.execute(stmt) # from_db = ret.mappings().all() # assert len(from_db) == 1, "not excactly one company initial recording obtained" # data_from_db = dict(from_db[0]) # geloescht = data_from_db["geloescht"] # del data_from_db["geloescht"] # return InitRec( # ids=EntityIds(pers_id=id_), # # rec_id=id_, # TODO remove # geloescht=geloescht, # db_data=data_from_db, # ) # TODO check removal # def initrec_company_get_initial_recording( # id_: RecId, # ) -> dict[str, Any]: # logger.debug("[Call backend] get_initial_recording") # stmt = db.grunderfassung_unternehmen.select().where( # db.grunderfassung_unternehmen.c.un_id == id_ # ) # with db.ENGINE.connect() as conn: # ret = conn.execute(stmt) # results = ret.mappings().all() # if not results: # raise KeyError(f"Database ID {id_} not found") # assert len(results) == 1, "more than one company initial recording obtained" # row = results[0] # assert row, "row was not obtained" # return dict(row) def _rename_db_data( db_data: dict[str, Any], mapping: DualDict, ) -> dict[str, Any]: new_db_data = copy.deepcopy(db_data) for k, v in db_data.items(): if k in mapping: new_db_data[mapping[k]] = v del new_db_data[k] if k in mapping.inverted: new_db_data[mapping.inverted[k]] = v del new_db_data[k] return new_db_data def page_initrec_to_db( data: InitRec, ) -> InitRec: with db.ENGINE.begin() as conn: # person dump_data = _rename_db_data( data.form_data.person, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN ) dump_data["geloescht"] = data.geloescht if data.ids.pers_id is None: logger.debug("[InitRec -- backend] Insert...") stmt = sa.insert(db.t_grunderfassung_personen).returning( db.t_grunderfassung_personen.c.pers_id, db.t_grunderfassung_personen.c.aktualisiert, db.t_grunderfassung_personen.c.geloescht, ) stmt_compiled = str(stmt.compile(db.ENGINE)) logger.debug( "[InitRec -- backend] Data to insert:\nStatement: %s\n%s", stmt_compiled, pformat(dump_data), ) ret = conn.execute(stmt, dump_data) from_db = ret.mappings().fetchall() assert len(from_db) == 1, "expected excactly one returned row" data_from_db = dict(from_db[0]) data_from_db = _rename_db_data( data_from_db, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN ) data.ids.pers_id = data_from_db["pers_id"] data.geloescht = data_from_db["geloescht"] data.form_data.person["Metadaten_aktualisierung"] = data_from_db[ "Metadaten_aktualisierung" ] logger.debug("[AutoForm -- backend] Inserted InitRec Person successfully") else: logger.debug("[AutoForm -- backend] Update...") stmt = ( db.t_grunderfassung_personen.update() .where(db.t_grunderfassung_personen.c.pers_id == data.ids.pers_id) .returning( db.t_grunderfassung_personen.c.aktualisiert, db.t_grunderfassung_personen.c.geloescht, ) ) ret = conn.execute(stmt, dump_data) from_db = ret.mappings().fetchall() assert len(from_db) == 1, "expected excatly one returned row" data_from_db = dict(from_db[0]) data_from_db = _rename_db_data( data_from_db, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN ) data.geloescht = data_from_db["geloescht"] data.form_data.person["Metadaten_aktualisierung"] = data_from_db[ "Metadaten_aktualisierung" ] logger.debug( "[InitRec -- backend] Updated InitRec Person with ID %d successfully", data.ids.pers_id, ) # company if data.recording_type is RecordingType.WITH_COMPANY: # insert into company table and add additional information assert data.ids.un_id, "company ID not set" assert data.ids.an_id, "contact person (an) ID not set" assert data.ids.pers_id, "person ID not set" stmt = ( sqlite_insert(db.t_unternehmen) .values(un_id=data.ids.un_id, geloescht=None) .on_conflict_do_nothing(index_elements=["un_id"]) ) conn.execute(stmt) # additional information --> linking table dump_data = _rename_db_data( data.form_data.company, FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN ) logger.debug( "[InitRec -- backend] Company - additional info: Dump Data = \n%s", pformat(dump_data), ) if "un_id" not in dump_data: # only case when updated dump_data["un_id"] = data.ids.un_id dump_data["an_id"] = data.ids.an_id stmt = sqlite_insert(db.t_zuordnung_personen_unternehmen).values( **dump_data, pers_id=data.ids.pers_id ) # dump_data["geloescht"] = data.geloescht # logger.debug( # "[InitRec -- backend] Company - additional info: Dump Data = \n%s", # pformat(dump_data), # ) stmt = stmt.on_conflict_do_update( index_elements=["un_id", "pers_id"], index_where=sa.text("geloescht IS NULL"), set_={ "an_id": stmt.excluded.an_id, "Partnersuche__kanal_aufmerksamkeit": stmt.excluded.Partnersuche__kanal_aufmerksamkeit, "aktualisiert": datetime.datetime.now(datetime.UTC), "geloescht": data.geloescht, }, ).returning(db.t_zuordnung_personen_unternehmen.c.id) ret = conn.execute(stmt) from_db = ret.mappings().fetchall() assert len(from_db) == 1, "expected excatly one returned row" data_from_db = dict(from_db[0]) data_from_db = _rename_db_data( data_from_db, FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN ) data.ids.link_id = data_from_db["id"] return data def page_initrec_from_db( ids: EntityIds, ) -> InitRec: logger.debug("[InitRec -- backend] Call database loading routine...") recording_type = RecordingType.WITHOUT_COMPANY if ids.link_id is not None or ids.valid(EntityType.COMPANY): recording_type = RecordingType.WITH_COMPANY data_company_table: dict[str, Any] = {} if recording_type is RecordingType.WITH_COMPANY: logger.debug("[InitRec -- backend] With company. Link ID: %s", ids.link_id) if ids.link_id: stmt = sa.select( db.t_zuordnung_personen_unternehmen.c.pers_id, db.t_zuordnung_personen_unternehmen.c.un_id, db.t_zuordnung_personen_unternehmen.c.an_id, db.t_zuordnung_personen_unternehmen.c.Partnersuche__kanal_aufmerksamkeit, ).where( db.t_zuordnung_personen_unternehmen.c.id == ids.link_id, ) else: assert ids.un_id, "company ID not set" assert ids.pers_id, "person ID not set" stmt = sa.select( db.t_zuordnung_personen_unternehmen.c.pers_id, db.t_zuordnung_personen_unternehmen.c.un_id, db.t_zuordnung_personen_unternehmen.c.an_id, db.t_zuordnung_personen_unternehmen.c.Partnersuche__kanal_aufmerksamkeit, ).where( db.t_zuordnung_personen_unternehmen.c.pers_id == ids.pers_id, db.t_zuordnung_personen_unternehmen.c.un_id == ids.un_id, db.t_zuordnung_personen_unternehmen.c.geloescht.is_(None), ) with db.ENGINE.connect() as conn: ret = conn.execute(stmt) from_db = ret.mappings().all() assert len(from_db) == 1, "not excactly one company initial recording obtained" data_company_table = dict(from_db[0]) ids.pers_id = data_company_table.pop("pers_id") ids.un_id = data_company_table["un_id"] ids.an_id = data_company_table["an_id"] data_company_table = _rename_db_data( data_company_table, FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN ) # person assert ids.pers_id, "person ID not set" logger.debug("[InitRec -- backend] Person... Person ID: %s", ids.pers_id) stmt = db.t_grunderfassung_personen.select().where( db.t_grunderfassung_personen.c.pers_id == ids.pers_id ) with db.ENGINE.connect() as conn: ret = conn.execute(stmt) from_db = ret.mappings().all() assert len(from_db) == 1, "not excactly one company initial recording obtained" data_person_table = dict(from_db[0]) data_person_table = _rename_db_data( data_person_table, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN ) geloescht = data_person_table["geloescht"] del data_person_table["geloescht"] form_data = Page_InitRec_Form_Data( metadata={}, person=data_person_table, company=data_company_table, ) if geloescht is not None: logger.warning( "[InitRec -- backend] Loaded entries which seem to be deleted - IDs: %s", ids ) return InitRec( ids=ids, recording_type=recording_type, geloescht=geloescht, form_data=form_data, ) # def initrec_company_delete_initial_recording( # id_: RecId, # ) -> None: # logger.debug("[Call backend] delete_initial_recording") # stmt = db.grunderfassung_unternehmen.delete().where( # db.grunderfassung_unternehmen.c.un_id == id_ # ) # with db.ENGINE.begin() as conn: # ret = conn.execute(stmt) # if ret.rowcount == 0: # raise KeyError(f"Database ID {id_} not found for deletion") # def initrec_person_insert_initial_recording( # data: dict[str, Any], # ) -> RecId: # logger.debug("[Call backend] insert_initial_recording") # stmt = db.grunderfassung_personen.insert() # with db.ENGINE.begin() as conn: # ret = conn.execute(stmt, data) # if ret.rowcount == 0: # raise IOError("Entry was not inserted correctly") # prim_keys = ret.inserted_primary_key # assert prim_keys # return prim_keys[0] # def initrec_person_update_initial_recording( # id_: RecId, # data: dict[str, Any], # ) -> None: # logger.debug("[Call backend] update_initial_recording") # stmt = db.grunderfassung_personen.update().where( # db.grunderfassung_personen.c.pers_id == id_ # ) # with db.ENGINE.begin() as conn: # conn.execute(stmt, data) # def initrec_person_get_initial_recording( # id_: RecId, # ) -> dict[str, Any]: # logger.debug("[Call backend] get_initial_recording person") # stmt = db.grunderfassung_personen.select().where( # db.grunderfassung_personen.c.pers_id == id_ # ) # with db.ENGINE.connect() as conn: # ret = conn.execute(stmt) # results = ret.mappings().all() # if not results: # raise KeyError(f"Database ID {id_} not found") # assert len(results) == 1, "more than one person initial recording obtained" # row = results[0] # assert row, "row was not obtained" # return dict(row) # def initrec_person_delete_initial_recording( # id_: RecId, # ) -> None: # logger.debug("[Call backend] delete_initial_recording") # stmt = db.grunderfassung_personen.delete().where( # db.grunderfassung_personen.c.pers_id == id_ # ) # with db.ENGINE.begin() as conn: # ret = conn.execute(stmt) # if ret.rowcount == 0: # raise KeyError(f"Database ID {id_} not found for deletion") @wrap_result(10) def page_consulting_to_db( consultation_data: Beratungsgespraech_Vorgang, ) -> Beratungsgespraech_Vorgang: logger.debug("[Consulting Page] Call database saving routine...") with db.ENGINE.begin() as conn: if consultation_data.vorgang_id is None: # insert new "Vorgang" insert_data_process = consultation_data.model_dump( exclude={"vorgang_id", "beratungen"} ) logger.debug( "[Consulting Page] Call insert 'Vorgang' with data:\n%s", pformat(insert_data_process), ) stmt = db.t_beratung_vorgang.insert() ret = conn.execute(stmt, insert_data_process) if ret.rowcount == 0: raise IOError("Entry was not inserted correctly") prim_keys = ret.inserted_primary_key assert prim_keys consultation_data.vorgang_id = cast("ConsId", prim_keys[0]) logger.debug("[Consulting Page] Inserted 'Vorgang' successfully") else: logger.debug( "[Consulting Page] VorgangID already set. ID: %d. Update...", consultation_data.vorgang_id, ) update_data_process = consultation_data.model_dump(exclude={"beratungen"}) stmt = db.t_beratung_vorgang.update().where( db.t_beratung_vorgang.c.vorgang_id == consultation_data.vorgang_id ) conn.execute(stmt, update_data_process) rows_for_db_insert: list[dict[str, Any]] = [] rows_for_db_update: list[dict[str, Any]] = [] cons_sessions_inserted: list[Beratungsgespraech_Einzelgespraech] = [] for cons_session in consultation_data.beratungen: cons_session.vorgang_id = consultation_data.vorgang_id if cons_session.beratung_id is None: row_data = cons_session.model_dump(exclude={"beratung_id"}) rows_for_db_insert.append(row_data) cons_sessions_inserted.append(cons_session) else: row_data = cons_session.model_dump() # new bind param to avoid name clashes row_data["b_beratung_id"] = row_data["beratung_id"] del row_data["beratung_id"] rows_for_db_update.append(row_data) if rows_for_db_update: # ... update logger.debug( "[Consulting Page] Call update for sessions:\n%s", pformat(rows_for_db_update) ) stmt = db.t_beratung_einzelberatung.update().where( db.t_beratung_einzelberatung.c.beratung_id == sa.bindparam("b_beratung_id") ) conn.execute(stmt, rows_for_db_update) if rows_for_db_insert: # ... insert logger.debug( "[Consulting Page] Call insert for sessions:\n%s", pformat(rows_for_db_insert) ) stmt = db.t_beratung_einzelberatung.insert().returning( db.t_beratung_einzelberatung.c.beratung_id ) res = conn.execute(stmt, rows_for_db_insert) new_cons_session_ids = cast(list[int], [row[0] for row in res.fetchall()]) assert len(cons_sessions_inserted) == len(new_cons_session_ids) for cons_session, new_id in zip(cons_sessions_inserted, new_cons_session_ids): cons_session.beratung_id = new_id return consultation_data @wrap_result(10) def page_consulting_delete( cons_id: ConsId, ) -> None: timestamp_deleted = datetime.datetime.now(datetime.UTC) dump_data: dict[str, Any] = {"geloescht": timestamp_deleted} with db.ENGINE.begin() as conn: stmt = db.t_beratung_vorgang.update().where( db.t_beratung_vorgang.c.vorgang_id == cons_id ) conn.execute(stmt, dump_data) stmt = db.t_beratung_einzelberatung.update().where( db.t_beratung_einzelberatung.c.vorgang_id == cons_id, db.t_beratung_einzelberatung.c.geloescht.is_(None), ) conn.execute(stmt, dump_data) @wrap_result(11) def page_consulting_from_db( cons_id: ConsId, ) -> Beratungsgespraech_Vorgang: # get "Vorgang" and all associated sessions to instantiate Pydantic data model logger.debug("[Consulting Page] Call database reading routine...") with db.ENGINE.connect() as conn: # get the consultation process stmt = db.t_beratung_vorgang.select().where( db.t_beratung_vorgang.c.vorgang_id == cons_id ) ret = conn.execute(stmt) results = ret.mappings().all() if not results: raise KeyError(f"Database ID {cons_id} not found") assert len(results) == 1, "more than one consulting process obtained" consultation_data_db = results[0] consultation_data = Beratungsgespraech_Vorgang( vorgang_id=consultation_data_db["vorgang_id"], un_id=consultation_data_db["un_id"], pers_id=consultation_data_db["pers_id"], titel=consultation_data_db["titel"], beratungs_typ=consultation_data_db["beratungs_typ"], erstellt=consultation_data_db["erstellt"], aktualisiert=consultation_data_db["aktualisiert"], geloescht=consultation_data_db["geloescht"], beratungen=[], ) if consultation_data.geloescht is not None: logger.info( ( "[Consulting Page] Obtained entry for ID=%d. This entry was marked " "as deleted." ), cons_id, ) # get all consultation sessions of this process, only the ones # which are not marked as deleted stmt = db.t_beratung_einzelberatung.select().where( db.t_beratung_einzelberatung.c.vorgang_id == cons_id, db.t_beratung_einzelberatung.c.geloescht.is_(None), ) ret = conn.execute(stmt) cons_sessions = ret.mappings().all() # empty results possible if not cons_sessions: logger.debug("[Consulting Page] No sessions, return directly...") return consultation_data for session in cons_sessions: cons_session_pydantic = Beratungsgespraech_Einzelgespraech( beratung_id=session["beratung_id"], vorgang_id=session["vorgang_id"], nutzer_id=session["nutzer_id"], nutzer_name=session["nutzer_name"], zeitstempel=session["zeitstempel"], ansprechpartner=session["ansprechpartner"], kommunikationsweg=session["kommunikationsweg"], thema_crm_matrix=session["thema_crm_matrix"], anmerkungen=session["anmerkungen"], rueckmeldung=session["rueckmeldung"], erstellt=session["erstellt"], aktualisiert=session["aktualisiert"], ) consultation_data.beratungen.append(cons_session_pydantic) logger.debug("[Consulting Page] Returning with sessions attached...") return consultation_data # // consulting page interaction # TODO change to correct logic def page_consulting_linking_companies( un_id: RecId | None, ) -> list[LinkingConsultationsEntryCompany]: raise RuntimeError("TODOs to implement") with db.ENGINE.connect() as conn: q_relevant_companies = sa.select( db.t_unternehmen.c.un_id, db.t_unternehmen.c.Partnersuche__un_suche, ).where( db.t_unternehmen.c.geloescht.is_(None), db.t_unternehmen.c.Partnersuche__un_suche.is_not(None), ) if un_id is not None: q_relevant_companies = q_relevant_companies.where( db.grunderfassung_unternehmen.c.un_id == un_id ) q_relevant_companies = q_relevant_companies.subquery("relevant_companies") stmt = sa.select( q_relevant_companies.c.un_id, db.t_ext_crm_master_attach.c.ma_unternehmensname, ).select_from( q_relevant_companies.join( db.t_ext_crm_master_attach, q_relevant_companies.c.Partnersuche__un_suche == db.t_ext_crm_master_attach.c.ma_id, ) ) ret = conn.execute(stmt) company_names = ret.mappings().all() if not company_names: raise ValueError(f"No companies found for 'un_id': {un_id}") logger.debug("[Page Consulting -- Linking] Retrieved company names:\n%s", company_names) linking_entries: list[LinkingConsultationsEntryCompany] = [] for entry in company_names: linking_entries.append( LinkingConsultationsEntryCompany( un_id=entry["un_id"], display_name=entry["ma_unternehmensname"], ) ) return linking_entries def page_consulting_linking_persons( un_id: RecId | None, ) -> list[LinkingConsultationsEntryPerson]: raise RuntimeError("TODOs to implement") with db.ENGINE.connect() as conn: # if un_id None: list all persons # else: list just the persons associated with the company person_select = sa.select( db.t_grunderfassung_personen.c.pers_id, db.t_grunderfassung_personen.c.Stammdaten__name, db.t_grunderfassung_personen.c.Stammdaten__vorname, db.t_grunderfassung_personen.c.Stammdaten__ort, ) if un_id is not None: person_select = person_select.select_from( db.t_grunderfassung_personen.join( db.t_zuordnung_personen_unternehmen, db.t_grunderfassung_personen.c.pers_id == db.t_zuordnung_personen_unternehmen.c.pers_id, ) ).where(db.t_zuordnung_personen_unternehmen.c.un_id == un_id) ret = conn.execute(person_select) person_names = ret.mappings().all() if not person_names: logger.warning(f"No persons found for 'un_id': {un_id}") return [] logger.debug("[Page Consulting -- Linking] Retrieved person names:\n%s", person_names) linking_entries: list[LinkingConsultationsEntryPerson] = [] for entry in person_names: first_name = entry["Stammdaten__vorname"] last_name = entry["Stammdaten__name"] location_city = entry["Stammdaten__ort"] display_name = f"{first_name} {last_name}" if location_city: display_name += f" (Ort: {location_city})" linking_entries.append( LinkingConsultationsEntryPerson( pers_id=entry["pers_id"], display_name=display_name, ) ) return linking_entries # TODO wrap result def page_consulting_eastablish_link( vorgang_id: ConsId, un_id: RecId | None, pers_id: RecId | None, ) -> None: raise RuntimeError("TODOs to implement") stmt = ( sa.update(db.t_beratung_vorgang) .where(db.t_beratung_vorgang.c.vorgang_id == vorgang_id) .values(un_id=un_id, pers_id=pers_id) ) with db.ENGINE.begin() as conn: conn.execute(stmt) # // company profile interaction def companyprofile_page_get_consultations( un_id: RecId, ) -> CompanyProfileConsultations: logger.debug("[Call backend] _companyprofile_page_get_consultations") stmt = sa.select( db.t_beratung_vorgang.c.vorgang_id, db.t_beratung_vorgang.c.aktualisiert, db.t_beratung_vorgang.c.titel, db.t_beratung_vorgang.c.beratungs_typ, ).where( db.t_beratung_vorgang.c.un_id == un_id, db.t_beratung_vorgang.c.geloescht.is_(None) ) with db.ENGINE.connect() as conn: res = conn.execute(stmt) cons_entries_pauschal: list[CompanyProfileConsultationEntry] = [] cons_entries_individual: list[CompanyProfileConsultationEntry] = [] for entry in res.mappings(): cons_id = entry["vorgang_id"] assert cons_id, "no VorgangID defined" datetime_updated = cast(datetime.datetime, entry["aktualisiert"]) datetime_updated = datetime_updated.astimezone(TIMEZONE_CEST) base_title = entry["titel"] cons_type = ConsultingType(entry["beratungs_typ"]) con_entry = CompanyProfileConsultationEntry( cons_id=cons_id, title=base_title, date_updated=datetime_updated, cons_type=cons_type, ) if cons_type is ConsultingType.PAUSCHAL: cons_entries_pauschal.append(con_entry) elif cons_type is ConsultingType.INDIVIDUAL: cons_entries_individual.append(con_entry) else: raise TypeError(f"Unknown consulting type: {cons_type}") cons_entries_pauschal.sort(key=lambda x: x.date_updated, reverse=True) cons_entries_individual.sort(key=lambda x: x.date_updated, reverse=True) return CompanyProfileConsultations( pauschal=cons_entries_pauschal, individual=cons_entries_individual, ) # // initial recording interaction # TODO wrap result def initrec_link_person_company( un_id: RecId, pers_id: RecId, unlink: bool, ) -> None: if unlink: deleted_ts = datetime.datetime.now(tz=datetime.UTC) stmt = ( sa.update(db.t_zuordnung_personen_unternehmen) .where( db.t_zuordnung_personen_unternehmen.c.un_id == un_id, db.t_zuordnung_personen_unternehmen.c.pers_id == pers_id, db.t_zuordnung_personen_unternehmen.c.gueltig_bis.is_(None), ) .values(gueltig_bis=deleted_ts) ) else: stmt = ( sqlite_insert(db.t_zuordnung_personen_unternehmen) .values(un_id=un_id, pers_id=pers_id) .on_conflict_do_nothing( index_elements=["un_id", "pers_id"], index_where=sa.text("gueltig_bis IS NULL"), ) ) with db.ENGINE.begin() as conn: conn.execute(stmt) # // main page interaction # TODO change to correct logic def _main_page_get_company_list() -> list[MainPageEntry]: logger.debug("[Call backend] get_company_list") raise RuntimeError("TODOs to implement") stmt = sa.select( db.grunderfassung_unternehmen.c.un_id, db.grunderfassung_unternehmen.c.Partnersuche__un_suche, db.grunderfassung_unternehmen.c.Metadaten_aktualisierung, ).where(db.grunderfassung_unternehmen.c.geloescht.is_(None)) with db.ENGINE.connect() as conn: res = conn.execute(stmt) main_page_companies: list[MainPageEntry] = [] for entry in res.mappings(): rec_id = entry["un_id"] assert rec_id, "no RecID defined" ma_id_external = entry["Partnersuche__un_suche"] assert ma_id_external is not None, "external MA ID is NULL" datetime_akt = cast(datetime.datetime, entry["Metadaten_aktualisierung"]) datetime_akt = datetime_akt.astimezone(TIMEZONE_CEST) comp_info = external_company_get_info(ma_id_external) display_name = comp_info["ma_unternehmensname"] main_page_companies.append( MainPageEntry( rec_id=rec_id, display_name=display_name, Metadaten_aktualisierung=datetime_akt, type=RecordingType.WITH_COMPANY, ) ) return main_page_companies def _main_page_get_person_list() -> list[MainPageEntry]: logger.debug("[Call backend] get_company_list") stmt = sa.select( db.t_grunderfassung_personen.c.pers_id, db.t_grunderfassung_personen.c.Stammdaten__vorname, db.t_grunderfassung_personen.c.Stammdaten__name, db.t_grunderfassung_personen.c.Metadaten_aktualisierung, ).where(db.t_grunderfassung_personen.c.geloescht.is_(None)) with db.ENGINE.connect() as conn: res = conn.execute(stmt) main_page_persons: list[MainPageEntry] = [] for entry in res.mappings(): rec_id = entry["pers_id"] assert rec_id, "no RecID defined" datetime_akt = cast(datetime.datetime, entry["Metadaten_aktualisierung"]) datetime_akt = datetime_akt.astimezone() surname: str = entry["Stammdaten__name"] first_name: str = entry["Stammdaten__vorname"] names_to_join = [n for n in (first_name, surname) if n] display_name = " ".join(names_to_join) main_page_persons.append( MainPageEntry( rec_id=rec_id, display_name=display_name, Metadaten_aktualisierung=datetime_akt, type=RecordingType.WITHOUT_COMPANY, ) ) return main_page_persons def main_page_get_entries() -> list[MainPageEntry]: companies = _main_page_get_company_list() persons = _main_page_get_person_list() all_entries = companies + persons all_entries.sort(key=lambda x: x.Metadaten_aktualisierung, reverse=True) return all_entries