generated from dopt-python/py311
509 lines
18 KiB
Python
509 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
import datetime
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import polars as pl
|
|
import sqlalchemy as sa
|
|
from sqlalchemy import Column, Index, String, Table, TypeDecorator, event, sql
|
|
|
|
from wce_crm import constants
|
|
from wce_crm import types as t
|
|
|
|
|
|
# // declarations
|
|
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.datetime.fromisoformat(clean_value)
|
|
except ValueError:
|
|
# Fallback if it's still weird
|
|
return None
|
|
|
|
|
|
class UTCDateTime(TypeDecorator):
|
|
"""Safely coerces naive datetimes from SQLite into timezone-aware UTC."""
|
|
|
|
impl = sa.DateTime
|
|
cache_ok = True
|
|
|
|
def process_bind_param(self, value, dialect):
|
|
"""Runs when saving to the database."""
|
|
if value is not None:
|
|
# Ensure it's converted to UTC before saving
|
|
if value.tzinfo is None:
|
|
value = value.replace(tzinfo=datetime.timezone.utc)
|
|
else:
|
|
value = value.astimezone(datetime.timezone.utc)
|
|
return value
|
|
|
|
def process_result_value(self, value, dialect):
|
|
"""Runs when fetching from the database."""
|
|
if value is not None and value.tzinfo is None:
|
|
# Explicitly tell Python this data *is* UTC
|
|
value = value.replace(tzinfo=datetime.timezone.utc)
|
|
return value
|
|
|
|
|
|
MD_CRM = sa.MetaData()
|
|
MD_MAIN = sa.MetaData()
|
|
ENGINE = sa.create_engine(f"sqlite:///{constants.Config.DB_PATH_MAIN}")
|
|
|
|
|
|
@event.listens_for(ENGINE, "connect")
|
|
def attach_second_db(dbapi_connection, connection_record):
|
|
cursor = dbapi_connection.cursor()
|
|
# 'kontaktliste' is the alias for the second DB
|
|
cursor.execute(f"ATTACH DATABASE '{constants.Config.DB_PATH_CRM}' AS kontaktliste")
|
|
cursor.close()
|
|
|
|
|
|
ext_crm_master: Table = Table(
|
|
"Master",
|
|
MD_CRM,
|
|
Column("ma_id", sa.Integer, nullable=False, unique=True),
|
|
Column("wce_id", sa.ForeignKey("Nutzer.wce_id")),
|
|
Column("ma_unternehmensname", sa.Text, nullable=True),
|
|
Column("ma_branche", sa.Text, nullable=True),
|
|
Column("ma_strasse", sa.Text, nullable=True),
|
|
Column("ma_hausnummer", sa.Text, nullable=True),
|
|
Column("ma_plz", sa.Text, nullable=True),
|
|
Column("ma_ort", sa.Text, nullable=True),
|
|
Column("ma_plz_postfach", sa.Text, nullable=True),
|
|
Column("ma_postfach", sa.Text, nullable=True),
|
|
Column("ma_website", sa.Text, nullable=True),
|
|
Column("ma_mail", sa.Text, nullable=True),
|
|
Column("ma_telefonnummer", sa.Text, nullable=True),
|
|
Column("ma_faxnummer", sa.Text, nullable=True),
|
|
Column("ma_ersteintrag_datum", SafeDateTime, nullable=True),
|
|
Column("ma_aktualisierung_datum", SafeDateTime, nullable=True),
|
|
Column("ma_aktualisierung_nutzer", sa.Text, nullable=True),
|
|
Column("ma_sollprozess", sa.Text, nullable=True),
|
|
Column("ma_auslaendische_mitarbeiter", sa.Text, nullable=True),
|
|
Column("ma_quelle_information", sa.Text, nullable=True),
|
|
Column("ma_bemerkung", sa.Text, nullable=True),
|
|
Column("ma_kontakt", sa.Boolean, nullable=True),
|
|
Column("ma_schlagworte", sa.Text, nullable=True),
|
|
Column("ma_archiviert", sa.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 = sa.create_engine(f"sqlite:///{db_path}")
|
|
stmt = sa.select(ext_crm_master)
|
|
return pl.read_database(stmt, engine, schema_overrides=ext_crm_master_schema)
|
|
|
|
|
|
DF_CRM_MASTER = get_ext_crm_master(constants.Config.DB_PATH_CRM)
|
|
|
|
ext_crm_nutzer: Table = Table(
|
|
"Nutzer",
|
|
MD_CRM,
|
|
Column("wce_id", sa.Integer, nullable=False, unique=True),
|
|
Column("wce_name", sa.Text, nullable=True),
|
|
Column("wce_vorname", sa.Text, nullable=True),
|
|
Column("wce_kuerzel", sa.Text, nullable=True),
|
|
Column("wce_passwort", sa.Text, nullable=True),
|
|
Column("wce_angelegt_am", sa.DateTime, nullable=True),
|
|
Column("wce_rolle", sa.Text, nullable=True),
|
|
Column("wce_angelegt_von", sa.Text, nullable=True),
|
|
Column("wce_aktiv", sa.Boolean, nullable=True),
|
|
Column("wce_letzter_login", sa.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,
|
|
}
|
|
|
|
ext_crm_contact_person: Table = Table(
|
|
"Ansprechpartner",
|
|
MD_CRM,
|
|
Column("an_id", sa.Integer, nullable=False, unique=True),
|
|
Column("ma_id", sa.ForeignKey("Master.ma_id")),
|
|
Column("wce_id", sa.ForeignKey("Nutzer.wce_id")),
|
|
Column("st_id", sa.Integer, nullable=False),
|
|
Column("an_sachgebiet", sa.Text, nullable=True),
|
|
Column("an_anrede", sa.Text, nullable=True),
|
|
Column("an_titel", sa.Text, nullable=True),
|
|
Column("an_nachname", sa.Text, nullable=True),
|
|
Column("an_vorname", sa.Text, nullable=True),
|
|
Column("an_position", sa.Text, nullable=True),
|
|
Column("an_mail", sa.Text, nullable=True),
|
|
Column("an_festnetz", sa.Text, nullable=True),
|
|
Column("an_mobil", sa.Text, nullable=True),
|
|
Column("an_faxnummer", sa.Text, nullable=True),
|
|
Column("an_hauptansprechpartner", sa.Text, nullable=True),
|
|
Column("an_anrede_anschrift", sa.Text, nullable=True),
|
|
Column("an_bemerkung", sa.Text, nullable=True),
|
|
Column("an_aktualisierung_datum", SafeDateTime, nullable=True),
|
|
Column("an_aktualisierung_nutzer", sa.Text, nullable=True),
|
|
Column("an_letztes_kontaktdatum", SafeDateTime, nullable=True),
|
|
Column("an_ersteintrag_datum", SafeDateTime, nullable=True),
|
|
Column("an_archiviert", sa.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 = sa.create_engine(f"sqlite:///{db_path}")
|
|
stmt = sa.select(ext_crm_contact_person)
|
|
df = pl.read_database(stmt, engine, schema_overrides=ext_crm_contact_person_schema)
|
|
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)
|
|
DF_CONTACT_PERSON = get_ext_crm_contact_person(constants.Config.DB_PATH_CRM)
|
|
|
|
ext_crm_master_attach: Table = Table(
|
|
"Master",
|
|
MD_CRM,
|
|
Column("ma_id", sa.Integer, nullable=False, unique=True),
|
|
Column("wce_id", sa.ForeignKey("Nutzer.wce_id")),
|
|
Column("ma_unternehmensname", sa.Text, nullable=True),
|
|
Column("ma_branche", sa.Text, nullable=True),
|
|
Column("ma_strasse", sa.Text, nullable=True),
|
|
Column("ma_hausnummer", sa.Text, nullable=True),
|
|
Column("ma_plz", sa.Text, nullable=True),
|
|
Column("ma_ort", sa.Text, nullable=True),
|
|
Column("ma_plz_postfach", sa.Text, nullable=True),
|
|
Column("ma_postfach", sa.Text, nullable=True),
|
|
Column("ma_website", sa.Text, nullable=True),
|
|
Column("ma_mail", sa.Text, nullable=True),
|
|
Column("ma_telefonnummer", sa.Text, nullable=True),
|
|
Column("ma_faxnummer", sa.Text, nullable=True),
|
|
Column("ma_ersteintrag_datum", SafeDateTime, nullable=True),
|
|
Column("ma_aktualisierung_datum", SafeDateTime, nullable=True),
|
|
Column("ma_aktualisierung_nutzer", sa.Text, nullable=True),
|
|
Column("ma_sollprozess", sa.Text, nullable=True),
|
|
Column("ma_auslaendische_mitarbeiter", sa.Text, nullable=True),
|
|
Column("ma_quelle_information", sa.Text, nullable=True),
|
|
Column("ma_bemerkung", sa.Text, nullable=True),
|
|
Column("ma_kontakt", sa.Boolean, nullable=True),
|
|
Column("ma_schlagworte", sa.Text, nullable=True),
|
|
Column("ma_archiviert", sa.Boolean, nullable=True, default=False),
|
|
schema="kontaktliste",
|
|
)
|
|
|
|
|
|
grunderfassung_unternehmen: Table = Table(
|
|
"grunderfassung_unternehmen",
|
|
MD_MAIN,
|
|
Column(
|
|
"un_id",
|
|
sa.Integer,
|
|
primary_key=True,
|
|
autoincrement=True,
|
|
),
|
|
Column(
|
|
"geloescht",
|
|
UTCDateTime,
|
|
nullable=True,
|
|
),
|
|
Column(
|
|
"Metadaten_erstellung",
|
|
UTCDateTime,
|
|
nullable=True,
|
|
default=lambda: datetime.datetime.now(datetime.UTC),
|
|
),
|
|
Column(
|
|
"Metadaten_aktualisierung",
|
|
UTCDateTime,
|
|
nullable=True,
|
|
default=lambda: datetime.datetime.now(datetime.UTC),
|
|
onupdate=lambda: datetime.datetime.now(datetime.UTC),
|
|
),
|
|
Column("Metadaten_nutzer", sa.String(20), nullable=True),
|
|
Column("Grunderfassung_fallnummer", sa.Text, nullable=True),
|
|
Column("Grunderfassung_notiz", sa.Text, nullable=True),
|
|
Column("Partnersuche__kanal_aufmerksamkeit", sa.Text, nullable=True, default=None),
|
|
Column("Partnersuche__person_suche", sa.Integer, nullable=True, default=None),
|
|
Column("Partnersuche__un_suche", sa.Integer, nullable=True, default=None),
|
|
)
|
|
|
|
grunderfassung_personen: Table = Table(
|
|
"grunderfassung_personen",
|
|
MD_MAIN,
|
|
Column(
|
|
"pers_id",
|
|
sa.Integer,
|
|
primary_key=True,
|
|
autoincrement=True,
|
|
),
|
|
Column(
|
|
"geloescht",
|
|
UTCDateTime,
|
|
nullable=True,
|
|
),
|
|
Column(
|
|
"Metadaten_erstellung",
|
|
UTCDateTime,
|
|
nullable=True,
|
|
default=lambda: datetime.datetime.now(datetime.UTC),
|
|
),
|
|
Column(
|
|
"Metadaten_aktualisierung",
|
|
UTCDateTime,
|
|
nullable=True,
|
|
default=lambda: datetime.datetime.now(datetime.UTC),
|
|
onupdate=lambda: datetime.datetime.now(datetime.UTC),
|
|
),
|
|
Column("Metadaten_nutzer", sa.String(20), nullable=True),
|
|
Column("Metadaten_wiedereintrittsdatum", sa.Date, nullable=True, default=None),
|
|
Column("Arbeitserfahrung", sa.Text, nullable=True),
|
|
Column("Grunderfassung_fallnummer", sa.Text, nullable=True),
|
|
Column("Grunderfassung_notiz", sa.Text, nullable=True),
|
|
Column("HoehereBildung", sa.Text, nullable=True),
|
|
Column("Kontaktperson__KP_adresse", sa.Text, nullable=True),
|
|
Column("Kontaktperson__KP_anrede_anschrift", sa.Text, nullable=True),
|
|
Column("Kontaktperson__KP_email", sa.Text, nullable=True),
|
|
Column("Kontaktperson__KP_festnetznummer", sa.Text, nullable=True),
|
|
Column("Kontaktperson__KP_funktion_beziehung", sa.Text, nullable=True),
|
|
Column("Kontaktperson__KP_mobilfunknummer", sa.Text, nullable=True),
|
|
Column("Kontaktperson__KP_name", sa.Text, nullable=True),
|
|
Column(
|
|
"Kontaktperson__KP_name_partner", sa.Text, nullable=True
|
|
), # TODO: check if needed when set by trigger
|
|
Column("Kontaktperson__KP_titel", sa.Text, nullable=True),
|
|
Column("Kontaktperson__KP_vorname", sa.Text, nullable=True),
|
|
Column("Projektrelevanz__relevanz", sa.Text, nullable=True),
|
|
Column("Projektrelevanz__foerderperiode", sa.Text, nullable=True),
|
|
Column("Schulbildung", sa.Text, nullable=True),
|
|
Column("Sprachkenntnisse", sa.Text, nullable=True),
|
|
Column("Stammdaten__PLZ", sa.Text, nullable=True),
|
|
Column("Stammdaten__anrede_anschrift", sa.Text, nullable=True),
|
|
Column("Stammdaten__anzahl_kinder__alter", sa.Text, nullable=True),
|
|
Column("Stammdaten__anzahl_kinder__anzahl", sa.Integer, nullable=True),
|
|
Column("Stammdaten__aufenthaltsort", sa.Text, nullable=True),
|
|
Column("Stammdaten__bundesland", sa.Text, nullable=True),
|
|
Column("Stammdaten__land", sa.Text, nullable=True),
|
|
Column("Stammdaten__email", sa.Text, nullable=True),
|
|
Column("Stammdaten__familienstand", sa.Text, nullable=True),
|
|
Column("Stammdaten__festnetznummer", sa.Text, nullable=True),
|
|
Column("Stammdaten__geburtsdatum", sa.Date, nullable=True),
|
|
Column("Stammdaten__hausnummer", sa.Text, nullable=True),
|
|
Column("Stammdaten__herkunftsland", sa.Text, nullable=True),
|
|
Column("Stammdaten__mobilfunknummer", sa.Text, nullable=True),
|
|
Column("Stammdaten__name", sa.Text, nullable=True),
|
|
Column("Stammdaten__ort", sa.Text, nullable=True),
|
|
Column("Stammdaten__rueckkehrer", sa.Boolean, nullable=True),
|
|
Column("Stammdaten__staatsangehoerigkeit", sa.Text, nullable=True),
|
|
Column("Stammdaten__strasse", sa.Text, nullable=True),
|
|
Column("Stammdaten__titel", sa.Text, nullable=True),
|
|
Column("Stammdaten__vorname", sa.Text, nullable=True),
|
|
Column("WeitereInfos__WI_arbeitsstatus", sa.Text, nullable=True),
|
|
Column("WeitereInfos__WI_aufenthaltstitel", sa.Text, nullable=True),
|
|
Column("WeitereInfos__WI_deutsch_sprache", sa.Text, nullable=True),
|
|
Column("WeitereInfos__WI_gueltigkeit_aufenthaltstitel", sa.Date, nullable=True),
|
|
Column("WeitereInfos__WI_meldung_institution", sa.Text, nullable=True),
|
|
)
|
|
|
|
zuordnung_personen_unternehmen: Table = Table(
|
|
"zuordnung_personen_unternehmen",
|
|
MD_MAIN,
|
|
Column(
|
|
"id",
|
|
sa.Integer,
|
|
primary_key=True,
|
|
autoincrement=True,
|
|
),
|
|
Column("un_id", sa.ForeignKey("grunderfassung_unternehmen.un_id"), nullable=False),
|
|
Column("pers_id", sa.ForeignKey("grunderfassung_personen.pers_id"), nullable=False),
|
|
Column(
|
|
"gueltig_ab",
|
|
UTCDateTime,
|
|
nullable=False,
|
|
default=lambda: datetime.datetime.now(datetime.UTC),
|
|
),
|
|
Column(
|
|
"gueltig_bis",
|
|
UTCDateTime,
|
|
nullable=True,
|
|
),
|
|
Index(
|
|
"uq_aktive_zuordnung",
|
|
"un_id",
|
|
"pers_id",
|
|
unique=True,
|
|
sqlite_where=sql.text("gueltig_bis IS NULL"),
|
|
),
|
|
)
|
|
|
|
|
|
beratung_vorgang: Table = Table(
|
|
"beratung_vorgang",
|
|
MD_MAIN,
|
|
Column(
|
|
"vorgang_id",
|
|
sa.Integer,
|
|
primary_key=True,
|
|
autoincrement=True,
|
|
),
|
|
Column("un_id", sa.ForeignKey("grunderfassung_unternehmen.un_id"), nullable=True),
|
|
Column("pers_id", sa.ForeignKey("grunderfassung_personen.pers_id"), nullable=True),
|
|
Column("titel", sa.Text, nullable=False),
|
|
Column("beratungs_typ", sa.String(20), nullable=False),
|
|
Column(
|
|
"erstellt",
|
|
UTCDateTime,
|
|
nullable=False,
|
|
default=lambda: datetime.datetime.now(datetime.UTC),
|
|
),
|
|
Column(
|
|
"aktualisiert",
|
|
UTCDateTime,
|
|
nullable=False,
|
|
default=lambda: datetime.datetime.now(datetime.UTC),
|
|
onupdate=lambda: datetime.datetime.now(datetime.UTC),
|
|
),
|
|
Column(
|
|
"geloescht",
|
|
UTCDateTime,
|
|
nullable=True,
|
|
),
|
|
)
|
|
|
|
|
|
beratung_einzelberatung: Table = Table(
|
|
"beratung_einzelberatung",
|
|
MD_MAIN,
|
|
Column(
|
|
"beratung_id",
|
|
sa.Integer,
|
|
primary_key=True,
|
|
autoincrement=True,
|
|
),
|
|
Column("vorgang_id", sa.ForeignKey("beratung_vorgang.vorgang_id"), nullable=False),
|
|
Column("nutzer_id", sa.Integer, nullable=False),
|
|
Column("nutzer_name", sa.String(20), nullable=False),
|
|
Column(
|
|
"zeitstempel",
|
|
UTCDateTime,
|
|
nullable=False,
|
|
default=lambda: datetime.datetime.now(datetime.UTC),
|
|
),
|
|
Column("ansprechpartner", sa.Text, nullable=False),
|
|
Column("kommunikationsweg", sa.Text, nullable=False),
|
|
Column("thema_crm_matrix", sa.Text, nullable=True),
|
|
Column("anmerkungen", sa.Text, nullable=True),
|
|
Column("rueckmeldung", sa.Text, nullable=True),
|
|
Column(
|
|
"erstellt",
|
|
UTCDateTime,
|
|
nullable=False,
|
|
default=lambda: datetime.datetime.now(datetime.UTC),
|
|
),
|
|
Column(
|
|
"aktualisiert",
|
|
UTCDateTime,
|
|
nullable=False,
|
|
default=lambda: datetime.datetime.now(datetime.UTC),
|
|
onupdate=lambda: datetime.datetime.now(datetime.UTC),
|
|
),
|
|
Column(
|
|
"geloescht",
|
|
UTCDateTime,
|
|
nullable=True,
|
|
),
|
|
)
|