Files
GerbilManager/tools/import/compare_rpro3.py
Gulum 4e30dc7fe3 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>
2026-06-22 22:38:24 +02:00

475 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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()