prepare better DB interaction with reload

src/wce_crm/backend/initial_recording.py

@@ -34,6 +34,31 @@ class CompanyInfo(TypedDict):
     ma_archiviert: str
 
 
+class ContactPersonInfo(TypedDict):
+    an_id: str
+    ma_id: str
+    wce_id: str
+    st_id: str
+    an_sachgebiet: str
+    an_anrede: str
+    an_titel: str
+    an_nachname: str
+    an_vorname: str
+    an_position: str
+    an_mail: str
+    an_festnetz: str
+    an_mobil: str
+    an_faxnummer: str
+    an_hauptansprechpartner: str
+    an_anrede_anschrift: str
+    an_bemerkung: str
+    an_aktualisierung_datum: str
+    an_aktualisierung_nutzer: str
+    an_letztes_kontaktdatum: str
+    an_ersteintrag_datum: str
+    an_archiviert: str
+
+
 def _transform_for_gui_output(
     data: pl.DataFrame,
 ) -> pl.DataFrame:
@@ -53,18 +78,19 @@ def _transform_for_gui_output(
     return q.collect()
 
 
-def comp_search_choice_mapping() -> dict[str, int]:
+def comp_search_choices() -> tuple[tuple[str, int], ...]:
     # TODO no reload functionality
     q = db.df_crm_master.lazy()
     counter = pl.int_range(0, pl.len()).over(pl.col.ma_unternehmensname)
     q = q.with_columns(
-        ma_unternehmensname_dedupl=pl.when(counter == 0)
+        dedupl=pl.when(counter == 0)
         .then(pl.col.ma_unternehmensname)
         .otherwise(pl.format("{} ({})", pl.col.ma_unternehmensname, counter))
     )
     df = q.collect()
 
-    return dict(zip(df["ma_unternehmensname_dedupl"], df["ma_id"]))
+    # return dict(zip(df["dedupl"], df["ma_id"]))
+    return tuple(zip(df["dedupl"], df["ma_id"]))
 
 
 def comp_search_get_info(
@@ -76,3 +102,42 @@ def comp_search_get_info(
 
     df = _transform_for_gui_output(df)
     return cast(CompanyInfo, df.row(0, named=True))
+
+
+def contact_person_search_choices(
+    ma_id: int | None,
+    use_both_names: bool,
+) -> tuple[tuple[str, int], ...]:
+    # TODO no reload functionality
+    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 dict(zip(df["dedupl"], df["an_id"]))
+    return tuple(zip(df["dedupl"], df["an_id"]))
+
+
+def contact_person_search_get_info(
+    an_id: int,
+) -> ContactPersonInfo:
+    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))

src/wce_crm/db.py

@@ -32,9 +32,12 @@ class SafeDateTime(TypeDecorator):
             return None
 
 
-md_kontaktliste = sql.MetaData()
 md_crm = sql.MetaData()
 
+# ---------- OLD "Kontaktliste" ----------
+
+md_kontaktliste = sql.MetaData()
+
 ext_kl_unternehmen: sql.Table = Table(
     "Unternehmen",
     md_kontaktliste,
@@ -107,6 +110,8 @@ def get_ext_kontaktliste(
 
 df_kontaktliste = get_ext_kontaktliste(None)
 
+# ----------------------------------------------------
+
 ext_crm_master: sql.Table = Table(
     "Master",
     md_crm,
@@ -211,3 +216,82 @@ ext_crm_nutzer_schema: t.PolarsSchema = {
     "wce_aktiv": pl.Boolean,
     "wce_letzter_login": pl.Datetime,
 }
+
+ext_crm_contact_person: sql.Table = Table(
+    "Ansprechpartner",
+    md_crm,
+    Column("an_id", sql.Integer, nullable=False, unique=True),
+    Column("ma_id", sql.ForeignKey("Master.ma_id")),
+    Column("wce_id", sql.ForeignKey("Nutzer.wce_id")),
+    Column("st_id", sql.Integer, nullable=False),
+    Column("an_sachgebiet", sql.Text, nullable=True),
+    Column("an_anrede", sql.Text, nullable=True),
+    Column("an_titel", sql.Text, nullable=True),
+    Column("an_nachname", sql.Text, nullable=True),
+    Column("an_vorname", sql.Text, nullable=True),
+    Column("an_position", sql.Text, nullable=True),
+    Column("an_mail", sql.Text, nullable=True),
+    Column("an_festnetz", sql.Text, nullable=True),
+    Column("an_mobil", sql.Text, nullable=True),
+    Column("an_faxnummer", sql.Text, nullable=True),
+    Column("an_hauptansprechpartner", sql.Text, nullable=True),
+    Column("an_anrede_anschrift", sql.Text, nullable=True),
+    Column("an_bemerkung", sql.Text, nullable=True),
+    Column("an_aktualisierung_datum", SafeDateTime, nullable=True),
+    Column("an_aktualisierung_nutzer", sql.Text, nullable=True),
+    Column("an_letztes_kontaktdatum", SafeDateTime, nullable=True),
+    Column("an_ersteintrag_datum", SafeDateTime, nullable=True),
+    Column("an_archiviert", sql.Boolean, nullable=True, default=0),
+)
+
+
+ext_crm_contact_person_schema: t.PolarsSchema = {
+    "an_id": pl.UInt64,
+    "ma_id": pl.UInt64,
+    "wce_id": pl.UInt64,
+    "st_id": pl.UInt64,
+    "an_sachgebiet": pl.String,
+    "an_anrede": pl.String,
+    "an_titel": pl.String,
+    "an_nachname": pl.String,
+    "an_vorname": pl.String,
+    "an_position": pl.String,
+    "an_mail": pl.String,
+    "an_festnetz": pl.String,
+    "an_mobil": pl.String,
+    "an_faxnummer": pl.String,
+    "an_hauptansprechpartner": pl.String,
+    "an_anrede_anschrift": pl.String,
+    "an_bemerkung": pl.String,
+    "an_aktualisierung_datum": pl.Datetime,
+    "an_aktualisierung_nutzer": pl.String,
+    "an_letztes_kontaktdatum": pl.Datetime,
+    "an_ersteintrag_datum": pl.Datetime,
+    "an_archiviert": pl.Boolean,
+}
+
+
+def get_ext_crm_contact_person(
+    db_path: Path | None,
+) -> pl.DataFrame:
+    if db_path is None:
+        ENV_PTH = os.environ.get("DOPT_DB_CRM", None)
+        if ENV_PTH is None:
+            raise ValueError("No database path provided or found as ENV var.")
+        db_path = Path(ENV_PTH)
+
+    if not db_path.exists():
+        raise FileNotFoundError(f"Database not found under >{db_path}<")
+
+    engine = sql.create_engine(f"sqlite:///{db_path}")
+    stmt = sql.select(ext_crm_contact_person)
+    df = pl.read_database(stmt, engine, schema_overrides=ext_crm_contact_person_schema)
+    # TODO Database seems to contain entries with invalid characters like \t or \n
+    df = df.with_columns(
+        pl.col(pl.String).str.replace_all(r"[\r\t\n]", " ").str.strip_chars(" ")
+    )
+
+    return df
+
+
+df_contact_person = get_ext_crm_contact_person(None)