generated from dopt-python/py311
Fix data model #4
+113
-4
@@ -19,13 +19,122 @@ from PySide6.QtCore import QDate, Qt
|
||||
|
||||
from wce_crm import constants, db
|
||||
from wce_crm.backend import backend
|
||||
from wce_crm.types import EntityIds, RecordingType
|
||||
from wce_crm.types import EntityIds, MainPageEntry, RecordingType
|
||||
|
||||
# %%
|
||||
# backend.page_consulting_linking_companies(None)
|
||||
# backend.page_consulting_linking_persons(3)
|
||||
ret = backend.main_page_get_entries()
|
||||
pprint(ret)
|
||||
# %%
|
||||
with db.ENGINE.connect() as conn:
|
||||
q_relevant_entries = (
|
||||
sa.select(
|
||||
db.t_grunderfassung_personen.c.pers_id,
|
||||
db.t_grunderfassung_personen.c.aktualisiert,
|
||||
db.t_zuordnung_personen_unternehmen.c.id.label("link_id"),
|
||||
db.t_zuordnung_personen_unternehmen.c.un_id,
|
||||
db.t_zuordnung_personen_unternehmen.c.an_id,
|
||||
db.t_grunderfassung_personen.c.Stammdaten__name,
|
||||
db.t_grunderfassung_personen.c.Stammdaten__vorname,
|
||||
db.t_grunderfassung_personen.c.Stammdaten__ort,
|
||||
)
|
||||
.outerjoin(
|
||||
db.t_zuordnung_personen_unternehmen,
|
||||
db.t_grunderfassung_personen.c.pers_id
|
||||
== db.t_zuordnung_personen_unternehmen.c.pers_id,
|
||||
)
|
||||
.order_by(db.t_grunderfassung_personen.c.Stammdaten__name)
|
||||
).subquery("relevant_entries")
|
||||
|
||||
stmt = sa.select(
|
||||
q_relevant_entries,
|
||||
db.t_ext_crm_master_attach.c.ma_unternehmensname,
|
||||
).outerjoin(
|
||||
db.t_ext_crm_master_attach,
|
||||
q_relevant_entries.c.un_id == db.t_ext_crm_master_attach.c.ma_id,
|
||||
)
|
||||
|
||||
ret = conn.execute(stmt)
|
||||
|
||||
# results = list(ret.mappings())
|
||||
# print(f"Results are:\n{pformat(results)}")
|
||||
|
||||
main_page_entries: list[MainPageEntry] = []
|
||||
for entry in ret.mappings():
|
||||
ids = EntityIds(
|
||||
pers_id=entry["pers_id"],
|
||||
link_id=entry["link_id"],
|
||||
un_id=entry["un_id"],
|
||||
an_id=entry["an_id"],
|
||||
)
|
||||
recording_type = RecordingType.WITHOUT_COMPANY
|
||||
if ids.un_id or ids.link_id:
|
||||
recording_type = RecordingType.WITH_COMPANY
|
||||
|
||||
first_name = entry["Stammdaten__vorname"]
|
||||
last_name = entry["Stammdaten__name"]
|
||||
location_city = entry["Stammdaten__ort"]
|
||||
person_name = f"{first_name} {last_name}"
|
||||
if location_city:
|
||||
person_name += f" (Ort: {location_city})"
|
||||
|
||||
company_name = entry["ma_unternehmensname"]
|
||||
|
||||
display_name = person_name
|
||||
if recording_type is RecordingType.WITH_COMPANY:
|
||||
display_name = company_name
|
||||
|
||||
main_page_entries.append(
|
||||
MainPageEntry(
|
||||
ids=ids,
|
||||
display_name=display_name,
|
||||
dt_updated=entry["aktualisiert"],
|
||||
recording_type=recording_type,
|
||||
)
|
||||
)
|
||||
main_page_entries.sort(key=lambda x: x.dt_updated, reverse=True)
|
||||
pprint(main_page_entries)
|
||||
# %%
|
||||
|
||||
|
||||
# %%
|
||||
entry = results[0]
|
||||
ids = EntityIds(
|
||||
pers_id=entry["pers_id"],
|
||||
link_id=entry["link_id"],
|
||||
un_id=entry["un_id"],
|
||||
an_id=entry["an_id"],
|
||||
)
|
||||
recording_type = RecordingType.WITHOUT_COMPANY
|
||||
if ids.un_id or ids.link_id:
|
||||
recording_type = RecordingType.WITH_COMPANY
|
||||
|
||||
first_name = entry["Stammdaten__vorname"]
|
||||
last_name = entry["Stammdaten__name"]
|
||||
location_city = entry["Stammdaten__ort"]
|
||||
person_name = f"{first_name} {last_name}"
|
||||
if location_city:
|
||||
person_name += f" (Ort: {location_city})"
|
||||
|
||||
company_name = entry["ma_unternehmensname"]
|
||||
|
||||
display_name = person_name
|
||||
if recording_type is RecordingType.WITH_COMPANY:
|
||||
display_name = company_name
|
||||
|
||||
main_page_entry = MainPageEntry(
|
||||
ids=ids,
|
||||
display_name=display_name,
|
||||
dt_updated=entry["aktualisiert"],
|
||||
recording_type=recording_type,
|
||||
)
|
||||
|
||||
# %%
|
||||
pprint(main_page_entry)
|
||||
|
||||
|
||||
# %%
|
||||
# %%
|
||||
@dc.dataclass(slots=True, kw_only=True)
|
||||
class LinkingPerson:
|
||||
@@ -74,13 +183,13 @@ with db.ENGINE.connect() as conn:
|
||||
last_name = row.Stammdaten__name
|
||||
location_city = row.Stammdaten__ort
|
||||
|
||||
display_name = f"{first_name} {last_name}"
|
||||
person_name = f"{first_name} {last_name}"
|
||||
if location_city:
|
||||
display_name += f" (Ort: {location_city})"
|
||||
person_name += f" (Ort: {location_city})"
|
||||
|
||||
persons_map[p_id] = {
|
||||
"pers_id": p_id,
|
||||
"display_name": display_name,
|
||||
"display_name": person_name,
|
||||
"un_ids": set(),
|
||||
}
|
||||
|
||||
|
||||
+65
-388
@@ -134,232 +134,6 @@ def initrec_comp_contact_person_search_get_info(
|
||||
|
||||
|
||||
# // internals
|
||||
# TODO check removal
|
||||
# def initrec_company_insert_initial_recording(
|
||||
# data: dict[str, Any],
|
||||
# ) -> RecId:
|
||||
# logger.debug("[Call backend] insert_initial_recording")
|
||||
# stmt = db.grunderfassung_unternehmen.insert()
|
||||
# with db.ENGINE.begin() as conn:
|
||||
# ret = conn.execute(stmt, data)
|
||||
|
||||
# if ret.rowcount == 0:
|
||||
# raise IOError("Entry was not inserted correctly")
|
||||
|
||||
# prim_keys = ret.inserted_primary_key
|
||||
# assert prim_keys
|
||||
|
||||
# return prim_keys[0]
|
||||
|
||||
|
||||
# def initrec_company_update_initial_recording(
|
||||
# id_: RecId,
|
||||
# data: dict[str, Any],
|
||||
# ) -> None:
|
||||
# logger.debug("[Call backend] update_initial_recording")
|
||||
# stmt = db.grunderfassung_unternehmen.update().where(
|
||||
# db.grunderfassung_unternehmen.c.un_id == id_
|
||||
# )
|
||||
# with db.ENGINE.begin() as conn:
|
||||
# conn.execute(stmt, data)
|
||||
|
||||
|
||||
# def initrec_company_to_db(
|
||||
# auto_form_data: InitRec,
|
||||
# ) -> InitRec:
|
||||
# logger.debug("[AutoForm -- backend] Call database saving routine...")
|
||||
|
||||
# dump_data = copy.deepcopy(auto_form_data.db_data)
|
||||
# dump_data["geloescht"] = auto_form_data.geloescht
|
||||
|
||||
# with db.ENGINE.begin() as conn:
|
||||
# if auto_form_data.ids.un_id is None:
|
||||
# logger.debug("[AutoForm -- backend] Insert...")
|
||||
|
||||
# stmt = db.grunderfassung_unternehmen.insert().returning(
|
||||
# db.grunderfassung_unternehmen.c.un_id,
|
||||
# db.grunderfassung_unternehmen.c.Metadaten_aktualisierung,
|
||||
# db.grunderfassung_unternehmen.c.geloescht,
|
||||
# )
|
||||
# ret = conn.execute(stmt, dump_data)
|
||||
|
||||
# from_db = ret.mappings().fetchall()
|
||||
# assert len(from_db) == 1, "expected excactly one returned row"
|
||||
# data_from_db = from_db[0]
|
||||
# auto_form_data.ids.un_id = data_from_db["un_id"]
|
||||
# auto_form_data.geloescht = data_from_db["geloescht"]
|
||||
# auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[
|
||||
# "Metadaten_aktualisierung"
|
||||
# ]
|
||||
|
||||
# logger.debug("[AutoForm -- backend] Inserted InitRec Company successfully")
|
||||
|
||||
# else:
|
||||
# logger.debug("[AutoForm -- backend] Update...")
|
||||
# stmt = (
|
||||
# db.grunderfassung_unternehmen.update()
|
||||
# .where(db.grunderfassung_unternehmen.c.un_id == auto_form_data.ids.un_id)
|
||||
# .returning(
|
||||
# db.grunderfassung_unternehmen.c.Metadaten_aktualisierung,
|
||||
# db.grunderfassung_unternehmen.c.geloescht,
|
||||
# )
|
||||
# )
|
||||
# ret = conn.execute(stmt, dump_data)
|
||||
|
||||
# from_db = ret.mappings().fetchall()
|
||||
# assert len(from_db) == 1, "expected excatly one returned row"
|
||||
# data_from_db = from_db[0]
|
||||
|
||||
# auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[
|
||||
# "Metadaten_aktualisierung"
|
||||
# ]
|
||||
|
||||
# logger.debug(
|
||||
# "[AutoForm -- backend] Updated InitRec Company with ID %d successfully",
|
||||
# auto_form_data.ids.un_id,
|
||||
# )
|
||||
|
||||
# return auto_form_data
|
||||
|
||||
|
||||
# def initrec_company_from_db(
|
||||
# id_: RecId,
|
||||
# ) -> InitRec:
|
||||
# logger.debug("[AutoForm -- backend] Call database loading routine...")
|
||||
# stmt = db.grunderfassung_unternehmen.select().where(
|
||||
# db.grunderfassung_unternehmen.c.un_id == id_
|
||||
# )
|
||||
# with db.ENGINE.connect() as conn:
|
||||
# ret = conn.execute(stmt)
|
||||
|
||||
# from_db = ret.mappings().all()
|
||||
# assert len(from_db) == 1, "not excactly one company initial recording obtained"
|
||||
# data_from_db = dict(from_db[0])
|
||||
|
||||
# geloescht = data_from_db["geloescht"]
|
||||
# del data_from_db["geloescht"]
|
||||
|
||||
# return InitRec(
|
||||
# ids=EntityIds(un_id=id_),
|
||||
# # rec_id=id_, # TODO remove
|
||||
# geloescht=geloescht,
|
||||
# db_data=data_from_db,
|
||||
# )
|
||||
|
||||
|
||||
# def initrec_person_to_db(
|
||||
# auto_form_data: InitRec,
|
||||
# ) -> InitRec:
|
||||
# logger.debug("[AutoForm -- backend] Call database saving routine...")
|
||||
|
||||
# dump_data = copy.deepcopy(auto_form_data.db_data)
|
||||
# dump_data["geloescht"] = auto_form_data.geloescht
|
||||
|
||||
# with db.ENGINE.begin() as conn:
|
||||
# if auto_form_data.ids.pers_id is None:
|
||||
# logger.debug("[AutoForm -- backend] Insert...")
|
||||
|
||||
# stmt = sa.insert(db.t_grunderfassung_personen).returning(
|
||||
# db.t_grunderfassung_personen.c.pers_id,
|
||||
# db.t_grunderfassung_personen.c.Metadaten_aktualisierung,
|
||||
# db.t_grunderfassung_personen.c.geloescht,
|
||||
# )
|
||||
# stmt_compiled = str(stmt.compile(db.ENGINE))
|
||||
# logger.debug(
|
||||
# "[AutoForm -- backend] Data to insert:\nStatement: %s\n%s",
|
||||
# stmt_compiled,
|
||||
# pformat(dump_data),
|
||||
# )
|
||||
|
||||
# ret = conn.execute(stmt, dump_data)
|
||||
|
||||
# from_db = ret.mappings().fetchall()
|
||||
# assert len(from_db) == 1, "expected excactly one returned row"
|
||||
# data_from_db = from_db[0]
|
||||
# auto_form_data.ids.pers_id = data_from_db["pers_id"]
|
||||
# auto_form_data.geloescht = data_from_db["geloescht"]
|
||||
# auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[
|
||||
# "Metadaten_aktualisierung"
|
||||
# ]
|
||||
|
||||
# logger.debug("[AutoForm -- backend] Inserted InitRec Person successfully")
|
||||
|
||||
# else:
|
||||
# logger.debug("[AutoForm -- backend] Update...")
|
||||
# stmt = (
|
||||
# db.t_grunderfassung_personen.update()
|
||||
# .where(db.t_grunderfassung_personen.c.pers_id == auto_form_data.ids.pers_id)
|
||||
# .returning(
|
||||
# db.t_grunderfassung_personen.c.Metadaten_aktualisierung,
|
||||
# db.t_grunderfassung_personen.c.geloescht,
|
||||
# )
|
||||
# )
|
||||
# ret = conn.execute(stmt, dump_data)
|
||||
|
||||
# from_db = ret.mappings().fetchall()
|
||||
# assert len(from_db) == 1, "expected excatly one returned row"
|
||||
# data_from_db = from_db[0]
|
||||
|
||||
# auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[
|
||||
# "Metadaten_aktualisierung"
|
||||
# ]
|
||||
|
||||
# logger.debug(
|
||||
# "[AutoForm -- backend] Updated InitRec Person with ID %d successfully",
|
||||
# auto_form_data.ids.pers_id,
|
||||
# )
|
||||
|
||||
# return auto_form_data
|
||||
|
||||
|
||||
# def initrec_person_from_db(
|
||||
# id_: RecId,
|
||||
# ) -> InitRec:
|
||||
# logger.debug("[AutoForm -- backend] Call database loading routine...")
|
||||
# stmt = db.t_grunderfassung_personen.select().where(
|
||||
# db.t_grunderfassung_personen.c.pers_id == id_
|
||||
# )
|
||||
# with db.ENGINE.connect() as conn:
|
||||
# ret = conn.execute(stmt)
|
||||
|
||||
# from_db = ret.mappings().all()
|
||||
# assert len(from_db) == 1, "not excactly one company initial recording obtained"
|
||||
# data_from_db = dict(from_db[0])
|
||||
|
||||
# geloescht = data_from_db["geloescht"]
|
||||
# del data_from_db["geloescht"]
|
||||
|
||||
# return InitRec(
|
||||
# ids=EntityIds(pers_id=id_),
|
||||
# # rec_id=id_, # TODO remove
|
||||
# geloescht=geloescht,
|
||||
# db_data=data_from_db,
|
||||
# )
|
||||
|
||||
|
||||
# TODO check removal
|
||||
# def initrec_company_get_initial_recording(
|
||||
# id_: RecId,
|
||||
# ) -> dict[str, Any]:
|
||||
# logger.debug("[Call backend] get_initial_recording")
|
||||
# stmt = db.grunderfassung_unternehmen.select().where(
|
||||
# db.grunderfassung_unternehmen.c.un_id == id_
|
||||
# )
|
||||
# with db.ENGINE.connect() as conn:
|
||||
# ret = conn.execute(stmt)
|
||||
|
||||
# results = ret.mappings().all()
|
||||
# if not results:
|
||||
# raise KeyError(f"Database ID {id_} not found")
|
||||
|
||||
# assert len(results) == 1, "more than one company initial recording obtained"
|
||||
|
||||
# row = results[0]
|
||||
# assert row, "row was not obtained"
|
||||
|
||||
# return dict(row)
|
||||
|
||||
|
||||
def split_dict(
|
||||
dic: dict[K, V],
|
||||
split_keys: Container[K],
|
||||
@@ -612,85 +386,6 @@ def page_initrec_from_db(
|
||||
)
|
||||
|
||||
|
||||
# def initrec_company_delete_initial_recording(
|
||||
# id_: RecId,
|
||||
# ) -> None:
|
||||
# logger.debug("[Call backend] delete_initial_recording")
|
||||
# stmt = db.grunderfassung_unternehmen.delete().where(
|
||||
# db.grunderfassung_unternehmen.c.un_id == id_
|
||||
# )
|
||||
# with db.ENGINE.begin() as conn:
|
||||
# ret = conn.execute(stmt)
|
||||
|
||||
# if ret.rowcount == 0:
|
||||
# raise KeyError(f"Database ID {id_} not found for deletion")
|
||||
|
||||
|
||||
# def initrec_person_insert_initial_recording(
|
||||
# data: dict[str, Any],
|
||||
# ) -> RecId:
|
||||
# logger.debug("[Call backend] insert_initial_recording")
|
||||
# stmt = db.grunderfassung_personen.insert()
|
||||
# with db.ENGINE.begin() as conn:
|
||||
# ret = conn.execute(stmt, data)
|
||||
|
||||
# if ret.rowcount == 0:
|
||||
# raise IOError("Entry was not inserted correctly")
|
||||
|
||||
# prim_keys = ret.inserted_primary_key
|
||||
# assert prim_keys
|
||||
|
||||
# return prim_keys[0]
|
||||
|
||||
|
||||
# def initrec_person_update_initial_recording(
|
||||
# id_: RecId,
|
||||
# data: dict[str, Any],
|
||||
# ) -> None:
|
||||
# logger.debug("[Call backend] update_initial_recording")
|
||||
# stmt = db.grunderfassung_personen.update().where(
|
||||
# db.grunderfassung_personen.c.pers_id == id_
|
||||
# )
|
||||
# with db.ENGINE.begin() as conn:
|
||||
# conn.execute(stmt, data)
|
||||
|
||||
|
||||
# def initrec_person_get_initial_recording(
|
||||
# id_: RecId,
|
||||
# ) -> dict[str, Any]:
|
||||
# logger.debug("[Call backend] get_initial_recording person")
|
||||
# stmt = db.grunderfassung_personen.select().where(
|
||||
# db.grunderfassung_personen.c.pers_id == id_
|
||||
# )
|
||||
# with db.ENGINE.connect() as conn:
|
||||
# ret = conn.execute(stmt)
|
||||
|
||||
# results = ret.mappings().all()
|
||||
# if not results:
|
||||
# raise KeyError(f"Database ID {id_} not found")
|
||||
|
||||
# assert len(results) == 1, "more than one person initial recording obtained"
|
||||
|
||||
# row = results[0]
|
||||
# assert row, "row was not obtained"
|
||||
|
||||
# return dict(row)
|
||||
|
||||
|
||||
# def initrec_person_delete_initial_recording(
|
||||
# id_: RecId,
|
||||
# ) -> None:
|
||||
# logger.debug("[Call backend] delete_initial_recording")
|
||||
# stmt = db.grunderfassung_personen.delete().where(
|
||||
# db.grunderfassung_personen.c.pers_id == id_
|
||||
# )
|
||||
# with db.ENGINE.begin() as conn:
|
||||
# ret = conn.execute(stmt)
|
||||
|
||||
# if ret.rowcount == 0:
|
||||
# raise KeyError(f"Database ID {id_} not found for deletion")
|
||||
|
||||
|
||||
@wrap_result(10)
|
||||
def page_consulting_to_db(
|
||||
consultation_data: Beratungsgespraech_Vorgang,
|
||||
@@ -1031,7 +726,7 @@ def companyprofile_page_get_consultations(
|
||||
|
||||
|
||||
# // initial recording interaction
|
||||
# TODO wrap result
|
||||
# TODO wrap result, add to corresponding component
|
||||
def initrec_link_person_company(
|
||||
un_id: RecId,
|
||||
pers_id: RecId,
|
||||
@@ -1063,89 +758,71 @@ def initrec_link_person_company(
|
||||
|
||||
|
||||
# // main page interaction
|
||||
# TODO change to correct logic
|
||||
def _main_page_get_company_list() -> list[MainPageEntry]:
|
||||
logger.debug("[Call backend] get_company_list")
|
||||
|
||||
raise RuntimeError("TODOs to implement")
|
||||
|
||||
stmt = sa.select(
|
||||
db.grunderfassung_unternehmen.c.un_id,
|
||||
db.grunderfassung_unternehmen.c.Partnersuche__un_suche,
|
||||
db.grunderfassung_unternehmen.c.Metadaten_aktualisierung,
|
||||
).where(db.grunderfassung_unternehmen.c.geloescht.is_(None))
|
||||
|
||||
with db.ENGINE.connect() as conn:
|
||||
res = conn.execute(stmt)
|
||||
|
||||
main_page_companies: list[MainPageEntry] = []
|
||||
|
||||
for entry in res.mappings():
|
||||
rec_id = entry["un_id"]
|
||||
assert rec_id, "no RecID defined"
|
||||
ma_id_external = entry["Partnersuche__un_suche"]
|
||||
assert ma_id_external is not None, "external MA ID is NULL"
|
||||
datetime_akt = cast(datetime.datetime, entry["Metadaten_aktualisierung"])
|
||||
datetime_akt = datetime_akt.astimezone(TIMEZONE_CEST)
|
||||
|
||||
comp_info = external_company_get_info(ma_id_external)
|
||||
display_name = comp_info["ma_unternehmensname"]
|
||||
|
||||
main_page_companies.append(
|
||||
MainPageEntry(
|
||||
rec_id=rec_id,
|
||||
display_name=display_name,
|
||||
Metadaten_aktualisierung=datetime_akt,
|
||||
type=RecordingType.WITH_COMPANY,
|
||||
)
|
||||
)
|
||||
|
||||
return main_page_companies
|
||||
|
||||
|
||||
def _main_page_get_person_list() -> list[MainPageEntry]:
|
||||
logger.debug("[Call backend] get_company_list")
|
||||
stmt = sa.select(
|
||||
db.t_grunderfassung_personen.c.pers_id,
|
||||
db.t_grunderfassung_personen.c.Stammdaten__vorname,
|
||||
db.t_grunderfassung_personen.c.Stammdaten__name,
|
||||
db.t_grunderfassung_personen.c.Metadaten_aktualisierung,
|
||||
).where(db.t_grunderfassung_personen.c.geloescht.is_(None))
|
||||
|
||||
with db.ENGINE.connect() as conn:
|
||||
res = conn.execute(stmt)
|
||||
|
||||
main_page_persons: list[MainPageEntry] = []
|
||||
|
||||
for entry in res.mappings():
|
||||
rec_id = entry["pers_id"]
|
||||
assert rec_id, "no RecID defined"
|
||||
datetime_akt = cast(datetime.datetime, entry["Metadaten_aktualisierung"])
|
||||
datetime_akt = datetime_akt.astimezone()
|
||||
|
||||
surname: str = entry["Stammdaten__name"]
|
||||
first_name: str = entry["Stammdaten__vorname"]
|
||||
names_to_join = [n for n in (first_name, surname) if n]
|
||||
display_name = " ".join(names_to_join)
|
||||
|
||||
main_page_persons.append(
|
||||
MainPageEntry(
|
||||
rec_id=rec_id,
|
||||
display_name=display_name,
|
||||
Metadaten_aktualisierung=datetime_akt,
|
||||
type=RecordingType.WITHOUT_COMPANY,
|
||||
)
|
||||
)
|
||||
|
||||
return main_page_persons
|
||||
|
||||
|
||||
def main_page_get_entries() -> list[MainPageEntry]:
|
||||
companies = _main_page_get_company_list()
|
||||
persons = _main_page_get_person_list()
|
||||
with db.ENGINE.connect() as conn:
|
||||
q_relevant_entries = (
|
||||
sa.select(
|
||||
db.t_grunderfassung_personen.c.pers_id,
|
||||
db.t_grunderfassung_personen.c.aktualisiert,
|
||||
db.t_zuordnung_personen_unternehmen.c.id.label("link_id"),
|
||||
db.t_zuordnung_personen_unternehmen.c.un_id,
|
||||
db.t_zuordnung_personen_unternehmen.c.an_id,
|
||||
db.t_grunderfassung_personen.c.Stammdaten__name,
|
||||
db.t_grunderfassung_personen.c.Stammdaten__vorname,
|
||||
db.t_grunderfassung_personen.c.Stammdaten__ort,
|
||||
)
|
||||
.outerjoin(
|
||||
db.t_zuordnung_personen_unternehmen,
|
||||
db.t_grunderfassung_personen.c.pers_id
|
||||
== db.t_zuordnung_personen_unternehmen.c.pers_id,
|
||||
)
|
||||
.where(db.t_grunderfassung_personen.c.geloescht.is_(None))
|
||||
.order_by(db.t_grunderfassung_personen.c.Stammdaten__name)
|
||||
).subquery("relevant_entries")
|
||||
|
||||
all_entries = companies + persons
|
||||
stmt = sa.select(
|
||||
q_relevant_entries,
|
||||
db.t_ext_crm_master_attach.c.ma_unternehmensname,
|
||||
).outerjoin(
|
||||
db.t_ext_crm_master_attach,
|
||||
q_relevant_entries.c.un_id == db.t_ext_crm_master_attach.c.ma_id,
|
||||
)
|
||||
|
||||
all_entries.sort(key=lambda x: x.Metadaten_aktualisierung, reverse=True)
|
||||
ret = conn.execute(stmt)
|
||||
|
||||
return all_entries
|
||||
main_page_entries: list[MainPageEntry] = []
|
||||
for entry in ret.mappings():
|
||||
ids = EntityIds(
|
||||
pers_id=entry["pers_id"],
|
||||
link_id=entry["link_id"],
|
||||
un_id=entry["un_id"],
|
||||
an_id=entry["an_id"],
|
||||
)
|
||||
recording_type = RecordingType.WITHOUT_COMPANY
|
||||
if ids.un_id or ids.link_id:
|
||||
recording_type = RecordingType.WITH_COMPANY
|
||||
|
||||
first_name = entry["Stammdaten__vorname"]
|
||||
last_name = entry["Stammdaten__name"]
|
||||
location_city = entry["Stammdaten__ort"]
|
||||
person_name = f"{first_name} {last_name}"
|
||||
if location_city:
|
||||
person_name += f" (Ort: {location_city})"
|
||||
|
||||
company_name = entry["ma_unternehmensname"]
|
||||
|
||||
display_name = person_name
|
||||
if recording_type is RecordingType.WITH_COMPANY:
|
||||
display_name = company_name
|
||||
|
||||
main_page_entries.append(
|
||||
MainPageEntry(
|
||||
ids=ids,
|
||||
display_name=display_name,
|
||||
dt_updated=entry["aktualisiert"],
|
||||
recording_type=recording_type,
|
||||
)
|
||||
)
|
||||
main_page_entries.sort(key=lambda x: x.dt_updated, reverse=True)
|
||||
|
||||
return main_page_entries
|
||||
|
||||
@@ -199,6 +199,14 @@ class Page_InitRec_Form_Data:
|
||||
company: dict[str, Any]
|
||||
|
||||
|
||||
def _create_empty_initrec_form_data() -> Page_InitRec_Form_Data:
|
||||
return Page_InitRec_Form_Data(
|
||||
metadata={},
|
||||
person={},
|
||||
company={},
|
||||
)
|
||||
|
||||
|
||||
# !! there must be exactly one InitRecForm for each recording type
|
||||
@dc.dataclass(slots=True, kw_only=True)
|
||||
class Page_InitRec_State(PageState["InitRecForm"]):
|
||||
@@ -208,7 +216,9 @@ class Page_InitRec_State(PageState["InitRecForm"]):
|
||||
# un_id: RecId | None = None
|
||||
locked: bool
|
||||
geloescht: bool = False
|
||||
form_data: Page_InitRec_Form_Data
|
||||
form_data: Page_InitRec_Form_Data = dc.field(
|
||||
default_factory=_create_empty_initrec_form_data
|
||||
)
|
||||
initrec_form_states: dict[EntityType, Page_InitRec_Form_State] = dc.field(
|
||||
init=False, default_factory=dict
|
||||
)
|
||||
|
||||
+40
-58
@@ -3510,15 +3510,14 @@ class Page_InitRec(QWidget):
|
||||
logger_initrecform.debug("[Page -- InitRec] Entity IDs: %s", ids)
|
||||
self._reset()
|
||||
|
||||
ids = EntityIds(
|
||||
# un_id=3,
|
||||
# an_id=7,
|
||||
pers_id=5,
|
||||
)
|
||||
|
||||
# ids = EntityIds(
|
||||
# link_id=7,
|
||||
# # un_id=3,
|
||||
# # an_id=7,
|
||||
# pers_id=5,
|
||||
# )
|
||||
# # ids = EntityIds(
|
||||
# # link_id=7,
|
||||
# # )
|
||||
self.STATE.ids.update(ids, ignore_none=False)
|
||||
|
||||
if ids.empty():
|
||||
@@ -6239,7 +6238,7 @@ class MainWindow(QMainWindow):
|
||||
self.initrec.back_main_requested.connect(self.show_main_page)
|
||||
self.initrec.back_requested.connect(self.simple_go_back)
|
||||
self.initrec.company_profile_requested.connect(self.show_page_company_profile)
|
||||
# self.newinitrec.update_triggered.connect(self.update_grid)
|
||||
self.initrec.update_triggered.connect(self.update_grid)
|
||||
self.stack.addWidget(self.initrec)
|
||||
|
||||
# SITE: 'Unternehmensprofil'
|
||||
@@ -6344,70 +6343,53 @@ class MainWindow(QMainWindow):
|
||||
for col_idx, title in enumerate(headers):
|
||||
self.header_grid.addWidget(HeaderCell(title), self.current_row, col_idx)
|
||||
|
||||
# self.update_grid()
|
||||
self.update_grid()
|
||||
|
||||
return main_widget
|
||||
|
||||
def _clear_layout(self) -> None:
|
||||
clear_layout(self.grid)
|
||||
|
||||
# def update_grid(self) -> None:
|
||||
# clear_layout(self.grid)
|
||||
def update_grid(self) -> None:
|
||||
clear_layout(self.grid)
|
||||
|
||||
# data = backend.main_page_get_entries()
|
||||
data = backend.main_page_get_entries()
|
||||
|
||||
# for entry in data:
|
||||
# self.add_row_to_grid(entry)
|
||||
for entry in data:
|
||||
self.add_row_to_grid(entry)
|
||||
|
||||
# TODO rework
|
||||
# def add_row_to_grid(
|
||||
# self,
|
||||
# entry: types.MainPageEntry,
|
||||
# ):
|
||||
# row = self.current_row
|
||||
def add_row_to_grid(
|
||||
self,
|
||||
entry: types.MainPageEntry,
|
||||
):
|
||||
row = self.current_row
|
||||
|
||||
# # NAME
|
||||
# cell_name = MainPage_ClickableCell(entry.display_name, entry)
|
||||
# cell_name.clicked.connect(self.goto_initial_recording)
|
||||
# self.grid.addWidget(cell_name, row, 0)
|
||||
# # DATE
|
||||
# cell_date_name = entry.Metadaten_aktualisierung.date().strftime(DATE_FMT)
|
||||
# cell_date = MainPage_ClickableCell(cell_date_name, entry)
|
||||
# cell_date.clicked.connect(self.goto_initial_recording)
|
||||
# self.grid.addWidget(cell_date, row, 4)
|
||||
# NAME
|
||||
cell_name = MainPage_ClickableCell(entry.display_name, entry)
|
||||
cell_name.clicked.connect(self.goto_initial_recording)
|
||||
self.grid.addWidget(cell_name, row, 0)
|
||||
# DATE
|
||||
cell_date_name = entry.dt_updated.date().strftime(DATE_FMT)
|
||||
cell_date = MainPage_ClickableCell(cell_date_name, entry)
|
||||
cell_date.clicked.connect(self.goto_initial_recording)
|
||||
self.grid.addWidget(cell_date, row, 4)
|
||||
|
||||
# self.current_row += 1 # Zähler für den nächsten Eintrag erhöhen
|
||||
self.current_row += 1 # Zähler für den nächsten Eintrag erhöhen
|
||||
|
||||
# TODO rework
|
||||
# def goto_initial_recording(
|
||||
# self,
|
||||
# data: types.MainPageEntry,
|
||||
# ):
|
||||
# target: QWidget
|
||||
# if data.type is types.RecordingType.COMPANY:
|
||||
# ids = EntityIds(un_id=data.rec_id, type=RecordingType.COMPANY)
|
||||
# new_state = Page_InitRecCompany_State(
|
||||
# session=self.STATE.session,
|
||||
# ids=ids,
|
||||
# # un_id=data.rec_id, # TODO remove
|
||||
# locked=True,
|
||||
# )
|
||||
# self.initrec_company.load_state(new_state)
|
||||
# target = self.initrec_company
|
||||
# elif data.type is types.RecordingType.PERSON:
|
||||
# ids = EntityIds(pers_id=data.rec_id, type=RecordingType.PERSON)
|
||||
# new_state = Page_InitRecPerson_State(
|
||||
# session=self.STATE.session,
|
||||
# ids=ids,
|
||||
# # pers_id=data.rec_id, # TODO remove
|
||||
# locked=True,
|
||||
# )
|
||||
# self.initrec_person.load_state(new_state)
|
||||
# target = self.initrec_person
|
||||
# else:
|
||||
# raise TypeError(f"Unknown init rec type: {data.type}")
|
||||
|
||||
# self._navigate_to_target(target)
|
||||
def goto_initial_recording(
|
||||
self,
|
||||
data: types.MainPageEntry,
|
||||
):
|
||||
new_state = Page_InitRec_State(
|
||||
session=self.STATE.session,
|
||||
ids=data.ids,
|
||||
recording_type=data.recording_type,
|
||||
locked=True,
|
||||
)
|
||||
self.initrec.load_state(new_state)
|
||||
self._navigate_to_target(self.initrec)
|
||||
|
||||
def _navigate_to_target(
|
||||
self,
|
||||
|
||||
@@ -244,10 +244,10 @@ class ContactPersonInfo(TypedDict):
|
||||
# ** internals
|
||||
@dc.dataclass(slots=True)
|
||||
class MainPageEntry:
|
||||
rec_id: RecId
|
||||
ids: EntityIds
|
||||
display_name: str
|
||||
Metadaten_aktualisierung: datetime.datetime
|
||||
type: RecordingType
|
||||
dt_updated: datetime.datetime
|
||||
recording_type: RecordingType
|
||||
|
||||
|
||||
@dc.dataclass(slots=True)
|
||||
|
||||
Reference in New Issue
Block a user