tx_activity.py

#!/usr/bin/env python3
"""Wer hat in den letzten Tagen welche Datei in Transifex bearbeitet?

Fragt die Transifex-API v3 ab und listet pro Datei auf, welcher Benutzer
dort Uebersetzungen geaendert hat, woher sie kamen und wann zuletzt.

Der Token kommt aus TX_TOKEN oder aus ~/.transifexrc -- er wird nirgends
ausgegeben.

Herkunft (Feld "origin" der API):
    EDITOR   im Web-Editor eingegeben
    TM       aus dem Translation Memory uebernommen
    MT       maschineller Vorschlag
    UPLOAD   per Datei-Upload oder tx push eingespielt
    API      ueber die API gesetzt

Aufruf (im Repo, im Branch, dessen Projekt gemeint ist):
    python tools/tx_activity.py --probe          # erst die API-Form klaeren
    python tools/tx_activity.py --bestand        # was wuerde ein Force-Push loeschen?
    python tools/tx_activity.py                  # letzte 7 Tage
    python tools/tx_activity.py --tage 1
    python tools/tx_activity.py --ohne-upload    # eigene Pushes ausblenden
    python tools/tx_activity.py --csv aktivitaet.csv
    python tools/tx_activity.py --pause 1.0      # langsamer, wenn gedrosselt
    python tools/tx_activity.py --ab library--os # Lauf fortsetzen
    python tools/tx_activity.py --alle           # ohne Vorauswahl (langsam)

Vorauswahl: die Sammelstatistik resource_language_stats nennt in vier
Anfragen, welche Ressourcen ueberhaupt angefasst wurden. Nur die werden
danach einzeln abgefragt -- sonst waeren es 553 Anfragen.
    python tools/tx_activity.py --debug          # Rohdatensatz zum Pruefen

Transifex drosselt am Gateway. Bei 429/503 wird laenger gewartet und
danach abgebrochen statt die restlichen Ressourcen gegen dieselbe Wand
zu fahren; die Meldung nennt den Slug zum Fortsetzen.

Es werden nur Lesezugriffe gemacht; nichts wird veraendert.
"""

import argparse
import configparser
import csv
import json
import os
import re
import ssl
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timedelta, timezone
from pathlib import Path

API = "https://rest.api.transifex.com"

# Wie oft ein voruebergehender Serverfehler wiederholt wird.
VERSUCHE = 4

# Mindestabstand zwischen zwei Anfragen in Sekunden (--pause).
# Ohne Pause antwortet das Gateway nach einigen hundert Anfragen mit 503.
TAKT = 0.35

# Am 2026-10-03 gegen python-newest geprueft: das Attribut heisst
# datetime_translated, der zugehoerige Filter date_translated.
DATUMSFILTER = [
    "filter[date_translated][gt]",
    "filter[datetime_translated][gt]",
]

# Herkuenfte, die keine Handarbeit im Editor sind.
MASCHINELL = {"UPLOAD", "API"}


def ssl_kontext() -> ssl.SSLContext:
    """Wurzelzertifikate: certifi, falls vorhanden, sonst die des Systems."""
    try:
        import certifi
        return ssl.create_default_context(cafile=certifi.where())
    except ImportError:
        return ssl.create_default_context()


KONTEXT = ssl_kontext()


def token() -> str:
    t = os.environ.get("TX_TOKEN")
    if t:
        return t.strip()
    rc = Path.home() / ".transifexrc"
    if rc.is_file():
        cfg = configparser.ConfigParser()
        cfg.read(rc)
        for abschnitt in cfg.sections():
            wert = cfg[abschnitt].get("token")
            if wert:
                return wert.strip()
    sys.exit("Kein API-Token gefunden. Setze TX_TOKEN oder pflege ~/.transifexrc.")


def git(*args: str) -> str:
    e = subprocess.run(["git", *args], capture_output=True, text=True)
    return e.stdout if e.returncode == 0 else ""


def repo_wurzel() -> Path:
    aus = git("rev-parse", "--show-toplevel").strip()
    if not aus:
        sys.exit("Kein Git-Repository -- bitte im python-docs-de-Verzeichnis starten.")
    return Path(aus)


def ressourcen(wurzel: Path) -> tuple:
    """({resource-slug: po-pfad}, organisation, projekt) aus .tx/config."""
    pfad = wurzel / ".tx" / "config"
    if not pfad.is_file():
        sys.exit(f"{pfad} nicht gefunden.")
    cfg = configparser.ConfigParser()
    cfg.read(pfad, encoding="utf-8")

    karte, org, projekt = {}, None, None
    for name in cfg.sections():
        m = re.match(r"o:([^:]+):p:([^:]+):r:(.+)", name)
        if not m:
            continue
        org, projekt, slug = m.groups()
        datei = (cfg[name].get("trans.de") or cfg[name].get("file_filter", ""))
        karte[slug] = datei.replace("<lang>", "de").strip()
    if not karte:
        sys.exit("In .tx/config wurden keine Ressourcen gefunden.")
    return karte, org, projekt


class Gedrosselt(RuntimeError):
    """Das Gateway weist uns ab (429/503).

    Das ist keine Eigenschaft der einzelnen Ressource, sondern unseres
    Anfragetempos insgesamt -- die naechste Ressource trifft dieselbe Wand.
    Der Aufrufer soll deshalb aufhoeren, nicht weiterprobieren.
    """


_letzte_anfrage = 0.0


def takt_halten() -> None:
    """Mindestabstand zur vorherigen Anfrage einhalten."""
    global _letzte_anfrage
    rest = TAKT - (time.monotonic() - _letzte_anfrage)
    if rest > 0:
        time.sleep(rest)
    _letzte_anfrage = time.monotonic()


def wartezeit(kopf, standard: int) -> int:
    """'Retry-After' auswerten, falls der Server es mitschickt."""
    wert = (kopf.get("Retry-After") or "").strip()
    if wert.isdigit():
        return max(1, min(300, int(wert)))
    return standard


def hole(url: str, tok: str) -> tuple:
    """Eine Seite abrufen. Liefert (daten, kopfzeilen).

    502/504 sind kurze Aussetzer -- da genuegt eine knappe Pause.
    429/503 ist Drosselung: laenger warten, 'Retry-After' beachten, und
    wenn es anhaelt, Gedrosselt ausloesen statt die restlichen Ressourcen
    gegen die Wand zu fahren (das verschaerft die Drosselung nur).
    """
    for versuch in range(1, VERSUCHE + 1):
        req = urllib.request.Request(url, headers={
            "Authorization": f"Bearer {tok}",
            "Accept": "application/vnd.api+json",
        })
        takt_halten()
        try:
            with urllib.request.urlopen(req, timeout=90, context=KONTEXT) as antwort:
                return json.load(antwort), dict(antwort.headers)
        except urllib.error.HTTPError as e:
            drosselung = e.code in (429, 503)
            if e.code < 500 and not drosselung:
                rumpf = e.read().decode("utf-8", "replace")[:400]
                raise RuntimeError(f"HTTP {e.code} bei {url}\n{rumpf}") from None
            if versuch == VERSUCHE:
                rumpf = e.read().decode("utf-8", "replace")[:200]
                art = Gedrosselt if drosselung else RuntimeError
                raise art(f"HTTP {e.code} bei {url}\n{rumpf}") from None
            warte = wartezeit(e.headers,
                              15 * 2 ** (versuch - 1) if drosselung else 2 ** versuch)
            print(f"  HTTP {e.code}, neuer Versuch in {warte}s "
                  f"({versuch}/{VERSUCHE - 1})", file=sys.stderr)
            time.sleep(warte)
        except urllib.error.URLError as e:
            if "CERTIFICATE_VERIFY_FAILED" in str(e.reason):
                sys.exit(
                    "Wurzelzertifikate fehlen. Einmalig ausfuehren:\n"
                    '  open "/Applications/Python 3.14/Install Certificates.command"\n'
                    "oder im aktiven venv:  pip install certifi"
                )
            if versuch < VERSUCHE:
                warte = 2 ** versuch
                print(f"  Netzwerkfehler, neuer Versuch in {warte}s", file=sys.stderr)
                time.sleep(warte)
                continue
            raise RuntimeError(f"Netzwerkfehler bei {url}: {e.reason}") from None
    raise RuntimeError(f"Nach {VERSUCHE} Versuchen aufgegeben: {url}")


_grenze_gemeldet = False


def grenze(kopf: dict) -> None:
    """Die Drosselungs-Kopfzeilen einmal ausgeben, falls der Server welche setzt.

    Damit muss das Anfragetempo nicht geraten werden.
    """
    global _grenze_gemeldet
    if _grenze_gemeldet:
        return
    treffer = {k: v for k, v in kopf.items()
               if "ratelimit" in k.lower().replace("-", "").replace("_", "")}
    if treffer:
        print("  Drosselung laut Server: "
              + ", ".join(f"{k}={v}" for k, v in sorted(treffer.items())),
              file=sys.stderr)
    _grenze_gemeldet = True


def seiten(pfad: str, params: dict, tok: str):
    """Alle Seiten einer JSON:API-Sammlung durchlaufen."""
    url = API + pfad + ("?" + urllib.parse.urlencode(params) if params else "")
    while url:
        daten, kopf = hole(url, tok)
        grenze(kopf)
        for eintrag in daten.get("data", []):
            yield eintrag
        url = (daten.get("links") or {}).get("next")


def benutzer(eintrag: dict) -> str:
    """Name des Uebersetzers, z. B. 'u:JystBreisgau' -> 'JystBreisgau'."""
    for rolle in ("translator", "reviewer", "proofreader"):
        bez = ((eintrag.get("relationships") or {}).get(rolle) or {}).get("data")
        if bez and bez.get("id"):
            return bez["id"].split(":", 1)[-1]
    return "(unbekannt)"


def zeitpunkt(eintrag: dict) -> str:
    attr = eintrag.get("attributes") or {}
    for schluessel in ("datetime_translated", "datetime_reviewed", "datetime_created"):
        if attr.get(schluessel):
            return attr[schluessel]
    return ""


def herkunft(eintrag: dict) -> str:
    return (eintrag.get("attributes") or {}).get("origin") or "?"


def uebersetzungen(res_id: str, seit: str, tok: str, filtername: list) -> tuple:
    """Seit *seit* geaenderte Uebersetzungen einer Ressource."""
    basis = {"filter[resource]": res_id, "filter[language]": "l:de"}

    for name in list(filtername):
        try:
            return list(seiten("/resource_translations",
                               {**basis, name: seit}, tok)), name
        except RuntimeError as fehler:
            if "HTTP 400" in str(fehler) or "HTTP 409" in str(fehler):
                filtername.remove(name)
                continue
            raise

    alle = list(seiten("/resource_translations", basis, tok))
    return [e for e in alle if zeitpunkt(e) >= seit], None


# Felder der Sammelstatistik, die menschliche Arbeit anzeigen.
# last_update bleibt bewusst aussen vor: das wandert auch bei
# Aenderungen am englischen Quelltext und wuerde das Netz unnoetig weiten.
STATFELDER = ("last_translation_update", "last_review_update",
              "last_proofread_update")


def slug_von(eintrag: dict) -> str:
    """Ressourcen-Slug aus einem Statistik-Eintrag.

    Die id hat die Form o:ORG:p:PROJ:r:SLUG:l:de; falls sich das aendert,
    dient die Beziehung zur Ressource als Rueckfallebene.
    """
    treffer = re.search(r":r:(.+?):l:", eintrag.get("id") or "")
    if treffer:
        return treffer.group(1)
    bez = ((eintrag.get("relationships") or {}).get("resource") or {}).get("data")
    if bez and bez.get("id"):
        return bez["id"].rsplit(":r:", 1)[-1]
    return ""


def frische_ressourcen(org: str, projekt: str, seit: str, tok: str) -> set:
    """Ressourcen, an deren Uebersetzung seit *seit* gearbeitet wurde.

    Die Sammelstatistik liefert 150 Ressourcen je Seite, also vier Anfragen
    fuer das ganze Projekt. Die Alternative ist eine Anfrage je Ressource --
    553 Stueck, und genau daran ist der Lauf in die Drosselung gelaufen.
    """
    frisch, gesehen = set(), 0
    params = {"filter[project]": f"o:{org}:p:{projekt}",
              "filter[language]": "l:de"}
    for e in seiten("/resource_language_stats", params, tok):
        gesehen += 1
        attr = e.get("attributes") or {}
        juengste = max((attr.get(f) or "") for f in STATFELDER)
        if juengste >= seit:
            slug = slug_von(e)
            if slug:
                frisch.add(slug)
    print(f"  Vorauswahl: {len(frisch)} von {gesehen} Ressourcen angefasst",
          file=sys.stderr)
    return frisch


# Ressource, die nachweislich Handarbeit enthaelt. Sie dient als Kontrolle:
# ein Herkunftsfilter, der hier nichts findet, ist kaputt -- und "nichts
# gefunden" ist genau die Antwort, die uns zum Loeschen verleiten wuerde.
KONTROLLE = "reference--datamodel"


def nur_editor(res_id: str, tok: str) -> int:
    """Anzahl der im Web-Editor eingegebenen Uebersetzungen einer Ressource."""
    return sum(1 for _ in seiten("/resource_translations", {
        "filter[resource]": res_id,
        "filter[language]": "l:de",
        "filter[origin]": "EDITOR",
    }, tok))


def slugs_aus_bericht(pfad: Path) -> list:
    """Ressourcen-Slugs aus einem --bestand-Bericht lesen."""
    text = pfad.read_text(encoding="utf-8")
    return [m.group(1) for m in
            re.finditer(r'^\| `[^`]+` \| (\S+) \|', text, re.M)]


def herkunft_pruefen(org: str, projekt: str, slugs: list, tok: str,
                     ziel=None) -> int:
    """Pruefen, ob in *slugs* Handarbeit steckt, die wir verlieren wuerden.

    Erst wird der Filter an KONTROLLE belegt. Ohne diesen Nachweis ist ein
    leeres Ergebnis wertlos, weil ein defekter Filter dasselbe liefert.

    Rueckgabe: Anzahl der Ressourcen mit EDITOR-Strings.
    """
    print(f"Kontrolle: Herkunftsfilter an {KONTROLLE} belegen ...",
          file=sys.stderr)
    probe = nur_editor(f"o:{org}:p:{projekt}:r:{KONTROLLE}", tok)
    if probe == 0:
        sys.exit(
            f"Abbruch: der Filter findet in {KONTROLLE} keine EDITOR-Strings,\n"
            f"obwohl dort nachweislich Handarbeit liegt. Der Filter ist also\n"
            f"nicht verlaesslich -- ein leeres Ergebnis waere wertlos.")
    print(f"  {probe} EDITOR-Strings gefunden, der Filter arbeitet.\n",
          file=sys.stderr)

    treffer = []
    for nr, slug in enumerate(slugs, 1):
        if nr % 25 == 0:
            print(f"  {nr}/{len(slugs)}", file=sys.stderr)
        if slug == KONTROLLE:
            continue
        try:
            n = nur_editor(f"o:{org}:p:{projekt}:r:{slug}", tok)
        except Gedrosselt as fehler:
            print(f"\n!! Abbruch bei {slug}: {str(fehler).splitlines()[0]}",
                  file=sys.stderr)
            sys.exit(2)
        except RuntimeError as fehler:
            print(f"  !! {slug}: {str(fehler).splitlines()[0]}", file=sys.stderr)
            continue
        if n:
            treffer.append((n, slug))

    treffer.sort(reverse=True)
    aus = [f"# Handarbeit in {len(slugs)} geprueften Ressourcen\n",
           f"- Kontrolle {KONTROLLE}: {probe} EDITOR-Strings (Filter belegt)",
           f"- Ressourcen mit EDITOR-Strings: {len(treffer)}",
           f"- EDITOR-Strings insgesamt: {sum(t[0] for t in treffer)}\n"]
    if treffer:
        aus.append("| Ressource | EDITOR-Strings |")
        aus.append("|---|---:|")
        aus += [f"| {s} | {n} |" for n, s in treffer]
        aus.append("\nDiese Stellen vor einem Force-Push ansehen.")
    else:
        aus.append("Keine Handarbeit gefunden -- alles dort ist TM, MT oder "
                   "Hochladung. Ein Force-Push loescht keine menschliche Arbeit.")
    text = "\n".join(aus)
    if ziel:
        Path(ziel).write_text(text + "\n", encoding="utf-8")
        print(f"\nBericht geschrieben: {ziel}", file=sys.stderr)
    else:
        print("\n" + text)
    return len(treffer)


def po_uebersetzt(pfad: Path) -> tuple:
    """(uebersetzt, gesamt) einer lokalen PO-Datei, ohne Fremdbibliothek."""
    if not pfad.is_file():
        return (0, 0)
    txt = pfad.read_text(encoding="utf-8", errors="replace")
    gesamt = uebersetzt = 0
    for block in txt.split("\n\n"):
        if not block.strip() or block.lstrip().startswith("#~"):
            continue
        m = re.search(r'^msgid ((?:"[^"]*"\s*)+)', block, re.M)
        if not m or m.group(1).strip() == '""':
            continue
        gesamt += 1
        s = re.search(r'^msgstr ((?:"[^"]*"\s*)+)', block, re.M)
        if s and s.group(1).strip() != '""':
            uebersetzt += 1
    return (uebersetzt, gesamt)


def bestand(org: str, projekt: str, karte: dict, wurzel: Path, tok: str,
            ziel=None) -> int:
    """Vergleicht je Ressource den Transifex-Bestand mit dem Repo-Stand.

    Ein Force-Push schiebt unseren Stand ueber alles. Wo Transifex mehr
    uebersetzte Strings hat als wir, gehen genau diese verloren -- auch
    solche, die aelter sind als jedes Aktivitaetsfenster. Die Sammel-
    statistik beantwortet das fuer das ganze Projekt in vier Anfragen.

    Rueckgabe: Anzahl der Ressourcen, bei denen wir Inhalt verlieren.
    """
    params = {"filter[project]": f"o:{org}:p:{projekt}",
              "filter[language]": "l:de"}
    zeilen, fehlen = [], []
    tx_summe = repo_summe = 0

    for e in seiten("/resource_language_stats", params, tok):
        slug = slug_von(e)
        if not slug:
            continue
        attr = e.get("attributes") or {}
        tx_ue = attr.get("translated_strings") or 0
        tx_ges = attr.get("total_strings") or 0
        datei = karte.get(slug)
        if datei is None:
            fehlen.append((slug, tx_ue))
            continue
        repo_ue, repo_ges = po_uebersetzt(wurzel / datei)
        tx_summe += tx_ue
        repo_summe += repo_ue
        if tx_ue > repo_ue:
            zeilen.append((tx_ue - repo_ue, datei, slug, tx_ue, repo_ue, tx_ges))

    zeilen.sort(reverse=True)
    verlust = sum(z[0] for z in zeilen)

    aus = []
    aus.append(f"# Bestandsvergleich {org}:{projekt} gegen das Repo\n")
    aus.append(f"- Transifex uebersetzt (gesamt): {tx_summe}")
    aus.append(f"- Repo uebersetzt (gesamt):      {repo_summe}")
    aus.append(f"- Dateien mit Verlustgefahr:     {len(zeilen)}")
    aus.append(f"- Strings, die ein Force-Push loeschen wuerde: {verlust}\n")
    if zeilen:
        aus.append("| Datei | Ressource | TX | Repo | Verlust | TX gesamt |")
        aus.append("|---|---|---:|---:|---:|---:|")
        for d, datei, slug, tx_ue, repo_ue, tx_ges in zeilen:
            aus.append(f"| `{datei}` | {slug} | {tx_ue} | {repo_ue} | **{d}** | {tx_ges} |")
    else:
        aus.append("Keine Datei, bei der Transifex mehr hat als wir.")
    if fehlen:
        aus.append(f"\nNicht in .tx/config ({len(fehlen)}): "
                   + ", ".join(f"{s} ({n})" for s, n in sorted(fehlen)[:20]))

    text = "\n".join(aus)
    if ziel:
        Path(ziel).write_text(text + "\n", encoding="utf-8")
        print(f"Bericht geschrieben: {ziel}", file=sys.stderr)
    else:
        print(text)
    return len(zeilen)


def probe_anfrage(was: str, pfad: str, params: dict, tok: str) -> tuple:
    """Eine einzelne Seite holen und beschreiben, ohne zu blaettern."""
    url = API + pfad + ("?" + urllib.parse.urlencode(params) if params else "")
    print(f"\n{was}", file=sys.stderr)
    try:
        daten, kopf = hole(url, tok)
    except RuntimeError as fehler:
        print(f"  -> geht nicht: {str(fehler).splitlines()[0]}", file=sys.stderr)
        return None, {}
    menge = daten.get("data", [])
    weiter = bool((daten.get("links") or {}).get("next"))
    print(f"  -> {len(menge)} Eintraege auf Seite 1, weitere Seiten: "
          f"{'ja' if weiter else 'nein'}", file=sys.stderr)
    grenze(kopf)
    return daten, kopf


def probe(org: str, projekt: str, karte: dict, seit: str, tok: str) -> None:
    """Die offenen Fragen zur API mit wenigen Anfragen klaeren.

    Der vollstaendige Lauf stellt 553 Anfragen und bricht nach 20 Minuten
    ab. Vorher muessen drei Dinge feststehen: wirkt der Datumsfilter
    ueberhaupt, wie viel davon ist unsere eigene Hochladung, und laesst
    sich die Vorauswahl billiger beantworten als Ressource fuer Ressource.
    """
    slug = sorted(karte)[0]
    res_id = f"o:{org}:p:{projekt}:r:{slug}"
    basis = {"filter[resource]": res_id, "filter[language]": "l:de"}

    print(f"Probelauf gegen {org}:{projekt}, Ressource {slug}, seit {seit}",
          file=sys.stderr)

    # 1. Wirkt der Datumsfilter, oder wird er stillschweigend ignoriert?
    mit, filt = None, None
    for name in DATUMSFILTER:
        mit, _ = probe_anfrage(f"1. Uebersetzungen MIT Datumsfilter ({name})",
                               "/resource_translations", {**basis, name: seit}, tok)
        if mit is not None:
            filt = name
            break
    if filt is None:
        print("\n  Kein Datumsfilter wird angenommen -- es muss clientseitig"
              "\n  gefiltert werden, und das heisst: ganzer Katalog.", file=sys.stderr)

    ohne, _ = probe_anfrage(
        "2. Dieselbe Ressource OHNE Datumsfilter (zum Vergleich)",
        "/resource_translations", basis, tok)

    if mit is not None and ohne is not None:
        a, b = len(mit.get("data", [])), len(ohne.get("data", []))
        weiter_mit = bool((mit.get("links") or {}).get("next"))
        weiter_ohne = bool((ohne.get("links") or {}).get("next"))
        if a == b and weiter_mit == weiter_ohne:
            print("\n  ACHTUNG: beide Abfragen liefern gleich viel. Der Filter"
                  "\n  wirkt vermutlich nicht -- dann laedt der Lauf den ganzen"
                  "\n  Katalog und das erklaert die Drosselung.", file=sys.stderr)
        else:
            print(f"\n  Der Filter wirkt: {a} statt {b} Eintraege auf Seite 1.",
                  file=sys.stderr)

    # 2. Woher kommen die Eintraege im Zeitraum? Eine eigene Hochladung
    #    stempelt jeden String neu und blaeht die Antwort auf.
    if mit and mit.get("data"):
        herkuenfte, juengste, aelteste = {}, "", "9"
        for e in mit["data"]:
            q = herkunft(e)
            herkuenfte[q] = herkuenfte.get(q, 0) + 1
            z = zeitpunkt(e)
            if z:
                juengste, aelteste = max(juengste, z), min(aelteste, z)
        print("\n3. Herkunft auf Seite 1: "
              + ", ".join(f"{k}={v}" for k, v in sorted(herkuenfte.items()))
              + f"\n   Zeitraum: {aelteste[:16]} bis {juengste[:16]}",
              file=sys.stderr)
        print("   Felder des ersten Datensatzes: "
              + ", ".join(sorted((mit["data"][0].get("attributes") or {}))),
              file=sys.stderr)

    # 3. Laesst sich serverseitig auf Handarbeit einschraenken? Das waere
    #    der groesste Hebel: eine eigene Hochladung stempelt jeden String
    #    neu, und die werfen wir bisher erst nach dem Laden weg.
    #    Ohne Datumsfilter, damit eine Vergleichsmenge da ist -- gegen eine
    #    leere Antwort laesst sich nicht pruefen, ob der Filter wirkt.
    nur_editor, _ = probe_anfrage(
        "4. Laesst sich nach Herkunft filtern (filter[origin]=EDITOR,"
        " ohne Datumsfilter)?", "/resource_translations",
        {**basis, "filter[origin]": "EDITOR"}, tok)
    if nur_editor is not None and ohne is not None:
        a, b = len(nur_editor.get("data", [])), len(ohne.get("data", []))
        if a == b:
            print("   -> gleich viele wie ungefiltert: der Filter wirkt nicht.",
                  file=sys.stderr)
        else:
            print(f"   -> wirkt: {a} statt {b}. Damit liesse sich die eigene"
                  "\n      Hochladung serverseitig ausblenden.", file=sys.stderr)

    # 4. Die Vorauswahl in wenigen Anfragen statt in 553.
    stats, _ = probe_anfrage(
        "5. Sammelstatistik je Ressource (resource_language_stats)",
        "/resource_language_stats",
        {"filter[project]": f"o:{org}:p:{projekt}", "filter[language]": "l:de"}, tok)
    if stats and stats.get("data"):
        attr = stats["data"][0].get("attributes") or {}
        print("   Felder: " + ", ".join(sorted(attr)), file=sys.stderr)
        zeitfelder = [k for k in attr if "last" in k or "update" in k]
        if zeitfelder:
            feld = zeitfelder[0]
            frisch = [e for e in stats["data"]
                      if ((e.get("attributes") or {}).get(feld) or "") >= seit]
            print(f"   Mit {feld} >= {seit[:10]}: {len(frisch)} von "
                  f"{len(stats['data'])} auf Seite 1", file=sys.stderr)

    print("\nProbelauf fertig -- es wurde nichts veraendert.", file=sys.stderr)


def main() -> None:
    global TAKT
    p = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("--tage", type=int, default=7, help="Zeitraum rueckwaerts (Vorgabe 7)")
    p.add_argument("--projekt", help="Projekt-Slug, sonst aus .tx/config")
    p.add_argument("--organisation", default=None, help="sonst aus .tx/config")
    p.add_argument("--ohne-upload", action="store_true", dest="ohne_upload",
                   help="Herkunft UPLOAD und API ausblenden (eigene Pushes)")
    p.add_argument("--csv", metavar="DATEI", help="Ergebnis zusaetzlich als CSV")
    p.add_argument("--pause", type=float, default=TAKT, metavar="SEK",
                   help=f"Mindestabstand zwischen Anfragen (Vorgabe {TAKT})")
    p.add_argument("--ab", metavar="SLUG",
                   help="erst ab diesem Ressourcen-Slug beginnen (Fortsetzung)")
    p.add_argument("--alle", action="store_true",
                   help="ohne Vorauswahl jede Ressource einzeln abfragen (langsam)")
    p.add_argument("--probe", action="store_true",
                   help="nur wenige Anfragen stellen und die API-Form pruefen")
    p.add_argument("--bestand", action="store_true",
                   help="je Datei Transifex-Bestand gegen Repo-Stand stellen "
                        "(zeigt, was ein Force-Push loeschen wuerde)")
    p.add_argument("--herkunft", metavar="BERICHT",
                   help="Ressourcen aus einem --bestand-Bericht auf Handarbeit "
                        "(EDITOR) pruefen; belegt den Filter erst an "
                        + KONTROLLE)
    p.add_argument("-o", "--out", metavar="DATEI",
                   help="Bericht in eine Datei schreiben statt auf die Konsole")
    p.add_argument("--debug", action="store_true",
                   help="ersten Rohdatensatz ausgeben (Feldnamen pruefen)")
    args = p.parse_args()

    TAKT = max(0.0, args.pause)

    tok = token()
    wurzel = repo_wurzel()
    karte, org, projekt = ressourcen(wurzel)
    org = args.organisation or org
    projekt = args.projekt or projekt
    branch = git("branch", "--show-current").strip()

    seit = (datetime.now(timezone.utc) - timedelta(days=args.tage)).strftime(
        "%Y-%m-%dT%H:%M:%SZ")

    if args.probe:
        probe(org, projekt, karte, seit, tok)
        return

    if args.herkunft:
        bericht = Path(args.herkunft)
        if not bericht.is_file():
            sys.exit(f"{bericht} nicht gefunden -- erst --bestand laufen lassen.")
        slugs = slugs_aus_bericht(bericht)
        if not slugs:
            sys.exit(f"In {bericht} stehen keine Ressourcen.")
        print(f"{len(slugs)} Ressourcen aus {bericht}, Takt {TAKT}s",
              file=sys.stderr)
        betroffen = herkunft_pruefen(org, projekt, slugs, tok, args.out)
        sys.exit(1 if betroffen else 0)

    if args.bestand:
        print(f"Bestandsvergleich {org}:{projekt} (Branch {branch}) ...",
              file=sys.stderr)
        betroffen = bestand(org, projekt, karte, wurzel, tok, args.out)
        # Rueckgabewert 1, damit ein Skript den Push anhalten kann
        sys.exit(1 if betroffen else 0)

    print(f"Projekt {org}:{projekt} (Branch {branch}), Sprache de, seit {seit}",
          file=sys.stderr)

    liste = sorted(karte)
    if not args.alle:
        frisch = frische_ressourcen(org, projekt, seit, tok)
        unbekannt = frisch - set(karte)
        if unbekannt:
            print(f"  Hinweis: {len(unbekannt)} Ressource(n) in Transifex stehen"
                  f" nicht in .tx/config, z. B. {sorted(unbekannt)[0]}",
                  file=sys.stderr)
        liste = [s for s in liste if s in frisch]
        if not liste:
            print(f"\nKeine Aenderungen in den letzten {args.tage} Tagen gefunden.")
            return

    if args.ab:
        if args.ab not in liste:
            sys.exit(f"Ressource {args.ab} ist nicht in der Auswahl.")
        liste = liste[liste.index(args.ab):]

    print(f"{len(liste)} Ressourcen werden einzeln geprueft, Takt {TAKT}s ...",
          file=sys.stderr)

    filtername = list(DATUMSFILTER)
    gemeldet = False
    zeilen = []

    for nr, slug in enumerate(liste, 1):
        if nr % 50 == 0:
            print(f"  {nr}/{len(liste)} ({slug})", file=sys.stderr)
        res_id = f"o:{org}:p:{projekt}:r:{slug}"
        try:
            eintraege, benutzt = uebersetzungen(res_id, seit, tok, filtername)
        except Gedrosselt as fehler:
            # Das Gateway drosselt uns. Weiterlaufen heisst 500-mal gegen
            # dieselbe Wand fahren und die Drosselung verlaengern.
            print(f"\n!! Abbruch bei {slug}: {str(fehler).splitlines()[0]}\n"
                  f"   Transifex drosselt die Anfragen. Bisheriges Ergebnis\n"
                  f"   bleibt erhalten. Spaeter fortsetzen mit:\n"
                  f"     --ab {slug} --pause {max(1.0, TAKT * 3):.1f}",
                  file=sys.stderr)
            break
        except RuntimeError as fehler:
            print(f"  !! {slug}: {fehler}", file=sys.stderr)
            continue

        if not gemeldet and eintraege:
            print(f"  Datumsfilter: {benutzt or 'clientseitig (langsam)'}",
                  file=sys.stderr)
            gemeldet = True

        if args.debug and eintraege:
            print(json.dumps(eintraege[0], indent=2, ensure_ascii=False))
            return

        gruppen = {}
        for e in eintraege:
            quelle = herkunft(e)
            if args.ohne_upload and quelle in MASCHINELL:
                continue
            schluessel = (benutzer(e), quelle)
            anzahl, letzte = gruppen.get(schluessel, (0, ""))
            gruppen[schluessel] = (anzahl + 1, max(letzte, zeitpunkt(e)))

        for (wer, quelle), (anzahl, letzte) in sorted(gruppen.items()):
            zeilen.append((karte[slug] or slug, slug, wer, quelle, anzahl,
                           letzte[:16].replace("T", " ")))

    if not zeilen:
        print(f"\nKeine Aenderungen in den letzten {args.tage} Tagen gefunden.")
        return

    zeilen.sort(key=lambda z: (z[5], z[0]), reverse=True)

    print("\n| Datei | Ressource | Benutzer | Herkunft | Strings | zuletzt (UTC) |")
    print("|---|---|---|---|---:|---|")
    for datei, slug, wer, quelle, anzahl, wann in zeilen:
        print(f"| `{datei}` | {slug} | {wer} | {quelle} | {anzahl} | {wann} |")

    print("\n### Nach Benutzer\n")
    print("| Benutzer | Dateien | Strings |")
    print("|---|---:|---:|")
    proWer = {}
    for datei, _slug, wer, _quelle, anzahl, _wann in zeilen:
        dateien, strings = proWer.get(wer, (set(), 0))
        dateien.add(datei)
        proWer[wer] = (dateien, strings + anzahl)
    for wer, (dateien, strings) in sorted(proWer.items(), key=lambda x: -x[1][1]):
        print(f"| {wer} | {len(dateien)} | {strings} |")

    print("\n### Nach Herkunft\n")
    print("| Herkunft | Strings |")
    print("|---|---:|")
    proQuelle = {}
    for _datei, _slug, _wer, quelle, anzahl, _wann in zeilen:
        proQuelle[quelle] = proQuelle.get(quelle, 0) + anzahl
    for quelle, anzahl in sorted(proQuelle.items(), key=lambda x: -x[1]):
        print(f"| {quelle} | {anzahl} |")

    if args.csv:
        with open(args.csv, "w", newline="", encoding="utf-8") as f:
            w = csv.writer(f)
            w.writerow(["datei", "ressource", "benutzer", "herkunft", "strings",
                        "zuletzt_utc"])
            w.writerows(zeilen)
        print(f"\nCSV geschrieben: {args.csv}", file=sys.stderr)


if __name__ == "__main__":
    main()