#!/usr/bin/env python3 """FEAT-Contracts Stage 1 — Abgabeverträge (.docx) extrahieren. Liest die Sammlung der Vermittlungs-/Abgabeverträge ("Zucht der kleinen Chaoten _ … (Tier.Namen) - _.docx") vom Netzlaufwerk und erzeugt: output/contracts.json — strukturierte Vertrags-Datensätze Reines stdlib-Python (kein pip), nach Vorbild von extract_docx.py: .docx ist ein ZIP mit word/document.xml; daraus werden die "Label: Wert"-Zeilen des Vertragskörpers gelesen. Der Dateiname dient als Zusatz-Signal für die Tier-Rufnamen und den Farbschlag, wenn der Körper sie nicht hergibt. Zuverlässig extrahierbar (beobachtete Abdeckung über 250er-Stichprobe): Name (Tier): 100 % — Body-Label "Name:" Geschlecht: 100 % — Body-Label "Geschlecht:" Geburtsdatum: 98 % — Body-Label "Geburtsdatum:" Abgabedatum: 78 % — Body-Label "Abgabedatum:" Preis: 78 % — "Gesamtpreis/Schutzgebühr/Kaufpreis" Käufer (Body): 79 % — Block "Abnehmer/Empfänger" → "Name:" Farbschlag(Body): 13 % — Body-Label "Farbschlag:" (oft im Dateinamen) Übersprungen werden: * die Vorlage "Vertrags mustter new.docx" * Abstammungsnachweise / Geburtsurkunden (kein Verkauf, eigener Doc-Typ) * Dateien, die kein lesbares ZIP sind (alte .doc als .docx getarnt) Ausführung: python extract_contracts.py [--dir PFAD] [--limit N] Idempotent: mehrfaches Ausführen überschreibt output/contracts.json. """ import os import re import sys import json import zipfile import argparse HERE = os.path.dirname(os.path.abspath(__file__)) DEFAULT_DIR = r"\\truenas\Datengrab\Rennmäuse\Verträge" OUT = os.path.join(HERE, "output") TEMPLATE_MARKERS = ("mustter", "muster new") # Doc types that are NOT sale contracts (skip): NON_CONTRACT_MARKERS = ("Abstammungsnachweis", "Geburtsurkunde") # --- Body label regexes (work on whitespace-collapsed plain text) ---------- # Generic "Label: value up to next known label or end". _KNOWN_LABELS = ( r"Name|Geburtsdatum|Geschlecht|Farbschlag|Farbe|Abgabedatum|Abgabe\s*am|" r"Gesamtpreis|Schutzgeb\w+|Kaufpreis|Preis|Zuchtname|Zuchtbuchnummer|" r"Stra\w+e|Wohnort|Fon|Festnetz|Handy|Handynummer|Telefon|E-?Mail|Homepage|" r"Facebook|Mutter|Vater|Linie|Wurf|Bemerkung|Z\w+chter\w*|Abnehmer|Empf\w+nger" ) def _norm_date(d: str) -> str: """German DD.MM.YY[YY] → ISO YYYY-MM-DD (or '' if unparseable).""" if not d: return "" m = re.search(r"(\d{1,2})\.(\d{1,2})\.(\d{2,4})", d) if not m: return "" day, mon, yr = m.group(1), m.group(2), m.group(3) if len(yr) == 2: yr = "20" + yr try: di, mi, yi = int(day), int(mon), int(yr) if not (1 <= di <= 31 and 1 <= mi <= 12 and 1900 <= yi <= 2100): return "" except ValueError: return "" return f"{yi:04d}-{mi:02d}-{di:02d}" def _unescape(t: str) -> str: for a, b in (("&", "&"), ("<", "<"), (">", ">"), ("'", "'"), (""", '"'), (" ", " ")): t = t.replace(a, b) return t def _para_text(p_xml: str) -> str: """Plain text of one . IMPORTANT: Word frequently splits a single word across multiple / runs (formatting/spell-check artefacts). Run boundaries are NOT word boundaries, so we concatenate contents directly (no separator) and only turn explicit tabs/breaks into spaces. This avoids mangling "Frau"→"F r au" or "30,00"→"3 0,00". """ p_xml = re.sub(r"]*/?>", " ", p_xml) p_xml = re.sub(r"]*/?>", " ", p_xml) texts = re.findall(r"]*>(.*?)", p_xml, re.DOTALL) t = _unescape("".join(texts)) return re.sub(r"\s+", " ", t).strip() def _full_text(docx_path: str) -> str: """Return whitespace-collapsed plain text of the document body. Paragraphs (and table cells, also wrapped in ) are joined by a single space so adjacent labels stay separable. """ with zipfile.ZipFile(docx_path) as z: xml = z.read("word/document.xml").decode("utf-8", errors="replace") paras = [_para_text(p) for p in re.findall(r"].*?", xml, re.DOTALL)] return re.sub(r"\s+", " ", " ".join(p for p in paras if p)).strip() def _paragraphs(docx_path: str) -> list[str]: """Return per-paragraph plain text (preserves the Label/value line breaks).""" with zipfile.ZipFile(docx_path) as z: xml = z.read("word/document.xml").decode("utf-8", errors="replace") out = [] for p in re.findall(r"].*?", xml, re.DOTALL): t = _para_text(p) if t: out.append(t) return out def _label_value(text: str, label_rx: str) -> str: """Find 'Label: value' in collapsed text; stop at the next known label.""" rx = re.compile( rf"(?:{label_rx})\s*:\s*(.+?)(?=\s*(?:{_KNOWN_LABELS})\s*:|$)", re.IGNORECASE, ) m = rx.search(text) return m.group(1).strip() if m else "" def _clean_money(raw: str) -> str: """'27,50 Euro (Überweisung)' → '27,50'. Returns '' if no number.""" m = re.search(r"(\d+(?:[.,]\d{1,2})?)", raw) if not m: return "" return m.group(1).replace(".", ",") def parse_filename(fname: str) -> dict: """Best-effort structured pieces from the contract filename. Returns dict with keys: color, animals (list of call-names), buyer (may be empty). Robust to the many messy variants observed. """ base = re.sub(r"\.docx$", "", fname, flags=re.IGNORECASE) # Strip the leading cattery prefix and the leading/trailing underscores. base = re.sub(r"^\s*(?:Zucht der kleinen Chaoten|Clan[^_]*Chaoten)\s*", "", base, flags=re.IGNORECASE) base = base.strip().strip("_").strip() color = "" animals: list[str] = [] buyer = "" # Animals are inside the (…) group, dot-separated call-names. pm = re.search(r"\(([^)]*)\)", base) if pm: inner = pm.group(1).strip() # split on dot (call-name separator) but keep multi-word names animals = [a.strip() for a in inner.split(".") if a.strip()] # color = text before "(" color = base[:pm.start()].strip(" -–_") # buyer = text after ")" after = base[pm.end():].strip() bm = re.match(r"\s*[-–]\s*(.+)", after) if bm: buyer = bm.group(1).strip().strip("_").strip() else: # No parens. Two shapes: # " - " (token is animal-name or color) # "" (just a name) dm = re.split(r"\s*[-–]\s*", base, maxsplit=1) if len(dm) == 2 and dm[1].strip(): color = "" # ambiguous; treat the left token as an animal name animals = [dm[0].strip().strip("_")] if dm[0].strip() else [] buyer = dm[1].strip().strip("_").strip() else: tok = base.strip().strip("_").strip() # comma list like "Speedy,Agouti" → first is the name if tok: animals = [tok.split(",")[0].strip()] # Clean buyer: drop trailing numbering "2", file-version noise buyer = re.sub(r"\s*\(?\d+\)?$", "", buyer).strip() if buyer else "" return {"color": color, "animals": animals, "buyer": buyer} # Seller is always the breeder Drazena Rimac (the cattery owner); never a buyer. _SELLER_RX = re.compile(r"^(?:(?:Herr|Frau|Familie)\s+)?Drazena\b", re.IGNORECASE) _ADDR_STOP = r"Stra\w+e|Wohnort|Fon|E-?Mail|Handy|Festnetz|Telefon|Homepage|Zuchtname|Facebook" _BUYER_REJECT_RX = re.compile( r"^(?:Stra\w+e|Wohnort|Fon|E-?Mail|Handy|Festnetz|Telefon|Homepage|" r"Zuchtname|Facebook|Name)\s*:?\s*$|^[\d\s/]+$", re.IGNORECASE, ) def _clean_person(name: str) -> str: name = re.sub(r"\s+", " ", name).strip().strip(",").strip() # Reject label leakage / non-person artefacts (e.g. "Straße:", phone runs). if _BUYER_REJECT_RX.match(name): return "" return name def _extract_buyer(text: str, paras: list[str]) -> str: """Buyer (Abnehmer/Empfänger) name from the body. Two document layouts exist: A) Side-by-side columns (two table cells): "Name:" and "Name:" appear as two separate runs in the full text. B) Stacked blocks: a "Empfänger/Abnehmer:" header precedes the buyer's "Name:". Strategy: collect every "Name:" value that looks like a person and is not the seller (Drazena Rimac). The buyer is such a value. """ # Layout B: explicit Abnehmer/Empfänger header followed by Name:. bm = re.search( r"(?:Empf\w+nger|Abnehmer)[^:]*:\s*(?:[^:]*?\s)?Name\s*:\s*" rf"(.+?)(?=\s*(?:{_ADDR_STOP})\s*:|$)", text, re.IGNORECASE, ) if bm: cand = _clean_person(bm.group(1)) if cand and not _SELLER_RX.match(cand): return cand # Layout A / fallback: scan all Name: values; pick the first non-seller one. for m in re.finditer( rf"Name\s*:\s*(.+?)(?=\s*(?:{_ADDR_STOP})\s*:|\s*Name\s*:|$)", text, re.IGNORECASE, ): cand = _clean_person(m.group(1)) if not cand or _SELLER_RX.match(cand): continue # Skip the animal block: animal Name is immediately followed by # Geburtsdatum/Geschlecht/Farbschlag. tail = text[m.end():m.end() + 40] if re.match(r"\s*(?:Geburtsdatum|Geschlecht|Farbschlag)\s*:", tail, re.IGNORECASE): continue return cand return "" def parse_contract(docx_path: str) -> dict | None: """Parse one .docx. Returns a contract record or None if not a contract.""" fname = os.path.basename(docx_path) try: text = _full_text(docx_path) paras = _paragraphs(docx_path) except (zipfile.BadZipFile, KeyError, OSError): return None # unreadable / not a real docx if any(m in text for m in NON_CONTRACT_MARKERS): return None # Abstammungsnachweis / Geburtsurkunde — not a sale contract fn = parse_filename(fname) # --- animal block (body) --- body_name = _label_value(text, r"Name") # The first "Name:" in the body could be the seller's. The animal's name # appears under "Tierdaten:". Prefer the Name that directly precedes # Geburtsdatum/Geschlecht (the animal block). animal_name = "" # The animal block is headed by "Tierdaten:" (when present) and its Name is # the LAST "Name:" before "Geburtsdatum:". Anchor on Tierdaten if present, # then take the closest Name: to Geburtsdatum. scope = text ti = re.search(r"Tierdaten\s*:", text, re.IGNORECASE) if ti: scope = text[ti.end():] am = re.search( r"Name\s*:\s*(.*?)\s*Geburtsdatum\s*:", scope, re.IGNORECASE, ) if am: cand = am.group(1).strip() # Greedy guard: if it still spans multiple labels, keep only the tail # after the last embedded "Name:". if "Name:" in cand or re.search(r"Name\s*:", cand): cand = re.split(r"Name\s*:", cand)[-1].strip() # Drop any leading address-block leftovers. cand = re.split(rf"\s*(?:{_ADDR_STOP})\s*:", cand)[-1].strip() animal_name = cand # Body animal name is often blank (the call-name lives in the filename). if not animal_name or len(animal_name) > 40: animal_name = "" dob = _norm_date(_label_value(text, r"Geburtsdatum")) gender_raw = _label_value(text, r"Geschlecht").lower() if gender_raw.startswith("m"): gender = "Male" elif gender_raw.startswith("w"): gender = "Female" else: gender = "" color = _label_value(text, r"Farbschlag") or _label_value(text, r"Farbe") color = color.strip() if not color and fn["color"]: color = fn["color"] handover = _norm_date( _label_value(text, r"Abgabedatum") or _label_value(text, r"Abgabe\s*am") ) price = "" for lab in (r"Gesamtpreis", r"Schutzgeb\w+", r"Kaufpreis", r"Preis"): raw = _label_value(text, lab) if raw: price = _clean_money(raw) if price: break # contract date: trailing "Ort, [den ]DD.MM.YYYY" near signature line contract_date = "" cm = re.findall(r"[A-Za-zÄÖÜäöü.\- ]+,\s*(?:den\s*)?(\d{1,2}\.\d{1,2}\.\d{2,4})", text) if cm: contract_date = _norm_date(cm[-1]) buyer = _extract_buyer(text, paras) if not buyer and fn["buyer"]: buyer = fn["buyer"] # Animal call-names: prefer filename (the call-names the breeder filed by), # fall back to body animal name. animals = list(fn["animals"]) if not animals and animal_name: animals = [animal_name] # Reject if we have neither a buyer nor any animal name — useless record. if not buyer and not animals: return None return { "sourceFile": fname, "buyer": buyer, "animals": animals, "animalNameBody": animal_name, "color": color, "gender": gender, "dob": dob, "handoverDate": handover, "contractDate": contract_date, "price": price, } def scan(dir_path: str, limit: int | None = None): """Walk the share, parse every .docx. Returns (records, stats).""" records = [] stats = { "files_seen": 0, "template_skipped": 0, "non_contract_skipped": 0, "unreadable": 0, "parsed": 0, "unparseable": 0, "with_buyer": 0, "with_handover": 0, "with_price": 0, "with_color": 0, "with_dob": 0, "with_animals": 0, } for root, _, files in os.walk(dir_path): for f in sorted(files): if not f.lower().endswith(".docx"): continue stats["files_seen"] += 1 low = f.lower() if any(m in low for m in TEMPLATE_MARKERS): stats["template_skipped"] += 1 continue path = os.path.join(root, f) try: rec = parse_contract(path) except Exception: # never let one bad file kill the run rec = None if rec is None: # Distinguish unreadable vs non-contract vs genuinely unparseable try: _ = _full_text(path) txt = _ if any(m in txt for m in NON_CONTRACT_MARKERS): stats["non_contract_skipped"] += 1 else: stats["unparseable"] += 1 except Exception: stats["unreadable"] += 1 continue stats["parsed"] += 1 if rec["buyer"]: stats["with_buyer"] += 1 if rec["handoverDate"]: stats["with_handover"] += 1 if rec["price"]: stats["with_price"] += 1 if rec["color"]: stats["with_color"] += 1 if rec["dob"]: stats["with_dob"] += 1 if rec["animals"]: stats["with_animals"] += 1 records.append(rec) if limit and len(records) >= limit: return records, stats return records, stats def main(): try: sys.stdout.reconfigure(encoding="utf-8", errors="replace") except Exception: pass ap = argparse.ArgumentParser(description="Abgabevertrag-Extraktor") ap.add_argument("--dir", default=DEFAULT_DIR, help="Verträge-Ordner") ap.add_argument("--limit", type=int, default=None, help="Nur die ersten N Verträge (Stichprobe)") args = ap.parse_args() if not os.path.isdir(args.dir): print(f"Fehler: Ordner nicht gefunden: {args.dir}", file=sys.stderr) sys.exit(1) os.makedirs(OUT, exist_ok=True) print(f"Scanne: {args.dir}") records, stats = scan(args.dir, args.limit) out_path = os.path.join(OUT, "contracts.json") with open(out_path, "w", encoding="utf-8") as f: json.dump(records, f, ensure_ascii=False, indent=2) print(f"Dateien gesehen: {stats['files_seen']}") print(f" Vorlage übersprungen: {stats['template_skipped']}") print(f" Nicht-Vertrag (AN): {stats['non_contract_skipped']}") print(f" unlesbar (kein ZIP): {stats['unreadable']}") print(f" unparsbar: {stats['unparseable']}") print(f"Verträge geparst: {stats['parsed']}") print(f" mit Käufer: {stats['with_buyer']}") print(f" mit Tier(en): {stats['with_animals']}") print(f" mit Geburtsdatum: {stats['with_dob']}") print(f" mit Abgabedatum: {stats['with_handover']}") print(f" mit Preis: {stats['with_price']}") print(f" mit Farbschlag: {stats['with_color']}") print(f"Ausgabe: {out_path}") if __name__ == "__main__": main()