Gulum/GerbilManager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(provenance): Datenherkunft als chronologischer, datei-attribuierter Verlauf

Browse Source 
Die Herkunft liest sich jetzt wie eine History: pro Feld wird genannt, aus
welcher Quelldatei der Wert kam, plus Merge-/Entscheidungs-/Wurfchronik-Schritte.
Beispiel: „In ‚X.xlsx' gefunden." → „Geburtsdatum (…) aus ‚X.xlsx'." → „Auch in
‚Y.docx' gefunden → zusammengeführt." → „Eltern über Position erkannt (Quelle …)."

Import: build_entity_provenance() um history[] erweitert; ein field_source-
Tracking im Gerbil-Dedup hält fest, welcher Datensatz/welche Datei jedes Feld
(DOB/Genotyp/Geschlecht/Farbschlag/Sterbedatum) lieferte — auch über Empty-Fill
und Mehrheitsentscheid hinweg. Kontakte und Würfe erhalten analoge, datei-
attribuierte Verläufe. Kein DB-Migration nötig (history liegt im Provenance-JSON).

Frontend: ProvenanceDialog rendert den Verlauf als nummerierte Timeline.
Tests: history wird erzeugt, nennt Dateien, überlebt Ingest (dotnet 212,
vitest 129, playwright 17, python merge/extract grün).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Gulum
2026-06-22 16:21:24 +02:00
parent a31fea17fa
commit cf672c0933
9 changed files with 389 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

285
tools/import/merge_and_resolve.py
View File

@@ -23,14 +23,17 @@ def generate_guid(key_str):
return str(uuid.uuid5(uuid.NAMESPACE_DNS, key_str))
def build_entity_provenance(source_files, merged_record_count, notes=None,
from_wurfchronik=None, extra=None):
from_wurfchronik=None, extra=None, history=None):
"""Generic data-provenance JSON builder shared by gerbils, contacts and
litters. Mirrors the GerbilProvenance frontend contract:
{ sourceFiles, mergedRecordCount, fromWurfchronik, notes, ... }
{ sourceFiles, mergedRecordCount, fromWurfchronik, notes, history, ... }
`source_files` is any iterable of filenames; `from_wurfchronik` is auto-
derived from the filenames when left as None. `extra` may carry entity-
specific keys (e.g. parentMethod/parentConfidence for gerbils). Returns a
JSON string (stored on the nullable Provenance text column)."""
specific keys (e.g. parentMethod/parentConfidence for gerbils). `history`
is an ordered list of human-readable German lines that read like a
chronological log of where each fact came from (the primary content shown
in the Datenherkunft dialog). Returns a JSON string (stored on the nullable
Provenance text column)."""
files = sorted({f for f in source_files if f})
if from_wurfchronik is None:
from_wurfchronik = any("wurfchronik" in f.lower() for f in files)
@@ -39,6 +42,7 @@ def build_entity_provenance(source_files, merged_record_count, notes=None,
"mergedRecordCount": merged_record_count,
"fromWurfchronik": bool(from_wurfchronik),
"notes": list(notes or []),
"history": list(history or []),
}
if extra:
for k, v in extra.items():
@@ -46,6 +50,58 @@ def build_entity_provenance(source_files, merged_record_count, notes=None,
prov[k] = v
return json.dumps(prov, ensure_ascii=False)
def _quote_file(fname):
"""German typographic quotes around a source filename for history lines."""
return f"„{fname}“"
# Human-readable German labels for the significant fields we attribute to files.
PROV_FIELD_LABELS = {
"DateOfBirth": "Geburtsdatum",
"DateOfDeath": "Sterbedatum",
"Gender": "Geschlecht",
"Genotype": "Genotyp",
"ColorVarietyId": "Farbschlag",
"Name": "Name",
}
def _primary_file_of(record):
"""The single most representative source file of one raw record.
Prefers an explicit Stammbaum/Wurfchronik filename from ImportSource, else
the record's _filename. Used to attribute a field value to a concrete file
in the history log."""
fn = record.get("_filename")
if fn:
return fn
imp = record.get("ImportSource")
if imp:
first = str(imp).split(",")[0].strip()
if first:
return first
return None
def _record_source_files(g):
"""All distinct source files a single raw gerbil record drew from.
ImportSource is either a comma-joined Stammbaum file list, a single
Wurfchronik filename, or the per-litter filename; _filename is the primary
file. We union both so nothing is lost."""
files = set()
imp = g.get("ImportSource")
if imp:
for part in str(imp).split(","):
part = part.strip()
if part:
files.add(part)
fn = g.get("_filename")
if fn:
files.add(fn)
return files
def to_valid_guid(val):
if not val:
return None
@@ -657,6 +713,114 @@ def pick_parent_ref(parent_refs, role, child_dob, avoid_name=None, gender_of=Non
return role_refs[order[0]]
def _build_gerbil_history(records, best_g, field_source, parent_method=None,
any_decision=False, any_conflict=False, conflict_notes=None):
"""Build an ordered, file-attributed German history for a resolved gerbil.
Reads like a chronological log:
• „In ‚X.xlsx' gefunden."
• „Geburtsdatum (27.03.2022) aus ‚X.xlsx'."
• „Auch in ‚Y.xlsx' gefunden → Datensätze zusammengeführt."
• „Genotyp aus ‚Z.xlsx'."
• „Eltern über Position im Stammbaum erkannt (Quelle: ‚X.xlsx')."
• „Aus Wurfchronik übernommen."
`field_source` maps a field name to the raw record that supplied its final
value; when present we name that record's file, otherwise we fall back to
the primary record. The records are visited in a stable order (primary
first, then the rest sorted by file) so the log is deterministic."""
conflict_notes = conflict_notes or []
history = []
# Order records: best_g first, then others by primary file name (stable).
others = [r for r in records if r is not best_g]
others.sort(key=lambda r: (_primary_file_of(r) or ""))
ordered = [best_g] + others
best_file = _primary_file_of(best_g)
if best_file:
history.append(f"In {_quote_file(best_file)} gefunden.")
else:
history.append("Im Import gefunden.")
# Field-by-field attribution: name the file that supplied each fact.
def attr_line(field, formatter):
rec = field_source.get(field) or best_g
val = best_g.get(field)
if not val or val == "unknown":
return
fname = _primary_file_of(rec)
label = PROV_FIELD_LABELS.get(field, field)
text = formatter(label, val)
if fname:
history.append(f"{text} aus {_quote_file(fname)}.")
else:
history.append(f"{text} (Quelle unbekannt).")
attr_line("DateOfBirth", lambda label, v: f"{label} ({_de_date(v)})")
attr_line("Gender", lambda label, v: f"{label} ({_de_gender(v)})")
attr_line("Genotype", lambda label, v: f"{label}")
attr_line("ColorVarietyId", lambda label, v: f"{label}")
attr_line("DateOfDeath", lambda label, v: f"{label} ({_de_date(v)})")
# Merge step: every additional record that contributed.
for r in others:
fname = _primary_file_of(r)
if fname:
history.append(
f"Auch in {_quote_file(fname)} gefunden → Datensätze zusammengeführt."
)
else:
history.append("In weiterem Datensatz gefunden → Datensätze zusammengeführt.")
# Parent derivation.
if parent_method:
method_label = {
"chart-position": "Position im Stammbaum",
"geburtsdatum+eltern": "Geburtsdatum und Elternnamen",
"nur-geburtsdatum": "Geburtsdatum",
"decision": "manuelle Entscheidung",
}.get(parent_method, parent_method)
# The parent evidence comes from a Stammbaum chart — attribute to the
# primary record's file when it is a Stammbaum.
parent_file = best_file if best_file and "stammbaum" in best_file.lower() else None
if parent_file:
history.append(
f"Eltern über {method_label} erkannt (Quelle: {_quote_file(parent_file)})."
)
else:
history.append(f"Eltern über {method_label} erkannt.")
# Manual decisions / conflicts.
if any_decision:
history.append("Zuordnung per manueller Entscheidung getroffen.")
if any_conflict:
history.append("Konflikt per Entscheidung gelöst.")
for cn in conflict_notes:
if cn and cn not in history:
history.append(cn + ".")
# Wurfchronik provenance line.
if any("wurfchronik" in f.lower() for r in records for f in _record_source_files(r)):
history.append("Angaben aus der Wurfchronik übernommen.")
return history
def _de_date(iso):
"""YYYY-MM-DD → DD.MM.YYYY for display; pass through anything else."""
if not iso:
return iso
m = re.match(r"^(\d{4})-(\d{2})-(\d{2})$", str(iso))
if m:
return f"{m.group(3)}.{m.group(2)}.{m.group(1)}"
return iso
def _de_gender(g):
return {"male": "männlich", "female": "weiblich"}.get(g, g)
def main():
print("Loading color variety seeds...")
variety_map = {}
@@ -1721,30 +1885,14 @@ def main():
return True
def _record_source_files(g):
"""All distinct source files a single raw record drew from.
ImportSource is either a comma-joined Stammbaum file list, a single
Wurfchronik filename, or the per-litter filename; _filename is the
primary file. We union both so nothing is lost."""
files = set()
imp = g.get("ImportSource")
if imp:
for part in str(imp).split(","):
part = part.strip()
if part:
files.add(part)
fn = g.get("_filename")
if fn:
files.add(fn)
return files
def build_provenance(records, best_g, extra_notes=None):
def build_provenance(records, best_g, extra_notes=None, field_source=None):
"""Aggregate data-provenance across every raw record merged into one
resolved gerbil. Returns a JSON string (stored on the Gerbil entity as a
nullable text column) so the Rennmausakte can show where the entry came
from. `records` is the list of raw records that became this gerbil;
`best_g` is the chosen primary record."""
`best_g` is the chosen primary record. `field_source` maps a significant
field name (DateOfBirth/Genotype/…) to the raw record that supplied its
final value, so the history can name the exact file for each fact."""
source_files = set()
from_wurfchronik = False
any_conflict = False
@@ -1783,12 +1931,21 @@ def main():
if n and n not in notes:
notes.append(n)
history = _build_gerbil_history(
records, best_g, field_source or {},
parent_method=parent_method,
any_decision=any_decision,
any_conflict=any_conflict,
conflict_notes=extra_notes or [],
)
return build_entity_provenance(
source_files,
merged_count,
notes=notes,
from_wurfchronik=from_wurfchronik,
extra={"parentMethod": parent_method, "parentConfidence": parent_confidence},
history=history,
)
# Group gerbils by name to perform deduplication
@@ -1861,13 +2018,24 @@ def main():
merged_notes = []
if best_g["Notes"]:
merged_notes.append(best_g["Notes"])
# Merge photos
merged_photos = list(best_g.get("_photos", []))
# Track sources for debugging
sources = [best_g["_filename"]]
# Per-field file attribution: which raw record supplied each final
# field value. Seed with best_g for every field it already carries;
# the fill loop and voting loop update it as winners change.
ATTRIB_FIELDS = ["DateOfBirth", "DateOfDeath", "Gender", "Genotype",
"ColorVarietyId", "Name"]
field_source = {}
for fld in ATTRIB_FIELDS:
v = best_g.get(fld)
if v and v != "unknown":
field_source[fld] = best_g
for g in sub:
if g == best_g:
continue
@@ -1884,16 +2052,20 @@ def main():
best_g["LitterId"] = g["LitterId"]
if not best_g["DateOfBirth"] and g["DateOfBirth"]:
best_g["DateOfBirth"] = g["DateOfBirth"]
field_source["DateOfBirth"] = g
if not best_g["DateOfDeath"] and g["DateOfDeath"]:
best_g["DateOfDeath"] = g["DateOfDeath"]
field_source["DateOfDeath"] = g
if not best_g["CauseOfDeath"] and g["CauseOfDeath"]:
best_g["CauseOfDeath"] = g["CauseOfDeath"]
if not best_g["GoHomeDate"] and g["GoHomeDate"]:
best_g["GoHomeDate"] = g["GoHomeDate"]
if not best_g["Genotype"] and g["Genotype"]:
best_g["Genotype"] = g["Genotype"]
field_source["Genotype"] = g
if not best_g["ColorVarietyId"] and g["ColorVarietyId"]:
best_g["ColorVarietyId"] = g["ColorVarietyId"]
field_source["ColorVarietyId"] = g
if not best_g["OriginContactId"] and g["OriginContactId"]:
best_g["OriginContactId"] = g["OriginContactId"]
if not best_g["ReceiverContactId"] and g["ReceiverContactId"]:
@@ -1904,10 +2076,12 @@ def main():
# Reconcile Gender: prefer a known gender over unknown, and prefer stammbaum over other sources
if best_g["Gender"] == "unknown" and g["Gender"] != "unknown":
best_g["Gender"] = g["Gender"]
field_source["Gender"] = g
elif best_g["Gender"] != "unknown" and g["Gender"] != "unknown" and best_g["Gender"] != g["Gender"]:
if g["ImportSource"] and "stammbaum" in g["ImportSource"].lower():
if not best_g["ImportSource"] or "stammbaum" not in best_g["ImportSource"].lower():
best_g["Gender"] = g["Gender"]
field_source["Gender"] = g
# Status precedence: Deceased > GivenAway > Breeding/Pet
if g["Status"] == "Deceased":
@@ -1935,17 +2109,15 @@ def main():
# If the records disagreed on a field, the merge had to pick a
# winner — record that as a provenance note.
if len(votes) > 1:
FIELD_LABEL = {
"DateOfBirth": "Geburtsdatum",
"DateOfDeath": "Sterbedatum",
"Gender": "Geschlecht",
"Genotype": "Genotyp",
"ColorVarietyId": "Farbschlag",
}
conflict_notes.append(
f"Konflikt bei {FIELD_LABEL[field]} per Mehrheitsentscheidung gelöst"
f"Konflikt bei {PROV_FIELD_LABELS[field]} per Mehrheitsentscheidung gelöst"
)
best_g[field] = best_val
# Attribute the winning value to a record that actually holds
# it, so the history names the right file.
winner = next((g for g in sub if g.get(field) == best_val), None)
if winner is not None:
field_source[field] = winner
# Keep helper fields in sync if we changed DateOfBirth
if field == "DateOfBirth":
best_g["_birth_date"] = best_val
@@ -1955,7 +2127,9 @@ def main():
best_g["Notes"] = " | ".join(merged_notes)
best_g["_photos"] = merged_photos
best_g["Provenance"] = build_provenance(sub, best_g, extra_notes=conflict_notes)
best_g["Provenance"] = build_provenance(
sub, best_g, extra_notes=conflict_notes, field_source=field_source
)
# Print merge trace
print(f"Deduplicated same-animal name '{best_g['Name']}': merged {len(sub)} entries across files: {', '.join(sources)}")
@@ -2246,9 +2420,31 @@ def main():
if l_merged_count > 1:
l_notes.append(f"aus {l_merged_count} Datensätzen zusammengeführt")
l_notes.append("Geschwister-Würfe zusammengeführt")
# Chronological, file-attributed history for the litter.
l_files_sorted = sorted({f for f in l_source_files if f})
l_history = []
first_file = l_files_sorted[0] if l_files_sorted else None
if is_virtual and not l_from_wurfchronik:
if first_file:
l_history.append(
f"Aus Stammbaum-Diagramm rekonstruiert ({_quote_file(first_file)})."
)
else:
l_history.append("Aus Stammbaum-Diagramm rekonstruiert.")
elif first_file:
l_history.append(f"Wurf aus Wurfchronik {_quote_file(first_file)}.")
else:
l_history.append("Wurf im Import gefunden.")
for f in l_files_sorted[1:]:
l_history.append(
f"Auch in {_quote_file(f)} gefunden → Datensätze zusammengeführt."
)
if l_merged_count > 1:
l_history.append("Geschwister-Würfe zusammengeführt.")
l["Provenance"] = build_entity_provenance(
l_source_files, l_merged_count, notes=l_notes,
from_wurfchronik=l_from_wurfchronik,
from_wurfchronik=l_from_wurfchronik, history=l_history,
)
# Set IsBreeder and IsReceiver flags on contacts
@@ -2275,7 +2471,24 @@ def main():
c_notes.append("als Züchter erkannt")
if is_receiver:
c_notes.append("als Abnehmer erkannt")
c["Provenance"] = build_entity_provenance(c_source_files, c_merged_count, notes=c_notes)
# Chronological, file-attributed history for the contact.
c_files_sorted = sorted({f for f in c_source_files if f})
c_history = []
role_word = "Züchter" if is_breeder else "Abnehmer"
if c_files_sorted:
c_history.append(f"In {_quote_file(c_files_sorted[0])} als {role_word} erkannt.")
for f in c_files_sorted[1:]:
c_history.append(
f"Auch in {_quote_file(f)} gefunden → Datensätze zusammengeführt."
)
else:
c_history.append(f"Im Import als {role_word} erkannt.")
if is_breeder and is_receiver:
c_history.append("Sowohl als Züchter als auch als Abnehmer geführt.")
c["Provenance"] = build_entity_provenance(
c_source_files, c_merged_count, notes=c_notes, history=c_history
)
# Set and map gerbilPhotos
resolved_photos = []

64
tools/import/test_merge_resolve.py
View File

@@ -213,6 +213,70 @@ mr = m.pick_parent_ref(molly_refs, "mother", "13.09.2021", avoid_name=fr["name"]
check("pick(gender): mother = Arya (female)", mr and mr["name"].startswith("Arya"))
# ── Provenance history: chronological, file-attributed German log ──
import json as _json
def _hist(prov_json):
return _json.loads(prov_json)["history"]
# build_entity_provenance carries history through verbatim.
_prov = _json.loads(
m.build_entity_provenance(["A.xlsx"], 1, notes=["x"], history=["line one"])
)
check("build_entity_provenance includes history key", _prov.get("history") == ["line one"])
check("build_entity_provenance defaults history to []",
_json.loads(m.build_entity_provenance(["A.xlsx"], 1)).get("history") == [])
# Single-record gerbil history: names the file and the per-field facts.
g_single = {
"Name": "Picus", "DateOfBirth": "2022-03-27", "Gender": "male",
"Genotype": "aa", "ColorVarietyId": None, "DateOfDeath": None,
"ImportSource": "Stammbaum von Picus Son.xlsx",
"_filename": "Stammbaum von Picus Son.xlsx", "parentRefs": [],
}
h = m._build_gerbil_history([g_single], g_single, {})
check("history: first line names the source file",
h[0] == "In „Stammbaum von Picus Son.xlsx“ gefunden.")
check("history: dob line names file + formatted date",
"Geburtsdatum (27.03.2022) aus „Stammbaum von Picus Son.xlsx“." in h)
check("history: gender line is German + file-attributed",
"Geschlecht (männlich) aus „Stammbaum von Picus Son.xlsx“." in h)
check("history: genotype line file-attributed",
"Genotyp aus „Stammbaum von Picus Son.xlsx“." in h)
# Merged gerbil: a field sourced from a DIFFERENT file is attributed to THAT file.
g_best = {
"Name": "Solice", "DateOfBirth": "2022-03-27", "Gender": "male",
"Genotype": None, "ColorVarietyId": None, "DateOfDeath": None,
"ImportSource": "Stammbaum von Picus Son.xlsx",
"_filename": "Stammbaum von Picus Son.xlsx", "parentRefs": [],
}
g_other = {
"Name": "Solice", "DateOfBirth": "2022-03-27", "Gender": "male",
"Genotype": "aa", "ColorVarietyId": None, "DateOfDeath": None,
"ImportSource": "Wurfchronik-Detail.docx",
"_filename": "Wurfchronik-Detail.docx", "parentRefs": [],
}
# Genotype was filled from g_other → its line must name the docx file.
g_best["Genotype"] = "aa"
fs = {"DateOfBirth": g_best, "Gender": g_best, "Genotype": g_other}
h2 = m._build_gerbil_history([g_best, g_other], g_best, fs)
check("history(merge): genotype attributed to the file that supplied it",
"Genotyp aus „Wurfchronik-Detail.docx“." in h2)
check("history(merge): dob attributed to primary file",
"Geburtsdatum (27.03.2022) aus „Stammbaum von Picus Son.xlsx“." in h2)
check("history(merge): merge line names the absorbed file",
"Auch in „Wurfchronik-Detail.docx“ gefunden → Datensätze zusammengeführt." in h2)
check("history(merge): Wurfchronik line present",
"Angaben aus der Wurfchronik übernommen." in h2)
# Date formatting helper.
check("_de_date: ISO → DD.MM.YYYY", m._de_date("2022-03-27") == "27.03.2022")
check("_de_date: passes through non-ISO", m._de_date("unbekannt") == "unbekannt")
if check.failed:
print(f"\n{check.failed} test(s) FAILED")
sys.exit(1)