diff --git a/alembic/versions/8000d15faf03_major_change_in_data_model.py b/alembic/versions/8000d15faf03_major_change_in_data_model.py new file mode 100644 index 0000000..5242e0a --- /dev/null +++ b/alembic/versions/8000d15faf03_major_change_in_data_model.py @@ -0,0 +1,317 @@ +"""major change in data model + +Revision ID: 8000d15faf03 +Revises: 665259cdfdc0 +Create Date: 2026-07-24 15:13:57.443684 + +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +import wce_crm.db + +# revision identifiers, used by Alembic. +revision: str = "8000d15faf03" +down_revision: Union[str, Sequence[str], None] = "665259cdfdc0" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_table("grunderfassung_personen") + op.drop_table("beratung_vorgang") + op.drop_table("grunderfassung_unternehmen") + with op.batch_alter_table("zuordnung_personen_unternehmen", schema=None) as batch_op: + batch_op.drop_index( + batch_op.f("uq_aktive_zuordnung"), sqlite_where=sa.text("gueltig_bis IS NULL") + ) + + op.drop_table("zuordnung_personen_unternehmen") + op.drop_table("beratung_einzelberatung") + + op.create_table( + "Grunderfassung_Personen", + sa.Column("pers_id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("erstellt", wce_crm.db.UTCDateTime(), nullable=False), + sa.Column("aktualisiert", wce_crm.db.UTCDateTime(), nullable=False), + sa.Column("geloescht", wce_crm.db.UTCDateTime(), nullable=True), + sa.Column("Metadaten_nutzer", sa.String(length=20), nullable=True), + sa.Column("Metadaten_wiedereintrittsdatum", sa.Date(), nullable=True), + sa.Column("Arbeitserfahrung", sa.Text(), nullable=True), + sa.Column("Grunderfassung_fallnummer", sa.Text(), nullable=True), + sa.Column("Grunderfassung_notiz", sa.Text(), nullable=True), + sa.Column("HoehereBildung", sa.Text(), nullable=True), + sa.Column("Kontaktperson__KP_adresse", sa.Text(), nullable=True), + sa.Column("Kontaktperson__KP_anrede_anschrift", sa.Text(), nullable=True), + sa.Column("Kontaktperson__KP_email", sa.Text(), nullable=True), + sa.Column("Kontaktperson__KP_festnetznummer", sa.Text(), nullable=True), + sa.Column("Kontaktperson__KP_funktion_beziehung", sa.Text(), nullable=True), + sa.Column("Kontaktperson__KP_mobilfunknummer", sa.Text(), nullable=True), + sa.Column("Kontaktperson__KP_name", sa.Text(), nullable=True), + sa.Column("Kontaktperson__KP_name_partner", sa.Text(), nullable=True), + sa.Column("Kontaktperson__KP_titel", sa.Text(), nullable=True), + sa.Column("Kontaktperson__KP_vorname", sa.Text(), nullable=True), + sa.Column("Projektrelevanz__relevanz", sa.Text(), nullable=True), + sa.Column("Projektrelevanz__foerderperiode", sa.Text(), nullable=True), + sa.Column("Schulbildung", sa.Text(), nullable=True), + sa.Column("Sprachkenntnisse", sa.Text(), nullable=True), + sa.Column("Stammdaten__PLZ", sa.Text(), nullable=True), + sa.Column("Stammdaten__anrede_anschrift", sa.Text(), nullable=True), + sa.Column("Stammdaten__anzahl_kinder__alter", sa.Text(), nullable=True), + sa.Column("Stammdaten__anzahl_kinder__anzahl", sa.Integer(), nullable=True), + sa.Column("Stammdaten__aufenthaltsort", sa.Text(), nullable=True), + sa.Column("Stammdaten__bundesland", sa.Text(), nullable=True), + sa.Column("Stammdaten__land", sa.Text(), nullable=True), + sa.Column("Stammdaten__email", sa.Text(), nullable=True), + sa.Column("Stammdaten__familienstand", sa.Text(), nullable=True), + sa.Column("Stammdaten__festnetznummer", sa.Text(), nullable=True), + sa.Column("Stammdaten__geburtsdatum", sa.Date(), nullable=True), + sa.Column("Stammdaten__hausnummer", sa.Text(), nullable=True), + sa.Column("Stammdaten__herkunftsland", sa.Text(), nullable=True), + sa.Column("Stammdaten__mobilfunknummer", sa.Text(), nullable=True), + sa.Column("Stammdaten__name", sa.Text(), nullable=True), + sa.Column("Stammdaten__ort", sa.Text(), nullable=True), + sa.Column("Stammdaten__rueckkehrer", sa.Boolean(), nullable=True), + sa.Column("Stammdaten__staatsangehoerigkeit", sa.Text(), nullable=True), + sa.Column("Stammdaten__strasse", sa.Text(), nullable=True), + sa.Column("Stammdaten__titel", sa.Text(), nullable=True), + sa.Column("Stammdaten__vorname", sa.Text(), nullable=True), + sa.Column("WeitereInfos__WI_arbeitsstatus", sa.Text(), nullable=True), + sa.Column("WeitereInfos__WI_aufenthaltstitel", sa.Text(), nullable=True), + sa.Column("WeitereInfos__WI_deutsch_sprache", sa.Text(), nullable=True), + sa.Column("WeitereInfos__WI_gueltigkeit_aufenthaltstitel", sa.Date(), nullable=True), + sa.Column("WeitereInfos__WI_meldung_institution", sa.Text(), nullable=True), + sa.PrimaryKeyConstraint("pers_id"), + ) + op.create_table( + "Unternehmen", + sa.Column("un_id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("erstellt", wce_crm.db.UTCDateTime(), nullable=False), + sa.Column("geloescht", wce_crm.db.UTCDateTime(), nullable=True), + sa.PrimaryKeyConstraint("un_id"), + ) + op.create_table( + "Beratung_Vorgang", + sa.Column("vorgang_id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("un_id", sa.Integer(), nullable=True), + sa.Column("pers_id", sa.Integer(), nullable=True), + sa.Column("titel", sa.Text(), nullable=False), + sa.Column("beratungs_typ", sa.String(length=20), nullable=False), + sa.Column("erstellt", wce_crm.db.UTCDateTime(), nullable=False), + sa.Column("aktualisiert", wce_crm.db.UTCDateTime(), nullable=False), + sa.Column("geloescht", wce_crm.db.UTCDateTime(), nullable=True), + sa.ForeignKeyConstraint( + ["pers_id"], + ["Grunderfassung_Personen.pers_id"], + ), + sa.ForeignKeyConstraint( + ["un_id"], + ["Unternehmen.un_id"], + ), + sa.PrimaryKeyConstraint("vorgang_id"), + ) + op.create_table( + "Personen_Unternehmen_Zuordnung", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("un_id", sa.Integer(), nullable=False), + sa.Column("an_id", sa.Integer(), nullable=False), + sa.Column("pers_id", sa.Integer(), nullable=False), + sa.Column("erstellt", wce_crm.db.UTCDateTime(), nullable=False), + sa.Column("aktualisiert", wce_crm.db.UTCDateTime(), nullable=False), + sa.Column("geloescht", wce_crm.db.UTCDateTime(), nullable=True), + sa.ForeignKeyConstraint( + ["pers_id"], + ["Grunderfassung_Personen.pers_id"], + ), + sa.ForeignKeyConstraint( + ["un_id"], + ["Unternehmen.un_id"], + ), + sa.PrimaryKeyConstraint("id"), + ) + with op.batch_alter_table("Personen_Unternehmen_Zuordnung", schema=None) as batch_op: + batch_op.create_index( + "uq_aktive_zuordnung", + ["un_id", "pers_id"], + unique=True, + sqlite_where=sa.text("geloescht IS NULL"), + ) + + op.create_table( + "Beratung_Einzelberatung", + sa.Column("beratung_id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("vorgang_id", sa.Integer(), nullable=False), + sa.Column("nutzer_id", sa.Integer(), nullable=False), + sa.Column("nutzer_name", sa.String(length=20), nullable=False), + sa.Column("zeitstempel", wce_crm.db.UTCDateTime(), nullable=False), + sa.Column("ansprechpartner", sa.Text(), nullable=False), + sa.Column("kommunikationsweg", sa.Text(), nullable=False), + sa.Column("thema_crm_matrix", sa.Text(), nullable=True), + sa.Column("anmerkungen", sa.Text(), nullable=True), + sa.Column("rueckmeldung", sa.Text(), nullable=True), + sa.Column("erstellt", wce_crm.db.UTCDateTime(), nullable=False), + sa.Column("aktualisiert", wce_crm.db.UTCDateTime(), nullable=False), + sa.Column("geloescht", wce_crm.db.UTCDateTime(), nullable=True), + sa.ForeignKeyConstraint( + ["vorgang_id"], + ["Beratung_Vorgang.vorgang_id"], + ), + sa.PrimaryKeyConstraint("beratung_id"), + ) + + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_table("Beratung_Einzelberatung") + with op.batch_alter_table("Personen_Unternehmen_Zuordnung", schema=None) as batch_op: + batch_op.drop_index("uq_aktive_zuordnung", sqlite_where=sa.text("geloescht IS NULL")) + + op.drop_table("Personen_Unternehmen_Zuordnung") + op.drop_table("Beratung_Vorgang") + op.drop_table("Unternehmen") + op.drop_table("Grunderfassung_Personen") + + op.create_table( + "beratung_einzelberatung", + sa.Column("beratung_id", sa.INTEGER(), nullable=False), + sa.Column("vorgang_id", sa.INTEGER(), nullable=False), + sa.Column("nutzer_id", sa.INTEGER(), nullable=False), + sa.Column("nutzer_name", sa.VARCHAR(length=20), nullable=False), + sa.Column("zeitstempel", sa.DATETIME(), nullable=False), + sa.Column("ansprechpartner", sa.TEXT(), nullable=False), + sa.Column("kommunikationsweg", sa.TEXT(), nullable=False), + sa.Column("thema_crm_matrix", sa.TEXT(), nullable=True), + sa.Column("anmerkungen", sa.TEXT(), nullable=True), + sa.Column("rueckmeldung", sa.TEXT(), nullable=True), + sa.Column("erstellt", sa.DATETIME(), nullable=False), + sa.Column("aktualisiert", sa.DATETIME(), nullable=False), + sa.Column("geloescht", sa.DATETIME(), nullable=True), + sa.ForeignKeyConstraint( + ["vorgang_id"], + ["beratung_vorgang.vorgang_id"], + ), + sa.PrimaryKeyConstraint("beratung_id"), + ) + op.create_table( + "zuordnung_personen_unternehmen", + sa.Column("id", sa.INTEGER(), nullable=False), + sa.Column("un_id", sa.INTEGER(), nullable=False), + sa.Column("pers_id", sa.INTEGER(), nullable=False), + sa.Column("gueltig_ab", sa.DATETIME(), nullable=False), + sa.Column("gueltig_bis", sa.DATETIME(), nullable=True), + sa.ForeignKeyConstraint( + ["pers_id"], + ["grunderfassung_personen.pers_id"], + ), + sa.ForeignKeyConstraint( + ["un_id"], + ["grunderfassung_unternehmen.un_id"], + ), + sa.PrimaryKeyConstraint("id"), + ) + with op.batch_alter_table("zuordnung_personen_unternehmen", schema=None) as batch_op: + batch_op.create_index( + batch_op.f("uq_aktive_zuordnung"), + ["un_id", "pers_id"], + unique=1, + sqlite_where=sa.text("gueltig_bis IS NULL"), + ) + + op.create_table( + "grunderfassung_unternehmen", + sa.Column("un_id", sa.INTEGER(), nullable=False), + sa.Column("Metadaten_erstellung", sa.DATETIME(), nullable=True), + sa.Column("Metadaten_aktualisierung", sa.DATETIME(), nullable=True), + sa.Column("Metadaten_nutzer", sa.VARCHAR(length=20), nullable=True), + sa.Column("Grunderfassung_fallnummer", sa.TEXT(), nullable=True), + sa.Column("Grunderfassung_notiz", sa.TEXT(), nullable=True), + sa.Column("Partnersuche__kanal_aufmerksamkeit", sa.TEXT(), nullable=True), + sa.Column("Partnersuche__person_suche", sa.INTEGER(), nullable=True), + sa.Column("Partnersuche__un_suche", sa.INTEGER(), nullable=True), + sa.Column("geloescht", sa.DATETIME(), nullable=True), + sa.PrimaryKeyConstraint("un_id"), + ) + op.create_table( + "beratung_vorgang", + sa.Column("vorgang_id", sa.INTEGER(), nullable=False), + sa.Column("un_id", sa.INTEGER(), nullable=True), + sa.Column("pers_id", sa.INTEGER(), nullable=True), + sa.Column("titel", sa.TEXT(), nullable=False), + sa.Column("beratungs_typ", sa.VARCHAR(length=20), nullable=False), + sa.Column("erstellt", sa.DATETIME(), nullable=False), + sa.Column("aktualisiert", sa.DATETIME(), nullable=False), + sa.Column("geloescht", sa.DATETIME(), nullable=True), + sa.ForeignKeyConstraint( + ["pers_id"], + ["grunderfassung_personen.pers_id"], + ), + sa.ForeignKeyConstraint( + ["un_id"], + ["grunderfassung_unternehmen.un_id"], + ), + sa.PrimaryKeyConstraint("vorgang_id"), + ) + op.create_table( + "grunderfassung_personen", + sa.Column("pers_id", sa.INTEGER(), nullable=False), + sa.Column("Metadaten_erstellung", sa.DATETIME(), nullable=True), + sa.Column("Metadaten_aktualisierung", sa.DATETIME(), nullable=True), + sa.Column("Metadaten_nutzer", sa.VARCHAR(length=20), nullable=True), + sa.Column("Metadaten_wiedereintrittsdatum", sa.DATE(), nullable=True), + sa.Column("Arbeitserfahrung", sa.TEXT(), nullable=True), + sa.Column("Grunderfassung_fallnummer", sa.TEXT(), nullable=True), + sa.Column("Grunderfassung_notiz", sa.TEXT(), nullable=True), + sa.Column("HoehereBildung", sa.TEXT(), nullable=True), + sa.Column("Kontaktperson__KP_adresse", sa.TEXT(), nullable=True), + sa.Column("Kontaktperson__KP_anrede_anschrift", sa.TEXT(), nullable=True), + sa.Column("Kontaktperson__KP_email", sa.TEXT(), nullable=True), + sa.Column("Kontaktperson__KP_festnetznummer", sa.TEXT(), nullable=True), + sa.Column("Kontaktperson__KP_funktion_beziehung", sa.TEXT(), nullable=True), + sa.Column("Kontaktperson__KP_mobilfunknummer", sa.TEXT(), nullable=True), + sa.Column("Kontaktperson__KP_name", sa.TEXT(), nullable=True), + sa.Column("Kontaktperson__KP_name_partner", sa.TEXT(), nullable=True), + sa.Column("Kontaktperson__KP_titel", sa.TEXT(), nullable=True), + sa.Column("Kontaktperson__KP_vorname", sa.TEXT(), nullable=True), + sa.Column("Projektrelevanz__relevanz", sa.TEXT(), nullable=True), + sa.Column("Projektrelevanz__foerderperiode", sa.TEXT(), nullable=True), + sa.Column("Schulbildung", sa.TEXT(), nullable=True), + sa.Column("Sprachkenntnisse", sa.TEXT(), nullable=True), + sa.Column("Stammdaten__PLZ", sa.TEXT(), nullable=True), + sa.Column("Stammdaten__anrede_anschrift", sa.TEXT(), nullable=True), + sa.Column("Stammdaten__anzahl_kinder__alter", sa.TEXT(), nullable=True), + sa.Column("Stammdaten__anzahl_kinder__anzahl", sa.INTEGER(), nullable=True), + sa.Column("Stammdaten__aufenthaltsort", sa.TEXT(), nullable=True), + sa.Column("Stammdaten__bundesland", sa.TEXT(), nullable=True), + sa.Column("Stammdaten__land", sa.TEXT(), nullable=True), + sa.Column("Stammdaten__email", sa.TEXT(), nullable=True), + sa.Column("Stammdaten__familienstand", sa.TEXT(), nullable=True), + sa.Column("Stammdaten__festnetznummer", sa.TEXT(), nullable=True), + sa.Column("Stammdaten__geburtsdatum", sa.DATE(), nullable=True), + sa.Column("Stammdaten__hausnummer", sa.TEXT(), nullable=True), + sa.Column("Stammdaten__herkunftsland", sa.TEXT(), nullable=True), + sa.Column("Stammdaten__mobilfunknummer", sa.TEXT(), nullable=True), + sa.Column("Stammdaten__name", sa.TEXT(), nullable=True), + sa.Column("Stammdaten__ort", sa.TEXT(), nullable=True), + sa.Column("Stammdaten__rueckkehrer", sa.BOOLEAN(), nullable=True), + sa.Column("Stammdaten__staatsangehoerigkeit", sa.TEXT(), nullable=True), + sa.Column("Stammdaten__strasse", sa.TEXT(), nullable=True), + sa.Column("Stammdaten__titel", sa.TEXT(), nullable=True), + sa.Column("Stammdaten__vorname", sa.TEXT(), nullable=True), + sa.Column("WeitereInfos__WI_arbeitsstatus", sa.TEXT(), nullable=True), + sa.Column("WeitereInfos__WI_aufenthaltstitel", sa.TEXT(), nullable=True), + sa.Column("WeitereInfos__WI_deutsch_sprache", sa.TEXT(), nullable=True), + sa.Column("WeitereInfos__WI_gueltigkeit_aufenthaltstitel", sa.DATE(), nullable=True), + sa.Column("WeitereInfos__WI_meldung_institution", sa.TEXT(), nullable=True), + sa.Column("geloescht", sa.DATETIME(), nullable=True), + sa.PrimaryKeyConstraint("pers_id"), + ) + + # ### end Alembic commands ### diff --git a/prototypes/dual_dict.py b/prototypes/dual_dict.py new file mode 100644 index 0000000..17fc7b3 --- /dev/null +++ b/prototypes/dual_dict.py @@ -0,0 +1,54 @@ +# %% +from __future__ import annotations + +import copy +from typing import Any + +from dopt_basics.datastructures import DualDict + +from wce_crm.data_models import FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN + +# %% +dd: DualDict[str, str] = DualDict(peter="test") + + +# %% +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 + + +form_data = { + "Metadaten_erstellung": "t1", + "Metadaten_aktualisierung": "t2", + "Metadaten_nutzer": "t3", +} +db_data = rename_db_data(form_data, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN) +# %% +db_data + +# %% +cvt_form_data = rename_db_data(db_data, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN) +cvt_form_data +# %% +dd["test"] = "peter" +# %% +dd["test"] +# %% +dd.inverted["peter"] +# %% +dd +# %% diff --git a/prototypes/tests.py b/prototypes/tests.py index a2cb9a6..3644836 100644 --- a/prototypes/tests.py +++ b/prototypes/tests.py @@ -19,11 +19,11 @@ from PySide6.QtCore import QDate, Qt from wce_crm import constants, db from wce_crm.backend import backend -from wce_crm.types import NewEntityIds, RecordingType +from wce_crm.types import EntityIds, RecordingType # %% -raw_ent_ids = NewEntityIds() -db_ent_ids = NewEntityIds(un_id=3, an_id=2) +raw_ent_ids = EntityIds() +db_ent_ids = EntityIds(un_id=3, an_id=2) print(f"{raw_ent_ids=}\n{db_ent_ids=}") print(f"{raw_ent_ids.empty()=}") @@ -31,36 +31,36 @@ print(f"{raw_ent_ids.valid()=}") # %% raw_ent_ids.update(db_ent_ids) print(f"{raw_ent_ids=}") -db_ent_ids = NewEntityIds(un_id=None, an_id=2, link_id=1) +db_ent_ids = EntityIds(un_id=None, an_id=2, link_id=1) raw_ent_ids.update(db_ent_ids) print(f"{raw_ent_ids=}") print(f"{raw_ent_ids.empty()=}") print(f"{raw_ent_ids.valid()=}") -raw_ent_ids.update(NewEntityIds()) +raw_ent_ids.update(EntityIds()) print(f"{raw_ent_ids=}") print(f"{raw_ent_ids.empty()=}") print(f"{raw_ent_ids.valid()=}") # %% -raw_ent_ids.update(NewEntityIds(un_id=None, pers_id=2)) +raw_ent_ids.update(EntityIds(un_id=None, pers_id=2)) print(f"{raw_ent_ids=}") print(f"{raw_ent_ids.empty()=}") -print(f"{raw_ent_ids.empty(RecordingType.COMPANY)=}") -print(f"{raw_ent_ids.empty(RecordingType.PERSON)=}") -print(f"{raw_ent_ids.valid(RecordingType.COMPANY)=}") -print(f"{raw_ent_ids.valid(RecordingType.PERSON)=}") -raw_ent_ids.update(NewEntityIds(un_id=None, an_id=2, link_id=3, pers_id=2)) +print(f"{raw_ent_ids.empty(RecordingType.WITH_COMPANY)=}") +print(f"{raw_ent_ids.empty(RecordingType.WITHOUT_COMPANY)=}") +print(f"{raw_ent_ids.valid(RecordingType.WITH_COMPANY)=}") +print(f"{raw_ent_ids.valid(RecordingType.WITHOUT_COMPANY)=}") +raw_ent_ids.update(EntityIds(un_id=None, an_id=2, link_id=3, pers_id=2)) print(f"{raw_ent_ids=}") -print(f"{raw_ent_ids.empty(RecordingType.COMPANY)=}") -print(f"{raw_ent_ids.empty(RecordingType.PERSON)=}") -print(f"{raw_ent_ids.valid(RecordingType.COMPANY)=}") -print(f"{raw_ent_ids.valid(RecordingType.PERSON)=}") -raw_ent_ids.update(NewEntityIds(un_id=2, pers_id=None, an_id=2, link_id=3)) +print(f"{raw_ent_ids.empty(RecordingType.WITH_COMPANY)=}") +print(f"{raw_ent_ids.empty(RecordingType.WITHOUT_COMPANY)=}") +print(f"{raw_ent_ids.valid(RecordingType.WITH_COMPANY)=}") +print(f"{raw_ent_ids.valid(RecordingType.WITHOUT_COMPANY)=}") +raw_ent_ids.update(EntityIds(un_id=2, pers_id=None, an_id=2, link_id=3)) print(f"{raw_ent_ids=}") -print(f"{raw_ent_ids.empty(RecordingType.COMPANY)=}") -print(f"{raw_ent_ids.empty(RecordingType.PERSON)=}") -print(f"{raw_ent_ids.valid(RecordingType.COMPANY)=}") -print(f"{raw_ent_ids.valid(RecordingType.PERSON)=}") +print(f"{raw_ent_ids.empty(RecordingType.WITH_COMPANY)=}") +print(f"{raw_ent_ids.empty(RecordingType.WITHOUT_COMPANY)=}") +print(f"{raw_ent_ids.valid(RecordingType.WITH_COMPANY)=}") +print(f"{raw_ent_ids.valid(RecordingType.WITHOUT_COMPANY)=}") # %% db_path = constants.Config.DB_PATH_MAIN diff --git a/src/wce_crm/backend/backend.py b/src/wce_crm/backend/backend.py index 5ee9f50..f21406c 100644 --- a/src/wce_crm/backend/backend.py +++ b/src/wce_crm/backend/backend.py @@ -13,9 +13,12 @@ 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 ( @@ -24,14 +27,17 @@ from wce_crm.types import ( CompanyProfileConsultations, ConsultingType, ContactPersonInfo, + EntityIds, + EntityType, LinkingConsultationsEntryCompany, LinkingConsultationsEntryPerson, MainPageEntry, - NewEntityIds, RecordingType, ) if TYPE_CHECKING: + from dopt_basics.datastructures import DualDict + from wce_crm.types import ConsId, ExtAnId, ExtMaId, RecId @@ -348,14 +354,201 @@ def initrec_comp_contact_person_search_get_info( # 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: ... +) -> 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 = from_db[0] + + data.ids.pers_id = data_from_db["pers_id"] + data.geloescht = data_from_db["geloescht"] + data.form_data.person["Metadaten_aktualisierung"] = data_from_db["aktualisiert"] + + 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.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] + + 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 + dump_data = _rename_db_data( + data.form_data.company, FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN + ) + dump_data["geloescht"] = data.geloescht + stmt = sqlite_insert(db.t_zuordnung_personen_unternehmen).values( + **dump_data, pers_id=data.ids.pers_id + ) + 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), + }, + ).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 = from_db[0] + data.ids.link_id = data_from_db["id"] + + return data def page_initrec_from_db( - ids: NewEntityIds, -) -> InitRec: ... + ids: EntityIds, +) -> InitRec: + recording_type = RecordingType.WITHOUT_COMPANY + if ids.valid(EntityType.COMPANY): + recording_type = RecordingType.WITH_COMPANY + + # person + assert ids.pers_id, "person ID not set" + logger.debug("[AutoForm -- backend] Call database loading routine...") + 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"] + + data_company_table: dict[str, Any] = {} + if recording_type is RecordingType.WITH_COMPANY: + assert ids.un_id, "company ID not set" + + if ids.link_id: + stmt = sa.select( + 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: + stmt = sa.select( + 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]) + + data_company_table = _rename_db_data( + data_person_table, FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN + ) + + form_data = Page_InitRec_Form_Data( + metadata={}, + person=data_person_table, + company=data_company_table, + ) + + return InitRec( + ids=ids, + recording_type=recording_type, + geloescht=geloescht, + form_data=form_data, + ) # def initrec_company_delete_initial_recording( @@ -857,7 +1050,7 @@ def _main_page_get_company_list() -> list[MainPageEntry]: rec_id=rec_id, display_name=display_name, Metadaten_aktualisierung=datetime_akt, - type=RecordingType.COMPANY, + type=RecordingType.WITH_COMPANY, ) ) @@ -894,7 +1087,7 @@ def _main_page_get_person_list() -> list[MainPageEntry]: rec_id=rec_id, display_name=display_name, Metadaten_aktualisierung=datetime_akt, - type=RecordingType.PERSON, + type=RecordingType.WITHOUT_COMPANY, ) ) diff --git a/src/wce_crm/data_models.py b/src/wce_crm/data_models.py index 83c83bb..e52daeb 100644 --- a/src/wce_crm/data_models.py +++ b/src/wce_crm/data_models.py @@ -7,6 +7,7 @@ import json from collections.abc import Iterable, Sequence from typing import TYPE_CHECKING, Annotated, Any, Final, Generic, Protocol, TypeVar +from dopt_basics.datastructures import DualDict from pydantic import ( AwareDatetime, BaseModel, @@ -18,12 +19,12 @@ from pydantic import ( ) from pydantic_core import ErrorDetails -from wce_crm.form_defs import FormField -from wce_crm.types import NewEntityIds +from wce_crm.form_defs import INITREC_COMP, INITREC_PERSON, FormField +from wce_crm.types import EntityIds, EntityType, RecordingType if TYPE_CHECKING: from wce_crm.gui import InitRecForm, Page_Consulting_ConsultingSession # noqa: F401 - from wce_crm.types import ConsId, ConsultingType, RecId, RecordingType, UserId + from wce_crm.types import ConsId, ConsultingType, EntityType, RecId, UserId ValidAge = Annotated[int, Field(ge=0, le=99)] COLUMN_SEP: Final[str] = "__" @@ -163,7 +164,7 @@ class InitRecFormToDb(Protocol): class InitRecFormFromDb(Protocol): def __call__( self, - ids: NewEntityIds, + ids: EntityIds, ) -> InitRec: ... @@ -171,6 +172,7 @@ class InitRecFormFromDb(Protocol): def set_page_state( target: S, source: S, + ignore_attr: set[str] | None = None, deep: bool = False, ) -> None: """ @@ -182,8 +184,12 @@ def set_page_state( if type(target) is not type(source): raise TypeError("Both source and target must have the same type") + if ignore_attr is None: + ignore_attr = set() + ignore_attr.add("child_modules") + for field in dc.fields(source): - if field.name == "child_modules": + if field.name in ignore_attr: continue new_value = getattr(source, field.name) @@ -216,35 +222,63 @@ class Page_NewEntry_State(PageState[Module]): session: Session +INITREC_REC_TO_ENTITIES: Final[dict[RecordingType, frozenset[EntityType]]] = { + RecordingType.WITH_COMPANY: frozenset((EntityType.COMPANY, EntityType.PERSON)), + RecordingType.WITHOUT_COMPANY: frozenset((EntityType.PERSON,)), +} + + +@dc.dataclass(slots=True, kw_only=True) +class Page_InitRec_Form_Data: + metadata: dict[str, Any] + person: dict[str, Any] + company: dict[str, Any] + + # !! there must be exactly one InitRecForm for each recording type @dc.dataclass(slots=True, kw_only=True) class Page_InitRec_State(PageState["InitRecForm"]): session: Session - ids: NewEntityIds = dc.field(default_factory=NewEntityIds) - record_type: RecordingType + ids: EntityIds = dc.field(default_factory=EntityIds) + recording_type: RecordingType # un_id: RecId | None = None locked: bool geloescht: bool = False - form_data: dict[RecordingType, dict[str, Any]] = dc.field(default_factory=dict) - pydantic_models: dict[RecordingType, type[FlatBaseModel]] = dc.field(default_factory=dict) + form_data: Page_InitRec_Form_Data + initrec_form_states: dict[EntityType, Page_InitRec_Form_State] = dc.field( + init=False, default_factory=dict + ) + # pydantic_models: dict[RecordingType, type[FlatBaseModel]] = dc.field(default_factory=dict) + + def __post_init__(self) -> None: + relevant_entities = INITREC_REC_TO_ENTITIES[self.recording_type] + for ent_type in relevant_entities: + initrec_form_state = Page_InitRec_Form_State( + session=self.session, + cfg=INITREC_CONFIGS[ent_type], + ids=self.ids, + ent_type=ent_type, + locked=self.locked, + geloescht=self.geloescht, + ) + self.initrec_form_states[ent_type] = initrec_form_state @dc.dataclass(slots=True, kw_only=True) class Page_InitRec_Form_Config: pydantic_model: type[FlatBaseModel] - # to_db: InitRecFormToDb # TODO remove - # from_db: InitRecFormFromDb # TODO remove form_fields: Sequence[FormField] ignored_keys: Iterable[str] = tuple() add_buttons: bool = False + id_mapping: dict[str, str] @dc.dataclass(slots=True, kw_only=True) class Page_InitRec_Form_State(PageState[Module]): session: Session cfg: Page_InitRec_Form_Config - ids: NewEntityIds - recording_type: RecordingType + ids: EntityIds + ent_type: EntityType # rec_id: RecId | None = None form_data: dict[str, Any] = dc.field(default_factory=dict) locked: bool @@ -269,7 +303,6 @@ class Page_InitRec_Form_State(PageState[Module]): # locked: bool -# TODO change to entity IDs @dc.dataclass(slots=True, kw_only=True) class Page_Consulting_State(PageState[Module]): session: Session @@ -285,7 +318,6 @@ class Page_Consulting_State(PageState[Module]): locked: bool -# TODO change to entity IDs @dc.dataclass(slots=True, kw_only=True) class Page_Consulting_Linking_State(PageState[Module]): session: Session @@ -322,7 +354,7 @@ class Page_Consulting_Table_State(PageState["Page_Consulting_ConsultingSession"] @dc.dataclass(slots=True, kw_only=True) class Page_CompanyProfile_State(PageState[Module]): session: Session - ids: NewEntityIds + ids: EntityIds # rec_id: RecId | None = None @@ -441,10 +473,21 @@ class FlatBaseModel(BaseModel): # ** InitRec class InitRec(BaseModel): - ids: NewEntityIds - # rec_id: RecId | None + ids: EntityIds + recording_type: RecordingType geloescht: AwareDatetime | None = None - db_data: dict[RecordingType, dict[str, Any]] + form_data: Page_InitRec_Form_Data + + +# mapping between initrec form field keys and database properties +FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN: Final[DualDict[str, str]] = DualDict( + Metadaten_erstellung="erstellt", + Metadaten_aktualisierung="aktualisiert", +) +FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN: Final[DualDict[str, str]] = DualDict( + Partnersuche__un_suche="un_id", + Partnersuche__person_suche="an_id", +) class Grunderfassung_Unternehmen(FlatBaseModel): @@ -649,3 +692,34 @@ class Beratungsgespraech_Einzelgespraech(BaseModel): erstellt: AwareDatetime | None = Field(default=None, exclude=True) aktualisiert: AwareDatetime | None = Field(default=None, exclude=True) # see above geloescht: AwareDatetime | None = None + + +INITREC_CONFIGS: Final[dict[EntityType, Page_InitRec_Form_Config]] = { + EntityType.COMPANY: Page_InitRec_Form_Config( + pydantic_model=Grunderfassung_Unternehmen, + # to_db=backend.initrec_company_to_db, # TODO remove + # from_db=backend.initrec_company_from_db, # TODO remove + ignored_keys=( + "Metadaten_erstellung", + "Metadaten_aktualisierung", + ), + form_fields=INITREC_COMP, + add_buttons=False, + id_mapping={ + "Partnersuche__un_suche": "un_id", + "Partnersuche__person_suche": "an_id", + }, + ), + EntityType.PERSON: Page_InitRec_Form_Config( + pydantic_model=Grunderfassung_Personen, + # to_db=backend.initrec_person_to_db, # TODO remove + # from_db=backend.initrec_person_from_db, # TODO remove + ignored_keys=( + "Metadaten_erstellung", + "Metadaten_aktualisierung", + ), + form_fields=INITREC_PERSON, + add_buttons=False, + id_mapping={}, + ), +} diff --git a/src/wce_crm/db.py b/src/wce_crm/db.py index 1c6b883..d5450b3 100644 --- a/src/wce_crm/db.py +++ b/src/wce_crm/db.py @@ -490,6 +490,7 @@ t_zuordnung_personen_unternehmen: Table = Table( UTCDateTime, nullable=True, ), + Column("Partnersuche__kanal_aufmerksamkeit", sa.Text, nullable=True), # Column( # "gueltig_ab", # UTCDateTime, @@ -520,8 +521,8 @@ t_beratung_vorgang: Table = Table( 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("un_id", sa.ForeignKey("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( @@ -554,7 +555,7 @@ t_beratung_einzelberatung: Table = Table( primary_key=True, autoincrement=True, ), - Column("vorgang_id", sa.ForeignKey("beratung_vorgang.vorgang_id"), nullable=False), + 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( diff --git a/src/wce_crm/form_defs.py b/src/wce_crm/form_defs.py index 765f6e4..9650dc5 100644 --- a/src/wce_crm/form_defs.py +++ b/src/wce_crm/form_defs.py @@ -889,42 +889,42 @@ INITREC_LANGUAGES = [ INITREC_COMP = [ - FormField( - "Ersteintrag Datum", - FormFieldType.TEXT_DATETIME, - required=False, - key="Metadaten_erstellung", - readonly=True, - ignore_get_data=True, - ), - FormField( - "Aktualisierung Datum", - FormFieldType.TEXT_DATETIME, - required=False, - key="Metadaten_aktualisierung", - readonly=True, - ignore_get_data=True, - ), - FormField( - "Aktualisierung Nutzer", - FormFieldType.TEXT, - required=False, - key="Metadaten_nutzer", - readonly=True, - ignore_get_data=True, - ), - FormField( - "Fallnummer", - FormFieldType.TEXT, - required=True, - key="Grunderfassung_fallnummer", - ), - FormField( - "Notizen", - FormFieldType.LONGTEXT, - required=False, - key="Grunderfassung_notiz", - ), + # FormField( + # "Ersteintrag Datum", + # FormFieldType.TEXT_DATETIME, + # required=False, + # key="Metadaten_erstellung", + # readonly=True, + # ignore_get_data=True, + # ), + # FormField( + # "Aktualisierung Datum", + # FormFieldType.TEXT_DATETIME, + # required=False, + # key="Metadaten_aktualisierung", + # readonly=True, + # ignore_get_data=True, + # ), + # FormField( + # "Aktualisierung Nutzer", + # FormFieldType.TEXT, + # required=False, + # key="Metadaten_nutzer", + # readonly=True, + # ignore_get_data=True, + # ), + # FormField( + # "Fallnummer", + # FormFieldType.TEXT, + # required=True, + # key="Grunderfassung_fallnummer", + # ), + # FormField( + # "Notizen", + # FormFieldType.LONGTEXT, + # required=False, + # key="Grunderfassung_notiz", + # ), FormField( "Suche", FormFieldType.CUSTOM, diff --git a/src/wce_crm/gui.py b/src/wce_crm/gui.py index 9c82681..b9ef80f 100644 --- a/src/wce_crm/gui.py +++ b/src/wce_crm/gui.py @@ -85,17 +85,16 @@ from wce_crm.backend import backend from wce_crm.constants import TIMEZONE_CEST from wce_crm.data_models import ( COLUMN_SEP, + INITREC_CONFIGS, Beratungsgespraech_Einzelgespraech, Beratungsgespraech_Vorgang, - Grunderfassung_Personen, - Grunderfassung_Unternehmen, InitRec, Page_CompanyProfile_State, Page_Consulting_ConsultingSession_State, Page_Consulting_Linking_State, Page_Consulting_State, Page_Consulting_Table_State, - Page_InitRec_Form_Config, + Page_InitRec_Form_Data, Page_InitRec_Form_State, Page_InitRec_State, Page_MainPage_State, @@ -105,8 +104,6 @@ from wce_crm.data_models import ( translate_pydantic_errors, ) from wce_crm.form_defs import ( - INITREC_COMP, - INITREC_PERSON, FormField, FormFieldType, ) @@ -121,7 +118,8 @@ from wce_crm.logging import ( from wce_crm.types import ( CompanyProfileConsultationEntry, ConsultingType, - NewEntityIds, + EntityIds, + EntityType, RecId, RecordingType, ) @@ -1547,7 +1545,7 @@ class InitRecForm(QWidget): # self.save_btn.setText(self.edit_buttons.save_btn_txt_enabled) def relevant_ids_set(self) -> bool: - return self.STATE.ids.valid(self.STATE.recording_type) + return self.STATE.ids.valid(self.STATE.ent_type) # def _activate_delete(self) -> None: # if self.STATE.ids.valid(): @@ -1613,12 +1611,23 @@ class InitRecForm(QWidget): # pformat(self.STATE.form_data), # ) + def _load_state_form_data(self) -> None: + logger_initrecform.info("[InitRecForm] Load data method...") + logger_initrecform.debug( + "[InitRecForm] Loaded data dict:\n%s Passing to Pydantic...", + pformat(self.STATE.form_data), + ) + # TODO error catching? + model = self.cfg.pydantic_model(**self.STATE.form_data) + logger_initrecform.debug("[InitRecForm] Loaded to Pydantic.") + logger_initrecform.debug("[InitRecForm] Convert to GUI structure...") + form_data_gui = model.to_gui() + # TODO include set data in this method? + self._set_data(form_data_gui) + def reset_form(self) -> None: reset_form(self.widget_registry) - # self.STATE.ids.set_id(None, self.STATE.ent_type) - self.STATE.ids.update(NewEntityIds()) - # TODO remove - # self.STATE.rec_id = None + self.STATE.ids.update(EntityIds()) self.STATE.form_data.clear() self.STATE.geloescht = False @@ -1658,7 +1667,7 @@ class InitRecForm(QWidget): # !! loading from IDs is only performed on top-level modules aka pages if section is None or section == "form": if not self.STATE.geloescht: - self.load_from_ids(self.STATE.ids) + self._load_state_form_data() else: self.reset_form() @@ -1678,6 +1687,10 @@ class InitRecForm(QWidget): # TODO change usage of user ID (needs changes in model and backend) if self.SESSION_USER_KEY in form_data: form_data[self.SESSION_USER_KEY] = self.STATE.session.user_name + + for form_key, id_key in self.STATE.cfg.id_mapping.items(): + self.STATE.ids.set_by_name(id_key, form_data[form_key]) + self.STATE.form_data = form_data def validate(self) -> list[str]: @@ -1695,18 +1708,18 @@ class InitRecForm(QWidget): return self.STATE - def _load_from_pydantic( - self, - data: InitRec, - ) -> None: - # ent_type = self.STATE.recording_type - # self.STATE.ids.set_id(data.ids.get_id(ent_type), ent_type) - self.STATE.ids.update(data.ids) - # self.STATE.rec_id = data.rec_id - self.STATE.form_data.clear() - self.STATE.geloescht = True if data.geloescht else False + # def _load_from_pydantic( + # self, + # data: InitRec, + # ) -> None: + # # ent_type = self.STATE.recording_type + # # self.STATE.ids.set_id(data.ids.get_id(ent_type), ent_type) + # self.STATE.ids.update(data.ids) + # # self.STATE.rec_id = data.rec_id + # self.STATE.form_data.clear() + # self.STATE.geloescht = True if data.geloescht else False - self._sync_state_to_GUI(None) # TODO change to correct loading + # self._sync_state_to_GUI(None) # TODO change to correct loading # !! must be part of the managing page def save_data(self) -> None: @@ -2979,7 +2992,8 @@ class Page_NewEntry(QWidget): def _request_new_initrec(self) -> None: req_state = Page_InitRec_State( session=self.STATE.session, - record_type=RecordingType.COMPANY, + recording_type=RecordingType.WITH_COMPANY, + form_data=Page_InitRec_Form_Data(metadata={}, person={}, company={}), locked=False, ) logger_gui.debug("[Page -- NewInitRec] State to call: %s", req_state) @@ -3009,29 +3023,29 @@ class Page_NewEntry(QWidget): pass -CONFIG_GRUNDERFASSUNG_UNTERNEHMEN: Final[Page_InitRec_Form_Config] = Page_InitRec_Form_Config( - pydantic_model=Grunderfassung_Unternehmen, - # to_db=backend.initrec_company_to_db, # TODO remove - # from_db=backend.initrec_company_from_db, # TODO remove - ignored_keys=( - "Metadaten_erstellung", - "Metadaten_aktualisierung", - ), - form_fields=INITREC_COMP, - add_buttons=False, -) +# CONFIG_GRUNDERFASSUNG_UNTERNEHMEN: Final[Page_InitRec_Form_Config] = Page_InitRec_Form_Config( +# pydantic_model=Grunderfassung_Unternehmen, +# # to_db=backend.initrec_company_to_db, # TODO remove +# # from_db=backend.initrec_company_from_db, # TODO remove +# ignored_keys=( +# "Metadaten_erstellung", +# "Metadaten_aktualisierung", +# ), +# form_fields=INITREC_COMP, +# add_buttons=False, +# ) -CONFIG_GRUNDERFASSUNG_PERSONEN: Final[Page_InitRec_Form_Config] = Page_InitRec_Form_Config( - pydantic_model=Grunderfassung_Personen, - # to_db=backend.initrec_person_to_db, # TODO remove - # from_db=backend.initrec_person_from_db, # TODO remove - ignored_keys=( - "Metadaten_erstellung", - "Metadaten_aktualisierung", - ), - form_fields=INITREC_PERSON, - add_buttons=False, -) +# CONFIG_GRUNDERFASSUNG_PERSONEN: Final[Page_InitRec_Form_Config] = Page_InitRec_Form_Config( +# pydantic_model=Grunderfassung_Personen, +# # to_db=backend.initrec_person_to_db, # TODO remove +# # from_db=backend.initrec_person_from_db, # TODO remove +# ignored_keys=( +# "Metadaten_erstellung", +# "Metadaten_aktualisierung", +# ), +# form_fields=INITREC_PERSON, +# add_buttons=False, +# ) CUSTOM_WIDGETS: Final[dict[str, type[CustomFormWidget]]] = { "grunderfassung_unternehmen_suche": Grunderfassung_SuchWidget, @@ -3128,12 +3142,12 @@ class Page_InitRec(QWidget): back_btn_step.setMinimumWidth(200) back_btn_step.setMaximumWidth(200) - if self.STATE.record_type is RecordingType.COMPANY: + if self.STATE.recording_type is RecordingType.WITH_COMPANY: title = QLabel("Grunderfassung Unternehmen") - elif self.STATE.record_type is RecordingType.PERSON: + elif self.STATE.recording_type is RecordingType.WITHOUT_COMPANY: title = QLabel("Grunderfassung Individualperson") else: - raise RuntimeError(f"Unspecified recording type {self.STATE.record_type}") + raise RuntimeError(f"Unspecified recording type {self.STATE.recording_type}") title.setStyleSheet("font-size: 20px; font-weight: bold;") header_layout.setSpacing(5) @@ -3142,6 +3156,64 @@ class Page_InitRec(QWidget): header_layout.addWidget(title) vert_layout.addWidget(header_container) + if DEBUG: + separator1 = QFrame() + separator1.setFrameShape(QFrame.Shape.HLine) + separator1.setFrameShadow(QFrame.Shadow.Sunken) + header_layout.addWidget(separator1) + + btn_gui2state = QPushButton("Sync GUI to state") + btn_gui2state.clicked.connect(self._sync_GUI_to_state) + btn_gui2state.setFixedHeight(35) + btn_gui2state.setSizePolicy( + QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed + ) + header_layout.addWidget(btn_gui2state) + + btn_state2gui = QPushButton("Sync state to GUI") + btn_state2gui.clicked.connect(self._sync_state_to_GUI) + btn_state2gui.setFixedHeight(35) + btn_state2gui.setSizePolicy( + QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed + ) + header_layout.addWidget(btn_state2gui) + + # self.test_button = QPushButton("Initialisiere Laden") + # self.test_button.clicked.connect(self._load_from_id) + # self.test_button.setFixedHeight(35) + # self.test_button.setSizePolicy( + # QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed + # ) + # header_layout.addWidget(self.test_button) + + # button_get = QPushButton("GET DATA") + # button_get.setFixedHeight(35) + # button_get.clicked.connect(self._get_form_data) + # header_layout.addWidget(button_get) + # button_set = QPushButton("SET DATA") + # button_set.setFixedHeight(35) + # button_set.clicked.connect(self._set_form_data) + # header_layout.addWidget(button_set) + + # id_field_layout = QHBoxLayout() + # id_field_layout.setContentsMargins(0, 0, 0, 0) + # id_field_layout.setSpacing(5) + # id_field_label = QLabel("ID Datenbank:") + # self.id_field_input = QLineEdit() + # id_field_layout.addWidget(id_field_label) + # id_field_layout.addWidget(self.id_field_input) + # header_layout.addLayout(id_field_layout) + # button_db_index = QPushButton("Setze DB Index") + # button_db_index.setFixedHeight(35) + # button_db_index.clicked.connect(self._set_db_index) + # header_layout.addWidget(button_db_index) + + separator2 = QFrame() + separator2.setFrameShape(QFrame.Shape.HLine) + separator2.setFrameShadow(QFrame.Shadow.Sunken) + header_layout.addWidget(separator2) + header_layout.addSpacing(5) + # --- BANNER --- self.info_banners_container = QWidget() self.info_banners: list[InfoBanner] = [] @@ -3183,35 +3255,29 @@ class Page_InitRec(QWidget): vert_layout.addWidget(scroll_area) - container_layout = QVBoxLayout(container) - container_layout.setContentsMargins(0, 0, 0, 0) + self.container_layout = QVBoxLayout(container) + self.container_layout.setContentsMargins(0, 0, 0, 0) - # --- AUTO FORM LAYOUT 1 --- + # --- AUTO FORM LAYOUT --- # now we try to render the layout within the collapsable box - container_layout.addSpacing(20) + self.container_layout.addSpacing(20) - self.initrec_form_company: InitRecForm | None = None - if self.STATE.record_type is RecordingType.COMPANY: - # add company - pass + # ** person + self.initrec_form_person_box = CollapsibleBox("Angaben zur Person") + ent_type = EntityType.PERSON + self.initrec_form_person = InitRecForm(self.STATE.initrec_form_states[ent_type]) + + assert self.STATE.initrec_form_states[ent_type] is self.initrec_form_person.STATE - box1 = CollapsibleBox("Angaben zur Person") - initrec_form_state_person = Page_InitRec_Form_State( - session=self.STATE.session, - cfg=CONFIG_GRUNDERFASSUNG_PERSONEN, - ids=self.STATE.ids, - recording_type=RecordingType.PERSON, - locked=False, - ) - self.initrec_form_person = InitRecForm(initrec_form_state_person) self.STATE.child_modules.append(self.initrec_form_person) - box1.setContentWidget(self.initrec_form_person) - container_layout.addWidget(box1) + self.initrec_form_person_box.setContentWidget(self.initrec_form_person) + self.container_layout.addWidget(self.initrec_form_person_box) # container_layout.addSpacing(15) - self.initrec_form_person.update_triggered.connect( - lambda: self.update_triggered.emit() - ) - self.initrec_form_person.update_triggered.connect(self._auto_form_updated) + # TODO trigger needed? + # self.initrec_form_person.update_triggered.connect( + # lambda: self.update_triggered.emit() + # ) + # self.initrec_form_person.update_triggered.connect(self._auto_form_updated) # --- CUSTOM LOGIC --- # ** 'Bundesland' only if 'Inland' selected in 'Stammdaten' @@ -3233,7 +3299,68 @@ class Page_InitRec(QWidget): self.selection_county.setProperty("styleClass", "stempel") self.selection_county.setEnabled(False) - container_layout.addStretch() + # ** company + ent_type = EntityType.COMPANY + self.initrec_form_company_box = CollapsibleBox("Angaben zum Unternehmen") + self.initrec_form_company = InitRecForm(self.STATE.initrec_form_states[ent_type]) + self.STATE.child_modules.append(self.initrec_form_company) + + assert self.STATE.initrec_form_states[ent_type] is self.initrec_form_company.STATE + + target_idx = self.container_layout.indexOf(self.initrec_form_person_box) - 1 + self.initrec_form_company_box.setContentWidget(self.initrec_form_company) + self.container_layout.insertWidget(target_idx, self.initrec_form_company_box) + + # container_layout.addSpacing(15) + # TODO trigger needed? + # self.initrec_form_person.update_triggered.connect( + # lambda: self.update_triggered.emit() + # ) + # self.initrec_form_person.update_triggered.connect(self._auto_form_updated) + + # --- CUSTOM LOGIC --- + # ** fill 'Kontaktperson -> Namen Unternehmen' + search_res = search_widgets_by_key( + self.initrec_form_company.widget_registry, "Partnersuche" + ) + assert len(search_res) == 1 + search_widget = cast(Grunderfassung_SuchWidget, search_res[0]["widget"]) + self.search_widget_trigger = cast( + QLineEdit, search_widget.company_widgets["ma_unternehmensname"] + ) + + self.container_layout.addStretch() + self._update_layout() + + def _update_layout(self) -> None: + if self.STATE.recording_type is RecordingType.WITH_COMPANY: + # add company + self.initrec_form_company_box.setVisible(True) + else: + self.initrec_form_company_box.setVisible(False) + + # elif self.initrec_form_company_box is not None: + # self.container_layout.removeWidget(self.initrec_form_company_box) + # # ** state + # assert self.initrec_form_company + # self.STATE.child_modules.remove(self.initrec_form_company) + # del self.STATE.initrec_form_states[EntityType.COMPANY] + + # self.initrec_form_company_box = None + # self.initrec_form_company = None + + # def _set_new_initrec_state( + # self, + # ent_type: EntityType, + # ) -> None: + # initrec_form_state = Page_InitRec_Form_State( + # session=self.STATE.session, + # cfg=INITREC_CONFIGS[ent_type], + # ids=self.STATE.ids, + # ent_type=ent_type, + # locked=False, + # ) + # self.STATE.initrec_form_states[ent_type] = initrec_form_state # ** custom logic def _custom_county_selection(self, idx: int) -> None: @@ -3328,18 +3455,28 @@ class Page_InitRec(QWidget): # TODO staes for form in self.STATE.child_modules: form.reset_form() - self.STATE.ids = NewEntityIds() + self.STATE.ids = EntityIds() self.STATE.geloescht = False def _load_database( self, - ids: NewEntityIds, + ids: EntityIds, ) -> None: logger_initrecform.info("[Page -- InitRec] Load data method...") logger_initrecform.debug("[Page -- InitRec] Entity IDs: %s", ids) self._reset() - # TODO add loading logic - initrec: InitRec = ... + # TODO add loading logic with backend call + # !! just a dummy + initrec: InitRec = InitRec( + ids=ids, + recording_type=RecordingType.WITHOUT_COMPANY, + geloescht=None, + form_data=Page_InitRec_Form_Data( + metadata={}, + person={}, + company={}, + ), + ) self._load_from_pydantic(initrec, sync_to_GUI=False) @@ -3349,7 +3486,23 @@ class Page_InitRec(QWidget): update_only: bool = False, sync_to_GUI: bool = True, ) -> None: + logger_initrecform.info("[Page -- InitRec] Load from Pydantic model...") # TODO add logic to update internal state + self.STATE.recording_type = data.recording_type + self.STATE.form_data = data.form_data + self.STATE.geloescht = True if data.geloescht else False + + # set form data to correct initrec form states + for ent_type, state in self.STATE.initrec_form_states.items(): + match ent_type: + case EntityType.COMPANY: + state.form_data = self.STATE.form_data.company + # self.initrec_form_company.load_state(state) + case EntityType.PERSON: + state.form_data = self.STATE.form_data.person + # self.initrec_form_person.load_state(state) + case _: + raise RuntimeError(f"Unspecified entity type {ent_type}") if update_only: ... @@ -3368,19 +3521,20 @@ class Page_InitRec(QWidget): ) if section is None or section == "content": + self._update_layout() if not self.STATE.geloescht: self._load_database(self.STATE.ids) else: self._reset() - initrec_form_state_person = Page_InitRec_Form_State( - session=self.STATE.session, - cfg=CONFIG_GRUNDERFASSUNG_PERSONEN, - ids=self.STATE.ids, - recording_type=RecordingType.PERSON, - locked=False, - ) - self.initrec_form_person.load_state(initrec_form_state_person) + # initrec_form_state_person = Page_InitRec_Form_State( + # session=self.STATE.session, + # cfg=CONFIG_GRUNDERFASSUNG_PERSONEN, + # ids=self.STATE.ids, + # ent_type=RecordingType.WITHOUT_COMPANY, + # locked=False, + # ) + # self.initrec_form_person.load_state(initrec_form_state_person) if section is None or section == "banners": self._clear_info_banners() @@ -3401,22 +3555,65 @@ class Page_InitRec(QWidget): form.unlock() def _sync_GUI_to_state(self) -> None: + logger_page_initrec.debug("[Page -- InitRec] Syncing GUI to state...") + + logger_page_initrec.debug( + "[Page -- InitRec] Syncing GUI to state: Child modules...\n%s", + pformat(self.STATE.child_modules), + ) + + logger_page_initrec.debug( + "[Page -- InitRec] Syncing GUI to state: Existing form states keys...\n%s", + pformat(self.STATE.initrec_form_states.keys()), + ) + for form in self.STATE.child_modules: # ?? is this the same reference to use a similar pattern as # ?? the consulting page (initialise once with the same state # ?? as reference which are then automatically updated) + # !! check if this is working as expected + # new entries' states are linked to this instance by the page state form_state = form.get_state() - self.STATE.ids.update(form_state.ids, ignore_none=True) - self.STATE.form_data[form_state.recording_type] = form_state.form_data - self.STATE.pydantic_models[form_state.recording_type] = ( - form_state.cfg.pydantic_model + logger_page_initrec.debug( + "[Page -- InitRec] Syncing GUI to state: Form state ent_type %s", + form_state.ent_type, ) + logger_page_initrec.debug( + "[Page -- InitRec] Syncing GUI to state: identity of form states: %s", + (self.STATE.initrec_form_states[form_state.ent_type] is form_state), + ) + assert self.STATE.initrec_form_states[form_state.ent_type] is form_state, ( + "passing by reference not working" + ) + self.STATE.ids.update(form_state.ids, ignore_none=True) + # self.STATE.form_data[form_state.ent_type] = form_state.form_data + match form_state.ent_type: + case EntityType.COMPANY: + self.STATE.form_data.company = form_state.form_data + case EntityType.PERSON: + self.STATE.form_data.person = form_state.form_data + case _: + raise RuntimeError(f"Unspecified entity type {form_state.ent_type}") + + logger_page_initrec.debug( + "[Page -- InitRec] Syncing GUI to state: IDs after...\n%s", + pformat(self.STATE.ids), + ) + logger_page_initrec.debug( + "[Page -- InitRec] Syncing GUI to state: Form data after...\n%s", + pformat(self.STATE.form_data), + ) + def load_state( self, new_state: Page_InitRec_State, ) -> None: - set_page_state(self.STATE, new_state) + set_page_state( + self.STATE, + new_state, + ignore_attr={"form_data", "initrec_form_states"}, + ) self._sync_state_to_GUI(None) def get_state(self) -> Page_InitRec_State: @@ -3444,6 +3641,8 @@ class Page_InitRec(QWidget): def save_data(self) -> None: logger_page_initrec.debug("[Page -- InitRec] Save button clicked...") + return + if self.STATE.ids.empty() and self.STATE.geloescht: logger_page_initrec.debug( "[Page -- InitRec] This was never saved in the database " @@ -3460,7 +3659,7 @@ class Page_InitRec(QWidget): to_db = InitRec( ids=self.STATE.ids, geloescht=deleted_datetime, - db_data={}, + form_data={}, ) returned = self.cfg.to_db(to_db) self._load_from_pydantic(returned) @@ -3520,7 +3719,7 @@ class Page_InitRec(QWidget): ids=self.STATE.ids, # rec_id=self.STATE.rec_id, geloescht=deleted_datetime, - db_data=db_data, + form_data=db_data, ) returned = self.cfg.to_db(to_db) @@ -5939,8 +6138,9 @@ class MainWindow(QMainWindow): # SITE: 'Grunderfassung' newinitrec_state = Page_InitRec_State( session=session, - ids=NewEntityIds(), - record_type=RecordingType.COMPANY, + ids=EntityIds(), + recording_type=RecordingType.WITH_COMPANY, + form_data=Page_InitRec_Form_Data(metadata={}, person={}, company={}), locked=False, ) self.newinitrec = Page_InitRec(newinitrec_state) @@ -5973,7 +6173,7 @@ class MainWindow(QMainWindow): # self.stack.addWidget(self.initrec_person) # SITE: 'Unternehmensprofil' comp_profile_state = Page_CompanyProfile_State( - session=self.STATE.session, ids=NewEntityIds() + session=self.STATE.session, ids=EntityIds() ) self.company_profile = Page_CompanyProfile(comp_profile_state) self.company_profile.back_main_requested.connect(self.show_main_page) diff --git a/src/wce_crm/types.py b/src/wce_crm/types.py index c53072f..c24c900 100644 --- a/src/wce_crm/types.py +++ b/src/wce_crm/types.py @@ -5,7 +5,7 @@ import dataclasses as dc import datetime import enum from collections.abc import Generator -from typing import TYPE_CHECKING, TypeAlias, TypedDict +from typing import TYPE_CHECKING, Final, TypeAlias, TypedDict if TYPE_CHECKING: import polars as pl @@ -24,6 +24,12 @@ PolarsSchema: TypeAlias = dict[str, type["pl.DataType"]] # TODO rename: these are not entity types or entities but different forms of # initial recordings class RecordingType(enum.IntEnum): + WITH_COMPANY = enum.auto() + WITHOUT_COMPANY = enum.auto() + + +class EntityType(enum.IntEnum): + METADATA = enum.auto() COMPANY = enum.auto() PERSON = enum.auto() @@ -34,19 +40,36 @@ class ConsultingType(enum.StrEnum): @dc.dataclass(slots=True, kw_only=True) -class NewEntityIds: +class EntityIds: un_id: ExtMaId | None = None an_id: ExtAnId | None = None pers_id: RecId | None = None link_id: IntLinkId | None = None + _field_names: frozenset[str] = dc.field(init=False, repr=False, compare=False) + + def __post_init__(self) -> None: + self._field_names = frozenset(f.name for f in dc.fields(EntityIds)) + + def set_by_name( + self, + attr_name: str, + value: int | None, + ) -> None: + if attr_name not in self._field_names: + raise KeyError( + f"Attribute {attr_name} not member of this class. Valid " + f"options: {self._field_names}" + ) + + setattr(self, attr_name, value) def update( self, - ids: NewEntityIds, + ids: EntityIds, ignore_none: bool = False, deep: bool = False, ) -> None: - if not isinstance(ids, NewEntityIds): + if not isinstance(ids, EntityIds): raise TypeError("New IDs must be have the same type as this object") for field in dc.fields(ids): @@ -62,18 +85,18 @@ class NewEntityIds: def _get_relevant_fields( self, - recording_type: RecordingType | None = None, + ent_type: EntityType | None = None, ) -> Generator[int | None, None, None]: - if recording_type is None: + if ent_type is None: relevant_fields = (getattr(self, field.name) for field in dc.fields(self)) - elif recording_type is RecordingType.COMPANY: + elif ent_type is EntityType.COMPANY: RELEVANT_ATTRIBUTES = {"un_id", "an_id", "link_id"} relevant_fields = ( getattr(self, field.name) for field in dc.fields(self) if field.name in RELEVANT_ATTRIBUTES ) - elif recording_type is RecordingType.PERSON: + elif ent_type is EntityType.PERSON: RELEVANT_ATTRIBUTES = {"pers_id"} relevant_fields = ( getattr(self, field.name) @@ -81,23 +104,23 @@ class NewEntityIds: if field.name in RELEVANT_ATTRIBUTES ) else: - raise RuntimeError(f"Unspecified recording type {recording_type}") + raise RuntimeError(f"Unspecified entity type {ent_type}") return relevant_fields def empty( self, - recording_type: RecordingType | None = None, + ent_type: EntityType | None = None, ) -> bool: - relevant_fields = self._get_relevant_fields(recording_type=recording_type) + relevant_fields = self._get_relevant_fields(ent_type=ent_type) return not any(relevant_fields) def valid( self, - recording_type: RecordingType | None = None, + ent_type: EntityType | None = None, ) -> bool: - relevant_fields = self._get_relevant_fields(recording_type=recording_type) + relevant_fields = self._get_relevant_fields(ent_type=ent_type) return all(relevant_fields)