#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
kw_ebenen.py - Ebenenvergleich Gemeinde-/Stadtratswahl vs. Ortsrats-/Ortsbeiratswahl
auf Wahlbezirksebene (Schwerpunkt Briefwahlbezirke), Kommunalwahl Niedersachsen.

Datenquelle: votemanager Open-Data-CSV (z. B. KDO: https://votemanager.kdo.de/<termin>/<ags>/)
Format (Semikolon, UTF-8), verifiziert am 20.09.2026:
  datum;wahl;ags;gebiet-nr;gebiet-name;max-schnellmeldungen;anz-schnellmeldungen;
  A1;A2;A3;A;B;B1;C1;C2;D;D1_1;D1_3;D1_4;D1_2_1;...;D2_1;D2_3;D2_4;D2_2_1;...
  B  = Waehler, B1 = Waehler mit Wahlschein
  C1 = ungueltige Stimmzettel, C2 = gueltige Stimmzettel (ein blankes "C" gibt es nicht)
  D  = gueltige Stimmen; D<n>_1 Listenstimmen, D<n>_3 Kandidatenstimmen,
  D<n>_4 = _1 + _3 (die hier verwendete Listensumme), D<n>_2_<k> einzelne Kandidaten.
  Es gilt Summe D<n>_4 == D. Die Listennummern sind nicht lueckenlos.
Listennamen: open_data.json -> dateifelder[].parteien, je Wahl (kein manuelles parties.csv noetig).

Fremdquellen (nds/fremdquellen.csv): Nicht jede Kommune veroeffentlicht bei votemanager.kdo.de.
  system=votemanager  eigener votemanager-Server, gleiche API; discover nimmt ihn wie eine Instanz auf.
  system=ivu          IVU-Wahlpraesentation (gesamtergebnis.csv + kandidatenstimmen.zip je Wahl);
                      read_vm_csv rechnet sie beim Einlesen ins votemanager-Format um (_ivu_lesen).
  html/pdf/unbekannt  keine maschinenlesbaren Bezirksdaten gefunden - nur fuer die Liste der Luecken.
  Die Rohdateien bleiben unveraendert im Stand; <Stand>/<ags>/herkunft.json haelt fest, woher sie
  stammen und welche Felder abgeleitet sind (bei IVU: C2 = B - C, also ist C1 + C2 == B dort kein Test).

Subcommands
  probe      Findet votemanager-Instanzen fuer AGS-Praefixe (Phase 1: 03459, Phase 2: ganz NDS)
  discover   Listet Open-Data-CSVs je Instanz  -> sources.csv
  download   Laedt die CSVs aus sources.csv
  inspect    Zeigt Struktur einer CSV (Spalten, Wahlname, Bezirksnummern)
  plausi     Prueft je Bezirk Stimmzettel gegen Stimmen (ungenutzte/fehlende Stimmen)
  parties    Erzeugt parties.csv (Code D<n> -> Listenname) zum Pruefen/Ueberschreiben;
             normalerweise nicht noetig, die Namen kommen aus open_data.json
  analyze    Merge, Deltas, Aggregation, Top-N, Markdown-Report
  demo       Erzeugt SYNTHETISCHE Testdaten und laeuft analyze (Selbsttest)

Beispiel Phase 1:
  python kw_ebenen.py probe    --termin 20260913 --prefix 03459 --out instances.csv
  python kw_ebenen.py discover --instances instances.csv --out sources.csv
  python kw_ebenen.py download --sources sources.csv --dir data/raw
  python kw_ebenen.py analyze  --dir data/raw --out out/

Abhaengigkeiten: pandas, numpy, requests (nur fuer probe/discover/download)
"""
from __future__ import annotations

import argparse
import csv
import hashlib
import io
import json
import os
import re
import sys
import tempfile
import threading
import time
import unicodedata
import zipfile
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from zoneinfo import ZoneInfo
from email.utils import parsedate_to_datetime
from pathlib import Path

import numpy as np
import pandas as pd

try:
    import requests
except ImportError:  # analyze/demo laufen ohne requests
    requests = None

# --------------------------------------------------------------------------- Konfiguration
VM_BASE = "https://votemanager.kdo.de/{termin}/{ags}/"
# Kandidaten fuer die Dateiliste einer Instanz (werden der Reihe nach probiert).
INDEX_PATHS = ["daten/opendata/open_data.json", "daten/api/termin.json"]
USER_AGENT = "kw-ebenen-analyse/1.0 (Zettelcheck; https://zettelcheck.gitlab.io/; f.a.w.ostmann+zettelcheck@gmail.com)"
# Quellen ausserhalb von votemanager.kdo.de (eigene votemanager-Server, IVU, Luecken), siehe oben.
FREMDQUELLEN = "nds/fremdquellen.csv"
# Zwischenzertifikate, die einzelne Wahlserver nicht mitliefern (Heidekreis: GlobalSign RSA OV SSL CA
# 2018). Browser laden sie ueber die AIA-Angabe nach, requests nicht; sie ergaenzen certifi.
CERT_DIR = Path(__file__).resolve().parent / "nds" / "certs"
REQUEST_PAUSE_S = 0.5

# Kreisschluessel Niedersachsen (Phase 2). Region Hannover = 03241.
NDS_KREISE = [
    "03101", "03102", "03103", "03151", "03153", "03154", "03155", "03157", "03158", "03159",
    "03241", "03251", "03252", "03254", "03255", "03256", "03257",
    "03351", "03352", "03353", "03354", "03355", "03356", "03357", "03358", "03359", "03360", "03361",
    "03401", "03402", "03403", "03404", "03405",
    "03451", "03452", "03453", "03454", "03455", "03456", "03457", "03458", "03459", "03460", "03461", "03462",
]

META_COLS = ["datum", "wahl", "ags", "gebiet-nr", "gebiet-name"]
# Kommunalwahl NDS (Kumulieren/Panaschieren): D<n>_4 = Gesamtstimmen der Liste n
#   D<n>_1 = Stimmen fuer die Gesamtliste, D<n>_3 = Summe Kandidatenstimmen,
#   D<n>_4 = D<n>_1 + D<n>_3, D<n>_2_<k> = einzelne Kandidaten.
# Einfache Wahlen (1 Stimme, z. B. Demo-Format): D<n>.
PARTY_COL_SUM = re.compile(r"^D(\d+)_4$")     # Listensumme Kommunalwahl
PARTY_COL_PLAIN = re.compile(r"^D(\d+)$")     # Listensumme einfaches Format
PARTY_COL = PARTY_COL_PLAIN                  # Rueckwaertskompatibilitaet
MIN_VOTES_DEFAULT = 300                      # Mindest-Stimmen fuer Ranking (Rauschen)
CONGRUENCE_TOL = 0.02                        # max. rel. Abweichung Waehler GR vs. OR
SMALL_BRIEF_VOTERS = 50                      # Hinweis-Schwelle kleine Briefwahlbezirke (pruefen!)
MAJOR = ["CDU", "SPD", "GRÜNE", "FDP", "AFD", "LINKE", "BSW", "VOLT"]

RE_OR = re.compile(r"ortsrat|ortsbeirat|stadtbezirksrat|bezirksrat", re.I)
RE_SKIP = re.compile(r"kreis|landrat|regionsversammlung|b(ü|ue)rgermeister|direktwahl|stichwahl|"
                     r"oberb|samtgemeinde", re.I)
# "Wahl des Rates der Stadt Melle": \brat\b matcht "Rates" nicht -> flektierte Formen zulassen.
# "Kommunalwahl der Stadt Georgsmarienhuette" ist dort der Name der Ratswahl.
RE_GR = re.compile(r"gemeinde|stadtrat|ratswahl|\brat(es|e|s)?\b|stadtwahl|kommunalwahl", re.I)


def classify_election(name: str) -> str:
    """GR = Gemeinde-/Stadtrat, OR = Ortsrat/Ortsbeirat/Stadtbezirksrat, X = sonstiges."""
    if RE_OR.search(name):
        return "OR"
    if RE_SKIP.search(name):
        return "X"
    if RE_GR.search(name):
        return "GR"
    return "X"


# --------------------------------------------------------------------------- Netzwerk
_local = threading.local()


_ca_sperre = threading.Lock()
_ca_pfad: list[str] = []


def _ca_bundle():
    """certifi plus nds/certs/*.pem als eine Datei (requests nimmt nur einen Pfad)."""
    extra = sorted(CERT_DIR.glob("*.pem")) if CERT_DIR.is_dir() else []
    if not extra:
        return True
    with _ca_sperre:
        if not _ca_pfad:
            import certifi
            f = tempfile.NamedTemporaryFile("w", suffix=".pem", delete=False, encoding="utf-8")
            f.write(Path(certifi.where()).read_text(encoding="utf-8"))
            for p in extra:
                f.write("\n" + p.read_text(encoding="utf-8"))
            f.close()
            _ca_pfad.append(f.name)
    return _ca_pfad[0]


def _session():
    if getattr(_local, "s", None) is None:
        _local.s = requests.Session()
        _local.s.headers["User-Agent"] = USER_AGENT
        _local.s.verify = _ca_bundle()
    return _local.s


def _get(url: str, **kw):
    if requests is None:
        sys.exit("requests fehlt: pip install requests")
    time.sleep(REQUEST_PAUSE_S)   # Pause je Worker, nicht global
    return _session().get(url, timeout=30, **kw)


def _parallel(fn, items, jobs):
    """fn auf items anwenden; jobs <= 1 laeuft seriell."""
    if jobs <= 1:
        return [fn(x) for x in items]
    with ThreadPoolExecutor(max_workers=jobs) as ex:
        return list(ex.map(fn, items))


def cmd_probe(a):
    prefixes = NDS_KREISE if a.prefix == "NDS" else a.prefix.split(",")
    cands = []
    for kr in prefixes:
        cands += [f"{kr}{i:03d}" for i in range(0, a.max_gemeinde + 1)]
        cands += [f"{kr}{5400 + i}" for i in range(1, a.max_samtgemeinde + 1)]  # Samtgemeinden

    def probe_one(ags):
        base = VM_BASE.format(termin=a.termin, ags=ags)
        try:
            if _get(base + "praesentation/index.html").status_code == 200:
                print("gefunden:", base, flush=True)
                return {"ags": ags, "base": base}
        except Exception:
            pass
        return None

    print(f"{len(cands)} AGS-Kandidaten, {a.jobs} parallele Abfragen", flush=True)
    rows = [r for r in _parallel(probe_one, cands, a.jobs) if r]
    Path(a.out).parent.mkdir(parents=True, exist_ok=True)
    df = pd.DataFrame(rows)
    (df.sort_values("ags") if len(df) else df).to_csv(a.out, index=False)
    print(f"{len(rows)} Instanzen -> {a.out}")


def _walk_strings(obj):
    if isinstance(obj, dict):
        for v in obj.values():
            yield from _walk_strings(v)
    elif isinstance(obj, list):
        for v in obj:
            yield from _walk_strings(v)
    elif isinstance(obj, str):
        yield obj


FNAME = re.compile(r"Open-Data-(?P<ags>\d+)-(?P<wahl>.+?)-(?P<level>Stimmbezirk|Wahlbezirk|Briefwahlbezirk"
                   r"|Gruppierungen?|[A-Za-zÄÖÜäöü]+)\.csv", re.I)


def cmd_discover(a):
    """Liest daten/opendata/open_data.json je Instanz.

    Verifizierte Struktur (KDO, Stand 20.09.2026):
      {"csvs": [{"wahl": "Gemeinderatswahl", "ebene": "Wahlbezirk", "url": "Open-Data-....csv"}, ...],
       "dateifelder": [{"name": "<Wahlname>", "felder": [...], "parteien": [{"feld": "D1", "wert": "CDU ..."}]}],
       "wahllokal_felder": [...]}
    Der Wahlname kommt aus dem JSON, nicht aus dem Dateinamen (dort sind Umlaute transliteriert).
    open_data.json selbst wird als type=META mitgefuehrt: es liefert spaeter die Listennamen.

    Dazu kommen die Fremdquellen (--fremd): eigene votemanager-Server wie Instanzen, IVU-Wahlen mit je
    zwei Dateien (gesamtergebnis.csv, kandidatenstimmen.zip) unter eigenem Dateinamen (Spalte datei).
    """
    inst = pd.read_csv(a.instances, dtype=str)
    fremd = _fremdquellen(a.fremd)
    vm_fremd = fremd[fremd.system == "votemanager"].rename(columns={"url": "base"})[["ags", "base"]]
    inst = pd.concat([inst[~inst.ags.isin(vm_fremd.ags)], vm_fremd], ignore_index=True)

    def one(row):
        out = []
        od, base_od = None, row.base + "daten/opendata/"
        try:
            r = _get(row.base + INDEX_PATHS[0])
            if r.status_code == 200:
                od = r.json()
        except Exception as e:
            print("  Index nicht lesbar:", row.base + INDEX_PATHS[0], e)
        if od and od.get("csvs"):
            for c in od["csvs"]:
                u = c["url"]
                out.append({"ags": row.ags, "url": u if u.startswith("http") else base_od + u,
                            "wahl": c.get("wahl", ""), "level": c.get("ebene", ""),
                            "type": classify_election(c.get("wahl", ""))})
            # Metadateien: open_data.json (Listennamen) und Wahllokale (Bezirk-Art W/B).
            out.append({"ags": row.ags, "url": row.base + INDEX_PATHS[0],
                        "wahl": "", "level": "open_data.json", "type": "META"})
            if od.get("wahllokal_felder"):
                out.append({"ags": row.ags, "url": base_od + "opendata-wahllokale.csv",
                            "wahl": "", "level": "Wahllokale", "type": "META"})
            print(f"[{row.ags}] {len(od['csvs'])} CSVs", flush=True)
            return out
        # Fallback: irgendeinen Index nach .csv-Strings durchsuchen
        files = set()
        for p in INDEX_PATHS:
            try:
                r = _get(row.base + p)
                if r.status_code != 200:
                    continue
                for s in _walk_strings(r.json()):
                    if s.lower().split("?")[0].endswith(".csv"):
                        files.add(s)
            except Exception as e:
                print("  Index nicht lesbar:", row.base + p, e)
        if not files:
            print(f"[{row.ags}] keine CSV-Liste gefunden -> URLs manuell in sources.csv eintragen")
        for f in sorted(files):
            url = f if f.startswith("http") else base_od + f.split("/")[-1]
            m = FNAME.search(url)
            wahl = m.group("wahl") if m else ""
            out.append({"ags": row.ags, "url": url, "wahl": wahl,
                        "level": m.group("level") if m else "",
                        "type": classify_election(wahl)})
        return out

    print(f"{len(inst)} votemanager-Instanzen (davon {len(vm_fremd)} auf eigenen Servern), "
          f"{a.jobs} parallele Abfragen", flush=True)
    out = [r for chunk in _parallel(one, [r for _, r in inst.iterrows()], a.jobs) for r in chunk]
    for r in out:
        r["system"] = "votemanager"
    ivu = [r for _, f in fremd[fremd.system == "ivu"].iterrows() for r in _ivu_quellen(f)]
    print(f"IVU: {len(ivu) // 3} Wahlen", flush=True)
    df = pd.DataFrame(out + ivu, columns=QUELLEN_SPALTEN)
    Path(a.out).parent.mkdir(parents=True, exist_ok=True)
    df.to_csv(a.out, index=False)
    print(df.groupby(["type", "level"]).size() if len(df) else "leer")


def cmd_download(a):
    src = _quellen_lesen(a.sources)
    # Nur unterste Ebene der GR/OR-Wahlen + Metadateien (open_data.json, Wahllokale).
    lvl_ok = src.level.str.contains("bezirk", case=False) | (src.level == "")
    typen = ["GR", "OR", "X"] if a.alle else ["GR", "OR"]
    keep = src[(src.type.isin(typen) & lvl_ok)
               | (src.type == "META")
               | src.level.str.contains("ruppier", case=False)]
    print(f"{len(keep)} von {len(src)} Eintraegen werden geladen"
          f"{' (alle Wahlarten)' if a.alle else ' (nur Gemeinde-/Ortsratswahlen)'}"
          f", {a.jobs} parallel", flush=True)
    zaehler = {"ok": 0, "err": 0, "da": 0}
    sperre = threading.Lock()

    def hole(r):
        target = Path(a.dir) / r.ags / r.datei
        if target.exists() and not a.force:
            with sperre:
                zaehler["da"] += 1
            return
        target.parent.mkdir(parents=True, exist_ok=True)
        try:
            resp = _get(r.url)
        except Exception as e:
            with sperre:
                zaehler["err"] += 1
            print("ERR", e, r.url, flush=True)
            return
        with sperre:
            if resp.status_code == 200:
                target.write_bytes(resp.content)
                zaehler["ok"] += 1
                if zaehler["ok"] % 250 == 0:
                    print(f"  ... {zaehler['ok']} geladen", flush=True)
            else:
                zaehler["err"] += 1
                print("ERR", resp.status_code, r.url, flush=True)

    _parallel(hole, [r for _, r in keep.iterrows()], a.jobs)
    _herkunft_schreiben(Path(a.dir), keep)
    print(f"geladen: {zaehler['ok']} · bereits vorhanden: {zaehler['da']} · Fehler: {zaehler['err']}")


# --------------------------------------------------------------------------- Fremdquellen
QUELLEN_SPALTEN = ["ags", "url", "wahl", "level", "type", "system", "datei", "teile"]
FREMD_SPALTEN = ["ags", "name", "system", "url", "wahl", "teile", "hinweis"]


def _quellen_lesen(sources_csv: str) -> pd.DataFrame:
    """sources.csv lesen; aeltere Fassungen ohne system/datei/teile bekommen die Standardwerte."""
    src = pd.read_csv(sources_csv, dtype=str).fillna("")
    for c in QUELLEN_SPALTEN:
        if c not in src.columns:
            src[c] = ""
    src.loc[src.system == "", "system"] = "votemanager"
    ohne = src.datei == ""
    src.loc[ohne, "datei"] = src.loc[ohne, "url"].str.split("/").str[-1].str.split("?").str[0]
    return src


def _fremdquellen(pfad: str) -> pd.DataFrame:
    if not pfad or not Path(pfad).exists():
        return pd.DataFrame(columns=FREMD_SPALTEN)
    return pd.read_csv(pfad, dtype=str).fillna("")


def _ivu_slug(url: str) -> str:
    """Kennung einer IVU-Wahl aus den letzten beiden Pfadteilen: .../kw2026/gr_lac/ -> kw2026-gr_lac."""
    return re.sub(r"[^A-Za-z0-9_]+", "-", "-".join(url.rstrip("/").split("/")[-2:]))


def _ivu_quellen(f) -> list[dict]:
    """Zwei Quellzeilen je IVU-Wahl. Der Dateiname traegt ags und Kennung, weil alle IVU-Wahlen
    gleich heissende Dateien haben."""
    stamm = f"IVU-{f.ags}-{_ivu_slug(f.url)}"
    basis = {"ags": f.ags, "wahl": f.wahl, "system": "ivu", "teile": f.teile}
    return [{**basis, "url": f.url + "gesamtergebnis.csv", "level": "Stimmbezirk",
             "type": classify_election(f.wahl), "datei": f"{stamm}-gesamtergebnis.csv"},
            {**basis, "url": f.url + "kandidatenstimmen.zip", "level": "", "type": "META",
             "datei": f"{stamm}-kandidatenstimmen.zip"},
            {**basis, "url": f.url + "parteistimmen.csv", "level": "", "type": "META",
             "datei": f"{stamm}-parteistimmen.txt"}]


def _herkunft_schreiben(stand: Path, keep: pd.DataFrame):
    """<ags>/herkunft.json fuer jede Kommune, deren Dateien nicht von votemanager.kdo.de kommen.

    Interner Hinweis fuer alle, die die Rohdaten lesen, und fuer _ivu_lesen (Wahlname, Teilgemeinden).
    Kein Abruf, sondern aus sources.csv abgeleitet; steht deshalb nicht im Manifest.
    """
    fremd = keep[(keep.system != "votemanager") | ~keep.url.str.contains("//votemanager.kdo.de/", regex=False)]
    for ags, g in fremd.groupby("ags"):
        d = stand / ags
        if not d.is_dir():
            continue
        dateien = {}
        for _, r in g.iterrows():
            e = {"system": r.system, "url": r.url}
            if r.system == "ivu" and r.type != "META":
                e.update(wahl=r.wahl, teile=r.teile, abgeleitet=["C2"],
                         umrechnung="IVU gesamtergebnis.csv -> votemanager-Spalten beim Einlesen (_ivu_lesen)")
            dateien[r.datei] = e
        systeme = sorted(set(g.system))
        (d / "herkunft.json").write_text(json.dumps({
            "hinweis": "Nicht von votemanager.kdo.de. "
                       + ("IVU-Wahlpraesentation: C2 (gueltige Stimmzettel) ist aus B - C abgeleitet, "
                          "die Pruefung C1 + C2 == B deshalb dort ohne Aussage. " if "ivu" in systeme else "")
                       + "Rohdateien unveraendert.",
            "systeme": systeme, "dateien": dateien}, ensure_ascii=False, indent=1), encoding="utf-8")


# Wolfsburg (eigener votemanager-Server) benennt die Listenspalten anders:
#   D<n>_liste, D<n>_summe_kandidaten, D<n>_summe_liste_kandidaten, D<n>_<k> (Kandidat k)
# statt D<n>_1, D<n>_3, D<n>_4, D<n>_2_<k>. D<n>_4 bedeutet dort also Kandidat 4 - ohne
# Umbenennung wuerde party_cols still die falschen Spalten summieren.
RE_LISTE_VARIANTE = re.compile(r"^D(\d+)_(liste|summe_kandidaten|summe_liste_kandidaten|\d+)$")


def _spalten_angleichen(df: pd.DataFrame) -> pd.DataFrame:
    if not any(c.endswith("_liste") for c in df.columns):
        return df
    ziel = {"liste": "1", "summe_kandidaten": "3", "summe_liste_kandidaten": "4"}
    neu = {}
    for c in df.columns:
        m = RE_LISTE_VARIANTE.match(c)
        if m:
            n, t = m.groups()
            neu[c] = f"D{n}_{ziel[t]}" if t in ziel else f"D{n}_2_{t}"
    df = df.rename(columns=neu)
    df.attrs["variante"] = "votemanager, Spalten D<n>_liste/_summe_* umbenannt"
    return df


def _ivu_lesen(path: Path) -> pd.DataFrame:
    """IVU gesamtergebnis.csv (eine Zeile je Stimm-/Briefwahlbezirk) -> votemanager-Spalten.

      A1, A2, A3, A      Wahlberechtigte ohne/mit Wahlschein/nicht im WVZ/gesamt
      B, B1              Waehler gesamt, Waehler mit Wahlschein
      C1                 "Stimmen ungueltige (C)" - laut Praesentation "Ungueltige Stimmzettel"
      C2 = B - C1        abgeleitet (IVU nennt die gueltigen Stimmzettel nicht)
      D, D<n>_4          gueltige Stimmen, Stimmen der Liste n (IVU: D<n>)
      D<n>_2_<k>         Kandidat k (IVU: D<n>_<k>); D<n>_3 = deren Summe
      D<n>_1             Listenstimmen aus kandidatenstimmen.zip; fehlt sie, D<n> - D<n>_3
    Listennamen: Die Nummer n ordnet IVU nicht ausdruecklich zu. Sie ergibt sich aus dem Vergleich der
    Stimmen je Bezirk: in kandidatenstimmen.zip (Listen- plus Kandidatenstimmen je Gruppe) fuer die
    Listenstimmen, in parteistimmen.csv (Kurzname, Stimmen) fuer den angezeigten Namen. Die
    parteistimmen liegen als .txt im Stand, damit sie nicht als Bezirksdatei gelesen werden.
    Mehrere Gemeinden in einer Datei (Raete der Mitgliedsgemeinden): Spalte Gemeinde, Namen aus
    herkunft.json (teile), Wahlname "Gemeinderatswahl {teil}". Nicht ausgezaehlte Bezirke
    (Ausbleibend = J, mit einem anderen Bezirk zusammen gezaehlt) entfallen.
    """
    her = json.loads((path.parent / "herkunft.json").read_text(encoding="utf-8"))["dateien"][path.name]
    roh = pd.read_csv(path, sep=";", dtype=str, encoding="utf-8-sig").fillna("")
    roh = roh[roh["Ausbleibend"].str.upper() != "J"]
    teile = dict(t.split("=", 1) for t in her.get("teile", "").split("|") if t)
    listen = sorted({int(m.group(1)) for c in roh.columns if (m := re.fullmatch(r"D(\d+)", c))})

    def zahl(v):
        return float(v) if str(v).strip() != "" else np.nan

    # kandidatenstimmen.zip: je (Bezirk, Gruppe) Listenstimmen und Summe; dazu die Bezirksnamen
    kand, namen, dateiname = {}, {}, {}
    zp = path.with_name(path.name.replace("-gesamtergebnis.csv", "-kandidatenstimmen.zip"))
    if zp.exists():
        with zipfile.ZipFile(zp) as z:
            for n in z.namelist():
                if not n.lower().endswith(".csv"):
                    continue
                k = pd.read_csv(io.TextIOWrapper(z.open(n), encoding="utf-8-sig"), sep=";", dtype=str).fillna("")
                k = k[k.Zeilenart == "ERFASSUNGSEINHEIT"]
                # Samtgemeinde Aue: Gruppe-Schluessel nur "1", "2" ...; der Dateiname nennt die Liste
                for g in set(k["Gruppe-Schluessel"]):
                    if g.strip().isdigit():
                        dateiname[g] = Path(n).stem.rsplit("-", 1)[-1].replace("_", " ").upper()
                kc = [c for c in k.columns[13:] if c.strip()]
                for _, r in k.iterrows():
                    lst = zahl(r.Listenstimmen)
                    ges = (0 if np.isnan(lst) else lst) + sum(0 if np.isnan(zahl(r[c])) else zahl(r[c]) for c in kc)
                    kand[(r.Gebietsnummer, r["Gruppe-Schluessel"])] = (lst, ges)
                    namen.setdefault(r.Gebietsnummer, r.Gebietsname)
    # parteistimmen.csv: Gebietsart;Gebietsnummer;A1;A2;A3;Waehler ohne/mit Wahlschein;Briefwaehler in
    # Urnenwahlgebiet;gueltige Stimmen;ungueltige; dann Paare Kurzname;Stimmen. Die Gebietsnummer ist
    # dort anders geschrieben als in gesamtergebnis.csv (Harburg: 0335034003 statt 033530340003),
    # deshalb Zuordnung ueber die Kennzahlen des Bezirks.
    def kennung(werte):
        return tuple(0.0 if str(v).strip() == "" else float(v) for v in werte)
    kurz_k = {}
    pp = path.with_name(path.name.replace("-gesamtergebnis.csv", "-parteistimmen.txt"))
    if pp.exists():
        for zeile in pp.read_text(encoding="utf-8-sig").splitlines()[1:]:
            t = zeile.split(";")
            if len(t) > 11:
                paare = [(n.strip(), float(v)) for n, v in zip(t[10::2], t[11::2]) if n.strip() and v.strip()]
                kurz_k.setdefault(kennung(t[2:7] + t[8:10]), []).append(paare)
    kurz = {}
    for _, r in roh.iterrows():
        k = kennung([r[c] for c in ("Wahlberechtigte ohne Wahlschein (A1)", "Wahlberechtigte mit Wahlschein (A2)",
                                    "Wahlberechtigte nicht im WVZ (A3)", "Waehler ohne Wahlschein",
                                    "Waehler mit Wahlschein", "Stimmen gueltige (D)", "Stimmen ungueltige (C)")])
        if len(kurz_k.get(k, [])) == 1:   # nur eindeutige Bezirke
            kurz[r.Erfassungsgebietsnummer.strip()] = kurz_k[k][0]

    # Liste n -> Gruppe: die Gruppe, deren Stimmen in den meisten Bezirken D<n> entsprechen. Je
    # Gemeinde getrennt: In den Raeten der Mitgliedsgemeinden steht dieselbe Nummer n fuer
    # verschiedene Waehlergruppen, beim Kreistag je Wahlbereich (der eine Gemeinde nicht teilt).
    def zuordnen(nach_bezirk):
        erg = {}
        for gem, rg in roh.groupby("Gemeinde"):
            for n in listen:
                treffer = {}
                for _, r in rg.iterrows():
                    v = zahl(r[f"D{n}"])
                    if np.isnan(v) or v == 0:
                        continue
                    kand_gr = [gr for gr, ges in nach_bezirk.get(r.Erfassungsgebietsnummer.strip(), []) if ges == v]
                    if len(kand_gr) == 1:
                        treffer[kand_gr[0]] = treffer.get(kand_gr[0], 0) + 1
                if treffer:
                    erg[(gem, n)] = max(treffer, key=treffer.get)
        return erg

    nach_bezirk = {}
    for (nr, gr), (_, ges) in kand.items():
        nach_bezirk.setdefault(nr, []).append((gr, ges))
    gruppen = zuordnen(nach_bezirk)     # (Gemeinde, n) -> Gruppe in kandidatenstimmen.zip
    anzeige = zuordnen(kurz)            # (Gemeinde, n) -> Kurzname aus parteistimmen.csv

    zeilen, abgeleitet, parteien = [], {"C2"}, {}
    for _, r in roh.iterrows():
        nr = r.Erfassungsgebietsnummer.strip()
        teil = teile.get(r.Gemeinde.strip(), "")
        wahl = her["wahl"].replace("{teil}", teil or r.Gemeinde.strip())
        B, C = zahl(r["Waehler gesamt (B)"]), zahl(r["Stimmen ungueltige (C)"])
        z = {"datum": "13.09.2026", "wahl": wahl, "ags": path.parent.name, "gebiet-nr": nr,
             "gebiet-name": namen.get(nr, ""),
             "A1": zahl(r["Wahlberechtigte ohne Wahlschein (A1)"]),
             "A2": zahl(r["Wahlberechtigte mit Wahlschein (A2)"]),
             "A3": zahl(r["Wahlberechtigte nicht im WVZ (A3)"]), "A": zahl(r["Wahlberechtigte gesamt (A)"]),
             "B": B, "B1": zahl(r["Waehler mit Wahlschein"]), "C1": C, "C2": B - C,
             "D": zahl(r["Stimmen gueltige (D)"])}
        for n in listen:
            ges = zahl(r[f"D{n}"])
            kc = [c for c in roh.columns if re.fullmatch(rf"D{n}_\d+", c)]
            ks = [zahl(r[c]) for c in kc]
            if np.isnan(ges):
                continue   # Liste tritt in diesem Bezirk nicht an
            s3 = sum(0 if np.isnan(v) else v for v in ks)
            gr = gruppen.get((r.Gemeinde, n))
            name = anzeige.get((r.Gemeinde, n)) or gr
            name = dateiname.get(str(name).strip(), name)
            if name:
                parteien.setdefault(nr, {})[f"D{n}"] = _kurzname(name)
            lst = kand.get((nr, gr), (np.nan, np.nan))[0]
            if np.isnan(lst):
                lst = ges - s3
                abgeleitet.add(f"D{n}_1")
            z.update({f"D{n}_1": lst, f"D{n}_3": s3, f"D{n}_4": ges})
            z.update({f"D{n}_2_{c.split('_')[1]}": v for c, v in zip(kc, ks)})
        zeilen.append(z)
    df = pd.DataFrame(zeilen)
    for c in META_COLS:
        df[c] = df[c].astype(str)
    # Listennamen je Bezirk: Beim Kreistag vergibt IVU die Nummern je Wahlbereich
    df.attrs.update(system="ivu", abgeleitet=sorted(abgeleitet), parteien_je_bezirk=parteien)
    return df


# --------------------------------------------------------------------------- Laden
def read_vm_csv(path: Path) -> pd.DataFrame:
    """votemanager-Open-Data-CSV lesen; IVU-Dateien und Spaltenvarianten werden angeglichen."""
    path = Path(path)
    with open(path, encoding="utf-8-sig") as f:
        kopf = f.readline()
    if "Erfassungsgebietsnummer" in kopf:
        return _ivu_lesen(path)
    df = pd.read_csv(path, sep=";", dtype=str, encoding="utf-8-sig")
    df.columns = [c.strip() for c in df.columns]
    df = _spalten_angleichen(df)
    for c in df.columns:
        if c not in META_COLS:
            df[c] = pd.to_numeric(df[c], errors="coerce")
    df["gebiet-nr"] = df["gebiet-nr"].astype(str).str.strip()
    df["ags"] = df["ags"].astype(str).str.strip()
    namen = _wahlnamen(path.parent)
    if namen and "wahl" in df.columns:
        df["wahl"] = df["wahl"].map(lambda w: namen.get(str(w).strip(), w))
    return df


def party_cols(df):
    """Listensummen-Spalten. Bevorzugt D<n>_4 (Kommunalwahl), sonst D<n>."""
    cols = [c for c in df.columns if PARTY_COL_SUM.match(c)]
    if cols:
        return sorted(cols, key=lambda c: int(PARTY_COL_SUM.match(c).group(1)))
    cols = [c for c in df.columns if PARTY_COL_PLAIN.match(c)]
    return sorted(cols, key=lambda c: int(c[1:]))


def party_code(col: str) -> str:
    """Spaltenname -> Listencode der parties.csv: 'D6_4' -> 'D6', 'D6' -> 'D6'."""
    m = PARTY_COL_SUM.match(col) or PARTY_COL_PLAIN.match(col)
    return f"D{m.group(1)}" if m else col


def invalid_col(df) -> str | None:
    """Spalte 'ungueltig': C1 = ungueltige Stimmzettel (Kommunalwahl), sonst C."""
    for c in ("C1", "C"):
        if c in df.columns:
            return c
    return None


def is_brief(nr: str, name: str) -> bool:
    """Heuristik: 'B'-Praefix, 'Brief' im Namen oder letzte drei Ziffern >= 900 (z. B. 0990, 05903)."""
    if re.match(r"^\s*B", str(nr), re.I) or re.search(r"brief", str(name), re.I):
        return True
    digits = re.sub(r"\D", "", str(nr))
    return len(digits) >= 3 and int(digits[-3:]) >= 900


def numbering_pattern(nr: str) -> str:
    s = str(nr)
    if re.match(r"^B", s, re.I):
        return "B-Präfix"
    d = re.sub(r"\D", "", s)
    if len(d) >= 3 and d[-3:-1] == "99":
        return "99x"
    if len(d) >= 3 and d[-3] == "9":
        return "9xx"
    return "andere"


def load_all(data_dir: Path):
    frames = []
    for p in sorted(Path(data_dir).rglob("*.csv")):
        if re.search(r"wahllokal|strassen|gruppier|parties", p.name, re.I):
            continue
        m = FNAME.search(p.name)
        if m and m.group("level").lower() not in ("wahlbezirk", "stimmbezirk", "briefwahlbezirk"):
            continue  # nur unterste Ebene
        try:
            df = read_vm_csv(p)
        except Exception as e:
            print("Uebersprungen:", p, e)
            continue
        if not {"wahl", "gebiet-nr"}.issubset(df.columns):
            continue
        df["source_file"] = str(p)
        df["etype"] = df["wahl"].map(classify_election)
        frames.append(df)
    if not frames:
        sys.exit(f"Keine votemanager-CSVs in {data_dir}")
    return frames


def load_parties(path: Path | None, data_dir: Path) -> pd.DataFrame:
    """parties.csv: ags;wahl;code;name  (code = D1, D2, ...)."""
    if path and Path(path).exists():
        pm = pd.read_csv(path, sep=None, engine="python", dtype=str)
        pm.columns = [c.strip().lower() for c in pm.columns]
        return pm[["ags", "wahl", "code", "name"]].dropna(subset=["name"])
    # Fallback 1: open_data.json der Instanz -> dateifelder[].parteien (offizielle Listennamen)
    rows = []
    for p in sorted(Path(data_dir).rglob("open_data*.json")):
        try:
            d = json.loads(p.read_text(encoding="utf-8"))
        except Exception as e:
            print("open_data.json nicht lesbar:", p, e)
            continue
        ags = p.parent.name
        for eintrag in d.get("dateifelder", []):
            for f in eintrag.get("parteien", []):
                rows.append({"ags": ags, "wahl": eintrag.get("name", ""),
                             "code": f["feld"], "name": f["wert"]})
    if rows:
        return pd.DataFrame(rows).drop_duplicates(subset=["ags", "wahl", "code"])

    # Fallback 2: Gruppierungs-Dateien heuristisch lesen
    rows = []
    for p in Path(data_dir).rglob("*ruppier*.csv"):
        g = pd.read_csv(p, sep=";", dtype=str, encoding="utf-8-sig")
        cols = {c.lower(): c for c in g.columns}
        ncol = next((cols[c] for c in cols if re.search(r"kurz|name|bezeich", c)), None)
        ccol = next((cols[c] for c in cols if re.search(r"^nr|nummer|kennz|spalte|position", c)), None)
        if ncol and ccol:
            for _, r in g.iterrows():
                code = str(r[ccol]).strip()
                code = code if code.startswith("D") else f"D{int(float(code))}"
                rows.append({"ags": r.get("ags", ""), "wahl": r.get("wahl", ""), "code": code,
                             "name": r[ncol]})
    return pd.DataFrame(rows, columns=["ags", "wahl", "code", "name"])


# --------------------------------------------------------------------------- Parteinamen
ALIASES = {
    "BUNDNIS90DIEGRUNEN": "GRÜNE", "GRUNE": "GRÜNE", "DIEGRUNEN": "GRÜNE", "B90GRUNE": "GRÜNE",
    "DIELINKE": "LINKE", "LINKE": "LINKE", "AFDNIEDERSACHSEN": "AFD", "AFD": "AFD",
    "CDU": "CDU", "SPD": "SPD", "FDP": "FDP", "BSW": "BSW", "VOLT": "VOLT", "DIEBASIS": "BASIS",
}


def _fold(s):
    s = unicodedata.normalize("NFKD", str(s)).encode("ascii", "ignore").decode()
    return re.sub(r"[^A-Z0-9]", "", s.upper())


def norm_party(name: str) -> str:
    if re.search(r"einzelbewerb|^eb[\s:.-]", str(name), re.I):
        return "EB: " + re.sub(r"(?i)einzelbewerber(in)?[:\s-]*|^eb[\s:.-]+", "", str(name)).strip()
    return ALIASES.get(_fold(name), str(name).strip())


def party_cat(p: str) -> str:
    if p.startswith("EB: "):
        return "Einzelbewerber"
    if p in MAJOR:
        return "Partei (überregional)"
    if re.search(r"wg|wähler|waehler|bürger|buerger|unabh|liste|uwg", p, re.I):
        return "Wählergruppe"
    return "Sonstige"


# --------------------------------------------------------------------------- Kern
def to_long(df: pd.DataFrame, pm: pd.DataFrame, side: str) -> pd.DataFrame:
    cols = party_cols(df)
    if not cols:
        raise KeyError(f"{df.ags.iloc[0]} / {df.wahl.iloc[0]}: keine Listenspalten (D<n>_4 bzw. D<n>) gefunden")
    ic = invalid_col(df)
    keep = ["ags", "wahl", "gebiet-nr", "gebiet-name", "B"]
    base = df[keep + [ic]].rename(columns={ic: "C"}).copy() if ic else df[keep].assign(C=np.nan)
    long = df.melt(id_vars=["ags", "wahl", "gebiet-nr"], value_vars=cols, var_name="code", value_name="votes")
    long["code"] = long.code.map(party_code)
    mp = pm[(pm.ags == df.ags.iloc[0]) & (pm.wahl == df.wahl.iloc[0])].set_index("code")["name"]
    missing = sorted(set(long.code) - set(mp.index))
    if missing:
        raise KeyError(f"{df.ags.iloc[0]} / {df.wahl.iloc[0]}: kein Listenname fuer {missing} "
                       f"-> parties.csv ergaenzen (python kw_ebenen.py parties ...)")
    long["party"] = long.code.map(mp).map(norm_party)
    long["votes"] = long.votes.fillna(0)
    long = long.groupby(["ags", "wahl", "gebiet-nr", "party"], as_index=False).votes.sum()
    long = long.merge(base, on=["ags", "wahl", "gebiet-nr"], how="left")
    return long.rename(columns={"votes": f"votes_{side}", "B": f"B_{side}", "C": f"C_{side}",
                                "wahl": f"wahl_{side}"})


def build_pairs(frames, pm):
    """Long-Tabelle: je (AGS, Ortsrat, Bezirk, Partei) Stimmen auf beiden Ebenen."""
    # Samtgemeinden fuehren mehrere Gemeinderatswahlen (je Mitgliedsgemeinde) unter
    # derselben AGS. Darum je AGS eine Liste und Auswahl ueber die Bezirksueberlappung.
    gr = {}
    for f in frames:
        if f.etype.iloc[0] == "GR":
            gr.setdefault(f.ags.iloc[0], []).append(f)
    ors = [f for f in frames if f.etype.iloc[0] == "OR"]
    out, log = [], []
    for o in ors:
        ags = o.ags.iloc[0]
        cands = gr.get(ags, [])
        if not cands:
            log.append(f"{ags} {o.wahl.iloc[0]}: keine Gemeindewahl-Datei")
            continue
        onr = set(o["gebiet-nr"])
        g = max(cands, key=lambda f: len(onr & set(f["gebiet-nr"])))
        if not (onr & set(g["gebiet-nr"])):
            log.append(f"{ags} {o.wahl.iloc[0]}: keine gemeinsamen Bezirksnummern mit einer "
                       f"Gemeindewahl ({len(cands)} Kandidat(en))")
            continue
        if len(cands) > 1:
            log.append(f"{ags} {o.wahl.iloc[0]}: zugeordnet zu '{g.wahl.iloc[0]}' "
                       f"({len(onr & set(g['gebiet-nr']))} von {len(onr)} Bezirken)")
        g_sub = g[g["gebiet-nr"].isin(o["gebiet-nr"])]
        unmatched = sorted(set(o["gebiet-nr"]) - set(g["gebiet-nr"]))
        if unmatched:
            log.append(f"{ags} {o.wahl.iloc[0]}: Bezirke ohne GR-Gegenstueck: {unmatched}")
        try:
            lg = to_long(g_sub, pm, "gr")
            lo = to_long(o[o["gebiet-nr"].isin(g_sub["gebiet-nr"])], pm, "or")
        except KeyError as e:
            log.append(str(e))
            continue
        on_or = set(lo.party)
        on_gr = set(lg.party)
        m = lg.drop(columns=["gebiet-name"]).merge(
            lo, on=["ags", "gebiet-nr", "party"], how="outer")
        # Bezirksattribute vervollstaendigen
        for c in ["B_gr", "C_gr", "wahl_gr"]:
            m[c] = m.groupby("gebiet-nr")[c].transform("first")
        for c in ["B_or", "C_or", "wahl_or", "gebiet-name"]:
            m[c] = m.groupby("gebiet-nr")[c].transform("first")
        m[["votes_gr", "votes_or"]] = m[["votes_gr", "votes_or"]].fillna(0)
        m["on_or"] = m.party.isin(on_or)
        m["on_gr"] = m.party.isin(on_gr)
        m["ortsrat"] = o.wahl.iloc[0]
        out.append(m)
    if not out:
        sys.exit("Keine GR/OR-Paare gebildet.\n" + "\n".join(log))
    L = pd.concat(out, ignore_index=True)
    L["is_brief"] = [is_brief(n, nm) for n, nm in zip(L["gebiet-nr"], L["gebiet-name"])]
    L["cat"] = L.party.map(party_cat)
    L["orphan_votes_gr"] = np.where(L.on_or, 0, L.votes_gr)
    L["kreis"] = L.ags.str[:5]
    return L, log


def district_table(L: pd.DataFrame):
    keys = ["kreis", "ags", "ortsrat", "gebiet-nr", "gebiet-name", "is_brief"]
    L = L.copy()
    tot = L.groupby(keys).agg(sum_gr=("votes_gr", "sum"), sum_or=("votes_or", "sum"))
    L = L.join(tot, on=keys)
    L["pct_gr"] = 100 * L.votes_gr / L.sum_gr
    L["pct_or"] = 100 * L.votes_or / L.sum_or
    L["d_votes"] = L.votes_or - L.votes_gr
    L["d_pp"] = L.pct_or - L.pct_gr
    orphan = L.groupby(keys).apply(lambda x: x.loc[~x.on_or, "pct_gr"].sum(), include_groups=False)
    L = L.join(orphan.rename("orphan_pp"), on=keys)
    both = L.on_or & L.on_gr
    L["exp_or_pp"] = np.where(both, L.pct_gr / (1 - L.orphan_pp / 100), np.nan)
    L["excess_pp"] = L.pct_or - L.exp_or_pp

    def agg(x):
        i = x.d_pp.abs().idxmax()
        return pd.Series({
            "B_gr": x.B_gr.iloc[0], "B_or": x.B_or.iloc[0],
            "sum_gr": x.sum_gr.iloc[0], "sum_or": x.sum_or.iloc[0],
            "stimmen_je_waehler_gr": x.sum_gr.iloc[0] / x.B_gr.iloc[0] if x.B_gr.iloc[0] else np.nan,
            "stimmen_je_waehler_or": x.sum_or.iloc[0] / x.B_or.iloc[0] if x.B_or.iloc[0] else np.nan,
            "ungueltig_gr_pct": 100 * x.C_gr.iloc[0] / x.B_gr.iloc[0] if x.B_gr.iloc[0] else np.nan,
            "ungueltig_or_pct": 100 * x.C_or.iloc[0] / x.B_or.iloc[0] if x.B_or.iloc[0] else np.nan,
            "tvd_pp": 0.5 * x.d_pp.abs().sum(),
            "orphan_pp": x.orphan_pp.iloc[0],
            "or_only_pp": x.loc[~x.on_gr, "pct_or"].sum(),
            "fehlende_listen": ", ".join(sorted(x.loc[~x.on_or & (x.votes_gr > 0), "party"])),
            "eb_gr_pp": x.loc[x.cat == "Einzelbewerber", "pct_gr"].sum(),
            "groesste_verschiebung": f"{x.party[i]} {x.d_pp[i]:+.1f}",
        })

    D = L.groupby(keys).apply(agg, include_groups=False).reset_index()
    D["waehler_ratio"] = D.B_or / D.B_gr
    D["kongruent"] = (D.waehler_ratio - 1).abs() <= CONGRUENCE_TOL
    # Deckt EIN Gemeindewahl-Bezirk mehrere Ortsraete ab (typisch: ein Briefwahlbezirk je
    # Gemeinde, aber je Ortschaft ein eigener Ortsrats-Briefwahlbezirk gleicher Nummer),
    # dann vergleicht das Paar unterschiedliche Grundgesamtheiten und die Gemeindewahl-
    # Stimmen wuerden mehrfach in die Aggregate eingehen.
    mehrfach = D.groupby(["ags", "gebiet-nr"]).ortsrat.transform("nunique") > 1
    D["gr_bezirk_mehrfach"] = mehrfach
    D["belastbar"] = D.kongruent & ~mehrfach
    return L, D


AGG_COLS = ["stimmen_gr", "stimmen_or", "tvd_pp", "orphan_pp", "groesste_verschiebung"]
LONG_COLS = ["party", "cat", "votes_gr", "votes_or", "orphan_votes_gr", "on_or_any",
             "pct_gr", "pct_or", "d_votes", "d_pp"]


def aggregate(L: pd.DataFrame, keys):
    if L.empty:
        return pd.DataFrame(columns=keys + LONG_COLS), pd.DataFrame(columns=keys + AGG_COLS)
    g = L.groupby(keys + ["party", "cat"], as_index=False).agg(
        votes_gr=("votes_gr", "sum"), votes_or=("votes_or", "sum"),
        orphan_votes_gr=("orphan_votes_gr", "sum"), on_or_any=("on_or", "any"))
    t = g.groupby(keys)[["votes_gr", "votes_or", "orphan_votes_gr"]].transform("sum")
    g["pct_gr"] = 100 * g.votes_gr / t.votes_gr
    g["pct_or"] = 100 * g.votes_or / t.votes_or
    g["d_votes"] = g.votes_or - g.votes_gr
    g["d_pp"] = g.pct_or - g.pct_gr

    def agg(x):
        i = x.d_pp.abs().idxmax()
        return pd.Series({
            "stimmen_gr": x.votes_gr.sum(), "stimmen_or": x.votes_or.sum(),
            "tvd_pp": 0.5 * x.d_pp.abs().sum(),
            "orphan_pp": 100 * x.orphan_votes_gr.sum() / x.votes_gr.sum(),
            "groesste_verschiebung": f"{x.party[i]} {x.d_pp[i]:+.1f}",
        })
    s = g.groupby(keys).apply(agg, include_groups=False).reset_index()
    return g, s


def absorption(Ld: pd.DataFrame, min_votes: int):
    """Wer absorbiert 'verwaiste' GR-Stimmen? Ueberschuss ggue. proportionaler Umverteilung."""
    d = Ld[(Ld.orphan_pp > 0) & Ld.on_or & Ld.on_gr & (Ld.sum_or >= min_votes)].copy()
    if d.empty:
        return pd.DataFrame()
    keys = ["kreis", "ags", "ortsrat", "gebiet-nr"]
    miss = (Ld[~Ld.on_or & (Ld.votes_gr > 0)].groupby(keys).party
            .apply(lambda s: "+".join(sorted(s))).rename("fehlend"))
    d = d.join(miss, on=keys)
    d["w_excess"] = d.excess_pp * d.sum_or
    d["w_orphan"] = d.orphan_pp * d.sum_or
    per_d = d.groupby(keys + ["fehlend"]).w_orphan.first().reset_index()
    denom = per_d.groupby("fehlend").w_orphan.sum()
    r = d.groupby(["fehlend", "party"]).w_excess.sum().reset_index()
    r["anteil_am_verwaisten_pct"] = 100 * r.w_excess / r.fehlend.map(denom)
    r["n_bezirke"] = r.fehlend.map(per_d.groupby("fehlend").size())
    return r.drop(columns="w_excess").sort_values(["fehlend", "anteil_am_verwaisten_pct"],
                                                  ascending=[True, False])


def postal_structure(frames, lokale: pd.DataFrame | None):
    rows = []
    for f in frames:
        if f.etype.iloc[0] != "GR":
            continue
        brief = [is_brief(n, nm) for n, nm in zip(f["gebiet-nr"], f["gebiet-name"])]
        if lokale is not None and not lokale.empty and "Bezirk-Art" in lokale.columns:
            art = (lokale.dropna(subset=["ags", "Bezirk-Nr"])
                   .drop_duplicates(subset=["ags", "Bezirk-Nr"])
                   .set_index(["ags", "Bezirk-Nr"])["Bezirk-Art"])
            brief = [art.get((a, n), "B" if b else "W") == "B"
                     for a, n, b in zip(f.ags, f["gebiet-nr"], brief)]
        f = f.assign(brief=brief)
        b, u = f[f.brief], f[~f.brief]
        pats = b["gebiet-nr"].map(numbering_pattern).value_counts()
        rows.append({
            "kreis": f.ags.iloc[0][:5], "ags": f.ags.iloc[0], "wahl": f.wahl.iloc[0],
            "n_urne": len(u), "n_brief": len(b),
            "urne_je_brief": len(u) / len(b) if len(b) else np.nan,
            "waehler_urne": u.B.sum(), "waehler_brief": b.B.sum(),
            "briefwahl_anteil_pct": 100 * b.B.sum() / f.B.sum() if f.B.sum() else np.nan,
            "waehler_je_brief_mittel": b.B.mean(), "waehler_je_brief_min": b.B.min(),
            f"brief_unter_{SMALL_BRIEF_VOTERS}": int((b.B < SMALL_BRIEF_VOTERS).sum()),
            "nummernschema": pats.index[0] if len(pats) else "",
        })
    return pd.DataFrame(rows)



# --------------------------------------------------------------------------- Plausibilitaet
LIST_COL = re.compile(r"^D(\d+)_1$")       # Stimmen fuer die Gesamtliste
CAND_SUM_COL = re.compile(r"^D(\d+)_3$")   # Summe der Kandidatenstimmen
VOTES_PER_VOTER = 3                        # NKWG: drei Stimmen je Waehler
OUTLIER_TOL = 0.25                         # Abweichung vom Median der eigenen Wahl (Stimmen/Zettel)
# Kriterien und Schwellen von plausi/zettelcheck sind in README_kw_ebenen.md ("Wie geprueft
# wird") zusammengefasst - bei jeder Aenderung dort mitziehen.


def _vorstaende(d) -> list:
    """Fruehere Staende neben <d> (nds/data/stand/<Stand>), juengster zuerst; leer, wenn <d> kein Stand ist."""
    d = Path(d).resolve()
    if not RE_STAND.fullmatch(d.name):
        return []
    return sorted((p for p in d.parent.iterdir() if p.is_dir() and RE_STAND.fullmatch(p.name) and p.name < d.name),
                  reverse=True)


def _leere_auffuellen(df: pd.DataFrame, rel: Path, vorstaende: list):
    """Bezirke ohne Zahlen (Gemeinde hat sie auf dem Server geleert, z. B. zur Neueingabe) mit den Zahlen
    des juengsten frueheren Stands fuellen, in dem sie welche hatten. Ein geleerter Bezirk ist nicht
    geloest, sondern in Arbeit: Er zaehlt mit seinen letzten veroeffentlichten Zahlen weiter.
    Rueckgabe: df und je Zeile der Stand, aus dem die Zahlen stammen ("" = aktueller Stand)."""
    aus = pd.Series("", index=df.index)
    nr = df["gebiet-nr"].astype(str).str.strip()
    c2 = df["C2"] if "C2" in df.columns else pd.Series(np.nan, index=df.index)
    offen = set(df.index[nr.ne("") & nr.str.lower().ne("nan") & ~(c2 > 0)])
    schluessel = lambda r: (str(r["wahl"]).strip(), str(r["gebiet-nr"]).strip(), str(r["gebiet-name"]).strip())  # noqa: E731
    for v in vorstaende:
        if not offen:
            break
        q = v / rel
        if not q.exists():
            continue
        try:
            alt = read_vm_csv(q)
        except Exception:
            continue
        if "C2" not in alt.columns:
            continue
        altmap = {schluessel(r): r for _, r in alt[alt.C2 > 0].iterrows()}
        spalten = [c for c in df.columns if c in alt.columns and c not in META_COLS]
        for i in sorted(offen):
            r = altmap.get(schluessel(df.loc[i]))
            if r is not None:
                df.loc[i, spalten] = [r[c] for c in spalten]
                aus[i] = v.name
                offen.discard(i)
    return df, aus


def cmd_plausi(a):
    """Stimmzettel vs. Stimmen je Bezirk: harte Identitaeten und ungenutzte Stimmen.

    Geprueft wird je Bezirk:
      C1 + C2 == B          Stimmzettel (ungueltig + gueltig) = Waehler
      D <= 3 * C2           hoechstens drei Stimmen je gueltigem Stimmzettel
      Summe D<n>_4 == D     Listensummen = gueltige Stimmen
      D<n>_1 + D<n>_3 == D<n>_4
    Kennzahl: ungenutzte Stimmen = 3*C2 - D. Ein gewisser Anteil ist normal (Waehler
    schoepfen ihre drei Stimmen nicht aus); auffaellig ist die Abweichung vom Median
    der *eigenen* Wahl, weil die Stimmzettelgroesse das Niveau bestimmt.
    """
    out = Path(a.out)
    out.parent.mkdir(parents=True, exist_ok=True)
    rows, verstoss = [], []
    vorstaende = [] if a.ohne_vorstaende else _vorstaende(a.dir)
    for p in sorted(Path(a.dir).rglob("*.csv")):
        if re.search(r"wahllokal|strassen|parties", p.name, re.I):
            continue
        try:
            df = read_vm_csv(p)
        except Exception as e:
            print("Uebersprungen:", p, e)
            continue
        if "C2" not in df.columns or "D" not in df.columns:
            continue  # nur Kommunalwahl-Dateien mit Stimmzettelzaehlung
        df, zahlen_aus = _leere_auffuellen(df, p.relative_to(a.dir), vorstaende)
        # Stimmen je Waehler aus den Daten ableiten: Verhaeltniswahl 3, Direktwahl 1.
        quot = (df.D / df.C2).replace([np.inf, -np.inf], np.nan).dropna()
        quot = quot[quot > 0]
        if quot.empty:
            continue
        vpv = VOTES_PER_VOTER if quot.max() > 1.02 else 1
        if quot.max() > vpv + 1e-9:
            verstoss.append(f"{df.ags.iloc[0]} {df.wahl.iloc[0]}: D/C2={quot.max():.2f} "
                            f"> {vpv} Stimmen je Waehler")
        c4 = party_cols(df)
        # Abgeleitete Felder (IVU: C2 = B - C1) machen die zugehoerige Identitaet zur Tautologie
        abgeleitet = set(df.attrs.get("abgeleitet", ()))
        c1 = sorted([c for c in df.columns if LIST_COL.match(c)], key=lambda c: int(c[1:-2]))
        c3 = sorted([c for c in df.columns if CAND_SUM_COL.match(c)], key=lambda c: int(c[1:-2]))
        def num(v):
            """Leere Zelle -> None (Liste nicht angetreten), sonst Zahl."""
            return None if v is None or pd.isna(v) else float(v)

        for i, r in df.iterrows():
            A_wb, B, B1, C1, C2, D = (num(r.get(k)) or 0 for k in ("A", "B", "B1", "C1", "C2", "D"))
            if C2 <= 0:
                continue
            if not str(r["gebiet-nr"]).strip() or str(r["gebiet-nr"]).lower() == "nan":
                continue   # Summenzeile (Stadtbezirk, Gruppierung), kein Wahlbezirk
            s4 = sum(num(r.get(c)) or 0 for c in c4)
            s1 = sum(num(r.get(c)) or 0 for c in c1)
            s3 = sum(num(r.get(c)) or 0 for c in c3)
            n_leer = sum(1 for c in c4 if num(r.get(c)) is None)
            kenn = f"{r['ags']} {r['wahl']} Bez {r['gebiet-nr']}"
            if C1 + C2 != B and "C2" not in abgeleitet:
                verstoss.append(f"{kenn}: C1+C2={C1 + C2:.0f} != B={B:.0f}")
            if s4 != D:
                verstoss.append(f"{kenn}: Summe D<n>_4={s4:.0f} != D={D:.0f}")
            # je Liste, und nur wo es eine waehlbare Gesamtliste gibt (Einzelwahlvorschlaege
            # haben keine _1-Spalte, dort gilt _3 == _4)
            for c in c4:
                k = c[1:-2]
                v4 = num(r.get(c))
                if v4 is None:
                    continue  # Liste tritt hier nicht an
                if f"D{k}_1" in abgeleitet:
                    continue  # Listenstimmen selbst aus _4 - _3 berechnet, nichts zu pruefen
                has1, has3 = f"D{k}_1" in df.columns, f"D{k}_3" in df.columns
                if has1 and has3:
                    v = (num(r.get(f"D{k}_1")) or 0) + (num(r.get(f"D{k}_3")) or 0)
                elif has3:
                    v = num(r.get(f"D{k}_3")) or 0
                else:
                    continue
                if v != v4:
                    verstoss.append(f"{kenn}: D{k} Teilsummen={v:.0f} != D{k}_4={v4:.0f}")
            mx = vpv * C2
            rows.append({
                "ags": r["ags"], "wahl": r["wahl"], "gebiet-nr": r["gebiet-nr"],
                "gebiet-name": r["gebiet-name"], "waehlberechtigte": A_wb, "waehler": B,
                "waehler_wahlschein": B1, "ungueltige_zettel": C1, "stimmen_je_waehler": vpv,
                "wahlbeteiligung": B / A_wb if A_wb else np.nan,
                "wahlschein_anteil": B1 / B if B else np.nan,
                "gueltige_zettel": C2, "stimmen": D, "stimmen_max": mx,
                "ungenutzt": mx - D, "ungenutzt_pct": 100 * (mx - D) / mx,
                "stimmen_je_zettel": D / C2,
                "listenstimmen_je_zettel": s1 / C2 if c1 else np.nan,
                "kandidatenstimmen_je_zettel": s3 / C2 if c3 else np.nan,
                "ungueltig_pct": 100 * C1 / B if B else np.nan,
                "listen_nicht_angetreten": n_leer,
                # Stand, aus dem die Zahlen stammen, wenn der Bezirk im aktuellen Stand leer ist (Neueingabe)
                "zahlen_aus_stand": zahlen_aus[i],
            })
    if not rows:
        sys.exit(f"Keine auswertbaren Dateien in {a.dir}")
    P = pd.DataFrame(rows)
    # Referenz ist der Median der eigenen Wahl: die Stimmzettelgroesse bestimmt das Niveau.
    # Und zwar der Median der *uebrigen* Bezirke - ein Bezirk darf den Massstab nicht
    # mitbestimmen, an dem er gemessen wird. Bei einer Wahl mit zwei Bezirken zieht ein
    # Ausreisser den gemeinsamen Median sonst bis zur Haelfte zu sich heran und seine
    # Abweichung erscheint halb so gross (Ortsratswahl Hettensen: 132 statt 264 fehlende
    # Stimmen). Wahlen mit nur einem Bezirk haben keine Referenz und bleiben unbewertet.
    def _median_ohne_eigenen(g):
        v = g.to_numpy()
        return pd.Series([np.median(np.delete(v, i)) if len(v) > 1 else np.nan
                          for i in range(len(v))], index=g.index)

    med = (P.groupby(["ags", "wahl"]).stimmen_je_zettel
            .transform(lambda g: _median_ohne_eigenen(g)))
    P["median_der_wahl"] = med
    P["abw_vom_median"] = P.stimmen_je_zettel - med
    P["fehlend_ggue_median"] = (-P.abw_vom_median * P.gueltige_zettel).round()
    P["auffaellig"] = (P.abw_vom_median < -OUTLIER_TOL) & P.median_der_wahl.notna()

    # Quervergleich: derselbe Bezirk in den anderen gleichzeitigen Wahlen. Weicht dort die
    # Zahl der Stimmzettel ab, bezieht sich C2 auf ein anderes Gebiet - dann fehlen keine
    # Stimmen, sondern der Nenner passt nicht. Stimmt C2 ueberein und nur D ist niedrig,
    # sind tatsaechlich Stimmen unterrepraesentiert.
    grp = P.groupby(["ags", "gebiet-nr"])
    P["wahlen_im_bezirk"] = grp.gueltige_zettel.transform("size")
    P["zettel_median_bezirk"] = grp.gueltige_zettel.transform("median")
    P["zettel_rel"] = P.gueltige_zettel / P.zettel_median_bezirk

    def einordnen(r):
        if not r.auffaellig:
            return ""
        if r.wahlen_im_bezirk < 2:
            return "kein Quervergleich moeglich"
        if r.zettel_rel > 1.10:
            return "Stimmzettelzahl weicht ab (anderer Gebietsbezug)"
        if r.zettel_rel < 0.90:
            return "kleinerer Bezirkszuschnitt"
        return "Stimmen unterrepraesentiert"

    P["einordnung"] = P.apply(einordnen, axis=1)
    P = P.sort_values("ungenutzt_pct", ascending=False)
    P.to_csv(out, index=False)

    print(f"{len(P)} Bezirke aus {P.groupby(['ags', 'wahl']).ngroups} Wahlen -> {out}")
    ersatz = P[P.zahlen_aus_stand != ""]
    if len(ersatz):
        print(f"  davon {len(ersatz)} Bezirke in {ersatz.ags.nunique()} Gemeinden ohne aktuelle Zahlen (auf dem "
              f"Server geleert), gewertet mit den Zahlen aus: "
              + ", ".join(f"{k} ({v})" for k, v in ersatz.zahlen_aus_stand.value_counts().sort_index().items()))
    print(f"Harte Identitaeten (C1+C2=B, D<=n*C2, Summe D<n>_4=D, _1+_3=_4 je Liste): "
          f"{'alle erfuellt' if not verstoss else str(len(verstoss)) + ' VERSTOSSE'}")
    for v in verstoss[:20]:
        print("  !", v)
    print(f"ungenutzte Stimmen: Median {P.ungenutzt_pct.median():.1f} % · "
          f"Mittel {P.ungenutzt_pct.mean():.1f} % · max {P.ungenutzt_pct.max():.1f} %")
    for v, g in P.groupby("stimmen_je_waehler"):
        print(f"  {len(g)} Bezirke mit {int(v)} Stimme(n) je Waehler: "
              f"Median {g.stimmen_je_zettel.median():.2f} genutzt")
    print(f"\nAuffaellig (mehr als {OUTLIER_TOL} Stimmen/Zettel unter dem Median der eigenen "
          f"Wahl): {int(P.auffaellig.sum())}")
    for k, v in P[P.auffaellig].einordnung.value_counts().items():
        print(f"  {v:>5}  {k}")
    cols = ["ungenutzt_pct", "stimmen_je_zettel", "median_der_wahl", "fehlend_ggue_median",
            "gueltige_zettel", "zettel_rel", "einordnung", "ags", "wahl", "gebiet-nr"]
    P = P[P.auffaellig] if a.nur_auffaellig else P
    with pd.option_context("display.width", 200, "display.max_colwidth", 30):
        print(P.head(a.top)[cols].to_string(index=False,
              formatters={"ungenutzt_pct": "{:.1f}".format,
                          "stimmen_je_zettel": "{:.2f}".format,
                          "median_der_wahl": "{:.2f}".format,
                          "zettel_rel": "{:.2f}".format}))


# --------------------------------------------------------------------------- Vorwahl (2021)
# Die Kommunalwahl 2021 liegt auf denselben Instanzen, aber mit anderem Layout:
#   Termine je AGS       https://votemanager.kdo.de/<ags>/api/termine.json
#   Dateiliste 2021      <base>api/praesentation/open_data.json
#   CSV-Dateien 2021     <base>praesentation/<datei>.csv
#   (2026 dagegen: <base>daten/opendata/open_data.json und dieselben Ordner fuer die CSVs)
# Verifiziert am 23.09.2026 an 03459029 (Ostercappeln).
HIST_LAYOUTS = [("api/praesentation/open_data.json", "praesentation/"),   # 2021
                ("daten/opendata/open_data.json", "daten/opendata/")]     # 2026


def _termin_base(ags: str, datum: str) -> str | None:
    """URL der Instanz fuer einen Wahltermin (datum = TT.MM.JJJJ) aus api/termine.json."""
    url = f"https://votemanager.kdo.de/{ags}/api/termine.json"
    try:
        r = _get(url)
        if r.status_code != 200:
            return None
        for t in r.json().get("termine", []):
            if t.get("date") == datum:
                # "../20210912/03459029/praesentation/" -> ".../20210912/03459029/"
                u = t.get("url", "")
                m = re.search(r"(\d{8})/(\d+)/", u)
                if m:
                    return VM_BASE.format(termin=m.group(1), ags=m.group(2))
    except Exception as e:
        print("  termine.json nicht lesbar:", url, e, flush=True)
    return None


# Kreisschluessel -> Name (Land Niedersachsen, Stand 2026). Wird gegen die Wahlnamen in
# den Daten geprueft ("Kreistagswahl Landkreis Aurich"), Abweichungen meldet `wirkung`.
KREIS_NAMEN = {
    "03101": "Braunschweig (Stadt)", "03102": "Salzgitter (Stadt)", "03103": "Wolfsburg (Stadt)",
    "03151": "Gifhorn", "03153": "Goslar", "03154": "Helmstedt", "03155": "Northeim",
    "03157": "Peine", "03158": "Wolfenbüttel", "03159": "Göttingen",
    "03241": "Region Hannover", "03251": "Diepholz", "03252": "Hameln-Pyrmont",
    "03254": "Hildesheim", "03255": "Holzminden", "03256": "Nienburg (Weser)",
    "03257": "Schaumburg", "03351": "Celle", "03352": "Cuxhaven", "03353": "Harburg",
    "03354": "Lüchow-Dannenberg", "03355": "Lüneburg", "03356": "Osterholz",
    "03357": "Rotenburg (Wümme)", "03358": "Heidekreis", "03359": "Stade", "03360": "Uelzen",
    "03361": "Verden", "03401": "Delmenhorst (Stadt)", "03402": "Emden (Stadt)",
    "03403": "Oldenburg (Stadt)", "03404": "Osnabrück (Stadt)", "03405": "Wilhelmshaven (Stadt)",
    "03451": "Ammerland", "03452": "Aurich", "03453": "Cloppenburg", "03454": "Emsland",
    "03455": "Friesland", "03456": "Grafschaft Bentheim", "03457": "Leer",
    "03458": "Oldenburg (Landkreis)", "03459": "Osnabrück (Landkreis)", "03460": "Vechta",
    "03461": "Wesermarsch", "03462": "Wittmund",
}


def wahl_typ(name: str) -> str:
    """Gruppiert die Wahlnamen in die vier Ebenen mit eigener Grundgesamtheit."""
    n = _fold(str(name)).lower()
    if re.search(r"ortsrat|ortsbeirat|stadtbezirksrat|bezirksrat", n):
        return "Ortsrat"
    if re.search(r"kreis|regionsversammlung", n):   # Regionsversammlung: regionsweit wie ein Kreistag
        return "Kreistag"
    if re.search(r"samtgemeinde", n):
        return "Samtgemeinderat"
    return "Gemeinde-/Stadtrat"


def cmd_namen(a):
    """Holt je Instanz den Gemeindenamen aus api/termine.json (title); Fremdquellen aus deren Liste."""
    inst = pd.read_csv(a.instances, dtype=str)
    fremd = _fremdquellen(a.fremd)
    fremd = fremd[fremd.ags != ""].drop_duplicates("ags")
    def eine(row):
        try:
            r = _get(f"https://votemanager.kdo.de/{row.ags}/api/termine.json")
            if r.status_code == 200:
                t = str(r.json().get("title", ""))
                # "Wahltermine Gemeinde Ostercappeln" -> "Gemeinde Ostercappeln"
                return {"ags": row.ags, "gemeinde": re.sub(r"^Wahltermine\s+", "", t).strip()}
        except Exception as e:
            print("ERR", row.ags, e, flush=True)
        return {"ags": row.ags, "gemeinde": ""}
    rows = _parallel(eine, [r for _, r in inst[~inst.ags.isin(fremd.ags)].iterrows()], a.jobs)
    rows += [{"ags": f.ags, "gemeinde": f["name"]} for _, f in fremd.iterrows()]
    df = pd.DataFrame(rows).sort_values("ags")
    Path(a.out).parent.mkdir(parents=True, exist_ok=True)
    df.to_csv(a.out, index=False)
    print(f"{int((df.gemeinde != '').sum())} von {len(df)} Namen -> {a.out}")


def cmd_histdownload(a):
    """Laedt die Bezirksdateien eines frueheren Termins (Standard: Kommunalwahl 12.09.2021).

    AGS-Auswahl aus der Plausibilitaetsdatei: standardmaessig nur die Gemeinden mit
    auffaelligen Bezirken (--alle-ags laedt alle darin vorkommenden).
    """
    P = pd.read_csv(a.plausi, dtype={"ags": str})
    if not a.alle_ags:
        P = P[P.auffaellig == True]  # noqa: E712 - pandas
    ags_liste = sorted(P.ags.dropna().unique())
    ziel = Path(a.dir)
    ziel.mkdir(parents=True, exist_ok=True)
    print(f"{len(ags_liste)} AGS, Termin {a.datum}, {a.jobs} parallel", flush=True)

    def eine(ags):
        base = _termin_base(ags, a.datum)
        if not base:
            print(f"[{ags}] kein Termin {a.datum}", flush=True)
            return {"ags": ags, "base": "", "dateien": 0, "status": "kein Termin"}
        od, prefix = None, ""
        for idx_path, csv_prefix in HIST_LAYOUTS:
            try:
                r = _get(base + idx_path)
                if r.status_code == 200:
                    od, prefix = r.json(), csv_prefix
                    break
            except Exception:
                pass
        if not od or not od.get("csvs"):
            print(f"[{ags}] keine Dateiliste unter {base}", flush=True)
            return {"ags": ags, "base": base, "dateien": 0, "status": "keine Dateiliste"}
        n = 0
        for c in od["csvs"]:
            if not re.search(r"bezirk", str(c.get("ebene", "")), re.I):
                continue   # nur unterste Ebene; Gemeinde-/Ortsteilsummen brauchen wir nicht
            url = c["url"] if c["url"].startswith("http") else base + prefix + c["url"]
            target = ziel / ags / url.split("/")[-1].split("?")[0]
            if target.exists() and not a.force:
                n += 1
                continue
            target.parent.mkdir(parents=True, exist_ok=True)
            try:
                resp = _get(url)
            except Exception as e:
                print("ERR", e, url, flush=True)
                continue
            if resp.status_code == 200:
                target.write_bytes(resp.content)
                n += 1
            else:
                print("ERR", resp.status_code, url, flush=True)
        print(f"[{ags}] {n} Bezirksdateien", flush=True)
        return {"ags": ags, "base": base, "dateien": n, "status": "ok" if n else "leer"}

    rows = _parallel(eine, ags_liste, a.jobs)
    man = pd.DataFrame(rows)
    man.to_csv(ziel / "manifest.csv", index=False)
    print(man.status.value_counts().to_string())
    print(f"Manifest -> {ziel / 'manifest.csv'}")


# --------------------------------------------------------------------------- Zettelcheck
def bezirk_key(nr: str) -> str:
    """Bezirksnummer vergleichbar machen: 0003 und 003 sind derselbe Bezirk."""
    s = str(nr).strip()
    return s.lstrip("0") or "0"


def name_key(name: str) -> str:
    """Bezirksname vergleichbar machen: '0003 Ostercappeln - Ost' -> 'ostercappelnost'."""
    s = _fold(str(name)).lower()
    s = re.sub(r"^\s*\d+\s*", "", s)          # fuehrende Bezirksnummer im Namen
    return re.sub(r"[^a-z0-9]", "", s)


def zettel_varianten(n: float) -> dict[int, str]:
    """Zahlen, die sich von n durch genau einen Erfassungsfehler unterscheiden.

    Abgedeckt sind die drei haeufigen Tippfehler beim Uebertragen einer Zahl:
    eine Ziffer falsch (317 statt 217), zwei Ziffern vertauscht (317 statt 371)
    und eine Ziffer zu viel (1317 statt 317).
    """
    if not np.isfinite(n) or n <= 0:
        return {}
    s = str(int(round(n)))
    out: dict[int, str] = {}
    for i in range(len(s)):
        for d in "0123456789":
            if d != s[i]:
                v = s[:i] + d + s[i + 1:]
                if not v.startswith("0"):
                    out.setdefault(int(v), f"eine Ziffer ({s} statt {v})")
    for i in range(len(s) - 1):
        if s[i] != s[i + 1]:
            v = s[:i] + s[i + 1] + s[i] + s[i + 2:]
            if not v.startswith("0"):
                out.setdefault(int(v), f"Zahlendreher ({s} statt {v})")
    for i in range(len(s)):
        v = s[:i] + s[i + 1:]
        if v and not v.startswith("0"):
            out.setdefault(int(v), f"eine Ziffer zu viel ({s} statt {v})")
    return {k: v for k, v in out.items() if k > 0}


def _hist_frame(hist_dir: Path) -> pd.DataFrame:
    """Alle Bezirksdateien des Vergleichstermins als eine Tabelle (nur A/B/B1/C1/C2/D)."""
    rows = []
    for p in sorted(hist_dir.rglob("*.csv")):
        if p.name == "manifest.csv" or re.search(r"wahllokal|strassen", p.name, re.I):
            continue
        try:
            df = read_vm_csv(p)
        except Exception as e:
            print("Uebersprungen:", p, e)
            continue
        if "C2" not in df.columns or "D" not in df.columns:
            continue
        for c in ("A", "B", "B1", "C1"):
            if c not in df.columns:
                df[c] = np.nan
        df = df[["wahl", "gebiet-nr", "gebiet-name", "A", "B", "B1", "C1", "C2", "D"]].copy()
        df["ags"] = p.parent.name          # Dateiordner ist massgeblich (AGS-Spalte kann fehlen)
        rows.append(df)
    if not rows:
        return pd.DataFrame()
    H = pd.concat(rows, ignore_index=True)
    H = H[(H.C2 > 0) & H.D.notna()]
    H["stimmen_je_zettel"] = H.D / H.C2
    H = H[H.stimmen_je_zettel > 1.02]      # nur Verhaeltniswahlen (3 Stimmen), keine Direktwahlen
    H["bez"] = H["gebiet-nr"].map(bezirk_key)
    H["nkey"] = H["gebiet-name"].map(name_key)
    H["wkey"] = H["wahl"].map(name_key)
    # Urnen- und Briefwahlbezirk tragen haeufig dieselbe Nummer; Briefwahlbezirke haben
    # keine Wahlberechtigten. Ohne diese Trennung vergleicht man Dorf gegen Gemeinde.
    H["art"] = np.where(H.A.fillna(0) > 0, "W", "B")
    return H


def _median_map(df, keys, values, name):
    """Median von values je Schluesselkombination als mergebare Tabelle."""
    g = values.groupby([df[k] for k in keys]).median().rename(name)
    return g.reset_index()


def cmd_zettelcheck(a):
    """Prueft bei auffaelligen Bezirken, ob statt der Stimmen die Stimmzettelzahl falsch ist.

    Wenn in einem Bezirk die Stimmen je Stimmzettel weit unter dem Median der eigenen Wahl
    liegen, gibt es zwei Erklaerungen: es fehlen Stimmen (Zaehler zu klein) oder es stehen
    zu viele Stimmzettel in der Meldung (Nenner zu gross). Getrennt wird das ueber zwei
    voneinander unabhaengige Erwartungswerte fuer die Zahl der Stimmzettel:

      (1) Wahlberechtigte  A x uebliche Quote Stimmzettel/Wahlberechtigte derselben Wahl.
          A stammt aus dem Waehlerverzeichnis und haengt nicht an der Auszaehlung.
      (2) Vorwahl 2021     Stimmzettel desselben Bezirks 2021 x Veraenderung der Gemeinde.

    Wie weit ein unauffaelliger Bezirk von diesen Erwartungen normalerweise abweicht, misst
    die Streuung (MAD) der unauffaelligen Bezirke derselben Wahl. Daraus werden z-Werte:
    z_ist  = Abweichung der gemeldeten Zettelzahl, z_soll = Abweichung der Zettelzahl, die
    zu den gemeldeten Stimmen passen wuerde. Klein z_ist und grosses z_soll heisst "Stimmen
    fehlen", umgekehrt "Zettelzahl unplausibel".

    Zwei Stoerquellen werden vorher abgetrennt, weil sie eine hoehere Zettelzahl harmlos
    erklaeren: ein neuer Bezirkszuschnitt (A 2026 weicht stark von A 2021 ab) und eine
    geaenderte Briefwahlstruktur (Briefwaehler werden jetzt im Urnenbezirk mitgezaehlt,
    sichtbar am Anteil B1/B).
    """
    P = pd.read_csv(a.plausi, dtype={"ags": str, "gebiet-nr": str})
    H = _hist_frame(Path(a.hist_dir))
    if H.empty:
        sys.exit(f"keine Vergleichsdaten in {a.hist_dir} - erst 'histdownload' laufen lassen")
    for c in ("waehlberechtigte", "waehler_wahlschein", "wahlbeteiligung", "wahlschein_anteil"):
        if c not in P.columns:
            P[c] = np.nan
    P["bez"] = P["gebiet-nr"].map(bezirk_key)
    P["nkey"] = P["gebiet-name"].map(name_key)
    P["wkey"] = P["wahl"].map(name_key)
    P["art"] = np.where(P.waehlberechtigte.fillna(0) > 0, "W", "B")

    # ---- Zuordnung der Bezirke auf den Vergleichstermin ---------------------
    per_bez = (H.groupby(["ags", "bez", "art"])
                 .agg(h_nkey=("nkey", "first"), h_A=("A", "max"), h_B=("B", "median"),
                      h_B1=("B1", "median"), h_C2=("C2", "median"))
                 .reset_index())
    per_name = (H.groupby(["ags", "nkey", "art"])
                  .agg(h_A=("A", "max"), h_B=("B", "median"), h_B1=("B1", "median"),
                       h_C2=("C2", "median"))
                  .reset_index())
    per_wahl = (H.groupby(["ags", "wkey", "bez", "art"])
                  .agg(hw_C2=("C2", "median"), hw_B=("B", "median"))
                  .reset_index())

    m = P.merge(per_bez, on=["ags", "bez", "art"], how="left")
    m["treffer"] = np.where(m.h_C2.notna(),
                            np.where(m.nkey == m.h_nkey, "Nummer+Name", "Nummer"), "")
    fehlt = m.h_C2.isna()
    if fehlt.any():   # Bezirke, deren Nummer sich geaendert hat, ueber den Namen nachziehen
        nach = m.loc[fehlt, ["ags", "nkey", "art"]].merge(per_name, on=["ags", "nkey", "art"],
                                                          how="left")
        for col in ("h_A", "h_B", "h_B1", "h_C2"):
            m.loc[fehlt, col] = nach[col].values
        m.loc[fehlt & m.h_C2.notna(), "treffer"] = "Name"
    m.loc[m.h_C2.isna(), "treffer"] = "kein Treffer"
    m = m.merge(per_wahl, on=["ags", "wkey", "bez", "art"], how="left")
    m["v_C2"] = m.hw_C2.fillna(m.h_C2)     # gleiche Wahl bevorzugen, sonst Bezirksmittel
    m["v_B"] = m.hw_B.fillna(m.h_B)
    m["v_A"], m["v_B1"] = m.h_A, m.h_B1

    unauff = m.auffaellig != True          # noqa: E712 - pandas-Vergleich
    ref = m[unauff & (m.gueltige_zettel > 0)]

    # ---- Erwartung 1: Wahlberechtigte ---------------------------------------
    rq = ref[ref.waehlberechtigte > 0]
    quote = rq.gueltige_zettel / rq.waehlberechtigte
    m = m.merge(_median_map(rq, ["ags", "wahl"], quote, "q_wahl"), on=["ags", "wahl"], how="left")
    m = m.merge(_median_map(rq, ["ags"], quote, "q_ags"), on="ags", how="left")
    m["quote"] = m.q_wahl.fillna(m.q_ags)
    m["erw_wb"] = np.where(m.waehlberechtigte > 0, m.waehlberechtigte * m.quote, np.nan)

    # ---- Erwartung 2: Vorwahl -----------------------------------------------
    rf = ref[ref.v_C2 > 0]
    f = rf.gueltige_zettel / rf.v_C2
    m = m.merge(_median_map(rf, ["ags", "wahl"], f, "f_wahl"), on=["ags", "wahl"], how="left")
    m = m.merge(_median_map(rf, ["ags"], f, "f_ags"), on="ags", how="left")
    m["faktor"] = m.f_wahl.fillna(m.f_ags)
    m["faktor_n"] = m.ags.map(rf.groupby("ags").size())
    m["erw_21"] = m.v_C2 * m.faktor

    # ---- Streuung der unauffaelligen Bezirke (MAD der Log-Abweichung) -------
    def streuung(spalte, name):
        r = m[unauff & (m[spalte] > 0) & (m.gueltige_zettel > 0)].copy()
        r["res"] = np.log(r.gueltige_zettel / r[spalte])
        mad = r.groupby(["ags", "wahl"]).res.apply(lambda x: (x - x.median()).abs().median())
        s = (1.4826 * mad).rename(name).reset_index()
        glob = float(np.nanmax([1.4826 * (r.res - r.res.median()).abs().median(), 0.05]))
        return s, glob

    for spalte, name in (("erw_wb", "s_wb"), ("erw_21", "s_21")):
        s, glob = streuung(spalte, name)
        m = m.merge(s, on=["ags", "wahl"], how="left")
        m[name] = m[name].fillna(glob).clip(lower=0.05)

    # ---- Kennzahlen je Bezirk ------------------------------------------------
    m["zettel_soll"] = m.stimmen / m.median_der_wahl      # Zettelzahl, die zu den Stimmen passt
    for anker, erw, s in (("wb", "erw_wb", "s_wb"), ("21", "erw_21", "s_21")):
        ok = m[erw] > 0
        m[f"z_ist_{anker}"] = np.where(ok, np.log(m.gueltige_zettel / m[erw]) / m[s], np.nan)
        m[f"z_soll_{anker}"] = np.where(ok, np.log(m.zettel_soll / m[erw]) / m[s], np.nan)
    m["groesse_rel_2021"] = np.where((m.waehlberechtigte > 0) & (m.v_A > 0),
                                     m.waehlberechtigte / m.v_A, np.nan)
    m["wahlbeteiligung_2021"] = np.where(m.v_A > 0, m.v_B / m.v_A, np.nan)
    m["wahlschein_anteil_2021"] = np.where(m.v_B > 0, m.v_B1 / m.v_B, np.nan)
    m["wahlschein_zuwachs"] = m.wahlschein_anteil - m.wahlschein_anteil_2021

    # Zuschnitt oder Briefwahlstruktur geaendert -> die hoehere Zettelzahl ist erklaerbar
    m["zuschnitt_neu"] = (m.groesse_rel_2021.notna()
                          & ((m.groesse_rel_2021 < 1 - a.toleranz)
                             | (m.groesse_rel_2021 > 1 + a.toleranz)))
    m["briefwahl_neu"] = m.wahlschein_zuwachs > 0.15

    # ---- Quervergleich im selben Bezirk (schaerfster Massstab) ---------------
    # Dieselben Waehler, derselbe Tag, derselbe Raum: die Zahl der Stimmzettel der
    # parallel ausgezaehlten Wahlen unterscheidet sich nur um die ungueltigen Zettel,
    # typisch wenige Prozent. Verglichen wird mit dem Median der *anderen* Wahlen im
    # Bezirk - der eigene Wert darf den Massstab nicht mitbestimmen.
    # In Briefwahlbezirken traegt der Ortsrat oft dieselbe Nummer, deckt aber nur eine
    # Ortschaft ab; dort wird deshalb nur innerhalb derselben Gebietsklasse verglichen.
    m["scope"] = np.where(m.wahl.map(classify_election) == "OR", "OR", "GEM")
    m["gkey"] = np.where(m.art == "B", m.ags + "|" + m.bez + "|B|" + m.scope,
                         m.ags + "|" + m.bez + "|W")

    def _geschwister(g):
        v = g.values
        n = len(v)
        out = []
        for i in range(n):
            rest = np.delete(v, i)
            if len(rest) == 0:
                out.append((np.nan, 0, np.nan))
                continue
            med = float(np.median(rest))
            spann = (rest.max() - rest.min()) / med if len(rest) > 1 and med else 0.0
            out.append((med, len(rest), spann))
        return pd.DataFrame(out, index=g.index,
                            columns=["zettel_geschwister", "n_geschwister", "spann_geschwister"])

    # Verglichen wird die Zahl der *Waehler*: die ist in einem Bezirk fuer alle
    # gleichzeitigen Wahlen dieselbe. Die gueltigen Stimmzettel unterscheiden sich
    # dagegen legitim um die ungueltigen, die je Stimmzettelart verschieden ausfallen.
    gk = (m.groupby("gkey").waehler.apply(_geschwister)
           .reset_index(level=0, drop=True))
    m = m.join(gk.rename(columns={"zettel_geschwister": "waehler_geschwister"}))
    m["rel_geschwister"] = m.waehler / m.waehler_geschwister
    zg = (m.groupby("gkey").gueltige_zettel.apply(_geschwister)
           .reset_index(level=0, drop=True))
    m["zettel_geschwister"] = zg.zettel_geschwister   # nur zur Rekonstruktion angezeigt

    # Sonderfall Briefwahlbezirk: Die gemeindeweiten Wahlen zaehlen dort alle Waehler des
    # Bezirks, jeder Ortsrat nur die seiner Ortschaft. Die Ortsrats-Waehler muessen sich
    # also auf die Bezirkszahl summieren. Meldet ein Ortsrat allein schon den ganzen
    # Bezirk, obwohl daneben weitere Ortsraete eigene Waehler ausweisen, ist seine Zahl
    # zu gross (Twistringen, Briefwahl 05901: Kreis und Stadtrat je 451, Ortsrat
    # Twistringen ebenfalls 451, Ortsrat Abbenhausen zusaetzlich 78).
    bw = m[m.art == "B"]
    summe_or = (bw[bw.scope == "OR"].groupby(["ags", "bez"]).waehler.sum()
                  .rename("or_summe").reset_index())
    median_gem = (bw[bw.scope == "GEM"].groupby(["ags", "bez"]).waehler.median()
                    .rename("gem_median").reset_index())
    m = (m.merge(summe_or, on=["ags", "bez"], how="left")
          .merge(median_gem, on=["ags", "bez"], how="left"))
    for spalte in ("or_summe", "gem_median"):
        m[spalte] = np.where(m.art == "B", m[spalte], np.nan)

    # Briefwahlbezirk, ueber die Gebietsklassen hinweg. Oben wird dort nur innerhalb derselben
    # Klasse verglichen und eine einzelne Parallelwahl verworfen, weil dieselbe Nummer
    # verschiedene Gebiete meinen kann. Zwei Faelle sind trotzdem eindeutig:
    #  - Gleiche Waehlerzahl: meldet eine andere Wahl im Bezirk (fast) dieselbe Zahl, ist es
    #    dasselbe Gebiet und die Zettelzahl bestaetigt (Hannover, Briefwahl 4052: Rat 774,
    #    Stadtbezirksrat 771 Waehler). Nur Uebereinstimmung sagt etwas; eine abweichende
    #    Zahl kann ein anderes Gebiet sein und bleibt ohne Aussage. Die Vergleichswahl muss
    #    selbst unauffaellig sein: im Briefwahlbezirk ist die Waehlerzahl (Wahlbriefe) fuer
    #    alle Wahlen dieselbe, zwei gleich niedrige Wahlen bestaetigen sich sonst gegenseitig,
    #    obwohl eher die gemeinsame Zettelzahl falsch ist (Verden, Briefwahl 240: Stadtrat und
    #    Kreis je 2,6 Stimmen je Zettel, Ortsrat Borstel im selben Bezirk unauffaellig).
    #  - Ortsrat deutlich groesser als die Gemeinde: meldet ein Ortsrat oder Stadtbezirksrat
    #    weit mehr Waehler als jede gemeindeweite Wahl im Bezirk und passen seine Stimmen zu
    #    deren Zettelzahl, ist seine Zettelzahl falsch (Hannover, Stadtbezirksrat Mitte,
    #    Briefwahl 0151: 2028 Waehler, 1627 Zettel, 2565 Stimmen; Rat im selben Bezirk 880
    #    Waehler, 871 Zettel, 2611 Stimmen). Einen maessigen Ueberschuss gibt es auch legitim,
    #    wenn Briefwahlbezirke je Wahl anders nummeriert sind (Braunschweig 21187: 649 gegen
    #    366, Stimmen dort normal); verlangt wird deshalb mindestens das bw_or_faktor-fache.
    m["_rid"] = np.arange(len(m))
    m["_unauff"] = m.auffaellig != True    # noqa: E712 - pandas-Vergleich
    bx = m.loc[(m.art == "B") & (m.waehler > 0),
               ["_rid", "ags", "bez", "wahl", "scope", "waehler", "gueltige_zettel", "zettel_soll",
                "_unauff"]]
    px = bx.merge(bx, on=["ags", "bez"], suffixes=("", "_p"))
    px = px[px._rid != px._rid_p]
    gleich = (px[((px.waehler_p / px.waehler - 1).abs() <= a.bw_gleich_tol) & px._unauff_p]
                .groupby("_rid")
                .agg(n_gleich_bw=("wahl_p", "size"), waehler_gleich_bw=("waehler_p", "median"),
                     zettel_gleich_bw=("gueltige_zettel_p", "median")))
    gem = px[px.scope_p == "GEM"]
    gem_max = gem.groupby("_rid").waehler_p.max()
    og = gem[(gem.scope == "OR")
             & (gem.waehler > gem._rid.map(gem_max) * a.bw_or_faktor)
             & (gem.gueltige_zettel_p > 0)
             & ((gem.zettel_soll / gem.gueltige_zettel_p - 1).abs() <= a.geschwister_tol)].copy()
    og["_abst"] = (og.zettel_soll / og.gueltige_zettel_p - 1).abs()
    og = (og.sort_values("_abst").drop_duplicates("_rid")
            .set_index("_rid")[["wahl_p", "waehler_p", "gueltige_zettel_p"]]
            .rename(columns={"wahl_p": "gem_wahl_bw", "waehler_p": "gem_waehler_bw",
                             "gueltige_zettel_p": "gem_zettel_bw"}))
    m = m.join(gleich, on="_rid").join(og, on="_rid").drop(columns=["_rid", "_unauff"])
    m["n_gleich_bw"] = m.n_gleich_bw.fillna(0)

    # Parallelwahlen im Bezirk, die selbst unauffaellig sind - zeigt, ob eine Abweichung
    # nur eine Wahl betrifft oder den ganzen Wahlvorstand.
    unauff = m.groupby("gkey").auffaellig.transform(lambda g: (~g.astype(bool)).sum())
    m["geschwister_unauffaellig"] = unauff - np.where(m.auffaellig, 0, 1)

    # ---- Urteil --------------------------------------------------------------
    zt, zv = a.z_treffer, a.z_verdacht
    gt = a.geschwister_tol

    def urteil(r):
        if not r.auffaellig:
            return pd.Series(["", "", np.nan, np.nan, "", ""])
        # Anker waehlen: Wahlberechtigte zuerst (unabhaengig von der Zuordnung auf 2021),
        # sonst die Vorwahl - aber nur, wenn der Bezirk noch derselbe ist.
        anker, z_ist, z_soll = "", np.nan, np.nan
        if np.isfinite(r.z_ist_wb):
            anker, z_ist, z_soll = "Wahlberechtigte", r.z_ist_wb, r.z_soll_wb
        elif np.isfinite(r.z_ist_21) and not r.zuschnitt_neu:
            anker, z_ist, z_soll = "Vorwahl 2021", r.z_ist_21, r.z_soll_21

        # Tippfehler-Kandidat: eine Zettelzahl, die sich durch einen einzelnen
        # Erfassungsfehler von der gemeldeten unterscheidet und sowohl zur Stimmenzahl
        # als auch zur Erwartung passt.
        erw = r.erw_wb if anker == "Wahlberechtigte" else r.erw_21
        s = r.s_wb if anker == "Wahlberechtigte" else r.s_21
        # Ortsrat meldet im Briefwahlbezirk den ganzen Bezirk, obwohl daneben weitere
        # Ortsraete eigene Waehler ausweisen (Beleg unter 1b).
        or_ganz = (r.art == "B" and r.scope == "OR" and np.isfinite(r.or_summe)
                   and np.isfinite(r.gem_median) and r.gem_median > 0
                   and r.or_summe > r.waehler          # es gibt ueberhaupt weitere Ortsraete
                   and r.or_summe > 1.02 * r.gem_median
                   and r.waehler >= 0.98 * r.gem_median)
        or_zu_gross = r.art == "B" and np.isfinite(r.gem_zettel_bw) and not or_ganz   # (1c)
        # Vergleich ueber gleiche Waehlerzahl im Briefwahlbezirk (siehe gleich oben), wenn der
        # Vergleich innerhalb der Gebietsklasse nicht traegt
        bw_gleich = (r.art == "B" and not (r.n_geschwister >= 2 and r.spann_geschwister <= 0.02)
                     and r.n_gleich_bw >= 1 and not or_ganz and not or_zu_gross)
        muster, best = "", None
        for c, txt in zettel_varianten(r.gueltige_zettel).items():
            quot = r.stimmen / c
            if abs(quot - r.median_der_wahl) > a.quot_tol:
                continue
            if np.isfinite(erw) and erw > 0 and abs(np.log(c / erw)) > zt * s:
                continue
            if np.isfinite(r.rel_geschwister) and abs(c / r.zettel_geschwister - 1) > gt:
                continue
            if bw_gleich and abs(c / r.zettel_gleich_bw - 1) > gt:
                continue
            d = abs(quot - r.median_der_wahl)
            if best is None or d < best[0]:
                best = (d, c, txt)
        if best:
            muster = best[2]

        belege, fuer_zettel, fuer_stimmen = [], 0, 0
        # (1) Parallelwahlen im selben Bezirk - engster Massstab, aber blind dafuer,
        #     wenn dieselbe falsche Zettelzahl in allen Wahlen steht. Nur *mehr* Zettel
        #     als in den Parallelwahlen erklaeren eine zu niedrige Stimmenquote.
        g = r.rel_geschwister
        n_vergleich, spann = r.n_geschwister, r.spann_geschwister
        geschwister_genutzt = False
        if r.art == "B" and not (r.n_geschwister >= 2 and r.spann_geschwister <= 0.02):
            # In Briefwahlbezirken tragen Wahlen mit unterschiedlichem Gebiet oft dieselbe
            # Nummer. Nur wenn mehrere Wahlen exakt dieselbe Waehlerzahl melden, ist
            # gesichert, dass sie dasselbe Gebiet meinen - gleich welcher Gebietsklasse.
            if bw_gleich:
                g = r.waehler / r.waehler_gleich_bw
                n_vergleich, spann = r.n_gleich_bw, 0.0
            else:
                if np.isfinite(g):
                    belege.append("Parallelwahlen im Briefwahlbezirk nicht vergleichbar")
                g = np.nan
        if np.isfinite(g):
            geschwister_genutzt = True
            # Gewicht: zwei einige Vergleichswahlen wiegen mehr als eine, uneinige weniger.
            if n_vergleich >= 2 and spann <= 0.10:
                gew, zusatz = 3, f"{int(n_vergleich)} Parallelwahlen einig"
            elif n_vergleich >= 2:
                gew, zusatz = 1, (f"{int(n_vergleich)} Parallelwahlen uneinig "
                                  f"(Spanne {spann:.0%})")
            else:
                gew, zusatz = 2, "1 Parallelwahl"
            if g > 1 + gt:
                belege.append(f"{g:.2f}x so viele Zettel wie in den Parallelwahlen ({zusatz})")
                fuer_zettel += gew
            elif g < 1 - gt:
                belege.append(f"nur {g:.2f}x der Zettel der Parallelwahlen (anderer Zuschnitt?)")
            else:
                belege.append(f"Zettelzahl deckt sich mit den Parallelwahlen "
                              f"({g:.2f}x, {zusatz})")
                fuer_stimmen += gew
        # (1b) Briefwahlbezirk: meldet ein Ortsrat den ganzen Bezirk statt seiner Ortschaft?
        #      Das ist ein echter Parallelwahl-Vergleich, er zaehlt deshalb auch dann, wenn
        #      der allgemeine Vergleich oben wegen unterschiedlicher Gebiete verworfen wurde.
        if or_ganz:
            andere = r.or_summe - r.waehler
            belege.append(f"Ortsrat meldet alle {r.waehler:.0f} Waehler des Briefwahlbezirks, "
                          f"obwohl dort weitere {andere:.0f} Ortsrats-Waehler gemeldet sind")
            fuer_zettel += 3
            geschwister_genutzt = True
        # (1c) Briefwahlbezirk: Ortsrat meldet mehr Waehler als jede gemeindeweite Wahl, und
        #      seine Stimmen passen zu deren Zettelzahl (Herleitung oben bei gem_zettel_bw).
        if or_zu_gross:
            belege.append(f"mehr Waehler ({r.waehler:.0f}) als die gemeindeweite Wahl im "
                          f"Briefwahlbezirk ({r.gem_waehler_bw:.0f}, {r.gem_wahl_bw}); die Stimmen "
                          f"passen zu deren {r.gem_zettel_bw:.0f} gueltigen Zetteln")
            fuer_zettel += 3
            geschwister_genutzt = True

        # (2) und (3) Wahlberechtigte bzw. Vorwahl. Liegen gemeldete *und* rechnerische
        #     Zettelzahl weit auf derselben Seite, passt der Anker nicht zum Bezirk.
        if anker and np.isfinite(z_ist):
            if (z_ist >= zv and z_soll >= zv) or (z_ist <= -zv and z_soll <= -zv):
                belege.append(f"{anker} passt nicht zum Bezirk (z={z_ist:+.1f}/{z_soll:+.1f})")
            else:
                # Ein Anker mit grosser Streuung kann bestaetigen, dass etwas *nicht* passt,
                # aber nicht, dass etwas passt - dafuer ist er zu unscharf.
                scharf = s <= a.s_max
                if abs(z_ist) <= zt and scharf:
                    belege.append(f"Zettelzahl passt zu {anker} (z={z_ist:+.1f})")
                    fuer_stimmen += 1
                elif z_ist >= zv:
                    belege.append(f"mehr Zettel als nach {anker} erwartbar (z={z_ist:+.1f})")
                    fuer_zettel += 2 if z_ist >= 2 * zv else 1
                if z_soll <= -zv:
                    belege.append(f"Stimmen reichen nicht zu {anker} (z={z_soll:+.1f})")
                    fuer_stimmen += 2 if z_soll <= -2 * zv else 1
                elif abs(z_soll) <= zt and z_ist > zt and scharf:
                    belege.append(f"Stimmen passen zu {anker} (z={z_soll:+.1f})")
                    fuer_zettel += 1
        if muster:
            # Hinweis auf den moeglichen Hergang, aber kein eigener Beleg: unter den rund
            # 30 Ziffernvarianten einer Zahl trifft rein zufaellig oft eine das Zielfenster.
            belege.append(f"Tippfehler denkbar: {muster}")
        if r.briefwahl_neu:
            belege.append(f"Briefwaehler jetzt im Urnenbezirk (+{r.wahlschein_zuwachs:.0%})")
        if r.zuschnitt_neu:
            belege.append(f"Wahlberechtigte {r.groesse_rel_2021:.2f}x gegenueber 2021")

        beleg = " · ".join(belege)
        if not anker and not np.isfinite(g) and not geschwister_genutzt:
            return pd.Series(["kein Vergleich moeglich", anker, z_ist, z_soll, muster, beleg])
        if r.briefwahl_neu and fuer_zettel > fuer_stimmen:
            return pd.Series(["Struktur geaendert (Briefwahl im Urnenbezirk)",
                              anker, z_ist, z_soll, muster, beleg])
        # Ohne brauchbaren Parallelwahl-Vergleich haengt ein Zettelzahl-Verdacht an einem
        # einzigen Anker. Das reicht nur bei einer extremen Abweichung: der Vergleich mit der
        # Vorwahl traegt nicht, wenn sich ein Bezirk seither stark veraendert hat (Verden,
        # Briefwahlbezirk 240: von 288 auf 490 Waehler seit 2021).
        if (fuer_zettel > fuer_stimmen and not geschwister_genutzt
                and not (np.isfinite(z_ist) and abs(z_ist) >= 2 * zv)):
            return pd.Series(["nicht entscheidbar (nur ein unsicherer Vergleich)",
                              anker, z_ist, z_soll, muster, beleg])
        if fuer_zettel >= 3 and fuer_zettel > fuer_stimmen:
            return pd.Series(["Zettelzahl unplausibel", anker, z_ist, z_soll, muster, beleg])
        if fuer_stimmen >= 3 and fuer_stimmen > fuer_zettel:
            return pd.Series(["Stimmen fehlen", anker, z_ist, z_soll, muster, beleg])
        if fuer_zettel > fuer_stimmen:
            return pd.Series(["Tendenz: Zettelzahl", anker, z_ist, z_soll, muster, beleg])
        if fuer_stimmen > fuer_zettel:
            return pd.Series(["Tendenz: Stimmen fehlen", anker, z_ist, z_soll, muster, beleg])
        if r.zuschnitt_neu:
            return pd.Series(["Bezirk neu zugeschnitten", anker, z_ist, z_soll, muster, beleg])
        return pd.Series(["nicht entscheidbar", anker, z_ist, z_soll, muster, beleg])

    m[["urteil", "anker", "z_ist", "z_soll", "tippfehler", "belege"]] = m.apply(urteil, axis=1)
    A = m[m.auffaellig == True].copy()   # noqa: E712
    A = A.sort_values(["urteil", "z_ist"], ascending=[True, False])

    out = Path(a.out)
    out.parent.mkdir(parents=True, exist_ok=True)
    cols = ["ags", "wahl", "gebiet-nr", "gebiet-name", "waehlberechtigte", "waehler",
            "wahlschein_anteil", "gueltige_zettel", "stimmen", "stimmen_je_zettel",
            "median_der_wahl", "fehlend_ggue_median", "einordnung", "treffer",
            "v_A", "v_B", "v_C2", "quote", "faktor", "faktor_n",
            "erw_wb", "erw_21", "s_wb", "s_21", "zettel_soll",
            "z_ist_wb", "z_soll_wb", "z_ist_21", "z_soll_21", "z_ist", "z_soll", "anker",
            "groesse_rel_2021", "wahlbeteiligung", "wahlbeteiligung_2021",
            "wahlschein_anteil_2021", "wahlschein_zuwachs", "zuschnitt_neu", "briefwahl_neu",
            "wahlen_im_bezirk", "waehler_geschwister", "zettel_geschwister", "n_geschwister",
            "geschwister_unauffaellig", "spann_geschwister", "rel_geschwister", "or_summe",
            "n_gleich_bw", "waehler_gleich_bw", "zettel_gleich_bw",
            "gem_wahl_bw", "gem_waehler_bw", "gem_zettel_bw",
            "tippfehler", "urteil", "belege"]
    cols = [c for c in cols if c in A.columns]
    A[cols].to_csv(out, index=False)

    print(f"{len(A)} auffaellige Bezirke gegen {a.datum_alt} geprueft -> {out}")
    print(f"Vergleichsdaten: {H.ags.nunique()} Gemeinden, {len(H)} Bezirksergebnisse")
    print(f"Schwellen: Treffer |z| <= {zt}, Verdacht |z| >= {zv}, "
          f"Zuschnitt-Toleranz {a.toleranz:.0%}\n")
    print("Bezirkszuordnung auf 2021:")
    print(A.treffer.value_counts().to_string())
    print("\nAnker:")
    print(A.anker.replace("", "keiner").value_counts().to_string())
    print("\nUrteil:")
    print(A.urteil.value_counts().to_string())

    show = ["ags", "wahl", "gebiet-nr", "gebiet-name", "gueltige_zettel", "zettel_soll",
            "rel_geschwister", "erw_wb", "erw_21", "z_ist", "z_soll", "stimmen_je_zettel",
            "tippfehler"]
    fmt = {"zettel_soll": "{:.0f}".format, "erw_wb": "{:.0f}".format, "erw_21": "{:.0f}".format,
           "rel_geschwister": "{:.2f}".format,
           "z_ist": "{:+.1f}".format, "z_soll": "{:+.1f}".format,
           "stimmen_je_zettel": "{:.2f}".format}
    for titel, teil in (("Zettelzahl unplausibel", A[A.urteil == "Zettelzahl unplausibel"]),
                        ("Tendenz: Zettelzahl", A[A.urteil == "Tendenz: Zettelzahl"]),
                        ("Stimmen fehlen", A[A.urteil == "Stimmen fehlen"])):
        if not len(teil):
            continue
        print(f"\n{titel} ({len(teil)}):")
        with pd.option_context("display.width", 220, "display.max_colwidth", 24):
            print(teil.head(a.top)[show].to_string(index=False, formatters=fmt))

    if a.report:
        _zettel_report(Path(a.report), A, H, a)
        print(f"\nReport -> {a.report}")


def _zettel_report(pfad: Path, A: pd.DataFrame, H: pd.DataFrame, a):
    """Markdown-Bericht: Methode, Zaehlung je Urteil, Tabellen der entscheidbaren Faelle."""
    pfad.parent.mkdir(parents=True, exist_ok=True)
    sp = ["ags", "wahl", "gebiet-nr", "gebiet-name", "gueltige_zettel", "zettel_soll",
          "rel_geschwister", "erw_wb", "erw_21", "z_ist", "z_soll", "stimmen_je_zettel",
          "tippfehler"]
    kurz = {"gueltige_zettel": "Zettel", "zettel_soll": "Zettel laut Stimmen",
            "rel_geschwister": "vs. Parallelwahlen",
            "erw_wb": "erwartet (Wahlber.)", "erw_21": "erwartet (2021)",
            "stimmen_je_zettel": "Stimmen/Zettel", "gebiet-nr": "Bezirk", "gebiet-name": "Name"}

    def tab(df, n=40):
        d = df.head(n)[[c for c in sp if c in df.columns]].copy()
        for c in ("zettel_soll", "erw_wb", "erw_21"):
            if c in d:
                d[c] = d[c].round(0)
        return md_table(d.rename(columns=kurz), floatfmt="{:.2f}")

    t = [f"# Zettelcheck: was steckt hinter den {len(A)} auffaelligen Bezirken?", "",
         f"Kommunalwahl 13.09.2026 gegen {a.datum_alt} · erzeugt mit `kw_ebenen.py zettelcheck`",
         f"Rohdaten: `{a.out}` · Vergleichsdaten: {H.ags.nunique()} Gemeinden, "
         f"{len(H)} Bezirksergebnisse aus der Vorwahl", "",
         "## Frage", "",
         "Ein Bezirk faellt auf, wenn die Stimmen je Stimmzettel weit unter dem Median der",
         "eigenen Wahl liegen. Dieser Quotient kann aus zwei Gruenden zu klein sein:", "",
         "1. **Es fehlen Stimmen** - der Zaehler ist zu klein. So war es in Ostercappeln:",
         "   uebertragen wurde die Zahl der Stimmzettel eines Stapels statt der Stimmen.",
         "2. **Es stehen zu viele Stimmzettel in der Meldung** - der Nenner ist zu gross,",
         "   etwa durch einen Zahlendreher oder einen anderen Gebietsbezug.", "",
         "Beide Faelle sehen in der Plausibilitaetspruefung gleich aus. Getrennt werden sie",
         "hier ueber drei Vergleiche.", "",
         "## Die drei Vergleiche", "",
         "**1. Parallelwahlen im selben Bezirk (schaerfster Massstab).** Am selben Tag, im",
         "selben Raum, mit denselben Waehlern werden zwei bis fuenf Wahlen ausgezaehlt. Die",
         "Zahl der **Waehler** ist dort dieselbe - verglichen wird deshalb `B`, nicht die",
         "gueltigen Stimmzettel, die sich legitim um die ungueltigen unterscheiden. Weicht",
         "eine Wahl nach oben ab, ist ihre Melde-Zahl verdaechtig; decken sich alle, ist die",
         "Zettelzahl bestaetigt und es fehlen wirklich Stimmen. Der Vergleich ist blind,",
         "wenn dieselbe falsche Zahl in allen Wahlen des Bezirks steht - dafuer gibt es 2.",
         "und 3.", "",
         "**2. Wahlberechtigte des eigenen Termins.** `A` stammt aus dem Waehlerverzeichnis",
         "und haengt nicht an der Auszaehlung. Erwartet werden `A` x uebliche Quote",
         "Stimmzettel je Wahlberechtigtem derselben Wahl.", "",
         f"**3. Vorwahl {a.datum_alt[-4:]}.** Die Stimmzettel desselben Bezirks, hochgerechnet",
         "auf die Veraenderung der Gemeinde. Zuordnung ueber Bezirksnummer und -name,",
         "getrennt nach Urnen- und Briefwahlbezirken.", "",
         "Wie weit ein unauffaelliger Bezirk von den Erwartungen 2 und 3 normalerweise",
         "abweicht, misst die Streuung (MAD) der unauffaelligen Bezirke derselben Wahl.",
         "Daraus werden z-Werte: `z_ist` fuer die gemeldete Zettelzahl, `z_soll` fuer die",
         "Zettelzahl, die zu den gemeldeten Stimmen passen wuerde.", "",
         "## Wie daraus ein Urteil wird", "",
         "Jeder Vergleich gibt Punkte fuer eine der beiden Erklaerungen. Die Parallelwahlen",
         "wiegen am schwersten (drei Punkte, wenn mindestens zwei von ihnen einig sind; einen",
         f"Punkt, wenn sie untereinander um mehr als 10 % streuen). Ein Anker zaehlt, wenn er",
         f"|z| >= {a.z_verdacht} widerlegt oder |z| <= {a.z_treffer} bestaetigt - bestaetigen",
         f"darf er nur, solange seine eigene Streuung unter {a.s_max:.0%} liegt. Drei Punkte",
         "und mehr als die Gegenseite ergeben ein Urteil, sonst nur eine Tendenz.", "",
         "Zwei harmlose Erklaerungen fuer eine hoehere Zettelzahl werden vorher abgetrennt:",
         f"ein neuer Bezirkszuschnitt (Wahlberechtigte weichen um mehr als {a.toleranz:.0%} von",
         "der Vorwahl ab) und eine geaenderte Briefwahlstruktur (Anteil Wahlschein-Waehler um",
         "mehr als 15 Prozentpunkte gestiegen, die Briefwaehler werden also jetzt im",
         "Urnenbezirk mitgezaehlt). In Briefwahlbezirken meint dieselbe Nummer oft verschiedene",
         "Gebiete. Verglichen wird dort innerhalb derselben Gebietsklasse, wenn sich mindestens",
         "zwei Wahlen auf 2 % einig sind, sonst mit einer unauffaelligen Wahl, die (+/-2 %)",
         "dieselbe Waehlerzahl meldet. Ein Ortsrat, der den ganzen Bezirk oder weit mehr",
         "Waehler als die gemeindeweiten Wahlen meldet, zaehlt als Zettelzahl-Beleg. Regeln im",
         "Einzelnen: README_kw_ebenen.md, Abschnitt \"Wie geprueft wird\".", "",
         "## Ergebnis", "", md_table(A.urteil.value_counts().rename_axis("Urteil")
                                     .reset_index(name="Bezirke"), floatfmt="{:.0f}"), ""]

    fall = A[A.urteil.isin(["Zettelzahl unplausibel", "Tendenz: Zettelzahl"])]
    t += [f"## Verdacht auf falsche Stimmzettel- oder Waehlerzahl ({len(fall)})", "",
          "Hier passt die Stimmenzahl zur Erwartung, die gemeldete Zettelzahl nicht.",
          "`vs. Parallelwahlen` ist der wichtigste Wert: er sagt, um welchen Faktor dieser",
          "Bezirk mehr Waehler meldet als dieselben Waehler in den anderen Wahlen desselben",
          "Tages. Die Spalte `tippfehler` nennt eine Zahl, die sich durch einen einzelnen",
          "Erfassungsfehler von der gemeldeten unterscheidet und zu Stimmen und Erwartung",
          "passt - ein Hinweis auf den moeglichen Hergang, kein Beleg: unter den rund 30",
          "Ziffernvarianten einer Zahl trifft auch zufaellig manchmal eine das Zielfenster.",
          "Die belastbare Rekonstruktion steht in `waehler_geschwister` der CSV.",
          "", tab(fall), ""]
    klar = A[A.urteil == "Stimmen fehlen"]
    tendenz = A[A.urteil == "Tendenz: Stimmen fehlen"]
    t += [f"## Stimmen fehlen ({len(klar)})", "",
          "Die Zettelzahl passt zu den Parallelwahlen, zu den Wahlberechtigten und zur",
          "Vorwahl - nur die Stimmen sind zu wenige. Das ist das Muster aus Ostercappeln.",
          "", tab(klar, 60), "",
          f"## Tendenz: Stimmen fehlen ({len(tendenz)})", "",
          "Dasselbe Muster, aber mit duennerer Belegkette - etwa weil nur ein Vergleich",
          "greift oder die Abweichung kleiner ist.",
          "", tab(tendenz, 40), ""]
    rest = A[~A.urteil.isin(["Zettelzahl unplausibel", "Tendenz: Zettelzahl",
                             "Stimmen fehlen", "Tendenz: Stimmen fehlen"])]
    t += [f"## Nicht entscheidbar ({len(rest)})", "",
          md_table(rest.groupby("urteil").size().rename_axis("Urteil")
                   .reset_index(name="Bezirke"), floatfmt="{:.0f}"), "",
          "Meist fehlt der Vergleich: Briefwahlbezirke ohne Wahlberechtigte, Bezirke ohne",
          "Parallelwahl oder ohne Entsprechung in der Vorwahl.", "", tab(rest, 30), "",
          "## Grenzen", "",
          "- Die Zuordnung auf die Vorwahl geht ueber Nummer und Namen. Umbenannte *und*",
          "  umnummerierte Bezirke bleiben unzugeordnet; dort tragen nur die anderen beiden",
          "  Vergleiche.",
          "- Briefwahlbezirke haben keine Wahlberechtigten (`A = 0`), und Wahlen mit",
          "  unterschiedlichem Gebiet tragen dort oft dieselbe Bezirksnummer. Die Aussagen",
          "  sind dort schwaecher als in Urnenbezirken.",
          "- Alle harten Identitaeten sind erfuellt (`C1 + C2 = B` in jedem Bezirk). Eine",
          "  falsche Zettelzahl steht also nie allein: Waehler, ungueltige und gueltige",
          "  Stimmzettel sind untereinander stimmig. Verdreht waere dann die ganze Meldung,",
          "  nicht ein einzelnes Feld.",
          "- Ein Verdacht ist eine Rueckfrage wert, kein Nachweis. Die Niederschriften der",
          "  Wahlvorstaende liegen bei den Wahlaemtern, nicht in den offenen Daten.", ""]
    pfad.write_text("\n".join(t), encoding="utf-8")


# --------------------------------------------------------------------------- Wirkung
def cmd_wirkung(a):
    """Rechnet die Befunde auf die jeweilige Wahl hoch: Landkreis, Wahl, Anteil.

    Eine absolute Zahl fehlender Stimmen sagt wenig. Entscheidend ist, woran man sie misst:
    400 fehlende Stimmen sind in einer Ortsratswahl mit 7000 Stimmen ein anderer Vorgang als
    in einer Kreistagswahl mit 400 000. Deshalb wird je Wahl die *eigene* Grundgesamtheit
    gebildet:

      Ortsrat / Gemeinde- bzw. Stadtrat / Samtgemeinderat  alle Bezirke dieser Wahl
      Kreistag                                             alle Bezirke des ganzen Kreises,
                                                           also ueber alle Gemeinden hinweg

    Gezaehlt werden nur die Bezirke, bei denen `zettelcheck` auf fehlende Stimmen kommt.
    Faelle mit Verdacht auf eine falsche Stimmzettelzahl stehen getrennt: dort fehlen keine
    Stimmen, dort stimmt die Meldung nicht.
    """
    P = pd.read_csv(a.plausi, dtype={"ags": str, "gebiet-nr": str})
    Z = pd.read_csv(a.zettelcheck, dtype={"ags": str, "gebiet-nr": str})
    namen = {}
    if Path(a.namen).exists():
        namen = (pd.read_csv(a.namen, dtype=str).fillna("")
                 .set_index("ags").gemeinde.to_dict())

    P = P[P.stimmen_je_waehler == VOTES_PER_VOTER].copy()   # nur Verhaeltniswahlen
    P["kreis"] = P.ags.str[:5]
    P["typ"] = P.wahl.map(wahl_typ)
    P["gemeinde"] = P.ags.map(namen).fillna("")
    # Grundgesamtheit: der Kreistag wird im ganzen Kreis gewaehlt, alles andere je Gemeinde.
    P["wahl_id"] = np.where(P.typ == "Kreistag", P.kreis + "|Kreistagswahl", P.ags + "|" + P.wahl)

    # Namenspruefung: die Wahlnamen nennen oft den Kreis ("Kreistagswahl Landkreis Aurich").
    for kreis, g in P[P.typ == "Kreistag"].groupby("kreis"):
        gefunden = {re.sub(r".*?(Landkreis|Kreis)\s+", "", w).strip()
                    for w in g.wahl.unique() if re.search(r"(Landkreis|Kreis)\s+\S", w)}
        soll = KREIS_NAMEN.get(kreis, "")
        for f in gefunden:
            if _fold(f).lower()[:6] not in _fold(soll).lower():
                print(f"  Hinweis: Kreis {kreis} heisst hier '{f}', Tabelle sagt '{soll}'")

    Z["fehlt"] = Z.urteil.str.contains("Stimmen fehlen")
    Z["zettel_verdacht"] = Z.urteil.str.contains("Zettelzahl")
    key = ["ags", "wahl", "gebiet-nr"]
    P = P.merge(Z[key + ["urteil", "fehlt", "zettel_verdacht", "belege"]], on=key, how="left")
    P["fehlt"] = P.fehlt.fillna(False)
    P["zettel_verdacht"] = P.zettel_verdacht.fillna(False)
    P["fehlend"] = np.where(P.fehlt, P.fehlend_ggue_median, 0)

    W = (P.groupby("wahl_id")
           .agg(kreis=("kreis", "first"), typ=("typ", "first"), wahl=("wahl", "first"),
                bezirke=("gebiet-nr", "size"), stimmen=("stimmen", "sum"),
                waehler=("waehler", "sum"), gemeinden=("ags", "nunique"),
                auffaellig=("auffaellig", "sum"), bezirke_fehlend=("fehlt", "sum"),
                fehlend=("fehlend", "sum"), bezirke_zettel=("zettel_verdacht", "sum"))
           .reset_index())
    W["landkreis"] = W.kreis.map(KREIS_NAMEN).fillna(W.kreis)
    # Gemeindename: beim Kreistag die betroffenen Gemeinden, sonst die eine Gemeinde
    betroffen = (P[P.auffaellig].groupby("wahl_id").gemeinde
                  .apply(lambda g: ", ".join(sorted(set(x for x in g if x)))))
    alle_gem = P.groupby("wahl_id").gemeinde.first()
    W["gemeinde"] = np.where(W.typ == "Kreistag",
                             W.wahl_id.map(betroffen).fillna(""),
                             W.wahl_id.map(alle_gem).fillna(""))
    W["anteil_pct"] = 100 * W.fehlend / W.stimmen.replace(0, np.nan)
    W["stimmen_je_bezirk"] = W.stimmen / W.bezirke
    B = W[(W.auffaellig > 0)].sort_values("anteil_pct", ascending=False)

    K = (B.groupby(["kreis", "landkreis"])
           .agg(wahlen=("wahl_id", "nunique"), bezirke_auffaellig=("auffaellig", "sum"),
                bezirke_fehlend=("bezirke_fehlend", "sum"),
                bezirke_zettel=("bezirke_zettel", "sum"), fehlend=("fehlend", "sum"),
                max_anteil_pct=("anteil_pct", "max"))
           .reset_index())
    gem_je_kreis = (P[P.auffaellig].assign(kreis=lambda d: d.ags.str[:5])
                     .groupby("kreis").gemeinde
                     .apply(lambda g: ", ".join(sorted(set(x for x in g if x)))))
    K["gemeinden"] = K.kreis.map(gem_je_kreis).fillna("")
    K["anzahl_gemeinden"] = K.gemeinden.str.count(",").add(1).where(K.gemeinden != "", 0)
    schlimmste = B.sort_values("anteil_pct", ascending=False).groupby("kreis").first()
    K["groesste_wirkung"] = K.kreis.map(
        schlimmste.apply(lambda r: f"{r.wahl} ({r.anteil_pct:.1f} %)", axis=1))
    K = K.sort_values("fehlend", ascending=False)

    out_w, out_k = Path(a.out), Path(a.out_kreise)
    out_w.parent.mkdir(parents=True, exist_ok=True)
    spalten_w = ["landkreis", "kreis", "gemeinde", "wahl", "typ", "bezirke", "gemeinden",
                 "stimmen", "auffaellig", "bezirke_fehlend", "fehlend", "anteil_pct",
                 "bezirke_zettel"]
    B[spalten_w].to_csv(out_w, index=False)
    spalten_k = ["landkreis", "kreis", "anzahl_gemeinden", "wahlen", "bezirke_auffaellig",
                 "bezirke_fehlend", "bezirke_zettel", "fehlend", "max_anteil_pct",
                 "groesste_wirkung", "gemeinden"]
    K[spalten_k].to_csv(out_k, index=False)

    print(f"{len(B)} betroffene Wahlen in {K.kreis.nunique()} Landkreisen -> {out_w}")
    print(f"Landkreis-Uebersicht -> {out_k}\n")
    print("Nach Wahlebene (nur Wahlen mit fehlenden Stimmen):")
    F = B[B.fehlend > 0]
    t = (F.groupby("typ").agg(wahlen=("wahl_id", "size"), bezirke=("bezirke_fehlend", "sum"),
                              fehlend=("fehlend", "sum"), median_pct=("anteil_pct", "median"),
                              max_pct=("anteil_pct", "max")))
    print(t.to_string(float_format=lambda v: f"{v:.2f}"))
    print("\nGroesste Wirkung (Anteil der fehlenden Stimmen an der Wahl):")
    z = B[B.fehlend > 0].head(a.top)[["landkreis", "gemeinde", "wahl", "typ", "stimmen",
                                      "fehlend", "anteil_pct", "bezirke_fehlend"]]
    with pd.option_context("display.width", 220, "display.max_colwidth", 26):
        print(z.to_string(index=False, formatters={"anteil_pct": "{:.2f}".format,
                                                   "fehlend": "{:.0f}".format,
                                                   "stimmen": "{:.0f}".format}))
    print("\nLandkreise:")
    with pd.option_context("display.width", 220, "display.max_colwidth", 30):
        print(K.head(a.top)[["landkreis", "anzahl_gemeinden", "wahlen", "bezirke_auffaellig",
                             "fehlend", "max_anteil_pct"]]
              .to_string(index=False, formatters={"fehlend": "{:.0f}".format,
                                                  "max_anteil_pct": "{:.2f}".format}))
    if a.report:
        _wirkung_report(Path(a.report), B, K, P, a)
        print(f"\nReport -> {a.report}")


def _wirkung_report(pfad: Path, B: pd.DataFrame, K: pd.DataFrame, P: pd.DataFrame, a):
    """Markdown fuer die Landkreise: was liegt bei uns, wie stark wirkt es."""
    pfad.parent.mkdir(parents=True, exist_ok=True)

    def t_wahlen(df, n=200):
        d = df.head(n)[["landkreis", "gemeinde", "wahl", "typ", "stimmen", "bezirke_fehlend",
                        "fehlend", "anteil_pct"]].copy()
        d["stimmen"] = d.stimmen.round(0).astype("Int64")
        d["fehlend"] = d.fehlend.round(0).astype("Int64")
        d["anteil_pct"] = d.anteil_pct.round(2)
        return md_table(d.rename(columns={"landkreis": "Landkreis", "gemeinde": "Gemeinde",
                                          "wahl": "Wahl", "typ": "Ebene",
                                          "stimmen": "Stimmen der Wahl",
                                          "bezirke_fehlend": "Bezirke",
                                          "fehlend": "fehlende Stimmen",
                                          "anteil_pct": "Anteil %"}), floatfmt="{:.2f}")

    kt = K[["landkreis", "anzahl_gemeinden", "wahlen", "bezirke_auffaellig", "bezirke_fehlend",
            "bezirke_zettel", "fehlend", "max_anteil_pct", "gemeinden"]].copy()
    kt["fehlend"] = kt.fehlend.round(0).astype("Int64")
    kt["max_anteil_pct"] = kt.max_anteil_pct.round(2)
    kt = kt.rename(columns={"landkreis": "Landkreis", "anzahl_gemeinden": "Gemeinden",
                            "wahlen": "Wahlen", "bezirke_auffaellig": "auff. Bezirke",
                            "bezirke_fehlend": "davon Stimmen fehlen",
                            "bezirke_zettel": "davon Zettelzahl",
                            "fehlend": "fehlende Stimmen", "max_anteil_pct": "max. Anteil %",
                            "gemeinden": "betroffene Gemeinden"})

    typ = (B[B.fehlend > 0].groupby("typ")
             .agg(Wahlen=("wahl_id", "size"), Bezirke=("bezirke_fehlend", "sum"),
                                fehlende_Stimmen=("fehlend", "sum"),
                                Median_Anteil=("anteil_pct", "median"),
                                max_Anteil=("anteil_pct", "max")).reset_index())
    for c in ("Wahlen", "Bezirke", "fehlende_Stimmen"):
        typ[c] = typ[c].round(0).astype("Int64")
    typ = typ.rename(columns={"typ": "Ebene", "Bezirke": "auff. Bezirke",
                              "fehlende_Stimmen": "fehlende Stimmen",
                              "Median_Anteil": "Median Anteil %", "max_Anteil": "max. Anteil %"})

    t = [f"# Wirkung der Befunde je Landkreis und Wahl", "",
         "Kommunalwahl Niedersachsen 13.09.2026 · erzeugt mit `kw_ebenen.py wirkung`",
         f"Grundlage: `{a.plausi}` und `{a.zettelcheck}` · vorlaeufige Ergebnisse", "",
         "## Wozu diese Tabelle", "",
         "Eine absolute Zahl fehlender Stimmen sagt wenig ueber die Tragweite. 400 fehlende",
         "Stimmen sind in einer Ortsratswahl mit 7000 Stimmen ein anderer Vorgang als in",
         "einer Kreistagswahl mit 400 000. Deshalb ist jede Wahl an ihrer *eigenen*",
         "Grundgesamtheit gemessen:", "",
         "| Ebene | Grundgesamtheit |",
         "|---|---|",
         "| Ortsrat, Gemeinde-/Stadtrat, Samtgemeinderat | alle Bezirke dieser Wahl |",
         "| Kreistag | alle Bezirke des ganzen Kreises, ueber alle Gemeinden hinweg |", "",
         "Gezaehlt sind nur Bezirke, bei denen der Abgleich mit Parallelwahlen,",
         "Wahlberechtigten und der Vorwahl 2021 auf **fehlende Stimmen** hinauslaeuft.",
         "Faelle mit Verdacht auf eine falsche **Stimmzettelzahl** stehen in einer eigenen",
         "Spalte: dort fehlen keine Stimmen, dort passt die Meldung nicht zu den",
         "Parallelwahlen. Auf die Sitzverteilung wirkt das nicht, auf die ausgewiesene",
         "Wahlbeteiligung schon.", "",
         "## Nach Wahlebene", "",
         "Nur Wahlen, in denen nach dem Abgleich Stimmen fehlen - die reinen",
         "Zettelzahl-Verdachtsfaelle wuerden den Median sonst mit Nullen nach unten ziehen.",
         "", md_table(typ, floatfmt="{:.2f}"), "",
         "Das ist der Kern: Auf Kreisebene verschwinden die Abweichungen im Nenner, auf",
         "Ortsratsebene koennen zwei Wahlbezirke das Ergebnis der ganzen Wahl verschieben.",
         "", "## Nach Landkreis", "",
         "Sortiert nach fehlenden Stimmen. Die letzte Spalte nennt die Gemeinden, in denen",
         "Bezirke auffaellig sind - das ist die Liste zum Nachfragen.", "",
         md_table(kt, floatfmt="{:.2f}"), "",
         "## Die Wahlen mit der groessten Wirkung", "",
         "Alle betroffenen Wahlen mit fehlenden Stimmen, nach Anteil sortiert.", "",
         t_wahlen(B[B.fehlend > 0]), "",
         "## Verdacht auf falsche Stimmzettelzahl", "",
         "Hier fehlen keine Stimmen; die gemeldete Zahl der Waehler bzw. Stimmzettel passt",
         "nicht zu den anderen Wahlen desselben Bezirks. Details je Fall in",
         f"`{a.zettelcheck}` (Spalte `belege`).", ""]
    zv = P[P.zettel_verdacht].copy()
    if len(zv):
        zv["landkreis"] = zv.ags.str[:5].map(KREIS_NAMEN)
        d = zv[["landkreis", "gemeinde", "wahl", "gebiet-nr", "gebiet-name", "waehler",
                "gueltige_zettel", "stimmen", "stimmen_je_zettel", "urteil"]].copy()
        d["stimmen_je_zettel"] = d.stimmen_je_zettel.round(2)
        t += [md_table(d.rename(columns={"landkreis": "Landkreis", "gemeinde": "Gemeinde",
                                         "wahl": "Wahl", "gebiet-nr": "Bezirk",
                                         "gebiet-name": "Name", "waehler": "Waehler",
                                         "gueltige_zettel": "gueltige Zettel",
                                         "stimmen": "Stimmen",
                                         "stimmen_je_zettel": "Stimmen/Zettel",
                                         "urteil": "Urteil"}), floatfmt="{:.2f}"), ""]
    t += ["## Grenzen", "",
          "- Beim Kreistag ist die Grundgesamtheit der ganze Kreis. Gewaehlt wird aber in",
          "  Wahlbereichen; die tatsaechliche Wirkung auf die Sitzverteilung ist dort also",
          "  groesser als der ausgewiesene Anteil.",
          "- Die fehlenden Stimmen sind gegen den Median der eigenen Wahl geschaetzt, nicht",
          "  gegen die Niederschrift. Die genaue Zahl steht in den Wahlunterlagen.",
          "- Vorlaeufige Ergebnisse. Nach Feststellung durch die Wahlausschuesse neu rechnen.",
          ""]
    pfad.write_text("\n".join(t), encoding="utf-8")


# --------------------------------------------------------------------------- Report
def md_table(df: pd.DataFrame, floatfmt="{:.1f}") -> str:
    if df.empty:
        return "_keine Daten_\n"
    cols = list(df.columns)
    out = ["| " + " | ".join(map(str, cols)) + " |", "|" + "---|" * len(cols)]
    for _, r in df.iterrows():
        cells = []
        for c in cols:
            v = r[c]
            if isinstance(v, (float, np.floating)):
                cells.append("" if pd.isna(v) else floatfmt.format(v))
            elif isinstance(v, (bool, np.bool_)):
                cells.append("ja" if v else "nein")
            else:
                cells.append(str(v))
        out.append("| " + " | ".join(cells) + " |")
    return "\n".join(out) + "\n"


def write_report(out: Path, D, muni_b, muni_all, kreis_s, top, absorb, post, log, top_n, demo,
                 dmask=None, scope="Briefwahlbezirke", dsel=None, basis=""):
    bd = D[D.is_brief] if dmask is None else D[dmask]
    ud = D[~D.is_brief]
    sel = bd if dsel is None else D[dsel]
    wmed = lambda x: np.nan if x.empty else float(np.median(x))
    lines = []
    if demo:
        lines.append("> **DEMO – SYNTHETISCHE DATEN. Keine realen Wahlergebnisse.**\n")
    lines += [
        "# Ebenenvergleich Gemeindewahl vs. Ortsratswahl",
        "",
        "## Kennzahlen",
        f"- Gepaarte Bezirke: {len(D)} (davon Briefwahl: {len(bd)}), Gemeinden: {D.ags.nunique()}, "
        f"Ortsrats-/Ortsbeiratswahlen: {D.ortsrat.nunique()}",
        f"- **Auswertungsebene: {scope}** ({len(bd)} von {len(D)} Bezirkspaaren)",
        f"- Median Dissimilarität (TVD) Auswertungsebene: {wmed(bd.tvd_pp):.1f} %p · "
        f"Urnenbezirke: {wmed(ud.tvd_pp):.1f} %p",
        f"- Paare mit abweichender Wählerzahl (>{CONGRUENCE_TOL:.0%}, Bezirkszuschnitt "
        f"nicht deckungsgleich): {int((~bd.kongruent).sum())} von {len(bd)}",
        f"- Gemeindewahl-Bezirke, die mehrere Ortsräte abdecken (Vergleich unterschiedlicher "
        f"Grundgesamtheiten): {int(bd.gr_bezirk_mehrfach.sum())} von {len(bd)}",
        f"- **Rangliste und Aggregate: {basis}** ({len(sel)} Paare)",
        f"- Bezirke mit fehlenden Listen auf Ortsratsebene: {int((D.orphan_pp > 0).sum())} "
        f"(mittlerer verwaister Anteil: {D.loc[D.orphan_pp > 0, 'orphan_pp'].mean():.1f} %p)",
        f"- Mittlere Stimmen je Wähler: GR {D.stimmen_je_waehler_gr.mean():.2f} · "
        f"OR {D.stimmen_je_waehler_or.mean():.2f} (max. 3)",
        "",
        f"## Top {top_n}: Gemeinden mit größter Abweichung ({scope})",
        f"Basis: {basis}.",
        "",
        md_table(top),
        "## Ausgeschlossene Bezirkspaare (abweichender Zuschnitt)",
        "Gleiche Bezirksnummer, aber unterschiedliche Grundgesamtheit – meist ein Briefwahlbezirk "
        "der Gemeindewahl gegen den Briefwahlbezirk einer einzelnen Ortschaft. Der TVD misst dort "
        "den Zuschnitt, nicht das Wahlverhalten.",
        "",
        md_table(bd[~bd.belastbar].sort_values("tvd_pp", ascending=False).head(top_n)[
            ["ags", "ortsrat", "gebiet-nr", "B_gr", "B_or", "waehler_ratio", "tvd_pp",
             "gr_bezirk_mehrfach"]]),
        f"## Landkreisebene ({scope})",
        md_table(kreis_s),
        "## Umverteilung verwaister Stimmen",
        "Anteil der verwaisten GR-Stimmen, den eine Liste *über* proportionale Umverteilung hinaus "
        "gewinnt (negativ = unterproportional). Aggregatdaten – kein Nachweis individuellen Splittings.",
        "",
        md_table(absorb.head(40)) if not absorb.empty else "_keine Bezirke mit fehlenden Listen_\n",
        "## Struktur der Briefwahlbezirke (Gemeindewahl)",
        md_table(post),
        f"## Bezirke mit größter Abweichung ({scope})",
        md_table(sel.sort_values("tvd_pp", ascending=False).head(top_n)[
            ["ags", "ortsrat", "gebiet-nr", "gebiet-name", "sum_or", "tvd_pp", "orphan_pp",
             "fehlende_listen", "groesste_verschiebung", "kongruent"]]),
        "## Methodik",
        "- **Δ %p** = Anteil Ortsratswahl − Anteil Gemeindewahl je Liste, bezogen auf gültige *Stimmen* "
        "(bis zu 3 je Wähler).",
        "- **TVD** (Dissimilaritätsindex) = ½ · Σ|Δ %p|: Anteil der Stimmen, die man verschieben müsste, "
        "um beide Verteilungen gleich zu machen.",
        "- **Verwaist** = GR-Stimmanteil von Listen, die auf dem Ortsrats-Stimmzettel fehlen. "
        "Erwartung bei proportionaler Umverteilung: Anteil_GR / (1 − verwaist).",
        "- Paarung über identische Bezirksnummer innerhalb der AGS; `kongruent` prüft gleiche "
        "Wählerzahl, `gr_bezirk_mehrfach` erkennt einen Gemeindewahl-Bezirk, der mehrere Ortsräte "
        "abdeckt. Nur `belastbar` = beides sauber geht in Rangliste und Aggregate ein.",
        "",
        "## Protokoll",
        *([f"- {x}" for x in log] or ["- keine Auffälligkeiten"]),
    ]
    (out / "report.md").write_text("\n".join(lines), encoding="utf-8")


def cmd_analyze(a):
    out = Path(a.out)
    out.mkdir(parents=True, exist_ok=True)
    frames = load_all(Path(a.dir))
    pm = load_parties(a.parties, Path(a.dir))
    L, log = build_pairs(frames, pm)
    Ld, D = district_table(L)
    key = ["ags", "ortsrat", "gebiet-nr"]
    Ld = Ld.merge(D[key + ["belastbar", "kongruent", "gr_bezirk_mehrfach"]], on=key, how="left")
    lok_files = list(Path(a.dir).rglob("*wahllokal*.csv"))
    lokale = pd.concat([pd.read_csv(p, sep=";", dtype=str) for p in lok_files]) if lok_files else None

    # Auswertungsebene: Briefwahlbezirke, sofern es sie gibt. Viele NDS-Gemeinden rechnen
    # die Briefwaehler stattdessen in die Urnenbezirke ein (erkennbar an B1) - dann waere
    # jede Briefwahl-Auswertung leer, und es wird ueber alle Bezirke gerechnet.
    n_brief = int(D.is_brief.sum())
    if a.scope == "brief" or (a.scope == "auto" and n_brief):
        lmask, dmask = Ld.is_brief, D.is_brief
        scope = "Briefwahlbezirke"
    else:
        lmask = pd.Series(True, index=Ld.index)
        dmask = pd.Series(True, index=D.index)
        scope = ("alle Wahlbezirke" if n_brief else
                 "alle Wahlbezirke (keine separaten Briefwahlbezirke vorhanden)")
    if not n_brief:
        log.append("Keine separaten Briefwahlbezirke gefunden - Briefwaehler sind vermutlich "
                   "den Urnenbezirken zugeordnet (Spalte B1). Auswertung ueber alle Bezirke.")
    print(f"Auswertungsebene: {scope} ({int(dmask.sum())} von {len(D)} Bezirkspaaren)")

    # Nicht belastbare Paare (unterschiedlicher Zuschnitt) wuerden die Rangliste anfuehren,
    # ohne dass dem ein inhaltlicher Unterschied entspraeche.
    if a.strict:
        lsel, dsel = lmask & Ld.belastbar.fillna(False), dmask & D.belastbar
        basis = "nur zuschnittsgleiche Paare"
    else:
        lsel, dsel = lmask, dmask
        basis = "alle Paare, auch mit abweichendem Zuschnitt"
    n_aus = int((dmask & ~D.belastbar).sum())
    if n_aus:
        log.append(f"{n_aus} Paare der Auswertungsebene sind nicht zuschnittsgleich "
                   f"({int((dmask & ~D.kongruent).sum())} abweichende Waehlerzahl, "
                   f"{int((dmask & D.gr_bezirk_mehrfach).sum())} Gemeindewahl-Bezirk fuer mehrere "
                   f"Ortsraete) - " + ("ausgeschlossen (--no-strict hebt das auf)" if a.strict
                                       else "enthalten (--strict schliesst sie aus)"))
    print(f"Ranking-Basis: {basis} ({int(dsel.sum())} Paare, {n_aus} ausgeschlossen)")

    _, muni_b = aggregate(Ld[lsel], ["kreis", "ags"])
    _, muni_all = aggregate(Ld, ["kreis", "ags", "is_brief"])
    kreis_long, kreis_s = aggregate(Ld[lsel], ["kreis"])
    wm = (D[dsel].assign(w=lambda x: x.tvd_pp * x.sum_or)
          .groupby("ags").agg(w=("w", "sum"), s=("sum_or", "sum")))
    if not muni_b.empty:
        muni_b["tvd_bezirke_gewichtet_pp"] = muni_b.ags.map(wm.w / wm.s)
        muni_b["n_bezirke"] = muni_b.ags.map(D[dsel].groupby("ags").size()).fillna(0).astype(int)
        top = (muni_b[muni_b.stimmen_or >= a.min_votes]
               .sort_values("tvd_bezirke_gewichtet_pp", ascending=False).head(a.top))
    else:
        top = muni_b.copy()
    absorb = absorption(Ld[lsel] if a.strict else Ld, a.min_votes)
    post = postal_structure(frames, lokale)

    Ld.to_csv(out / "bezirk_partei_long.csv", index=False)
    D.to_csv(out / "bezirk_summary.csv", index=False)
    muni_all.to_csv(out / "gemeinde_summary.csv", index=False)
    kreis_long.to_csv(out / "kreis_partei_briefwahl.csv", index=False)
    kreis_s.to_csv(out / "kreis_summary_briefwahl.csv", index=False)
    top.to_csv(out / f"top{a.top}_gemeinden_briefwahl.csv", index=False)
    absorb.to_csv(out / "umverteilung.csv", index=False)
    post.to_csv(out / "briefwahl_struktur.csv", index=False)
    write_report(out, D, muni_b, muni_all, kreis_s, top, absorb, post, log, a.top, a.demo_flag,
                 dmask, scope, dsel, basis)
    print(f"Fertig: {len(D)} Bezirkspaare -> {out}/report.md")
    for x in log:
        print("  !", x)


# --------------------------------------------------------------------------- Hilfsbefehle
def cmd_inspect(a):
    df = read_vm_csv(Path(a.file))
    print("Wahl:", df.wahl.unique(), "| Typ:", classify_election(df.wahl.iloc[0]))
    print("Spalten:", list(df.columns))
    print("Listen-Spalten:", party_cols(df))
    print(df[["gebiet-nr", "gebiet-name", "B"]].assign(
        brief=[is_brief(n, m) for n, m in zip(df["gebiet-nr"], df["gebiet-name"])]).to_string())


def cmd_parties(a):
    rows = []
    for f in load_all(Path(a.dir)):
        if f.etype.iloc[0] in ("GR", "OR"):
            for c in party_cols(f):
                rows.append({"ags": f.ags.iloc[0], "wahl": f.wahl.iloc[0], "code": c, "name": ""})
    tmpl = pd.DataFrame(rows)
    known = load_parties(None, Path(a.dir))
    if not known.empty:
        tmpl = tmpl.drop(columns="name").merge(known, on=["ags", "wahl", "code"], how="left")
    tmpl.to_csv(a.out, index=False)
    print(f"{len(tmpl)} Zeilen -> {a.out}; leere 'name'-Felder bitte aus der Ergebnisseite ergänzen.")


def cmd_demo(a):
    """Synthetische Daten im votemanager-Format + Lauf von analyze."""
    rng = np.random.default_rng(7)
    root = Path(a.dir)
    root.mkdir(parents=True, exist_ok=True)
    head = ["datum", "wahl", "ags", "gebiet-nr", "gebiet-name", "max-schnellmeldungen",
            "anz-schnellmeldungen", "A1", "A2", "A3", "A", "B", "B1", "C", "D"]
    gr_lists = ["CDU", "SPD", "GRÜNE", "FDP", "AfD Niedersachsen", "UWG", "Einzelbewerber Müller"]
    base_share = np.array([.36, .26, .13, .06, .10, .06, .03])
    pmap = []
    munis = {"03459901": ("Demo-Stadt A", [["CDU", "SPD", "GRÜNE", "UWG"], ["CDU", "SPD"]]),
             "03459902": ("Demo-Gemeinde B", [["CDU", "SPD", "GRÜNE", "FDP", "AfD Niedersachsen"]]),
             "03459903": ("Demo-Stadt C", [["CDU", "SPD", "GRÜNE", "FDP", "UWG"],
                                            ["SPD", "GRÜNE", "Bürgerliste Dorf"]])}
    for ags, (gname, or_ballots) in munis.items():
        gr_rows, or_rows = [], {i: [] for i in range(len(or_ballots))}
        for oi, ballot in enumerate(or_ballots):
            for k in range(3):
                for brief in (False, True):
                    nr = f"{(oi + 1) * (900 if brief else 100) + k + 1:04d}" if not brief else f"0{990 + oi * 3 + k}"
                    name = f"{'Briefwahl ' if brief else ''}{gname} Bezirk {nr}"
                    B = int(rng.integers(250, 700))
                    shares = rng.dirichlet(base_share * (60 if brief else 40))
                    stimmen_gr = rng.multinomial(int(B * 2.8), shares)
                    # Ortsrat: nur Listen des Stimmzettels, verwaiste Stimmen ueberproportional an CDU/lokal
                    idx = [gr_lists.index(p) for p in ballot if p in gr_lists]
                    w = np.zeros(len(gr_lists))
                    w[idx] = shares[idx]
                    orphan = shares.sum() - w.sum()
                    loc_extra = []
                    if "CDU" in ballot:
                        w[0] += orphan * 0.5
                    if "Bürgerliste Dorf" in ballot:
                        loc_extra = [orphan * 0.6 + 0.08]
                    vec = np.concatenate([w[idx], loc_extra])
                    vec = vec / vec.sum()
                    B_or = B - int(rng.integers(0, 4))
                    st_or = rng.multinomial(int(B_or * 2.6), vec)
                    common = ["13.09.2026", None, ags, nr, name, 1, 1, "", "", "", "" if brief else B + 200,
                              B, B if brief else int(B * .1), int(B * .015), B]
                    gr_rows.append([*common[:1], f"Gemeindewahl {gname}", *common[2:], *stimmen_gr])
                    or_rows[oi].append([*common[:1], f"Ortsratswahl {gname} {oi + 1}", *common[2:11],
                                        B_or, *common[12:13], int(B_or * .03), B_or, *st_or])
        d = root / ags
        d.mkdir(exist_ok=True)
        cols = head + [f"D{i + 1}" for i in range(len(gr_lists))]
        pd.DataFrame(gr_rows, columns=cols).to_csv(
            d / f"Open-Data-{ags}-Gemeindewahl-Wahlbezirk.csv", sep=";", index=False)
        pmap += [{"ags": ags, "wahl": f"Gemeindewahl {gname}", "code": f"D{i + 1}", "name": n}
                 for i, n in enumerate(gr_lists)]
        for oi, ballot in enumerate(or_ballots):
            cols = head + [f"D{i + 1}" for i in range(len(ballot))]
            pd.DataFrame(or_rows[oi], columns=cols).to_csv(
                d / f"Open-Data-{ags}-Ortsratswahl-{oi + 1}-Wahlbezirk.csv", sep=";", index=False)
            pmap += [{"ags": ags, "wahl": f"Ortsratswahl {gname} {oi + 1}", "code": f"D{i + 1}", "name": n}
                     for i, n in enumerate(ballot)]
    pd.DataFrame(pmap).to_csv(root / "parties_demo.csv", index=False)
    a.parties = str(root / "parties_demo.csv")
    a.demo_flag = True
    cmd_analyze(a)


# --------------------------------------------------------------------------- CLI
# --------------------------------------------------------------------------- Abgleich der Datenstaende
# Jeder Abruf der offenen Daten ist ein eingefrorener Stand <stand-dir>/<JJJJ-MM-TT_HHMM>/<ags>/<datei>
# mit manifest.csv (Bytes, SHA-256, ETag, Last-Modified des Servers, Abrufzeit, HTTP-Status).
# `abgleich` laedt den aktuellen Stand, vergleicht ihn mit dem vorigen und schreibt je geaendertem
# Bezirk eine Zeile: welche Zahlen sich geaendert haben, ob der Bezirk gemeldet war (zettelcheck.csv)
# und ob er mit den neuen Zahlen noch auffaellig ist - mit demselben Kriterium wie `plausi` (Median
# der uebrigen Bezirke derselben Wahl, OUTLIER_TOL). Dazu ein Bericht, fertige Eintraege fuer das
# Ereignis-Log der Webseite und eine Zeile im laufenden protokoll.csv.
#
# Fuer mehrere Laeufe am Tag: Der Abruf schickt ETag und Last-Modified des vorigen Standes mit
# (If-None-Match / If-Modified-Since); unveraenderte Dateien beantwortet der Server mit 304 ohne
# Inhalt und sie werden als Hardlink auf den vorigen Stand abgelegt. Ein Stand ist so vollstaendig
# und kostet nur den Platz der geaenderten Dateien. Staende sind unveraenderlich; <raw> ist ein
# Symlink auf den juengsten, damit `plausi --dir nds/data/raw` weiter den aktuellen Abruf meint.
ABGLEICH_SKIP = re.compile(r"wahllokal|strassen|parties", re.I)
RE_SCHNELLMELDUNG = re.compile(r"schnellmeldung", re.I)
RE_STAND = re.compile(r"\d{4}-\d{2}-\d{2}(_\d{4})?")
MANIFEST_SPALTEN = ["ags", "datei", "bytes", "sha256", "url", "status", "etag", "last_modified", "abgerufen"]
PROTOKOLL_SPALTEN = ["stand", "vergleich_mit", "abgerufen", "quellen", "geladen", "unveraendert_304", "fehler",
                     "dateien_geaendert", "bezirke_geaendert", "gemeldet_geaendert", "geloest",
                     "weiterhin_auffaellig", "neu_auffaellig", "dateien_neu", "dateien_entfernt", "bericht"]
KERNSPALTEN = ["A", "B", "B1", "C1", "C2", "D"]
BEZIRK_SPALTEN = ["ags", "gemeinde", "wahl", "gebiet-nr", "gebiet-name", "bewertung", "gemeldet",
                  "urteil_gemeldet", "fehlend_gemeldet", "stimmen_alt", "stimmen_neu", "delta_stimmen",
                  "zettel_alt", "zettel_neu", "delta_zettel", "waehler_alt", "waehler_neu",
                  "ungueltig_alt", "ungueltig_neu", "quot_alt", "quot_neu", "median_alt", "median_neu",
                  "abw_alt", "abw_neu", "auffaellig_alt", "auffaellig_neu", "listen", "aenderungen",
                  "n_zellen", "datei", "server_stand", "vergleich_stand"]
ZELLEN_SPALTEN = ["ags", "wahl", "gebiet-nr", "gebiet-name", "spalte", "liste", "alt", "neu", "delta", "datei"]
WIEDER = "unverändert wieder veröffentlicht"   # geleert und mit denselben Zahlen wieder eingestellt
BEWERTUNGEN = ["gelöst", "weiterhin auffällig", "Zahlen zurückgezogen", WIEDER, "neu auffällig",
               "war auffällig, nicht gemeldet", "erstmals mit Zahlen", "unauffällig geändert"]
KURZNAMEN = [
    (r"sahra wagenknecht", "BSW"), (r"christlich[ -]?demokratische", "CDU"),
    (r"sozialdemokratische", "SPD"), (r"b(ü|ue)ndnis ?90|gr(ü|ue)ne", "GRÜNE"),
    (r"freie demokratische", "FDP"), (r"alternative f(ü|ue)r deutschland", "AfD"),
    (r"\blinke\b", "Linke"), (r"\bvolt\b", "Volt"), (r"freie w(ä|ae)hler", "FW"),
    # Die PARTEI heisst ausgeschrieben "Partei fuer Arbeit, Rechtsstaat, Tierschutz, Elitenfoerderung ..."
    (r"\bpiraten", "Piraten"), (r"elitenf(ö|oe)rderung|\bdie partei\b", "PARTEI"), (r"tierschutz", "Tierschutz"),
]


def _kurzname(wert: str) -> str:
    for muster, kurz in KURZNAMEN:
        if re.search(muster, wert, re.I):
            return kurz
    w = re.sub(r"^einzelwahlvorschlag\s+", "EWV ", wert, flags=re.I).strip()
    return w if len(w) <= 48 else w[:46].rstrip() + "…"


def _sha256(p: Path) -> str:
    h = hashlib.sha256()
    with open(p, "rb") as f:
        for block in iter(lambda: f.read(1 << 20), b""):
            h.update(block)
    return h.hexdigest()


def _zahl(v):
    """Zelle -> float oder None (leer / nicht vorhanden)."""
    if v is None:
        return None
    try:
        if pd.isna(v):
            return None
    except (TypeError, ValueError):
        pass
    return float(v)


def _fmt(v, nk=0):
    if v is None or (isinstance(v, float) and np.isnan(v)):
        return ""
    return f"{v:.{nk}f}" if nk else f"{v:.0f}"


def _de(v, nk=2):
    return "" if v is None or (isinstance(v, float) and np.isnan(v)) else f"{v:.{nk}f}".replace(".", ",")


def _datum_de(iso: str) -> str:
    return f"{iso[8:10]}.{iso[5:7]}.{iso[0:4]}"


def _stand_de(stand: str) -> str:
    """'2026-09-24_1530' -> '24.09.2026 15:30', '2026-09-20' -> '20.09.2026'."""
    s = _datum_de(stand[:10])
    return f"{s} {stand[11:13]}:{stand[13:15]}" if len(stand) > 10 else s


def _server_stand(last_modified: str) -> str:
    """'Thu, 24 Sep 2026 12:27:49 GMT' -> '24.09.2026 12:27 UTC'."""
    if not last_modified:
        return ""
    try:
        return parsedate_to_datetime(last_modified).strftime("%d.%m.%Y %H:%M UTC")
    except Exception:
        return last_modified


def _server_tag(server_stand: str) -> str:
    """'01.10.2026 22:30 UTC' -> Kalendertag in Niedersachsen als JJJJ-MM-TT ('2026-10-02').
    Der UTC-Tag allein waere zwischen 0 und 2 Uhr MESZ (1 Uhr MEZ) der Vortag."""
    try:
        t = datetime.strptime(server_stand, "%d.%m.%Y %H:%M UTC").replace(tzinfo=timezone.utc)
        return t.astimezone(ZoneInfo("Europe/Berlin")).strftime("%Y-%m-%d")
    except ValueError:
        return server_stand[6:10] + "-" + server_stand[3:5] + "-" + server_stand[0:2]


def _abgleich_quellen(sources_csv: str) -> pd.DataFrame:
    """Dieselbe Auswahl wie `download --alle`: unterste Ebene aller Wahlen plus Metadateien."""
    src = _quellen_lesen(sources_csv)
    lvl_ok = src.level.str.contains("bezirk", case=False) | (src.level == "")
    return src[(src.type.isin(["GR", "OR", "X"]) & lvl_ok) | (src.type == "META")
               | src.level.str.contains("ruppier", case=False)].copy()


def _hardlink(quelle: Path, ziel: Path) -> bool:
    """ziel durch einen Hardlink auf quelle ersetzen (gleiche Datei, kein zweiter Platz)."""
    try:
        if ziel.exists():
            ziel.unlink()
        os.link(quelle, ziel)
        return True
    except OSError:
        return False


def _stand_laden(keep: pd.DataFrame, ziel: Path, jobs: int, vorher: Path | None,
                 M_vor: pd.DataFrame | None) -> list[dict]:
    """Alle Quellen nach ziel/<ags>/<datei> laden. Mit vorigem Stand: bedingte Abrufe (ETag /
    Last-Modified); 304 und inhaltsgleiche Dateien werden Hardlinks auf den vorigen Stand."""
    zeilen, sperre = [], threading.Lock()
    zaehler = {"ok": 0, "304": 0, "err": 0, "link": 0}
    vor = {}
    if M_vor is not None:
        vor = {(r.ags, r.datei): r for _, r in M_vor.iterrows() if r.status in ("200", "304")}

    def hole(r):
        target = ziel / r.ags / r.datei
        target.parent.mkdir(parents=True, exist_ok=True)
        v = vor.get((r.ags, r.datei))
        if v is not None and v.url and v.url != r.url:
            v = None   # Datei kommt jetzt von einer anderen URL (umbenannte Wahl): ETag/Datum gelten nicht
        v_datei = vorher / r.ags / r.datei if (v is not None and vorher) else None
        if v_datei is not None and not v_datei.exists():
            v, v_datei = None, None
        z = {"ags": r.ags, "datei": r.datei, "url": r.url, "bytes": "", "sha256": "", "status": "",
             "etag": "", "last_modified": "",
             "abgerufen": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")}
        kopf = {}
        if v is not None:
            if v.etag:
                kopf["If-None-Match"] = v.etag
            if v.last_modified:
                kopf["If-Modified-Since"] = v.last_modified
        try:
            resp = _get(r.url, headers=kopf)
            z["status"] = str(resp.status_code)
            if resp.status_code == 304 and v is not None:
                z.update(bytes=v.bytes, sha256=v.sha256, etag=v.etag, last_modified=v.last_modified)
                if not _hardlink(v_datei, target):
                    target.write_bytes(v_datei.read_bytes())
            elif resp.status_code == 200:
                target.write_bytes(resp.content)
                z["bytes"] = str(len(resp.content))
                z["sha256"] = _sha256(target)
                z["etag"] = resp.headers.get("ETag", "")
                z["last_modified"] = resp.headers.get("Last-Modified", "")
                if v is not None and v.sha256 == z["sha256"] and _hardlink(v_datei, target):
                    with sperre:
                        zaehler["link"] += 1
        except Exception as e:
            z["status"] = f"Fehler: {e}"
        with sperre:
            zeilen.append(z)
            if z["status"] == "200":
                zaehler["ok"] += 1
                if zaehler["ok"] % 500 == 0:
                    print(f"  ... {zaehler['ok']} geladen", flush=True)
            elif z["status"] == "304":
                zaehler["304"] += 1
            else:
                zaehler["err"] += 1
                print("ERR", z["status"], r.url, flush=True)

    _parallel(hole, [r for _, r in keep.iterrows()], jobs)
    print(f"geladen: {zaehler['ok']} (davon inhaltsgleich mit dem vorigen Stand: {zaehler['link']}) · "
          f"unveraendert laut Server (304): {zaehler['304']} · Fehler: {zaehler['err']}", flush=True)
    return zeilen


def _manifest(stand: Path, keep: pd.DataFrame, zeilen: list[dict] | None = None) -> pd.DataFrame:
    """manifest.csv eines Standes: lesen, aus Abrufzeilen schreiben oder (aeltere Staende ohne
    Abrufprotokoll) aus den Dateien nachbauen - dann ohne ETag und Server-Zeitstempel."""
    mf = stand / "manifest.csv"
    if zeilen is None and mf.exists():
        M = pd.read_csv(mf, dtype=str).fillna("")
        for c in MANIFEST_SPALTEN:
            if c not in M.columns:
                M[c] = ""
        return M[MANIFEST_SPALTEN]
    if zeilen is None:
        urls = {(r.ags, r.datei): r.url for _, r in keep.iterrows()}
        zeilen = []
        for p in sorted(stand.rglob("*")):
            if not p.is_file() or p.name == "manifest.csv":
                continue
            st = p.stat()
            zeilen.append({"ags": p.parent.name, "datei": p.name, "bytes": str(st.st_size),
                           "sha256": _sha256(p), "url": urls.get((p.parent.name, p.name), ""),
                           "status": "200", "etag": "", "last_modified": "",
                           "abgerufen": datetime.fromtimestamp(st.st_mtime, timezone.utc)
                           .strftime("%Y-%m-%dT%H:%M:%SZ")})
        print(f"manifest.csv fuer {stand} nachgebaut ({len(zeilen)} Dateien, ohne Server-Zeitstempel)")
    M = pd.DataFrame(zeilen, columns=MANIFEST_SPALTEN).astype(str).sort_values(["ags", "datei"])
    M.to_csv(mf, index=False)
    return M


def _bezirke_lesen(p: Path, df: pd.DataFrame | None = None):
    df = read_vm_csv(p) if df is None else df
    zeilen = {}
    for _, r in df.iterrows():
        nr = str(r["gebiet-nr"]).strip()
        nr = "" if nr.lower() == "nan" else nr
        zeilen[(nr, str(r["gebiet-name"]).strip())] = r
    return df, zeilen


def _loo_median(zeilen: dict):
    """Stimmen je gueltigem Zettel und Median der *uebrigen* Bezirke - wie in plausi."""
    q = {}
    for k, r in zeilen.items():
        c2, d = _zahl(r.get("C2")), _zahl(r.get("D"))
        if k[0] and c2 and c2 > 0 and d is not None:
            q[k] = d / c2
    med = {}
    for k in q:
        rest = [v for kk, v in q.items() if kk != k]
        med[k] = float(np.median(rest)) if rest else np.nan
    return q, med


def _parteien(od: Path, wahl: str, datei: str = "") -> dict:
    """Listencode -> Kurzname aus open_data.json (dateifelder[].parteien). Weicht der Wahlname dort
    vom CSV-Namen ab (Server benennt um), ueber den Dateinamen in csvs[] zuordnen."""
    try:
        d = json.loads(od.read_text(encoding="utf-8"))
        alias = ({wahl} | {c.get("wahl") for c in d.get("csvs", []) if datei and c.get("url") == datei}
                 | {n for n, k in _wahlnamen(od.parent).items() if k == wahl})
        eintraege = [e for e in d.get("dateifelder", []) if e.get("name") in alias]
        if not eintraege:   # alte Datei, Wahl inzwischen kuerzer benannt: eindeutiger Praefix ("Kreiswahl")
            eintraege = [e for e in d.get("dateifelder", [])
                         if e.get("name") and wahl.lower().startswith(e["name"].lower())]
            eintraege = eintraege if len(eintraege) == 1 else []
        if eintraege:
            return {p["feld"]: _kurzname(p["wert"]) for p in eintraege[0].get("parteien", [])}
    except Exception:
        pass
    return {}


def _dateiliste(od: Path) -> set:
    try:
        return {c.get("url", "") for c in json.loads(od.read_text(encoding="utf-8")).get("csvs", [])}
    except Exception:
        return set()


def _stand_vergleichen(alt: Path, neu: Path, gemeldet: dict, namen: dict, manifest_neu: pd.DataFrame):
    zellen, bezirke = [], []
    vorstaende = _vorstaende(alt)
    info = {"verglichen": 0, "geaendert": 0, "neu": [], "entfernt": [], "neu_laut_liste": []}
    lm = {(r.ags, r.datei): r.last_modified for _, r in manifest_neu.iterrows()}

    def csvs(stand):
        return {p.relative_to(stand): p for p in stand.rglob("*.csv")
                if not ABGLEICH_SKIP.search(p.name) and p.name != "manifest.csv"}
    ca, cn = csvs(alt), csvs(neu)
    info["neu"] = sorted(str(k) for k in cn.keys() - ca.keys())
    info["entfernt"] = sorted(str(k) for k in ca.keys() - cn.keys())
    for od in sorted(neu.glob("*/open_data.json")):
        alt_od = alt / od.parent.name / "open_data.json"
        if alt_od.exists():
            for f in sorted(_dateiliste(od) - _dateiliste(alt_od)):
                info["neu_laut_liste"].append(f"{od.parent.name}/{f}")

    for rel in sorted(cn.keys() & ca.keys()):
        info["verglichen"] += 1
        pa, pn = ca[rel], cn[rel]
        if pa.samefile(pn) or pa.read_bytes() == pn.read_bytes():
            continue
        try:
            # Im alten Stand geleerte Bezirke (Neueingabe) mit den zuletzt veroeffentlichten Zahlen fuellen:
            # Verglichen wird mit dem, was die Gemeinde zuletzt veroeffentlicht hat, nicht mit leeren Zellen.
            dfa, aus_a = _leere_auffuellen(read_vm_csv(pa), rel, vorstaende)
            dfa, za = _bezirke_lesen(pa, dfa)
            dfn, zn = _bezirke_lesen(pn)
        except Exception as e:
            print("Uebersprungen:", rel, e)
            continue
        aufgefuellt = {(str(dfa.at[i, "gebiet-nr"]).strip(), str(dfa.at[i, "gebiet-name"]).strip()): v
                       for i, v in aus_a.items() if v}
        ags = rel.parts[0]
        wahl = str(dfn.wahl.iloc[0]) if len(dfn) else (str(dfa.wahl.iloc[0]) if len(dfa) else "")
        # IVU: Listennamen je Bezirk (Nummern je Wahlbereich), und eine Datei kann mehrere Wahlen
        # enthalten (Raete der Mitgliedsgemeinden) - Wahl und Listennamen dann je Zeile
        pj = dfn.attrs.get("parteien_je_bezirk")
        parteien_datei = _parteien(neu / ags / "open_data.json", wahl, rel.name)
        qa, ma = _loo_median(za)
        qn, mn = _loo_median(zn)
        spalten = [c for c in dict.fromkeys(list(dfa.columns) + list(dfn.columns))
                   if c not in META_COLS and not RE_SCHNELLMELDUNG.search(c)]
        datei_geaendert = False
        for key in sorted(za.keys() | zn.keys()):
            nr, name = key
            ra, rn = za.get(key), zn.get(key)
            vgl = aufgefuellt.get(key, "")
            if vgl and rn is not None and not (_zahl(rn.get("D")) or 0):
                continue   # weiter leer: schon als "Zahlen zurueckgezogen" gemeldet, als der Bezirk geleert wurde
            wahl = str((rn if rn is not None else ra).get("wahl", wahl))
            parteien = pj.get(nr, {}) if pj is not None else parteien_datei
            basis = {"ags": ags, "wahl": wahl, "gebiet-nr": nr, "gebiet-name": name, "datei": str(rel)}
            if ra is None or rn is None:
                zellen.append({**basis, "spalte": "(Bezirk neu)" if ra is None else "(Bezirk entfernt)",
                               "liste": "", "alt": "", "neu": "", "delta": ""})
                datei_geaendert = True
                continue
            diffs = []
            for c in spalten:
                va, vn = _zahl(ra.get(c)), _zahl(rn.get(c))
                if va == vn or (va is None and vn is None):
                    continue
                diffs.append((c, va, vn))
            if not diffs:
                gm = gemeldet.get((ags, wahl, nr))
                if vgl and nr and gm is not None and (_zahl(rn.get("D")) or 0):
                    # geleert und mit denselben Zahlen wieder eingestellt: keine Aenderung, aber der Abschluss
                    # der Meldung "Zahlen auf dem Server geleert"
                    g_ = lambda r, c: _fmt(_zahl(r.get(c)))  # noqa: E731
                    bezirke.append({**dict.fromkeys(BEZIRK_SPALTEN, ""),
                        "ags": ags, "gemeinde": namen.get(ags, ""), "wahl": wahl, "gebiet-nr": nr,
                        "gebiet-name": name, "bewertung": WIEDER, "gemeldet": True,
                        "urteil_gemeldet": gm["urteil"], "fehlend_gemeldet": gm["fehlend_ggue_median"],
                        "stimmen_alt": g_(ra, "D"), "stimmen_neu": g_(rn, "D"), "delta_stimmen": "0",
                        "zettel_alt": g_(ra, "C2"), "zettel_neu": g_(rn, "C2"), "delta_zettel": "0",
                        "listen": "", "aenderungen": "", "n_zellen": 0, "datei": str(rel),
                        "server_stand": _server_stand(lm.get((ags, rel.name), "")), "vergleich_stand": vgl})
                continue
            datei_geaendert = True
            if not nr:
                # Summenzeile (Stadtbezirk, Gruppierung): nur in die Zellenliste, kein Bezirk
                for c, va, vn in diffs:
                    zellen.append({**basis, "spalte": c, "liste": "", "alt": _fmt(va), "neu": _fmt(vn),
                                   "delta": _fmt((vn or 0) - (va or 0))})
                continue
            listen, kern, sonst = [], [], 0
            for c, va, vn in diffs:
                m = PARTY_COL_SUM.match(c) or LIST_COL.match(c) or CAND_SUM_COL.match(c)
                liste = parteien.get(f"D{m.group(1)}", f"D{m.group(1)}") if m else ""
                delta = (vn or 0) - (va or 0)
                zellen.append({**basis, "spalte": c, "liste": liste, "alt": _fmt(va), "neu": _fmt(vn),
                               "delta": _fmt(delta)})
                if c in KERNSPALTEN:
                    kern.append(f"{c} {_fmt(va)}→{_fmt(vn)}")
                elif PARTY_COL_SUM.match(c):
                    listen.append(f"{liste} {delta:+.0f}")
                else:
                    sonst += 1
            g = lambda r, c: _zahl(r.get(c))  # noqa: E731
            qa_, qn_ = qa.get(key, np.nan), qn.get(key, np.nan)
            ma_, mn_ = ma.get(key, np.nan), mn.get(key, np.nan)
            abw_a, abw_n = qa_ - ma_, qn_ - mn_
            auff_a = bool(abw_a < -OUTLIER_TOL) if not np.isnan(abw_a) else False
            auff_n = bool(abw_n < -OUTLIER_TOL) if not np.isnan(abw_n) else False
            gm = gemeldet.get((ags, wahl, nr))
            erstmals = not (g(ra, "D") or 0) and bool(g(rn, "D") or 0)   # am alten Stand noch ohne Zahlen
            # Gemeinde hat den Bezirk auf dem Server geleert (anz-schnellmeldungen 0, z. B. Neueingabe des
            # endgueltigen Ergebnisses): nichts geloest, der Bezirk bleibt offen bis wieder Zahlen da sind.
            zurueck = bool(g(ra, "D") or 0) and not (g(rn, "D") or 0)
            if gm is not None and zurueck:
                bewertung = "Zahlen zurückgezogen"
            elif gm is not None:
                bewertung = "weiterhin auffällig" if auff_n else "gelöst"
            elif auff_n:
                bewertung = "neu auffällig"
            elif auff_a:
                bewertung = "war auffällig, nicht gemeldet"
            elif erstmals:
                bewertung = "erstmals mit Zahlen"
            else:
                bewertung = "unauffällig geändert"
            aend = kern + listen + ([f"{sonst} weitere Teilspalten"] if sonst else [])
            bezirke.append({
                "ags": ags, "gemeinde": namen.get(ags, ""), "wahl": wahl, "gebiet-nr": nr, "gebiet-name": name,
                "bewertung": bewertung, "gemeldet": gm is not None,
                "urteil_gemeldet": gm["urteil"] if gm is not None else "",
                "fehlend_gemeldet": gm["fehlend_ggue_median"] if gm is not None else "",
                "stimmen_alt": _fmt(g(ra, "D")), "stimmen_neu": _fmt(g(rn, "D")),
                "delta_stimmen": _fmt((g(rn, "D") or 0) - (g(ra, "D") or 0)),
                "zettel_alt": _fmt(g(ra, "C2")), "zettel_neu": _fmt(g(rn, "C2")),
                "delta_zettel": _fmt((g(rn, "C2") or 0) - (g(ra, "C2") or 0)),
                "waehler_alt": _fmt(g(ra, "B")), "waehler_neu": _fmt(g(rn, "B")),
                "ungueltig_alt": _fmt(g(ra, "C1")), "ungueltig_neu": _fmt(g(rn, "C1")),
                "quot_alt": _fmt(qa_, 3), "quot_neu": _fmt(qn_, 3),
                "median_alt": _fmt(ma_, 3), "median_neu": _fmt(mn_, 3),
                "abw_alt": _fmt(abw_a, 3), "abw_neu": _fmt(abw_n, 3),
                "auffaellig_alt": auff_a, "auffaellig_neu": auff_n,
                "listen": " · ".join(listen), "aenderungen": " · ".join(aend), "n_zellen": len(diffs),
                "datei": str(rel), "server_stand": _server_stand(lm.get((ags, rel.name), "")),
                "vergleich_stand": vgl,
            })
        if datei_geaendert:
            info["geaendert"] += 1
    return (pd.DataFrame(bezirke, columns=BEZIRK_SPALTEN),
            pd.DataFrame(zellen, columns=ZELLEN_SPALTEN), info)


def _ruecknahmen(out: Path, neu_s: str, B: pd.DataFrame) -> list[dict]:
    """Bezirke, die ein frueherer Abgleich als "gelöst" gefuehrt hat und die jetzt wieder auffaellig oder geleert
    sind. Die Webseite fuehrt sie dann noch als berichtigt; ob das zurueckgenommen wird, entscheidet Felix.
    Grundlage sind die Tabellen der frueheren Laeufe in out (<stand>.csv), juengste Loesung je Bezirk."""
    if not len(B):
        return []
    geloest = {}
    for f in sorted(out.glob("*.csv")):
        if not RE_STAND.fullmatch(f.stem) or f.stem >= neu_s:
            continue
        T = pd.read_csv(f, dtype=str).fillna("")
        for _, r in T[T.bewertung == "gelöst"].iterrows():
            geloest[(r.ags, r.wahl, r["gebiet-nr"])] = {"stand": f.stem, "stimmen_alt": r.stimmen_alt,
                                                         "stimmen_neu": r.stimmen_neu, "server_stand": r.server_stand}
    erg = []
    for _, r in B.iterrows():
        g = geloest.get((r.ags, r.wahl, r["gebiet-nr"]))
        if g is None:
            continue
        if not (str(r.auffaellig_neu) == "True" or r.bewertung == "Zahlen zurückgezogen"):
            continue
        erg.append({"ags": r.ags, "gemeinde": r.gemeinde, "wahl": r.wahl, "gebiet-nr": r["gebiet-nr"],
                    "gebiet-name": r["gebiet-name"], "bewertung": r.bewertung, **{f"geloest_{k}": v for k, v in g.items()},
                    "stimmen_jetzt": str(r.stimmen_neu), "server_stand": r.server_stand,
                    "wie_vorher": str(r.stimmen_neu) == g["stimmen_alt"] and g["stimmen_alt"] != ""})
    return erg


def _ruecknahme_text(x: dict) -> str:
    s = (f"{x['ags']} {x['gemeinde']}, {x['wahl']} {x['gebiet-name']}: am {_stand_de(x['geloest_stand'])} als gelöst "
         f"eingetragen ({x['geloest_stimmen_alt']} → {x['geloest_stimmen_neu']} Stimmen), jetzt {x['bewertung']}")
    if x["stimmen_jetzt"]:
        s += f" mit {x['stimmen_jetzt']} Stimmen"
    if x["wie_vorher"]:
        s += ", also wieder die Zahl von vor der Berichtigung"
    if x["server_stand"]:
        s += f" (Server-Stand {x['server_stand']})"
    return s


def _abgleich_bericht(pfad: Path, alt_s: str, neu_s: str, B: pd.DataFrame, info: dict,
                      abgerufen: str, n_quellen: int, fehler: int):
    def zeile(r):
        st = f"{r.stimmen_alt} → {r.stimmen_neu}"
        if r.delta_stimmen not in ("", "0"):
            st += f" ({float(r.delta_stimmen):+.0f})"
        zt = f"{r.zettel_alt} → {r.zettel_neu}"
        q = f"{_de(float(r.quot_alt)) if r.quot_alt else '–'} → {_de(float(r.quot_neu)) if r.quot_neu else '–'}"
        if r.median_neu:
            q += f" (Median {_de(float(r.median_neu))})"
        return (f"| {r.gemeinde or r.ags} | {r.wahl} | {r['gebiet-nr']} {r['gebiet-name']} | {st} | {zt} | "
                f"{r.fehlend_gemeldet or '–'} | {q} | {r.listen or '–'} | {r.bewertung} | {r.server_stand or '–'} |")

    kopf = ("| Gemeinde | Wahl | Bezirk | Stimmen | gültige Zettel | erwartet fehlend | Stimmen je Zettel | "
            "Listen | Bewertung | Server-Stand |\n|---|---|---|---|---|---|---|---|---|---|")
    G, W = B[B.gemeldet == True], B[B.gemeldet != True]  # noqa: E712
    n_gem = B[B.bewertung != WIEDER].ags.nunique() if len(B) else 0
    t = [f"# Datenabgleich {_stand_de(neu_s)} gegenüber {_stand_de(alt_s)}", "",
         f"Abruf der offenen Daten aller {n_quellen} Quellen aus `nds/sources.csv` am {abgerufen}"
         f"{f' ({fehler} Abrufe fehlgeschlagen, siehe manifest.csv)' if fehler else ''}. "
         f"Verglichen werden die eingefrorenen Stände `nds/data/stand/{alt_s}` und `nds/data/stand/{neu_s}`; "
         "jeder Stand trägt ein `manifest.csv` mit Größe, SHA-256, Abrufzeit und dem Zeitstempel, "
         "den der Server für die Datei meldet (Last-Modified).", "",
         "| | |", "|---|---|",
         f"| Ergebnisdateien verglichen | {info['verglichen']} |",
         f"| davon mit geänderten Zahlen | {info['geaendert']} in {n_gem} Gemeinden |",
         f"| Bezirke mit geänderten Zahlen | {(B.bewertung != WIEDER).sum()} |",
         f"| davon gemeldete Bezirke | {(G.bewertung != WIEDER).sum()} |",
         f"| … gelöst (nicht mehr auffällig) | {(G.bewertung == 'gelöst').sum()} |",
         f"| … weiterhin auffällig | {(G.bewertung == 'weiterhin auffällig').sum()} |",
         f"| … Zahlen zurückgezogen (Bezirk auf dem Server geleert, bleibt offen) | {(G.bewertung == 'Zahlen zurückgezogen').sum()} |",
         f"| gemeldete Bezirke, nach dem Leeren unverändert wieder veröffentlicht (keine Änderung) | {(G.bewertung == WIEDER).sum()} |",
         f"| nicht gemeldete Bezirke, jetzt auffällig | {(W.bewertung == 'neu auffällig').sum()} |",
         f"| Bezirke, die am alten Stand noch keine Zahlen hatten | {(W.bewertung == 'erstmals mit Zahlen').sum()} |",
         f"| Dateien neu / entfernt | {len(info['neu'])} / {len(info['entfernt'])} |", "",
         "Summenzeilen (Stadtbezirke, Gruppierungen) zählen nicht als Bezirke; ihre Änderungen stehen nur in "
         f"`{pfad.stem}_zellen.csv`.", ""]
    if info.get("ruecknahmen"):
        t += ["## Achtung: früher gelöst, jetzt wieder auffällig", "",
              "Diese Bezirke hat ein früherer Abgleich als gelöst geführt; auf der Webseite stehen sie deshalb als "
              "berichtigt. Der Eintrag „Zahlen geändert, Bezirk bleibt auffällig“ setzt den Status nicht zurück. "
              "Von Hand entscheiden, wie die Webseite damit umgeht.", ""]
        t += [f"- {_ruecknahme_text(x)}" for x in info["ruecknahmen"]] + [""]
    t += ["## Gemeldete Bezirke mit geänderten Zahlen", ""]
    t += [kopf] + [zeile(r) for _, r in G.iterrows()] if len(G) else ["Keine."]
    t += ["", "## Weitere Bezirke mit geänderten Zahlen (nicht gemeldet)", ""]
    t += [kopf] + [zeile(r) for _, r in W.iterrows()] if len(W) else ["Keine."]
    t += ["", "## Neue oder entfernte Dateien", ""]
    if info["neu"] or info["entfernt"]:
        t += [f"- neu: `{f}`" for f in info["neu"]] + [f"- entfernt: `{f}`" for f in info["entfernt"]]
    else:
        t += ["Keine."]
    v = info.get("verzeichnis") or {}
    t += ["", "## Verzeichnis der Dateien (open_data.json)", ""]
    if not v:
        t += ["Nicht abgeglichen (Lauf ohne Download)."]
    else:
        t += ["Geladen wird, was das Verzeichnis jeder votemanager-Instanz beim Abruf nennt. Eine Wahl ist an ihrer "
              "wahl_id erkennbar; benennt der Server sie um, liegt die neue Datei unter dem bisherigen Namen im "
              "Stand, und ihre Zahlen werden mit der alten verglichen (Register: `nds/wahlen_register.csv`).", ""]
        if v.get("doppelt"):
            t += ["**Achtung, mögliche Doppelung** (identische Bezirkszahlen in zwei Dateien einer Kommune, "
                  "die Wahl zählt dann zweimal):", ""]
            t += [f"- {d[0]}: `{d[1]}` und `{d[2]}`" for d in v["doppelt"]] + [""]
        if v.get("umbenannt"):
            t += [f"Umbenannt ({len(v['umbenannt'])} Dateien, geladen von der neuen Adresse):", ""]
            t += [f"- {u['ags']}: „{u['wahl_alt']}“ → „{u['wahl_neu']}“ (wahl_id {u['wahl_id']}, {u['ebene']}): "
                  f"`{u['datei_neu']}`" for u in v["umbenannt"]] + [""]
        if v.get("neu"):
            t += [f"Neu im Verzeichnis und mitgeladen ({len(v['neu'])}):", ""]
            t += [f"- {n['ags']}: {n['wahl']} (`{n['datei']}`)" for n in v["neu"]] + [""]
        ohne_st = [x for x in v.get("verwaist", []) if not RE_STICHWAHL.search(x["wahl"] + x["datei"])]
        if v.get("verwaist"):
            t += [f"Nicht mehr im Verzeichnis, ohne Nachfolger ({len(v['verwaist'])} Dateien, davon "
                  f"{len(v['verwaist']) - len(ohne_st)} Stichwahlen). Sie werden weiter geladen, damit nichts still "
                  "verschwindet; ob ihre Zahlen noch gepflegt werden, ist offen:", ""]
            t += [f"- {x['ags']}: {x['wahl']} (`{x['datei']}`)" for x in ohne_st] + [""]
        if v.get("stichwahl_neu"):
            t += [f"Neue Stichwahl-Dateien im Verzeichnis, nicht geladen: {v['stichwahl_neu']}.", ""]
        if v.get("ohne_verzeichnis"):
            t += [f"Verzeichnis nicht lesbar, Dateien laut `nds/sources.csv` geladen: "
                  f"{', '.join(v['ohne_verzeichnis'])}.", ""]
    t += ["", "## Lesart", "",
          "„Erwartet fehlend“ ist der Fehlbetrag aus dem Zettelcheck (Stimmen, die gegenüber dem Median der "
          "Wahl fehlten). Deckt sich die Änderung der Stimmen damit, hat die Gemeinde genau die Lücke geschlossen.",
          "",
          f"**Gelöst** heißt: Mit den neuen Zahlen liegt der Bezirk nicht mehr als {OUTLIER_TOL:.2f} Stimmen je "
          "gültigem Stimmzettel unter dem Median der übrigen Bezirke derselben Wahl – dasselbe Kriterium, mit dem "
          "er aufgefallen ist. Das ist ein Datenbefund. Ob dahinter eine Berichtigung durch den Wahlausschuss "
          "steht, sagt die Niederschrift, nicht die Datei.",
          "",
          "**Weiterhin auffällig**: Zahlen geändert, Kriterium weiter verletzt. **Neu auffällig**: ein bisher "
          "unauffälliger Bezirk verletzt es jetzt, oft einer, der am alten Stand noch keine Zahlen hatte "
          "(**erstmals mit Zahlen**). Der Server-Stand ist der Zeitpunkt, zu dem die Gemeinde die "
          "Datei zuletzt neu veröffentlicht hat (Angabe des Servers, UTC)."]
    pfad.write_text("\n".join(t) + "\n", encoding="utf-8")


def _yaml_str(s: str) -> str:
    return '"' + str(s).replace("\\", "\\\\").replace('"', '\\"') + '"'


def _abgleich_ereignisse(pfad: Path, alt_s: str, neu_s: str, B: pd.DataFrame, info: dict):
    """Fertige Eintraege fuer daten/ereignisse.yaml der Webseite (zum Pruefen und Anhaengen)."""
    a_de, n_de = _stand_de(alt_s), _stand_de(neu_s)
    quelle = f"Offene Daten der Gemeinden (votemanager), Abruf {n_de}, verglichen mit dem Stand vom {a_de}"
    beleg = f"/daten/abgleich/{neu_s}.csv"
    G = B[B.gemeldet == True]  # noqa: E712
    n_gel = int((G.bewertung == "gelöst").sum())
    n_wtr = int((G.bewertung == "weiterhin auffällig").sum())
    n_wdr = int((G.bewertung == WIEDER).sum())
    Bg, Gg = B[B.bewertung != WIEDER], G[G.bewertung != WIEDER]   # tatsaechlich geaenderte Zahlen
    L = [f"# Vorschläge für daten/ereignisse.yaml aus dem Datenabgleich vom {n_de} (gegen {a_de}).",
         "# Prüfen, gegebenenfalls kürzen, dann an ereignisse.yaml anhängen. Der Beleg zeigt auf die",
         "# Abgleichsdatei, die scripts/daten_importieren.py nach docs/public/daten/abgleich/ kopiert.", ""]
    teile = []
    if len(Bg):
        teile.append(f"{len(Bg)} Bezirke in {Bg.ags.nunique()} Gemeinden mit geänderten Zahlen")
        teile.append(f"davon {len(Gg)} gemeldete Bezirke, {n_gel} gelöst, {n_wtr} weiterhin auffällig")
    else:
        teile.append("keine geänderten Bezirksergebnisse")
    if n_wdr:
        teile.append(f"{n_wdr} gemeldete Bezirke nach dem Leeren unverändert wieder veröffentlicht")
    L += [f"- datum: {neu_s[:10]}", "  typ: daten",
          f"  titel: {_yaml_str(f'Abgleich der offenen Daten vom {n_de} mit dem Stand vom {a_de}: ' + '; '.join(teile))}",
          "  text: >-",
          f"    Alle {info['verglichen']} Ergebnisdateien neu abgerufen und Bezirk für Bezirk mit dem Stand vom {a_de}",
          "    verglichen. Ein gemeldeter Bezirk gilt als gelöst, wenn er mit den neuen Zahlen das Kriterium der",
          "    Auffälligkeit nicht mehr erfüllt. Was hinter einer Änderung steht, sagen die Wahlausschüsse; die",
          "    Datei zeigt nur, dass und wann die Gemeinde neue Zahlen veröffentlicht hat.",
          f"  quelle: {quelle}", f"  beleg: {beleg}", ""]

    def bezirk_name(r):
        return r["gebiet-name"] if r["gebiet-name"].startswith(r["gebiet-nr"]) else f"{r['gebiet-nr']} {r['gebiet-name']}"

    def satz(r):
        """Eine Zeile je Bezirk: Name (Wahl): Stimmen alt → neu (Δ; erwartet) bei Zetteln, je Zettel (Median). Listen."""
        d = f"{float(r.delta_stimmen):+.0f}" if r.delta_stimmen not in ("", "0") else "±0"
        s = f"{bezirk_name(r)} ({r.wahl}): {r.stimmen_alt} → {r.stimmen_neu} Stimmen ({d}"
        if r.fehlend_gemeldet:
            s += f"; erwartet rund {float(r.fehlend_gemeldet):.0f}"
        zt = f"{r.zettel_alt} → {r.zettel_neu}" if r.zettel_alt != r.zettel_neu else r.zettel_neu
        s += f") bei {zt} gültigen Zetteln"
        if r.quot_alt and r.quot_neu:
            s += f", {_de(float(r.quot_alt))} → {_de(float(r.quot_neu))} je Zettel"
            if r.median_neu:
                s += f" (Median {_de(float(r.median_neu))})"
        s += "."
        if r.listen:
            s += f" {r.listen.replace(' · ', ', ')}."
        return s

    def kurz(r):
        d = f" ({float(r.delta_stimmen):+.0f} Stimmen)" if r.delta_stimmen not in ("", "0") else ""
        return f"{r.wahl} {bezirk_name(r)}{d}"

    def server_datum(g):
        """Datum der Neuveroeffentlichung laut Server (juengste Datei der Gruppe), sonst Abrufdatum."""
        tage = sorted(_server_tag(t) for t in g.server_stand if t)
        return tage[-1] if tage else neu_s[:10]

    def eintrag(ags, g, typ, ergebnis, praefix):
        keys = [f"{r.wahl}|{r['gebiet-nr']}" for _, r in g.iterrows()]
        k = [kurz(r) for _, r in g.head(3).iterrows()]
        if len(g) > 3:
            k.append(f"{len(g) - 3} weitere Bezirke")
        L.extend([f"- datum: {server_datum(g)}", f"  typ: {typ}", f"  ags: \"{ags}\"",
                  f"  bezirke: {json.dumps(keys, ensure_ascii=False)}"])
        if ergebnis:   # "offen" wuerde den Bezirk auf "in Pruefung" setzen - das ist er ohne Antwort nicht
            L.append(f"  ergebnis: {ergebnis}")
        summe = sum(float(r.delta_stimmen or 0) for _, r in g.iterrows())
        if ergebnis == "korrigiert" and summe > 0:
            L.append(f"  stimmen: {summe:.0f}")
        L.append(f"  titel: {_yaml_str(praefix + ', '.join(k))}")
        stände = [_datum_de(t) for t in sorted({_server_tag(t) for t in g.server_stand if t})]
        wann = f", neu veröffentlicht laut Server am {' und '.join(stände)}" if stände else ""
        # Im alten Stand geleert: verglichen mit den zuletzt veroeffentlichten Zahlen eines frueheren Stands
        vgl = sorted({v for v in g.vergleich_stand if v})
        zuvor = (f" Die Zahlen waren zwischenzeitlich auf dem Server geleert; verglichen wird mit den zuletzt "
                 f"veröffentlichten vom {' bzw. '.join(_stand_de(v) for v in vgl)}.") if vgl else ""
        L.append(f"  text: Abgleich der offenen Daten vom {n_de} mit dem Stand vom {a_de}{wann}.{zuvor}")
        # Strukturiert je Bezirk (die Webseite stellt das als Tabelle dar): alt/neu, erwartet, Listen
        L.append("  aenderungen:")
        for _, r in g.iterrows():
            def z(v, nk=0):
                return "null" if v in ("", None) else (f"{float(v):.{nk}f}" if nk else f"{float(v):.0f}")
            listen = [[m.group(1).strip(), int(m.group(2))] for m in
                      (re.match(r"^(.*)\s([+-]\d+)$", t) for t in r.listen.split(" · ") if t) if m]
            L.append(f"    - bezirk: {_yaml_str(r.wahl + '|' + r['gebiet-nr'])}")
            L.append(f"      name: {_yaml_str(bezirk_name(r))}")
            L.append(f"      wahl: {_yaml_str(r.wahl)}")
            L.append(f"      stimmen: [{z(r.stimmen_alt)}, {z(r.stimmen_neu)}]")
            L.append(f"      erwartet: {z(r.fehlend_gemeldet)}")
            L.append(f"      zettel: [{z(r.zettel_alt)}, {z(r.zettel_neu)}]")
            L.append(f"      je_zettel: [{z(r.quot_alt, 2)}, {z(r.quot_neu, 2)}]")
            L.append(f"      median: {z(r.median_neu, 2)}")
            L.append(f"      listen: {json.dumps(listen, ensure_ascii=False)}")
        L.extend([f"  quelle: {quelle}", f"  beleg: {beleg}", ""])

    for ags, g in G.groupby("ags", sort=True):
        gel = g[g.bewertung == "gelöst"]
        wtr = g[g.bewertung == "weiterhin auffällig"]
        for x in info.get("ruecknahmen", []):
            if x["ags"] == ags:
                L.append(f"# ACHTUNG Rücknahme: {_ruecknahme_text(x)}. Status auf der Webseite bleibt sonst berichtigt.")
        if len(gel):
            eintrag(ags, gel, "korrektur", "korrigiert", "Berichtigte Zahlen in den offenen Daten: ")
        if len(wtr):
            eintrag(ags, wtr, "daten", "", "Zahlen in den offenen Daten geändert, Bezirk bleibt auffällig: ")
        wdr = g[g.bewertung == WIEDER]
        if len(wdr):
            k = [f"{r.wahl} {bezirk_name(r)}" for _, r in wdr.head(3).iterrows()]
            if len(wdr) > 3:
                k.append(f"{len(wdr) - 3} weitere Bezirke")
            keys = [f"{r.wahl}|{r['gebiet-nr']}" for _, r in wdr.iterrows()]
            vgl = sorted({v for v in wdr.vergleich_stand if v})
            L.extend([f"- datum: {server_datum(wdr)}", "  typ: daten", f"  ags: \"{ags}\"",
                      f"  bezirke: {json.dumps(keys, ensure_ascii=False)}",
                      f"  titel: {_yaml_str('Zahlen wieder veröffentlicht, unverändert; Bezirk bleibt auffällig: ' + ', '.join(k))}",
                      "  text: >-",
                      f"    Die zwischenzeitlich geleerten Zahlen stehen im Abruf vom {n_de} wieder auf dem Server, Zelle für",
                      f"    Zelle gleich den zuletzt veröffentlichten vom {' bzw. '.join(_stand_de(v) for v in vgl)}. Keine Berichtigung.",
                      f"  quelle: {quelle}", f"  beleg: {beleg}", ""])
        zur = g[g.bewertung == "Zahlen zurückgezogen"]
        if len(zur):
            # Kein Ergebnis und keine Aenderungstabelle: Leere Zellen sind keine Berichtigung; die Pruefung wertet
            # den Bezirk mit den zuletzt veroeffentlichten Zahlen weiter (plausi, Spalte zahlen_aus_stand).
            k = [f"{r.wahl} {bezirk_name(r)}" for _, r in zur.head(3).iterrows()]
            if len(zur) > 3:
                k.append(f"{len(zur) - 3} weitere Bezirke")
            keys = [f"{r.wahl}|{r['gebiet-nr']}" for _, r in zur.iterrows()]
            L.extend([f"- datum: {server_datum(zur)}", "  typ: daten", f"  ags: \"{ags}\"",
                      f"  bezirke: {json.dumps(keys, ensure_ascii=False)}",
                      f"  titel: {_yaml_str('Zahlen auf dem Server geleert, Neueingabe läuft; Bezirk bleibt offen: ' + ', '.join(k))}",
                      "  text: >-",
                      f"    Beim Abgleich der offenen Daten vom {n_de} mit dem Stand vom {a_de} waren die Zahlen dieser",
                      "    Bezirke leer. Das ist keine Berichtigung. Bis neue Zahlen veröffentlicht sind, wertet die Prüfung",
                      f"    die zuletzt veröffentlichten vom {a_de}.",
                      f"  quelle: {quelle}", f"  beleg: {beleg}", ""])
    pfad.write_text("\n".join(L) + "\n", encoding="utf-8")


def _protokoll(out: Path, zeile: dict):
    """Eine Zeile je Lauf in protokoll.csv; ein wiederholter Lauf fuer denselben Stand ersetzt seine Zeile."""
    pf = out / "protokoll.csv"
    P = pd.read_csv(pf, dtype=str).fillna("") if pf.exists() else pd.DataFrame(columns=PROTOKOLL_SPALTEN)
    P = P[P.stand != zeile["stand"]]
    P = pd.concat([P, pd.DataFrame([zeile], columns=PROTOKOLL_SPALTEN).astype(str)], ignore_index=True)
    P.sort_values("stand").to_csv(pf, index=False)


def _raw_symlink(raw: Path, neu: Path):
    """<raw> auf den juengsten Stand zeigen lassen; ein echtes Verzeichnis bleibt unangetastet. Ein nachtraeglicher
    Vergleich eines aelteren Stands (--ohne-download) setzt den Link nicht zurueck."""
    if any(RE_STAND.fullmatch(p.name) and p.name > neu.name for p in neu.parent.iterdir() if p.is_dir()):
        print(f"{raw} bleibt: es gibt einen juengeren Stand als {neu.name}")
        return
    if raw.is_symlink() or not raw.exists():
        if raw.is_symlink():
            raw.unlink()
        raw.symlink_to(os.path.relpath(neu, raw.parent))
        print(f"{raw} -> {os.readlink(raw)}")
    else:
        print(f"Hinweis: {raw} ist ein Verzeichnis, kein Symlink, und bleibt unveraendert. "
              f"Aktueller Stand: {neu}")


# --------------------------------------------------------------------------- Verzeichnis je Instanz
# votemanager legt die Dateien einer Wahl neu an, wenn sie umbenannt wird (seit 25.09.2026 etwa "Kreiswahl
# Landkreis Northeim" -> "Kreiswahl", "Wahl des Rates der Stadt Bremervoerde" -> "Wahl des Rates"). Die alte
# Datei bleibt unverlinkt liegen und wird nicht mehr gepflegt; Bremervoerde hat Berichtigungen nur in der neuen
# veroeffentlicht. Massgeblich ist deshalb bei jedem Abruf das Verzeichnis open_data.json (csvs), nicht die feste
# Liste sources.csv. Eine Wahl erkennt man an ihrer wahl_id (ergebnisgrafiken), die die Umbenennung uebersteht.
# Das Register haelt je (ags, wahl_id, ebene) den zuerst gesehenen Namen und Dateinamen fest: Unter diesem
# Dateinamen liegt die Datei im Stand (die URL steht im Manifest), und <stand>/<ags>/wahlnamen.json uebersetzt
# beim Lesen den neuen Wahlnamen auf den alten. Vergleich, Verfahrensliste und Webseite behalten so ihre
# Schluessel, und eine Berichtigung unter neuem Namen erscheint als gewoehnliche Aenderung.
WAHLEN_REGISTER = "nds/wahlen_register.csv"
REGISTER_SPALTEN = ["ags", "wahl_id", "ebene", "wahl_kanonisch", "datei_kanonisch", "wahl_aktuell",
                    "datei_aktuell", "erstmals", "zuletzt", "im_verzeichnis"]
RE_STICHWAHL = re.compile(r"stichwahl", re.I)
_WAHLNAMEN: dict = {}


def _wahlnamen(d: Path) -> dict:
    """Aktueller Wahlname -> Name im Register fuer die Instanz im Ordner d (aus wahlnamen.json)."""
    k = str(d)
    if k not in _WAHLNAMEN:
        p = d / "wahlnamen.json"
        try:
            _WAHLNAMEN[k] = json.loads(p.read_text(encoding="utf-8")).get("namen", {}) if p.exists() else {}
        except Exception:
            _WAHLNAMEN[k] = {}
    return _WAHLNAMEN[k]


def _vm_wahl_ids(od: dict) -> dict:
    """csvs-Wahlname -> wahl_id. ergebnisgrafiken nennt '<Wahl> - <Gebiet>'; der laengste passende Name gewinnt."""
    namen = sorted({c.get("wahl", "") for c in od.get("csvs", []) if c.get("wahl")}, key=len, reverse=True)
    ids = {}
    for g in od.get("ergebnisgrafiken", []):
        gw = g.get("wahl", "")
        for n in namen:
            if gw == n or gw.startswith(n + " - "):
                ids.setdefault(n, set()).add(str(g.get("wahl_id", "")))
                break
    return {n: next(iter(v)) for n, v in ids.items() if len(v) == 1 and "" not in v}


def _verzeichnis_eintraege(od: dict) -> list[dict]:
    """Dateien des Verzeichnisses, die der Abgleich laedt (unterste Ebene, Gruppierungen), mit wahl_id."""
    ids = _vm_wahl_ids(od)
    out = []
    for c in od.get("csvs", []):
        w, e, u = c.get("wahl", ""), c.get("ebene", ""), c.get("url", "")
        if u and ("bezirk" in e.lower() or "ruppier" in e.lower()):
            out.append({"wahl": w, "ebene": e, "datei": u.split("/")[-1].split("?")[0], "wahl_id": ids.get(w, "")})
    return out


def _register_aufbauen(stand_dir: Path) -> pd.DataFrame:
    """Register aus allen gespeicherten Staenden, aeltester zuerst: erster Name und Dateiname je Wahl."""
    reg = {}
    for s in sorted(p for p in stand_dir.iterdir() if p.is_dir() and RE_STAND.fullmatch(p.name)):
        for od_p in sorted(s.glob("*/open_data.json")):
            try:
                od = json.loads(od_p.read_text(encoding="utf-8"))
            except Exception:
                continue
            for e in _verzeichnis_eintraege(od):
                if not e["wahl_id"]:
                    continue
                k = (od_p.parent.name, e["wahl_id"], e["ebene"])
                if k not in reg:
                    reg[k] = {"ags": k[0], "wahl_id": k[1], "ebene": k[2], "wahl_kanonisch": e["wahl"],
                              "datei_kanonisch": e["datei"], "erstmals": s.name}
                reg[k].update(wahl_aktuell=e["wahl"], datei_aktuell=e["datei"], zuletzt=s.name, im_verzeichnis="ja")
    print(f"Wahlen-Register aus {stand_dir} aufgebaut: {len(reg)} Eintraege", flush=True)
    return pd.DataFrame(list(reg.values()), columns=REGISTER_SPALTEN).fillna("")


def _verzeichnis_abgleichen(keep: pd.DataFrame, register: Path, stand_dir: Path, stand: str,
                            jobs: int) -> tuple[pd.DataFrame, dict]:
    """keep (aus sources.csv) je votemanager-Instanz nach dem aktuellen Verzeichnis ausrichten.

    - Datei im Verzeichnis, Wahl bekannt: laden von der aktuellen URL, ablegen unter dem Dateinamen im Register
      (umbenannte Wahlen landen so auf ihrer bisherigen Datei).
    - Datei im Verzeichnis, Wahl unbekannt: neu aufnehmen (neue Stichwahlen nur zaehlen, sie aendern sonst die
      Gesamtzahlen ohne Anlass).
    - Datei aus sources.csv, die das Verzeichnis nicht mehr nennt und die keine Wahl ersetzt: weiter laden, damit
      nichts still verschwindet, und im Bericht als verwaist melden.
    Instanzen, deren Verzeichnis nicht lesbar ist, behalten ihre Zeilen aus sources.csv."""
    reg = (pd.read_csv(register, dtype=str).fillna("") if Path(register).exists() else _register_aufbauen(stand_dir))
    regmap = {(r.ags, r.wahl_id, r.ebene): r.to_dict() for _, r in reg.iterrows()}
    meta = keep[(keep.type == "META") & (keep.level == "open_data.json") & (keep.system == "votemanager")]

    def hole(r):
        try:
            resp = _get(r.url)
            return r.ags, r.url, (resp.json() if resp.status_code == 200 else None)
        except Exception:
            return r.ags, r.url, None

    print(f"Verzeichnisse von {len(meta)} votemanager-Instanzen lesen ...", flush=True)
    info = {"umbenannt": [], "verwaist": [], "neu": [], "stichwahl_neu": 0, "ohne_verzeichnis": []}
    zeilen, ersetzt = [], set()
    for ags, url, od in _parallel(hole, [r for _, r in meta.iterrows()], jobs):
        if not od or not od.get("csvs"):
            info["ohne_verzeichnis"].append(ags)
            continue
        base_od = url.rsplit("/", 1)[0] + "/"
        eigene = keep[(keep.ags == ags) & (keep.type != "META")]
        bekannt = set(eigene.datei)
        gesehen = set()
        for e in _verzeichnis_eintraege(od):
            g = regmap.get((ags, e["wahl_id"], e["ebene"])) if e["wahl_id"] else None
            if (g["datei_kanonisch"] if g else e["datei"]) not in bekannt:
                if RE_STICHWAHL.search(e["wahl"]):
                    info["stichwahl_neu"] += 1
                    continue
                info["neu"].append({"ags": ags, "wahl": e["wahl"], "datei": e["datei"]})
            if g is None and e["wahl_id"]:
                g = {"ags": ags, "wahl_id": e["wahl_id"], "ebene": e["ebene"], "wahl_kanonisch": e["wahl"],
                     "datei_kanonisch": e["datei"], "erstmals": stand}
                regmap[(ags, e["wahl_id"], e["ebene"])] = g
            if g is not None:
                g.update(wahl_aktuell=e["wahl"], datei_aktuell=e["datei"], zuletzt=stand, im_verzeichnis="ja")
                datei, wahl = g["datei_kanonisch"], g["wahl_kanonisch"]
            else:
                datei, wahl = e["datei"], e["wahl"]
            if datei != e["datei"]:
                info["umbenannt"].append({"ags": ags, "wahl_id": e["wahl_id"], "ebene": e["ebene"],
                                          "wahl_alt": wahl, "wahl_neu": e["wahl"],
                                          "datei_alt": datei, "datei_neu": e["datei"]})
            gesehen.add(datei)
            zeilen.append({"ags": ags, "url": base_od + e["datei"], "wahl": wahl, "level": e["ebene"],
                           "type": classify_election(wahl), "system": "votemanager", "datei": datei, "teile": ""})
        for _, r in eigene.iterrows():
            if r.datei in gesehen:
                ersetzt.add((ags, r.datei))
            elif r.system == "votemanager":
                info["verwaist"].append({"ags": ags, "wahl": r.wahl, "datei": r.datei})
        for k, g in regmap.items():
            if k[0] == ags and g.get("zuletzt") != stand:
                g["im_verzeichnis"] = "nein"
    alt = keep[[(r.ags, r.datei) not in ersetzt for r in keep.itertuples()]]
    keep = (pd.concat([alt, pd.DataFrame(zeilen, columns=keep.columns)], ignore_index=True)
              .drop_duplicates(["ags", "datei"], keep="last"))
    pd.DataFrame(list(regmap.values()), columns=REGISTER_SPALTEN).fillna("") \
        .sort_values(["ags", "wahl_id", "ebene"]).to_csv(register, index=False)
    print(f"Verzeichnis: {len(info['umbenannt'])} Dateien umbenannt (geladen unter bisherigem Namen), "
          f"{len(info['neu'])} neu aufgenommen, {len(info['verwaist'])} nicht mehr im Verzeichnis (weiter geladen), "
          f"{info['stichwahl_neu']} neue Stichwahl-Dateien nicht geladen, "
          f"{len(info['ohne_verzeichnis'])} Instanzen ohne lesbares Verzeichnis", flush=True)
    return keep, info


def _wahlnamen_schreiben(stand: Path, info: dict):
    """<stand>/<ags>/wahlnamen.json fuer Instanzen mit umbenannten Wahlen (aktueller Name -> Registername)."""
    je = {}
    for u in info.get("umbenannt", []):
        if u["wahl_neu"] != u["wahl_alt"]:
            je.setdefault(u["ags"], {})[u["wahl_neu"]] = u["wahl_alt"]
    for ags, namen in je.items():
        d = stand / ags
        if d.is_dir():
            (d / "wahlnamen.json").write_text(json.dumps({
                "hinweis": "Wahl auf dem Server umbenannt. Die Datei liegt unter ihrem bisherigen Namen (URL im "
                           "Manifest); beim Lesen wird der aktuelle Wahlname auf den bisherigen uebersetzt.",
                "namen": namen}, ensure_ascii=False, indent=1), encoding="utf-8")


def _doppelungen(stand: Path, keep: pd.DataFrame) -> list[tuple]:
    """Sicherheitsnetz: zwei Dateien einer Instanz mit identischen Bezirkszahlen (Nr., gueltige Zettel, Stimmen)
    sind fast sicher dieselbe Wahl unter zwei Namen - dann zaehlte sie doppelt."""
    funde = []
    for ags, g in keep[keep.type != "META"].groupby("ags"):
        sig = {}
        for datei in g.datei:
            p = stand / ags / datei
            if not p.exists() or p.suffix.lower() != ".csv":
                continue
            try:
                with open(p, encoding="utf-8-sig") as f:
                    rows = list(csv.DictReader(f, delimiter=";"))
            except Exception:
                continue
            s = tuple(sorted((r.get("gebiet-nr", "").strip(), r.get("C2", ""), r.get("D", ""))
                             for r in rows if (r.get("C2") or "").strip()))
            if len(s) < 2:
                continue
            if s in sig:
                funde.append((ags, sig[s], datei))
            else:
                sig[s] = datei
    return funde


def cmd_abgleich(a):
    stand_dir = Path(a.stand_dir)
    stand_dir.mkdir(parents=True, exist_ok=True)
    neu_s = a.datum or datetime.now().astimezone().strftime("%Y-%m-%d_%H%M")
    if not RE_STAND.fullmatch(neu_s):
        sys.exit(f"--datum {neu_s}: erwartet JJJJ-MM-TT oder JJJJ-MM-TT_HHMM")
    neu = stand_dir / neu_s
    keep = _abgleich_quellen(a.sources)
    verz = {}
    out = Path(a.out)
    out.mkdir(parents=True, exist_ok=True)

    # ---- Vergleichsstand: der juengste vor dem neuen ----------------------------
    staende = sorted(p.name for p in stand_dir.iterdir()
                     if p.is_dir() and RE_STAND.fullmatch(p.name) and p.name < neu_s)
    alt_s = a.alt or (staende[-1] if staende else None)
    alt = stand_dir / alt_s if alt_s else None
    if alt is not None and not alt.is_dir():
        sys.exit(f"Vergleichsstand {alt} fehlt")
    M_alt = _manifest(alt, keep) if alt is not None else None   # legt manifest.csv nach, falls es fehlt

    # ---- Stand laden --------------------------------------------------------------
    if a.ohne_download:
        if not neu.exists():
            sys.exit(f"{neu} fehlt (ohne --ohne-download wird er geladen)")
        M_neu = _manifest(neu, keep)
    else:
        if neu.exists() and not a.force:
            sys.exit(f"{neu} existiert bereits - Staende sind unveraenderlich. "
                     f"Nur vergleichen: --ohne-download; neu laden: --force")
        keep, verz = _verzeichnis_abgleichen(keep, Path(a.register), stand_dir, neu_s, a.jobs)
        print(f"{len(keep)} Quellen -> {neu} ({a.jobs} parallel"
              f"{', bedingt gegen ' + alt_s if alt_s else ''})", flush=True)
        M_neu = _manifest(neu, keep, _stand_laden(keep, neu, a.jobs, alt, M_alt))
        _herkunft_schreiben(neu, keep)
        _wahlnamen_schreiben(neu, verz)
        verz["doppelt"] = _doppelungen(neu, keep)
        for d in verz["doppelt"]:
            print(f"WARNUNG moegliche Doppelung {d[0]}: {d[1]} und {d[2]} haben identische Bezirkszahlen", flush=True)
    fehler = int((~M_neu.status.isin(["200", "304"])).sum())
    geladen = int((M_neu.status == "200").sum())
    n304 = int((M_neu.status == "304").sum())
    abgerufen = M_neu.abgerufen.max() if len(M_neu) else ""
    abgerufen_de = f"{_datum_de(abgerufen[:10])} {abgerufen[11:16]} UTC" if abgerufen else "?"

    if alt is None:
        print(f"Stand {neu_s} angelegt; kein aelterer Stand vorhanden, nichts zu vergleichen.")
        _protokoll(out, {"stand": neu_s, "vergleich_mit": "", "abgerufen": abgerufen, "quellen": len(keep),
                         "geladen": geladen, "unveraendert_304": n304, "fehler": fehler})
        _raw_symlink(Path(a.raw), neu)
        return

    # ---- Vergleich ------------------------------------------------------------------
    gemeldet = {}
    if Path(a.zettelcheck).exists():
        Z = pd.read_csv(a.zettelcheck, dtype=str).fillna("")
        gemeldet = {(r.ags, r.wahl, r["gebiet-nr"]): r for _, r in Z.iterrows()}
    namen = {}
    if Path(a.gemeinden).exists():
        namen = {r.ags: r.gemeinde for _, r in pd.read_csv(a.gemeinden, dtype=str).fillna("").iterrows()}

    print(f"Vergleich {alt_s} -> {neu_s} ...", flush=True)
    B, Zl, info = _stand_vergleichen(alt, neu, gemeldet, namen, M_neu)
    info["verzeichnis"] = verz
    info["ruecknahmen"] = _ruecknahmen(out, neu_s, B)
    if len(B):
        B = (B.assign(_o=B.bewertung.map({b: i for i, b in enumerate(BEWERTUNGEN)}))
              .sort_values(["_o", "ags", "wahl", "gebiet-nr"]).drop(columns="_o"))
    G = B[B.gemeldet == True] if len(B) else B  # noqa: E712
    etwas = bool(len(B) or len(Zl) or info["neu"] or info["entfernt"]
                 or any(verz.get(k) for k in ("umbenannt", "neu", "verwaist", "doppelt")))
    if etwas:
        B.to_csv(out / f"{neu_s}.csv", index=False)
        Zl.to_csv(out / f"{neu_s}_zellen.csv", index=False)
        _abgleich_bericht(out / f"{neu_s}.md", alt_s, neu_s, B, info, abgerufen_de, len(keep), fehler)
        _abgleich_ereignisse(out / f"{neu_s}_ereignisse.yaml", alt_s, neu_s, B, info)
    _protokoll(out, {
        "stand": neu_s, "vergleich_mit": alt_s, "abgerufen": abgerufen, "quellen": len(keep),
        "geladen": geladen, "unveraendert_304": n304, "fehler": fehler,
        "dateien_geaendert": info["geaendert"], "bezirke_geaendert": int((B.bewertung != WIEDER).sum()) if len(B) else 0,
        "gemeldet_geaendert": int((G.bewertung != WIEDER).sum()) if len(G) else 0,
        "geloest": int((G.bewertung == "gelöst").sum()) if len(G) else 0,
        "weiterhin_auffaellig": int((G.bewertung == "weiterhin auffällig").sum()) if len(G) else 0,
        "neu_auffaellig": int((B.bewertung == "neu auffällig").sum()) if len(B) else 0,
        "dateien_neu": len(info["neu"]), "dateien_entfernt": len(info["entfernt"]),
        "bericht": f"{neu_s}.md" if etwas else "",
    })

    print(f"\n{info['verglichen']} Dateien verglichen, {info['geaendert']} geaendert, "
          f"{len(B)} Bezirke mit geaenderten Zahlen in {B.ags.nunique() if len(B) else 0} Gemeinden")
    if len(B):
        print(f"gemeldete Bezirke darunter: {len(G)} · geloest: {(G.bewertung == 'gelöst').sum()} · "
              f"weiterhin auffaellig: {(G.bewertung == 'weiterhin auffällig').sum()} · "
              f"neu auffaellig: {(B.bewertung == 'neu auffällig').sum()}")
    if info["neu"] or info["entfernt"]:
        print(f"Dateien neu: {len(info['neu'])} · entfernt: {len(info['entfernt'])}")
    for _, r in B.iterrows():
        print(f"  {r.bewertung:<30} {r.ags} {r.wahl} {r['gebiet-nr']} {r['gebiet-name']}: {r.aenderungen}")
    for x in info["ruecknahmen"]:
        print(f"WARNUNG frueher geloest, jetzt wieder auffaellig: {_ruecknahme_text(x)}", flush=True)
    if etwas:
        print(f"-> {out / (neu_s + '.csv')}, {out / (neu_s + '_zellen.csv')}, {out / (neu_s + '.md')}, "
              f"{out / (neu_s + '_ereignisse.yaml')}")
    else:
        print("keine Aenderungen; nur protokoll.csv fortgeschrieben")
    _raw_symlink(Path(a.raw), neu)



# --------------------------------------------------------------------------- Verfahrensliste
def cmd_liste(a):
    """Verfahrensliste: Vereinigung der zettelcheck.csv aller Staende (<stand-dir>/<Stand>/zettelcheck.csv).

    Ein Bezirk steht mit den Zahlen des Standes darin, in dem er zuerst auffiel (Spalte
    `datenstand`); spaetere Staende fuegen nur neue Bezirke hinzu. Wer in einem spaeteren Stand
    nicht mehr auffaellt, bleibt drin - was aus ihm wurde, zeigen `abgleich` und das Ereignis-Log.

    Zweites Kriterium: Liegt neben der zettelcheck.csv eine zweite_suche.csv (nds/out/knapp_zuerst/zweite_suche.py,
    knappe Raete, kleinere doppelt belegte Abweichungen, Kippfall-Stufe A1-C), kommen deren Bezirke dazu.
    Spalte `kriterium`: schwelle_025 (plausi, 0,25 unter dem Median) oder zweite_suche. Innerhalb eines Standes
    geht die zettelcheck.csv vor; wer einmal drinsteht, behaelt Stand und Kriterium seines ersten Auftretens.
    """
    stand_dir = Path(a.stand_dir)
    staende = sorted(p for p in stand_dir.iterdir() if p.is_dir() and (p / "zettelcheck.csv").exists())
    if not staende:
        sys.exit(f"keine zettelcheck.csv unter {stand_dir}/<Stand>/")
    teile, gesehen = [], set()
    for st in staende:
        Z = pd.read_csv(st / "zettelcheck.csv", dtype=str).fillna("")
        Z["kriterium"] = "schwelle_025"
        if (st / "zweite_suche.csv").exists():
            Z2 = pd.read_csv(st / "zweite_suche.csv", dtype=str).fillna("")
            Z2["kriterium"] = "zweite_suche"
            schl1 = set(zip(Z.ags, Z.wahl, Z["gebiet-nr"]))
            Z2 = Z2[[k not in schl1 for k in zip(Z2.ags, Z2.wahl, Z2["gebiet-nr"])]]
            print(f"{st.name}: zweite Suche {len(Z2)} Bezirke")
            Z = pd.concat([Z, Z2], ignore_index=True).fillna("")
        Z["datenstand"] = st.name
        schl = list(zip(Z.ags, Z.wahl, Z["gebiet-nr"]))
        neu = Z[[k not in gesehen for k in schl]]
        weg = gesehen - set(schl)
        print(f"{st.name}: {len(Z)} auffaellige Bezirke, davon neu {len(neu)}"
              f"{f' · aus frueheren Staenden nicht mehr auffaellig: {len(weg)}' if gesehen else ''}")
        if gesehen:   # beim ersten Stand ist jeder Bezirk neu, das braucht keine Liste
            for _, r in neu.iterrows():
                print(f"  neu: {r.ags} {r.wahl} {r['gebiet-nr']} {r['gebiet-name']} ({r.urteil})")
        gesehen.update(schl)
        teile.append(neu)
    V = pd.concat(teile, ignore_index=True)
    V.to_csv(a.out, index=False)
    print(f"{len(V)} Bezirke in {V.ags.nunique()} Gemeinden -> {a.out}")


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    sp = ap.add_subparsers(dest="cmd", required=True)

    p = sp.add_parser("probe")
    p.add_argument("--termin", default="20260913")
    p.add_argument("--prefix", default="03459", help="Kreisschluessel, kommagetrennt, oder NDS")
    # Gemeindeschluessel reichen ueber 60 hinaus (Geestland 03352062, Wurster Nordseekueste 03352061)
    p.add_argument("--max-gemeinde", type=int, default=99)
    p.add_argument("--max-samtgemeinde", type=int, default=15)
    p.add_argument("--jobs", type=int, default=8, help="parallele Abfragen")
    p.add_argument("--out", default="instances.csv")
    p.set_defaults(func=cmd_probe)

    p = sp.add_parser("discover")
    p.add_argument("--fremd", default=FREMDQUELLEN, help="Quellen ausserhalb votemanager.kdo.de")
    p.add_argument("--instances", default="instances.csv")
    p.add_argument("--jobs", type=int, default=8, help="parallele Abfragen")
    p.add_argument("--out", default="sources.csv")
    p.set_defaults(func=cmd_discover)

    p = sp.add_parser("download")
    p.add_argument("--sources", default="sources.csv")
    p.add_argument("--dir", default="data/raw")
    p.add_argument("--force", action="store_true")
    p.add_argument("--alle", action="store_true",
                   help="auch Kreis-, Buergermeister- und Samtgemeindewahlen (fuer plausi)")
    p.add_argument("--jobs", type=int, default=8, help="parallele Downloads")
    p.set_defaults(func=cmd_download)

    p = sp.add_parser("plausi")
    p.add_argument("--dir", default="data/raw")
    p.add_argument("--out", default="out/plausibilitaet.csv")
    p.add_argument("--top", type=int, default=25)
    p.add_argument("--nur-auffaellig", action="store_true",
                   help="Ausgabe auf auffaellige Bezirke beschraenken")
    p.add_argument("--ohne-vorstaende", action="store_true",
                   help="geleerte Bezirke nicht mit Zahlen frueherer Staende (Nachbarordner von --dir) fuellen")
    p.set_defaults(func=cmd_plausi)

    p = sp.add_parser("namen", help="Gemeindenamen je Instanz holen")
    p.add_argument("--instances", default="nds/instances.csv")
    p.add_argument("--jobs", type=int, default=8)
    p.add_argument("--fremd", default=FREMDQUELLEN, help="Namen der Fremdquellen (Spalte name)")
    p.add_argument("--out", default="nds/out/gemeinden.csv")
    p.set_defaults(func=cmd_namen)

    p = sp.add_parser("histdownload", help="Bezirksdaten eines frueheren Termins laden (2021)")
    p.add_argument("--plausi", default="nds/out/plausibilitaet.csv")
    p.add_argument("--datum", default="12.09.2021", help="Wahltermin wie in api/termine.json")
    p.add_argument("--dir", default="nds/data/hist2021")
    p.add_argument("--alle-ags", action="store_true",
                   help="alle Gemeinden statt nur die mit auffaelligen Bezirken")
    p.add_argument("--force", action="store_true")
    p.add_argument("--jobs", type=int, default=8)
    p.set_defaults(func=cmd_histdownload)

    p = sp.add_parser("zettelcheck", help="auffaellige Bezirke gegen die Vorwahl pruefen")
    p.add_argument("--plausi", default="nds/out/plausibilitaet.csv")
    p.add_argument("--hist-dir", default="nds/data/hist2021")
    p.add_argument("--datum-alt", default="12.09.2021")
    p.add_argument("--toleranz", type=float, default=0.15,
                   help="ab welcher Abweichung der Wahlberechtigten ein Bezirk als neu "
                        "zugeschnitten gilt")
    p.add_argument("--z-treffer", type=float, default=2.0,
                   help="|z| bis zu dem ein Wert als zur Erwartung passend gilt")
    p.add_argument("--z-verdacht", type=float, default=3.0,
                   help="|z| ab dem ein Wert als unplausibel gilt")
    p.add_argument("--geschwister-tol", type=float, default=0.06,
                   help="erlaubte Abweichung der Zettelzahl von den Parallelwahlen im Bezirk")
    p.add_argument("--bw-gleich-tol", type=float, default=0.02,
                   help="Briefwahlbezirk: so nah muss die Waehlerzahl einer anderen Wahl liegen, "
                        "damit sie als dasselbe Gebiet gilt")
    p.add_argument("--bw-or-faktor", type=float, default=1.3,
                   help="Briefwahlbezirk: so viel mehr Waehler als jede gemeindeweite Wahl muss ein "
                        "Ortsrat melden, damit seine Zettelzahl als falsch gilt")
    p.add_argument("--s-max", type=float, default=0.20,
                   help="max. Streuung, bis zu der ein Anker eine Uebereinstimmung belegen darf")
    p.add_argument("--quot-tol", type=float, default=0.08,
                   help="erlaubter Abstand Stimmen/Zettel zum Median bei Tippfehler-Kandidaten")
    p.add_argument("--out", default="nds/out/zettelcheck.csv")
    p.add_argument("--report", default="nds/out/ZETTELCHECK.md")
    p.add_argument("--top", type=int, default=25)
    p.set_defaults(func=cmd_zettelcheck)

    p = sp.add_parser("wirkung", help="Befunde je Landkreis und Wahl hochrechnen")
    p.add_argument("--plausi", default="nds/out/plausibilitaet.csv")
    p.add_argument("--zettelcheck", default="nds/out/zettelcheck.csv")
    p.add_argument("--namen", default="nds/out/gemeinden.csv")
    p.add_argument("--out", default="nds/out/wirkung_wahlen.csv")
    p.add_argument("--out-kreise", default="nds/out/wirkung_kreise.csv")
    p.add_argument("--report", default="nds/out/WIRKUNG.md")
    p.add_argument("--top", type=int, default=20)
    p.set_defaults(func=cmd_wirkung)

    p = sp.add_parser("abgleich", help="offene Daten neu laden, als Stand einfrieren, mit dem vorigen vergleichen")
    p.add_argument("--sources", default="nds/sources.csv")
    p.add_argument("--register", default=WAHLEN_REGISTER,
                   help="Wahlen je Instanz mit wahl_id und bisherigem Namen (wird fortgeschrieben)")
    p.add_argument("--stand-dir", default="nds/data/stand", help="ein Unterordner je Abruf (JJJJ-MM-TT_HHMM)")
    p.add_argument("--raw", default="nds/data/raw", help="Symlink auf den juengsten Stand, wird umgesetzt")
    p.add_argument("--datum", help="Name des neuen Standes, JJJJ-MM-TT_HHMM (Standard: jetzt)")
    p.add_argument("--alt", help="Vergleichsstand (Standard: der juengste vor dem neuen)")
    p.add_argument("--ohne-download", action="store_true", help="nur vergleichen; der Stand muss vorhanden sein")
    p.add_argument("--force", action="store_true", help="vorhandenen Stand gleichen Namens neu laden")
    p.add_argument("--out", default="nds/out/abgleich")
    p.add_argument("--zettelcheck", default="nds/out/zettelcheck.csv", help="gemeldete Bezirke")
    p.add_argument("--gemeinden", default="nds/out/gemeinden.csv", help="Gemeindenamen je AGS")
    p.add_argument("--jobs", type=int, default=8, help="parallele Abrufe")
    p.set_defaults(func=cmd_abgleich)

    p = sp.add_parser("liste", help="Verfahrensliste: zettelcheck.csv aller Staende vereinigen (Spalte datenstand)")
    p.add_argument("--stand-dir", default="nds/out/stand", help="je Stand: <Stand>/zettelcheck.csv")
    p.add_argument("--out", default="nds/out/zettelcheck.csv")
    p.set_defaults(func=cmd_liste)

    p = sp.add_parser("inspect")
    p.add_argument("file")
    p.set_defaults(func=cmd_inspect)

    p = sp.add_parser("parties")
    p.add_argument("--dir", default="data/raw")
    p.add_argument("--out", default="parties.csv")
    p.set_defaults(func=cmd_parties)

    for name, fn, d in (("analyze", cmd_analyze, "data/raw"), ("demo", cmd_demo, "demo_data")):
        p = sp.add_parser(name)
        p.add_argument("--dir", default=d)
        p.add_argument("--parties", default="parties.csv")
        p.add_argument("--out", default="out" if name == "analyze" else "demo_out")
        p.add_argument("--top", type=int, default=20)
        p.add_argument("--min-votes", type=int, default=MIN_VOTES_DEFAULT)
        p.add_argument("--scope", choices=["auto", "brief", "alle"], default="auto",
                       help="Auswertungsebene: auto = Briefwahlbezirke, falls vorhanden")
        p.add_argument("--strict", action=argparse.BooleanOptionalAction, default=True,
                       help="Rangliste nur aus zuschnittsgleichen Bezirkspaaren (Standard)")
        p.set_defaults(func=fn, demo_flag=False)

    a = ap.parse_args()
    a.func(a)


if __name__ == "__main__":
    main()