tryton-db-migration/rsts_merge_dedup.py

#!/usr/bin/env python3
"""
rsts_merge_dedup.py

Liest alle von rst_to_dataset.py erzeugten *.yaml-Dateien aus einem
Verzeichnis ein, entfernt Records mit identischem model+fields (die Keys
unterscheiden sich zwar je Datei durch das Präfix, der Inhalt ist aber oft
gleich, z. B. dasselbe Basis-product.template in vielen Szenarien) und
schreibt eine einzige zusammengeführte, nach model sortierte Datei.

$ref-Verweise auf entfernte Duplikate werden automatisch auf den jeweils
behaltenen Record umgeschrieben.

ACHTUNG: Es wird bewusst GLOBAL dedupliziert, auch innerhalb derselben
Datei. Das kann bei Szenarien, die zwei absichtlich gleich aussehende
Records anlegen (z. B. scenario_party_replace.yaml: zwei Parteien namens
"Pam", um eine davon zu ersetzen), einen der beiden Records entfernen und
damit den Sinn des Szenarios verändern. Für den Zweck hier (grober
Testdatensatz) ist das laut Absprache in Ordnung - falls die Datei später
1:1 für szenario-treue Tests verwendet werden soll, nochmal prüfen.

Nutzung:
    python3 rsts_merge_dedup.py rsts/8e --output reports/8_testdata_merged.yaml
"""

import argparse
import heapq
import json
import sys
from pathlib import Path

try:
    import yaml
except ImportError:
    yaml = None


def load_records(path):
    text = path.read_text(encoding="utf-8")
    if yaml is not None:
        data = yaml.safe_load(text)
    else:
        data = json.loads(text)
    return data.get("records", [])


def rewrite_refs(value, rename):
    if isinstance(value, dict):
        if "$ref" in value:
            return {"$ref": rename.get(value["$ref"], value["$ref"])}
        return {k: rewrite_refs(v, rename) for k, v in value.items()}
    if isinstance(value, list):
        return [rewrite_refs(v, rename) for v in value]
    return value


def merge_and_dedupe(input_dir):
    all_records = []
    for path in sorted(Path(input_dir).glob("*.yaml")):
        for record in load_records(path):
            all_records.append((path.name, record))

    # Signatur = model + fields (Key bewusst ignoriert, da der sich je
    # Datei nur durchs Präfix unterscheidet).
    canonical_key = {}   # signatur -> key des ersten Fundes
    rename = {}          # jeder key -> key des ersten Fundes derselben Signatur
    kept = []
    removed = 0

    for source, record in all_records:
        signature = (record["model"], json.dumps(record["fields"], sort_keys=True))
        if signature in canonical_key:
            rename[record["key"]] = canonical_key[signature]
            removed += 1
            continue
        canonical_key[signature] = record["key"]
        rename[record["key"]] = record["key"]
        kept.append(record)

    for record in kept:
        record["fields"] = rewrite_refs(record["fields"], rename)

    kept = topo_sort_by_model(kept)

    return kept, removed


def find_refs(value):
    if isinstance(value, dict):
        if "$ref" in value:
            yield value["$ref"]
        else:
            for v in value.values():
                yield from find_refs(v)
    elif isinstance(value, list):
        for v in value:
            yield from find_refs(v)


def topo_sort_by_model(records):
    """Sortiert primär nach model, respektiert dabei aber zwingend die
    $ref-Abhängigkeiten (ein Record muss nach allem stehen, worauf er per
    $ref verweist). Kahn-Algorithmus: unter allen gerade freien Records
    (alle Abhängigkeiten schon platziert) wird immer der mit dem
    alphabetisch kleinsten model+key gewählt - das gruppiert gleiche
    models so weit, wie es die Abhängigkeiten zulassen."""
    by_key = {r["key"]: r for r in records}
    deps = {r["key"]: set() for r in records}
    dependents = {r["key"]: set() for r in records}

    for r in records:
        for ref in find_refs(r["fields"]):
            if ref in by_key and ref != r["key"]:
                deps[r["key"]].add(ref)
                dependents[ref].add(r["key"])

    indegree = {k: len(v) for k, v in deps.items()}
    heap = [(by_key[k]["model"], k) for k, deg in indegree.items() if deg == 0]
    heapq.heapify(heap)

    result = []
    while heap:
        _, key = heapq.heappop(heap)
        result.append(by_key[key])
        for dependent in dependents[key]:
            indegree[dependent] -= 1
            if indegree[dependent] == 0:
                heapq.heappush(heap, (by_key[dependent]["model"], dependent))

    if len(result) != len(records):
        stuck = set(by_key) - {r["key"] for r in result}
        raise ValueError(
            f"Zyklische oder unauflösbare $ref-Abhängigkeiten bei: {sorted(stuck)}"
        )

    return result


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("input_dir", help="Verzeichnis mit den *.yaml-Dateien (z. B. rsts/8e)")
    parser.add_argument("--output", required=True, help="Zieldatei für den zusammengeführten Datensatz")
    args = parser.parse_args()

    kept, removed = merge_and_dedupe(args.input_dir)

    doc = {"records": kept}
    out_path = Path(args.output)
    if yaml is not None:
        out_path.write_text(
            yaml.dump(doc, allow_unicode=True, sort_keys=False, default_flow_style=False),
            encoding="utf-8",
        )
    else:
        out_path.write_text(json.dumps(doc, indent=2, ensure_ascii=False), encoding="utf-8")

    print(f"{len(kept)} Records behalten, {removed} Duplikate entfernt -> {out_path}")


if __name__ == "__main__":
    sys.exit(main())