ai-bayarea/merge_buyers.py

#!/usr/bin/env python3
"""
merge_buyers.py - Fuehrt Buyer-Datensaetze auf PERSONENEBENE zusammen.

Problem: Eine Person hat oft mehrere Buyer Information Sheets (verschiedene
Daten, Text- und Notes-Varianten). Beispiel Sudduth: eine Text-Datei + eine
Notes-Scan-Datei = dieselbe Person. Oder Sahota mit 6 Dateien.

Dieses Script liest die rohe buyers_raw.jsonl (ein Datensatz je DATEI) und
erzeugt buyers_merged.jsonl (ein Datensatz je PERSON), mit:
  - allen Business-Kategorien ueber alle Sheets vereinigt
  - fruehestem und spaetestem Date-of-Introduction
  - je Kontaktfeld dem besten (nicht-leeren) Wert
  - Liste der Quelldateien zur Nachvollziehbarkeit

Die rohe Extraktion bleibt unangetastet (nachvollziehbar). Merge ist ein
separater, pruefbarer Schritt.

Aufruf:
  python merge_buyers.py ./poc_out/buyers_raw.jsonl ./poc_out/buyers_merged.jsonl
"""

import sys
import json
import re
from collections import defaultdict


def _empty(v):
    return v in (None, "", [], {})


def name_from_filename(source_file):
    """
    Extrahiert 'Nachname, Vorname' aus dem Dateinamen als robusten Fallback.
    Die Dateien folgen konsistent dem Schema 'Nachname, Vorname <datum> [Notes].pdf'.
    """
    if not source_file:
        return ""
    base = source_file.rsplit("/", 1)[-1]  # nur Dateiname
    base = re.sub(r"\.pdf$", "", base, flags=re.I)
    # Datum, 'Notes', Zahlen und Zusaetze abschneiden
    base = re.sub(r"\b\d{6,8}\b.*$", "", base)      # ab erstem Datum abschneiden
    base = re.sub(r"(?i)\bnotes\b.*$", "", base)
    base = base.strip(" -_")
    return base


def normalize_name(rec):
    """
    Personen-Schluessel. Bevorzugt den echten Namen (prospective_buyer),
    faellt auf name_company zurueck, dann auf den DATEINAMEN (robust, da
    konsistentes Schema). Normalisiert "Nachname, Vorname" und
    "Vorname Nachname" auf eine vergleichbare Form.
    """
    name = rec.get("prospective_buyer") or rec.get("name_company") or ""
    if not name.strip():
        # Fallback: aus Dateiname (gerade beim Text-Pfad oft noetig)
        name = name_from_filename(rec.get("source_file", ""))
    name = name.strip().lower()
    name = re.sub(r"\s+", " ", name)
    # "sudduth, henry" -> "henry sudduth" (Komma-Form angleichen)
    if "," in name:
        parts = [p.strip() for p in name.split(",", 1)]
        if len(parts) == 2 and parts[1]:
            name = f"{parts[1]} {parts[0]}"
    # Satzzeichen weg
    name = re.sub(r"[^\w\s]", "", name)
    return name.strip()


def pick_best(values):
    """Ersten nicht-leeren Wert aus einer Liste waehlen."""
    for v in values:
        if not _empty(v):
            return v
    return None


def merge_person(records):
    """Fuehrt alle Sheets EINER Person zu einem Datensatz zusammen."""
    # Kontaktfelder: bester nicht-leerer Wert (spaetere Sheets zuerst,
    # da meist aktueller - wir sortieren unten nach Datum absteigend)
    single_fields = ["name_company", "prospective_buyer", "company", "phone",
                     "cell", "email", "address", "state", "how_did_you_hear",
                     "background_experience"]
    out = {}
    for f in single_fields:
        out[f] = pick_best([r.get(f) for r in records])

    # interested_in_updates: wenn IRGENDEIN Sheet true/false sagt, nimm das
    # (bevorzugt das neueste eindeutige)
    upd = pick_best([r.get("interested_in_updates") for r in records
                     if r.get("interested_in_updates") is not None])
    out["interested_in_updates"] = upd

    # types_of_business: Vereinigung ueber alle Sheets
    cats = []
    for r in records:
        for c in (r.get("types_of_business") or []):
            c = c.strip()
            if c and c not in cats:
                cats.append(c)
    out["types_of_business"] = cats

    # Daten: alle gueltigen ISO-Daten sammeln
    dates = sorted(d for d in (r.get("date_of_introduction") for r in records)
                   if isinstance(d, str) and re.match(r"\d{4}-\d{2}-\d{2}", d))
    out["date_first_introduction"] = dates[0] if dates else None
    out["date_last_introduction"] = dates[-1] if dates else None
    out["all_introduction_dates"] = dates

    # Nachvollziehbarkeit
    out["source_files"] = [r.get("source_file") for r in records]
    out["n_sheets"] = len(records)
    return out


def main():
    if len(sys.argv) < 3:
        print("Aufruf: python merge_buyers.py <input.jsonl> <output.jsonl>")
        sys.exit(1)
    inp, outp = sys.argv[1], sys.argv[2]

    records = []
    with open(inp, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            r = json.loads(line)
            # Fehler und Nicht-Sheets ueberspringen
            if "_error" in r:
                continue
            if r.get("is_buyer_sheet") is False:
                continue
            records.append(r)

    # nach Person gruppieren
    groups = defaultdict(list)
    unkeyed = []
    for r in records:
        key = normalize_name(r)
        if key:
            groups[key].append(r)
        else:
            unkeyed.append(r)  # ohne erkennbaren Namen: einzeln behalten

    merged = []
    for key, recs in groups.items():
        # innerhalb der Person nach Datum absteigend (neuestes zuerst)
        recs_sorted = sorted(
            recs,
            key=lambda r: (r.get("date_of_introduction") or ""),
            reverse=True,
        )
        m = merge_person(recs_sorted)
        m["person_key"] = key
        merged.append(m)

    # namenlose einzeln anhaengen
    for r in unkeyed:
        m = merge_person([r])
        m["person_key"] = "(kein Name erkannt)"
        merged.append(m)

    # nach neuestem Datum sortieren
    merged.sort(key=lambda m: (m.get("date_last_introduction") or ""), reverse=True)

    with open(outp, "w", encoding="utf-8") as f:
        for m in merged:
            f.write(json.dumps(m, ensure_ascii=False) + "\n")

    # Statistik
    multi = [m for m in merged if m["n_sheets"] > 1]
    print(f"Eingelesen:            {len(records)} Datensaetze (Sheets)")
    print(f"Distinkte Personen:    {len(merged)}")
    print(f"davon mit >1 Sheet:    {len(multi)}")
    print(f"Ausgabe:               {outp}")
    if multi:
        print("\nBeispiele (Personen mit mehreren Sheets):")
        for m in sorted(multi, key=lambda x: -x["n_sheets"])[:8]:
            print(f"  {m['person_key']!r}: {m['n_sheets']} Sheets, "
                  f"Kategorien={m['types_of_business']}, "
                  f"Daten {m['date_first_introduction']}..{m['date_last_introduction']}")


if __name__ == "__main__":
    main()