Gulum
5e14124322
Neuer Parser extract_contracts.py liest die ~1,4k Abgabevertrags-DOCX (\truenas\…\Verträge): er extrahiert aus dem Dokument-Body (zuverlässiger als die Dateinamen) Käufer, Tier(e), Farbschlag, Abgabedatum und Preis — robust gegen Word-Run-Splits (z. B. „F r au"/„3 0,00"); überspringt Vorlage, Abstammungsnachweise und als .docx getarnte .doc. enrich_from_contracts() in merge_and_resolve.py: Käufer werden als Kontakte (IsReceiver) angelegt/zusammengeführt; Tiere werden KONSERVATIV per Rufname (+ DOB-Jahr bei Mehrdeutigkeit) auf eigene Bestandstiere gematcht und erhalten ReceiverContactId, GoHomeDate und Status „abgegeben" — nur wo nicht bereits gesetzt; Konflikte werden geloggt, nicht überschrieben. Jede Übernahme bekommt eine Herkunfts-Zeile („Abgabe an … aus Vertrag … übernommen."). Ergebnis: 1095 Verträge → 783 Tier-Treffer (400 mehrdeutige übersprungen), 274 neue Abnehmer-Kontakte, 153 Tiere mit Abnehmer, 49 mit Abgabedatum, 23 neu „abgegeben". Keine Backend-/Frontend-Änderung nötig (Akte zeigt Abnehmer/ Abgabedatum/Herkunft bereits). SaleContract-Records bewusst nicht erzeugt (bräuchte Migration + ingest-sichere Id — späterer Schritt). Tests: test_extract_contracts.py (Dateiname/Body/Run-Split/Skip-Regeln) + alle bestehenden grün; dotnet 212. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
447 lines
16 KiB
Python
447 lines
16 KiB
Python
#!/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) - <Käufer>_.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 <w:p>.
|
||
|
||
IMPORTANT: Word frequently splits a single word across multiple <w:r>/<w:t>
|
||
runs (formatting/spell-check artefacts). Run boundaries are NOT word
|
||
boundaries, so we concatenate <w:t> 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"<w:tab\b[^>]*/?>", " ", p_xml)
|
||
p_xml = re.sub(r"<w:br\b[^>]*/?>", " ", p_xml)
|
||
texts = re.findall(r"<w:t\b[^>]*>(.*?)</w:t>", 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 <w:p>) 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"<w:p[ >].*?</w:p>", 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"<w:p[ >].*?</w:p>", 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> - <Buyer>" (token is animal-name or color)
|
||
# "<AnimalName>" (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:<seller>" and
|
||
"Name:<buyer>" 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()
|