import hashlib, json

def split_rows(rows):
    parent = list(range(len(rows)))
    def root(i):
        while parent[i] != i:
            parent[i] = parent[parent[i]]
            i = parent[i]
        return i
    seen = {}
    for i, row in enumerate(rows):
        if not row.get("speaker_id") or not row.get("recording_id"):
            raise ValueError("Missing speaker_id or recording_id")
        for key in (("speaker", row["speaker_id"]), ("recording", row["recording_id"])):
            if key in seen:
                parent[root(i)] = root(seen[key])
            else:
                seen[key] = i
    components = {}
    for i, row in enumerate(rows):
        components.setdefault(root(i), []).append(row)
    groups = list(components.values())
    if len(groups) < 2:
        raise ValueError("Fewer than two independent components")
    def stable_key(group):
        ids = sorted((r["speaker_id"], r["recording_id"], r["file"]) for r in group)
        return hashlib.sha256(json.dumps(ids).encode()).hexdigest()
    groups.sort(key=stable_key)
    n_test = max(1, len(groups)//3)
    test = [r for g in groups[:n_test] for r in g]
    train = [r for g in groups[n_test:] for r in g]
    for key in ("speaker_id", "recording_id"):
        assert {r[key] for r in train}.isdisjoint({r[key] for r in test})
    return {"train": train, "test": test}

if __name__ == "__main__":
    rows = [
      {"file":"a1.wav","speaker_id":"s1","recording_id":"r1","seconds":3},
      {"file":"a2.wav","speaker_id":"s1","recording_id":"r2","seconds":4},
      {"file":"b1.wav","speaker_id":"s2","recording_id":"r2","seconds":2},
      {"file":"c1.wav","speaker_id":"s3","recording_id":"r3","seconds":5},
      {"file":"d1.wav","speaker_id":"s4","recording_id":"r4","seconds":6},
    ]
    result = split_rows(rows)
    print(json.dumps(result, ensure_ascii=False, indent=2))
    assert len(result["train"])+len(result["test"]) == len(rows)
    try:
        split_rows(rows[:3])
    except ValueError:
        print("single_component_rejected=True")
    else:
        raise AssertionError("Leakage case not rejected")