feat(import): RennmausPro-III-Analyse — Abgleich, Dedup, Feature-Gap, Ticket-Generierung

- compare_rpro3.py: RPRO3-SQLite ↔ aktueller Stand (Dedup Name+DOB+Farbe+Herkunft)
- rpro3-vergleich.md / rpro3-feature-gap.md: erzeugte Reports
- rpro3_tickets.py / rpro3_post_tickets.py: 413 Rückfrage-Tickets mit Herkunft + Links

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-22 22:38:24 +02:00
parent c15dd78bc7
commit 4e30dc7fe3
5 changed files with 1239 additions and 0 deletions

View File

@@ -0,0 +1,474 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
RennmausPro III (SQLite `_rpro3.db`) <-> aktueller GerbilManager-Stand: Abgleich + Dedup-Analyse.
Erzeugt NUR einen Report (schreibt NICHTS in die DB).
Zwei Teile:
(A) RPRO3-INTERNE DUBLETTEN: RennmausPro erkennt beim Import vorhandene Tiere nicht
-> dasselbe (v.a. externe) Tier kommt mehrfach vor. Zusammenlegung anhand
Name + Geburtsdatum (DOB) + Farbe + Herkunft. Sichere Cluster werden gemerged,
unsichere Faelle landen im Report.
(B) ABGLEICH vs. aktueller Stand (API-JSON): eigene Tiere (Eltern/Genotyp/fehlend),
sowie Zaehlungen fuer Wuerfe/Kontakte/externe Ahnen.
Aufruf:
python compare_rpro3.py <pfad/_rpro3.db> <scratch-dir-mit cur_*.json> [report.md]
"""
from __future__ import annotations
import sqlite3, sys, json, datetime, re, unicodedata
from collections import Counter, defaultdict
# ---------- Helpers ----------
PLACEHOLDER_NAMES = {"", "-", "n", "unbekannt", "unbekannt?", "?", "nn", "n.n.", "na"}
# Herkunft/Farbe-Werte, die "unbekannt" bedeuten -> wie leer behandeln (kompatibel mit allem)
UNKNOWN_VALUES = {"", "unbekannt", "unbek.", "unbek", "unbekannte zucht", "?", "n.n.",
"unbekannter farbschlag", "keine angabe", "k.a.", "na"}
def smart_decode(b):
"""RPRO3-DB ist gemischt kodiert: neuere Felder UTF-8, ältere Latin-1/CP1252.
Erst UTF-8 versuchen (valid -> nehmen), sonst CP1252."""
try:
return b.decode("utf-8")
except UnicodeDecodeError:
return b.decode("cp1252", "replace")
def jdn_to_date(v):
"""RPRO3 speichert Daten als astronomische Julianische Tageszahl (REAL). 0/None = unbekannt."""
if v is None:
return None
try:
v = float(v)
except (TypeError, ValueError):
return None
if v <= 0:
return None
try:
return datetime.date.fromordinal(int(round(v)) - 1721425)
except (ValueError, OverflowError):
return None
def norm_name(s):
if s is None:
return ""
s = unicodedata.normalize("NFKC", str(s)).strip().casefold()
s = re.sub(r"\s+", " ", s)
return s
def norm_geno(s):
if not s:
return ""
return re.sub(r"\s+", " ", str(s).strip())
def norm_farbe(s):
v = norm_name(s)
return "" if v in UNKNOWN_VALUES else v
def norm_origin(s):
v = norm_name(s)
return "" if v in UNKNOWN_VALUES else v
def iso(d):
return d.isoformat() if d else ""
# ---------- RPRO3 laden ----------
def load_rpro3(db_path):
c = sqlite3.connect(db_path)
c.text_factory = smart_decode
c.row_factory = sqlite3.Row
cur = c.cursor()
herk = {r["id"]: (r["_BEZ"] or r["_CLAN"] or f'{r["_VNAME"]} {r["_NNAME"]}'.strip())
for r in cur.execute("SELECT * FROM herk_tb")}
color_stamm = {str(r["id"]): r for r in cur.execute("SELECT * FROM color_tb")}
color_pup = {str(r["id"]): r for r in cur.execute("SELECT * FROM wurfcolor_tb")}
color_ext = {str(r["id"]): r for r in cur.execute("SELECT * FROM fremdcolor_tb")}
baum = {str(r["id"]): (r["_MID"], r["_PID"]) for r in cur.execute("SELECT * FROM baum_tb")}
stamm = {int(r["id"]): r for r in cur.execute("SELECT * FROM stamm_tb")}
fremd = {int(r["id"]): r for r in cur.execute("SELECT * FROM fremd_tb")}
wurftier = {str(r["id"]): r for r in cur.execute("SELECT * FROM wurftier_tb")}
wurf = {int(r["id"]): r for r in cur.execute("SELECT * FROM wurf_tb")}
def color_for(rid):
rid = str(rid)
if rid.startswith("u"):
row = color_ext.get(rid)
elif "j" in rid:
row = color_pup.get(rid)
else:
row = color_stamm.get(rid)
if not row:
return ("", "")
return (row["_FARBE"] or "", row["_FCODE"] or "")
def name_for(rid):
if rid in (None, "", "n", "NULL"):
return None
rid = str(rid)
if rid.startswith("u"):
try:
r = fremd.get(int(rid[1:]))
except ValueError:
return None
return r["_NAME"] if r else None
if "j" in rid:
r = wurftier.get(rid)
return r["_NAME"] if r else None
try:
r = stamm.get(int(rid))
except ValueError:
return None
return r["_NAME"] if r else None
animals = [] # einheitliche Records
# eigene Tiere (stamm)
for sid, r in stamm.items():
farbe, fcode = color_for(sid)
mid, pid = baum.get(str(sid), (None, None))
animals.append({
"rid": str(sid), "src": "stamm", "name": r["_NAME"],
"sex": r["_SEX"], "dob": jdn_to_date(r["_BIRTH"]),
"farbe": farbe, "fcode": fcode,
"origin": herk.get(r["_HERKUNFT"], ""),
"zb": r["_ZB"], "mid_raw": mid, "pid_raw": pid,
"mother": name_for(mid), "father": name_for(pid),
"wid": r["_WID"],
})
# externe Tiere (fremd)
for fid, r in fremd.items():
rid = f"u{fid}"
farbe, fcode = color_for(rid)
animals.append({
"rid": rid, "src": "fremd", "name": r["_NAME"],
"sex": r["_SEX"], "dob": jdn_to_date(r["_BIRTH"]),
"farbe": farbe, "fcode": fcode,
"origin": herk.get(r["_HERK"], ""),
"zb": r["_ZB"], "mid_raw": r["_MID"], "pid_raw": r["_PID"],
"mother": name_for(r["_MID"]), "father": name_for(r["_PID"]),
"wid": None,
})
return {
"animals": animals, "stamm": stamm, "fremd": fremd,
"wurf": wurf, "wurftier": wurftier, "herk": herk,
"n_color_stamm": len(color_stamm), "n_color_ext": len(color_ext),
"name_for": name_for,
}
# ---------- Dedup (Teil A) ----------
def dedup(animals):
"""Union-Find innerhalb gleicher (normalisierter) Namen.
merge(a,b) wenn DOB/Farbe/Herkunft kompatibel (gleich oder eine Seite unbekannt)
UND mindestens ein bekanntes Feld positiv uebereinstimmt.
Konflikt in einem bekannten Feld -> NICHT mergen (unsicher)."""
def compat(x, y):
return (not x) or (not y) or (x == y)
def positive(a, b):
agree = 0
if a["dobk"] and b["dobk"] and a["dobk"] == b["dobk"]:
agree += 1
if a["farbek"] and b["farbek"] and a["farbek"] == b["farbek"]:
agree += 1
if a["origink"] and b["origink"] and a["origink"] == b["origink"]:
agree += 1
return agree
def conflict(a, b):
c = 0
if a["dobk"] and b["dobk"] and a["dobk"] != b["dobk"]:
c += 1
if a["farbek"] and b["farbek"] and a["farbek"] != b["farbek"]:
c += 1
if a["origink"] and b["origink"] and a["origink"] != b["origink"]:
c += 1
return c
for a in animals:
a["namek"] = norm_name(a["name"])
a["dobk"] = a["dob"]
a["farbek"] = norm_farbe(a["farbe"])
a["origink"] = norm_origin(a["origin"])
by_name = defaultdict(list)
placeholders = []
for a in animals:
if a["namek"] in PLACEHOLDER_NAMES:
placeholders.append(a)
else:
by_name[a["namek"]].append(a)
parent = {}
def find(x):
while parent[x] != x:
parent[x] = parent[parent[x]]
x = parent[x]
return x
def union(x, y):
parent.setdefault(x, x); parent.setdefault(y, y)
parent[find(x)] = find(y)
for name, group in by_name.items():
for a in group:
parent.setdefault(a["rid"], a["rid"])
n = len(group)
for i in range(n):
for j in range(i + 1, n):
a, b = group[i], group[j]
pos = positive(a, b)
con = conflict(a, b)
comp = compat(a["dobk"], b["dobk"]) and compat(a["farbek"], b["farbek"]) and compat(a["origink"], b["origink"])
if comp and pos >= 1 and con == 0:
union(a["rid"], b["rid"]) # sicher zusammenlegen
# Cluster je Wurzel sammeln
clusters = defaultdict(list)
for a in animals:
if a["namek"] in PLACEHOLDER_NAMES:
continue
clusters[find(a["rid"])].append(a)
merge_clusters = {k: v for k, v in clusters.items() if len(v) > 1}
# Mehrdeutige Namen: Name löst sich nach sicherem Merge in >1 Cluster auf
name_to_roots = defaultdict(set)
for a in animals:
if a["namek"] in PLACEHOLDER_NAMES:
continue
name_to_roots[a["namek"]].add(find(a["rid"]))
ambiguous = [] # (anzeigename, [variant_sig,...])
for nm, roots in name_to_roots.items():
if len(roots) <= 1:
continue
variants = []
for root in roots:
recs = clusters[root]
r0 = recs[0]
variants.append({
"n": len(recs),
"rids": [r["rid"] for r in recs][:4],
"dob": sorted({iso(r["dob"]) for r in recs if r["dob"]}),
"farbe": sorted({r["farbe"] for r in recs if r["farbe"]}),
"origin": sorted({r["origin"] for r in recs if norm_origin(r["origin"])}),
"is_own": any(r["src"] == "stamm" for r in recs),
})
variants.sort(key=lambda v: -v["n"])
ambiguous.append((recs[0]["name"] if False else nm, variants))
# Anzeigenamen schöner: ersten Originalnamen je Gruppe nehmen
disp = {}
for a in animals:
disp.setdefault(a["namek"], a["name"])
ambiguous = [(disp.get(nm, nm), v) for nm, v in ambiguous]
ambiguous.sort(key=lambda x: -len(x[1]))
return merge_clusters, ambiguous, placeholders
# ---------- aktueller Stand ----------
def load_current(scratch):
def load(ep):
d = json.load(open(f"{scratch}\\cur_{ep}.json", encoding="utf-8"))
return d["items"] if isinstance(d, dict) else d
g = load("gerbils"); lit = load("litters"); con = load("contacts")
gid = {x["id"]: x for x in g}
litmap = {x["id"]: x for x in lit}
by_name = defaultdict(list)
for x in g:
by_name[norm_name(x["name"])].append(x)
def parents(x):
lid = x.get("litterId")
if not lid or lid not in litmap:
return (None, None)
l = litmap[lid]
f = gid.get(l.get("fatherId"), {}).get("name")
m = gid.get(l.get("motherId"), {}).get("name")
return (m, f)
return {"gerbils": g, "litters": lit, "contacts": con,
"by_name": by_name, "parents": parents}
def parse_dob(s):
if not s:
return None
try:
return datetime.date.fromisoformat(str(s)[:10])
except ValueError:
return None
# ---------- Report ----------
def main():
db = sys.argv[1]
scratch = sys.argv[2]
out = sys.argv[3] if len(sys.argv) > 3 else None
R = load_rpro3(db)
animals = R["animals"]
own = [a for a in animals if a["src"] == "stamm"]
ext = [a for a in animals if a["src"] == "fremd"]
merge_clusters, ambiguous, placeholders = dedup(animals)
cur = load_current(scratch)
L = []
w = L.append
w("# RennmausPro III — Abgleich- & Dedup-Report\n")
w("> Erzeugt von `tools/import/compare_rpro3.py`. **Schreibt nichts in die DB** — reine Analyse.\n")
# Überblick
w("## 1. Überblick\n")
w("| Datensatz | RennmausPro III | Aktueller Stand (DB) |")
w("|---|---:|---:|")
w(f"| Eigene Tiere (stamm) | {len(own)} | {sum(1 for x in cur['gerbils'] if x.get('isResident'))} (isResident) / {len(cur['gerbils'])} gesamt |")
w(f"| Externe Ahnen (fremd, roh) | {len(ext)} | — |")
w(f"| Würfe | {len(R['wurf'])} | {len(cur['litters'])} |")
w(f"| Kontakte (Herkunft+Abnehmer) | {len(R['herk'])} (+Abnehmer) | {len(cur['contacts'])} |")
w(f"| Genotypen hinterlegt | {R['n_color_stamm']} eigen / {R['n_color_ext']} extern | {sum(1 for x in cur['gerbils'] if x.get('genotype'))} |")
w("")
# Teil A: Dedup
collapsed = sum(len(v) for v in merge_clusters.values())
unique_after = collapsed - len(merge_clusters)
w("## 2. RPRO3-interne Dubletten (Teil A)\n")
w("RennmausPro erkennt beim Import vorhandene Tiere nicht → dasselbe Tier kommt mehrfach vor. "
"Zusammenlegung anhand **Name + Geburtsdatum + Farbe + Herkunft**.\n")
w(f"- **{len(merge_clusters)} sichere Cluster** fassen **{collapsed} Datensätze** zu **{len(merge_clusters)} Tieren** zusammen "
f"→ **{unique_after} Dubletten** fallen weg.")
w(f"- **{len(ambiguous)} mehrdeutige Namen** (gleicher Name → mehrere Varianten, manuelle Entscheidung nötig, s. u.).")
ph = Counter(norm_name(a['name']) for a in placeholders)
w(f"- **{len(placeholders)} Platzhalter-Namen** ohne Identität (nicht dedupbar): {dict(ph.most_common(6))}")
w("")
# größte sichere Cluster
w("### 2a. Größte sichere Zusammenlegungen (Top 25)\n")
w("| Tier | #Datensätze | DOB | Farbe | Herkunft |")
w("|---|---:|---|---|---|")
big = sorted(merge_clusters.values(), key=lambda v: -len(v))[:25]
for v in big:
a = v[0]
dobs = sorted({iso(x["dob"]) for x in v if x["dob"]})
farben = sorted({x["farbe"] for x in v if x["farbe"]})
orig = sorted({x["origin"] for x in v if x["origin"]})
w(f"| {a['name']} | {len(v)} | {', '.join(dobs) or ''} | {', '.join(farben)[:40] or ''} | {', '.join(orig)[:40] or ''} |")
w("")
# mehrdeutige Namen
w("### 2b. Mehrdeutige Namen — bitte entscheiden (gleicher Name, mehrere Varianten)\n")
w("Pro Name die nach sicherem Merge verbliebenen **Varianten**. Frage je Zeile: *Ist das dasselbe Tier "
"(→ zusammenlegen) oder verschiedene Tiere gleichen Namens (→ getrennt lassen)?* "
"★ = eigenes Tier (stamm). Top 60 nach Variantenzahl.\n")
w("Merkmalslose Datensätze (kein DOB, keine Farbe, keine Herkunft) sind als „**N× nur Name**\" "
"zusammengefasst. Sortiert nach Anzahl **informativer** Varianten.\n")
w("| Name | Informative Varianten (Anzahl · DOB · Farbe · Herkunft) | nur Name |")
w("|---|---|---:|")
# nach Zahl informativer Varianten sortieren
def split_variants(variants):
info, bare = [], 0
for v in variants:
if v["dob"] or v["farbe"] or v["origin"]:
info.append(v)
else:
bare += v["n"]
return info, bare
enriched = [(nm, *split_variants(vs)) for nm, vs in ambiguous]
enriched.sort(key=lambda x: (-len(x[1]), -x[2]))
shown = [e for e in enriched if len(e[1]) >= 1][:60]
for nm, info, bare in shown:
parts = []
for v in info:
star = "" if v["is_own"] else ""
dob = ", ".join(v["dob"]) or ""
farbe = ", ".join(v["farbe"])[:30] or ""
orig = ", ".join(v["origin"])[:30] or ""
parts.append(f"{star}{v['n']}× · {dob} · {farbe} · {orig}")
w(f"| {nm} | {'<br>'.join(parts) or ''} | {bare or ''} |")
only_bare = sum(1 for e in enriched if len(e[1]) == 0)
w(f"\n_Zusätzlich {only_bare} Namen, bei denen **alle** Varianten merkmalslos sind "
f"(reine Namens-Dubletten ohne Unterscheidungsmerkmale)._")
w("")
# Teil B: Abgleich eigene Tiere
w("## 3. Eigene Tiere — Abgleich gegen aktuellen Stand (Teil B)\n")
missing, parent_diffs, geno_diffs, dob_sex_diffs, matched = [], [], [], [], 0
for a in own:
cands = cur["by_name"].get(norm_name(a["name"]), [])
if not cands:
missing.append(a); continue
# besten Kandidaten nach DOB wählen
best = cands[0]
if a["dob"]:
scored = sorted(cands, key=lambda x: abs(((parse_dob(x.get("dateOfBirth")) or datetime.date(1900,1,1)) - a["dob"]).days))
best = scored[0]
matched += 1
m_cur, f_cur = cur["parents"](best)
if (norm_name(a["mother"]) != norm_name(m_cur)) or (norm_name(a["father"]) != norm_name(f_cur)):
parent_diffs.append((a, m_cur, f_cur))
cg = norm_geno(best.get("genotype"))
rg = norm_geno(a["fcode"])
if rg and cg != rg:
geno_diffs.append((a, best.get("genotype")))
cdob = parse_dob(best.get("dateOfBirth"))
if a["dob"] and cdob and abs((a["dob"]-cdob).days) > 14:
dob_sex_diffs.append((a, cdob))
w(f"- Gematcht (Name): **{matched}/{len(own)}**")
w(f"- **Fehlen im aktuellen Stand: {len(missing)}**")
w(f"- **Eltern unterschiedlich: {len(parent_diffs)}**")
w(f"- **Genotyp unterschiedlich/aktuell ohne: {len(geno_diffs)}**")
w(f"- **Geburtsdatum weicht >14 Tage ab: {len(dob_sex_diffs)}**")
w("")
w("### 3a. Eltern unterschiedlich (RPRO3 = autoritativ)\n")
w("| Tier | DOB | RPRO3 Mutter × Vater | Aktuell Mutter × Vater |")
w("|---|---|---|---|")
for a, mc, fc in parent_diffs[:120]:
w(f"| {a['name']} | {iso(a['dob'])} | {a['mother'] or ''} × {a['father'] or ''} | {mc or ''} × {fc or ''} |")
if len(parent_diffs) > 120:
w(f"\n_… und {len(parent_diffs)-120} weitere._")
w("")
w("### 3b. Eigene Tiere, die im aktuellen Stand fehlen\n")
w("| Tier | DOB | Geschlecht | Farbe | Herkunft | ZB |")
w("|---|---|---|---|---|---|")
for a in missing[:120]:
w(f"| {a['name']} | {iso(a['dob'])} | {a['sex']} | {a['farbe'] or ''} | {a['origin'] or ''} | {a['zb'] or ''} |")
if len(missing) > 120:
w(f"\n_… und {len(missing)-120} weitere._")
w("")
w("### 3c. Genotyp unterschiedlich / aktuell nicht vorhanden (Top 60)\n")
w("| Tier | RPRO3 Genotyp (autoritativ) | Aktuell |")
w("|---|---|---|")
for a, cg in geno_diffs[:60]:
w(f"| {a['name']} | `{a['fcode']}` | {('`'+cg+'`') if cg else ''} |")
if len(geno_diffs) > 60:
w(f"\n_… und {len(geno_diffs)-60} weitere._")
w("")
# Caveat + Empfehlung
w("## 4. Hinweise zum Matching & Empfehlung\n")
w("**Matching-Vorbehalt:** Der Abgleich in Abschnitt 3 vergleicht über den **Namen**. Der aktuelle "
"Stand führt Tiere oft mit Zucht-Suffix („… von den Kleinen Chaoten\", „of …\"), RPRO3 nur den "
"Kurznamen. Ein Teil der **180 „fehlend\"** ist daher nur ein Suffix-Mismatch, kein echtes Fehlen. "
"Der spätere Importer normalisiert Namen (Suffix abtrennen) und matcht zusätzlich über DOB+Farbe.\n")
w("**Kernbefund:** RPRO3 ist die **autoritative Quelle** — die echten Zuchtdaten der Software. Die "
"Eltern-Abweichungen (Abschnitt 3a) zeigen klar die bekannten xlsx-Heuristik-Fehler (z. B. Blacky = "
"*Bella × Snoopy* statt fälschlich *Yumi × Flori*). Ein Import aus RPRO3 behebt viele dieser Fehler "
"an der Wurzel.\n")
w("**Empfohlenes Vorgehen für den Import:**")
w("1. RPRO3-interne Dubletten **vor** dem Laden zusammenlegen (Abschnitt 2; merkmalslose Namens-Dubletten zu 1 Tier).")
w("2. Mehrdeutige Namen (2b) der Züchterin zur Entscheidung vorlegen.")
w("3. Eigene Tiere + Eltern + Genotypen + Würfe + Kontakte aus RPRO3 als **führend** übernehmen.")
w("4. Externe Ahnen (nach Dedup) für tiefe Stammbäume ergänzen.")
w("")
text = "\n".join(L)
if out:
open(out, "w", encoding="utf-8").write(text)
print(f"Report geschrieben: {out} ({len(text)} Zeichen)")
else:
print(text)
if __name__ == "__main__":
main()