further prototyping, added first DB interactions

src/wce_crm/__init__.py

@@ -0,0 +1,3 @@
+import wce_crm.env
+
+wce_crm.env.setup()

src/wce_crm/backend/initial_recording.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+from typing import TypedDict, cast
+
+import polars as pl
+
+from wce_crm import db
+
+
+class CompanyInfo(TypedDict):
+    ma_id: str
+    wce_id: str
+    ma_unternehmensname: str
+    ma_branche: str
+    ma_strasse: str
+    ma_hausnummer: str
+    ma_plz: str
+    ma_ort: str
+    ma_plz_postfach: str
+    ma_postfach: str
+    ma_website: str
+    ma_mail: str
+    ma_telefonnummer: str
+    ma_faxnummer: str
+    ma_ersteintrag_datum: str
+    ma_aktualisierung_datum: str
+    ma_aktualisierung_nutzer: str
+    ma_sollprozess: str
+    ma_auslaendische_mitarbeiter: str
+    ma_quelle_information: str
+    ma_bemerkung: str
+    ma_kontakt: str
+    ma_schlagworte: str
+    ma_archiviert: str
+
+
+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 comp_search_choice_mapping() -> dict[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)
+        .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"]))
+
+
+def comp_search_get_info(
+    ma_id: int,
+) -> CompanyInfo:
+    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))

src/wce_crm/db.py

@@ -0,0 +1,213 @@
+from __future__ import annotations
+
+import os
+import re
+from datetime import datetime
+from pathlib import Path
+
+import polars as pl
+import sqlalchemy as sql
+from sqlalchemy import Column, String, Table, TypeDecorator
+
+from wce_crm import types as t
+
+
+class SafeDateTime(TypeDecorator):
+    """Cleans non-standard ISO strings before parsing."""
+
+    impl = String  # We treat the underlying data as a String first
+
+    def process_result_value(self, value, dialect):
+        if value is None:
+            return None
+
+        # 1. Remove the trailing 'ff' (or any trailing letters)
+        # 2. Replace comma with dot (SQLAlchemy prefers . over ,)
+        clean_value = re.sub(r"[a-zA-Z]+$", "", value).replace(",", ".")
+
+        try:
+            return datetime.fromisoformat(clean_value)
+        except ValueError:
+            # Fallback if it's still weird
+            return None
+
+
+md_kontaktliste = sql.MetaData()
+md_crm = sql.MetaData()
+
+ext_kl_unternehmen: sql.Table = Table(
+    "Unternehmen",
+    md_kontaktliste,
+    Column("u_id", sql.Integer, nullable=False, unique=True),
+    Column("u_zeitstempel_eintrag", sql.DateTime, nullable=False),
+    Column("u_rechtsform", sql.Text, nullable=False),
+    Column("u_firmenname", sql.Text, nullable=False),
+    Column("u_strasse", sql.Text, nullable=False),
+    Column("u_hausnummer", sql.Text, nullable=False),
+    Column("u_adresszusatz", sql.Text, nullable=True),
+    Column("u_plz", sql.Text, nullable=False),
+    Column("u_ort", sql.Text, nullable=False),
+    Column("u_postfach", sql.Text, nullable=True),
+    Column("u_website", sql.Text, nullable=True),
+    Column("u_anrede", sql.Text, nullable=False),
+    Column("u_titel", sql.Text, nullable=True),
+    Column("u_vorname", sql.Text, nullable=False),
+    Column("u_nachname", sql.Text, nullable=False),
+    Column("u_funktion", sql.Text, nullable=False),
+    Column("u_mail", sql.Text, nullable=False),
+    Column("u_telefon", sql.Text, nullable=False),
+    Column("u_plz_postfach", sql.Text, nullable=True),
+    Column("u_einwilligung_inhaber", sql.Boolean, nullable=True),
+    Column("u_einwilligung_ansprechpartner", sql.Boolean, nullable=True),
+    Column("u_aktiv", sql.Boolean, nullable=False, default=1),
+)
+
+ext_kl_unternehmen_schema: t.PolarsSchema = {
+    "u_id": pl.UInt64,
+    "u_zeitstempel_eintrag": pl.Datetime,
+    "u_rechtsform": pl.String,
+    "u_firmenname": pl.String,
+    "u_strasse": pl.String,
+    "u_hausnummer": pl.String,
+    "u_adresszusatz": pl.String,
+    "u_plz": pl.String,
+    "u_ort": pl.String,
+    "u_postfach": pl.String,
+    "u_website": pl.String,
+    "u_anrede": pl.String,
+    "u_titel": pl.String,
+    "u_vorname": pl.String,
+    "u_nachname": pl.String,
+    "u_funktion": pl.String,
+    "u_mail": pl.String,
+    "u_telefon": pl.String,
+    "u_plz_postfach": pl.String,
+    "u_einwilligung_inhaber": pl.Boolean,
+    "u_einwilligung_ansprechpartner": pl.Boolean,
+    "u_aktiv": pl.Boolean,
+}
+
+
+def get_ext_kontaktliste(
+    db_path: Path | None,
+) -> pl.DataFrame:
+    if db_path is None:
+        ENV_PTH = os.environ.get("DOPT_DB_KONTAKTLISTE", 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_kl_unternehmen)
+    return pl.read_database(stmt, engine, schema_overrides=ext_kl_unternehmen_schema)
+
+
+df_kontaktliste = get_ext_kontaktliste(None)
+
+ext_crm_master: sql.Table = Table(
+    "Master",
+    md_crm,
+    Column("ma_id", sql.Integer, nullable=False, unique=True),
+    Column("wce_id", sql.ForeignKey("Nutzer.wce_id")),
+    Column("ma_unternehmensname", sql.Text, nullable=True),
+    Column("ma_branche", sql.Text, nullable=True),
+    Column("ma_strasse", sql.Text, nullable=True),
+    Column("ma_hausnummer", sql.Text, nullable=True),
+    Column("ma_plz", sql.Text, nullable=True),
+    Column("ma_ort", sql.Text, nullable=True),
+    Column("ma_plz_postfach", sql.Text, nullable=True),
+    Column("ma_postfach", sql.Text, nullable=True),
+    Column("ma_website", sql.Text, nullable=True),
+    Column("ma_mail", sql.Text, nullable=True),
+    Column("ma_telefonnummer", sql.Text, nullable=True),
+    Column("ma_faxnummer", sql.Text, nullable=True),
+    Column("ma_ersteintrag_datum", SafeDateTime, nullable=True),
+    Column("ma_aktualisierung_datum", SafeDateTime, nullable=True),
+    Column("ma_aktualisierung_nutzer", sql.Text, nullable=True),
+    Column("ma_sollprozess", sql.Text, nullable=True),
+    Column("ma_auslaendische_mitarbeiter", sql.Text, nullable=True),
+    Column("ma_quelle_information", sql.Text, nullable=True),
+    Column("ma_bemerkung", sql.Text, nullable=True),
+    Column("ma_kontakt", sql.Boolean, nullable=True),
+    Column("ma_schlagworte", sql.Text, nullable=True),
+    Column("ma_archiviert", sql.Boolean, nullable=True, default=False),
+)
+
+
+ext_crm_master_schema: t.PolarsSchema = {
+    "ma_id": pl.UInt64,
+    "wce_id": pl.UInt64,
+    "ma_unternehmensname": pl.String,
+    "ma_branche": pl.String,
+    "ma_strasse": pl.String,
+    "ma_hausnummer": pl.String,
+    "ma_plz": pl.String,
+    "ma_ort": pl.String,
+    "ma_plz_postfach": pl.String,
+    "ma_postfach": pl.String,
+    "ma_website": pl.String,
+    "ma_mail": pl.String,
+    "ma_telefonnummer": pl.String,
+    "ma_faxnummer": pl.String,
+    "ma_ersteintrag_datum": pl.Datetime,
+    "ma_aktualisierung_datum": pl.Datetime,
+    "ma_aktualisierung_nutzer": pl.String,
+    "ma_sollprozess": pl.String,
+    "ma_auslaendische_mitarbeiter": pl.String,
+    "ma_quelle_information": pl.String,
+    "ma_bemerkung": pl.String,
+    "ma_kontakt": pl.Boolean,
+    "ma_schlagworte": pl.String,
+    "ma_archiviert": pl.Boolean,
+}
+
+
+def get_ext_crm_master(
+    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_master)
+    return pl.read_database(stmt, engine, schema_overrides=ext_crm_master_schema)
+
+
+df_crm_master = get_ext_crm_master(None)
+
+ext_crm_nutzer: sql.Table = Table(
+    "Nutzer",
+    md_crm,
+    Column("wce_id", sql.Integer, nullable=False, unique=True),
+    Column("wce_name", sql.Text, nullable=True),
+    Column("wce_vorname", sql.Text, nullable=True),
+    Column("wce_kuerzel", sql.Text, nullable=True),
+    Column("wce_passwort", sql.Text, nullable=True),
+    Column("wce_angelegt_am", sql.DateTime, nullable=True),
+    Column("wce_rolle", sql.Text, nullable=True),
+    Column("wce_angelegt_von", sql.Text, nullable=True),
+    Column("wce_aktiv", sql.Boolean, nullable=True),
+    Column("wce_letzter_login", sql.DateTime, nullable=True),
+)
+
+ext_crm_nutzer_schema: t.PolarsSchema = {
+    "wce_id": pl.UInt64,
+    "wce_name": pl.String,
+    "wce_vorname": pl.String,
+    "wce_kuerzel": pl.String,
+    "wce_passwort": pl.String,
+    "wce_angelegt_am": pl.Datetime,
+    "wce_rolle": pl.String,
+    "wce_angelegt_von": pl.String,
+    "wce_aktiv": pl.Boolean,
+    "wce_letzter_login": pl.Datetime,
+}

src/wce_crm/env.py

@@ -0,0 +1,15 @@
+import os
+from pathlib import Path
+
+PROJECT_ROOT = Path(__file__).parents[2]
+DB_PATH = PROJECT_ROOT / "data/db"
+
+DB_KONTAKTLISTE = DB_PATH / "wce_kontaktliste.db"
+assert DB_KONTAKTLISTE.exists()
+DB_CRM = DB_PATH / "wce_crm.db"
+assert DB_CRM.exists()
+
+
+def setup():
+    os.environ["DOPT_DB_KONTAKTLISTE"] = str(DB_KONTAKTLISTE)
+    os.environ["DOPT_DB_CRM"] = str(DB_CRM)

src/wce_crm/env_vars.txt

@@ -0,0 +1,2 @@
+DOPT_DB_KONTAKTLISTE: Pfad zur Datenbank der Kontaktliste, falls nicht direkt übergeben (Prototypenphase)
+DOPT_DB_CRM: Pfad zur CRM-Datenbank, falls nicht direkt übergeben (Prototypenphase)

src/wce_crm/types.py

@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, TypeAlias
+
+if TYPE_CHECKING:
+    import polars as pl
+
+
+PolarsSchema: TypeAlias = dict[str, type["pl.DataType"]]