fix(rawlogs): .gz-Member im Tarball zweimal entpacken

Die .gz-Dateien INNERHALB eines Backfill-Tarballs sind nochmals
gzip-komprimiert; tarfile entpackt nur die aeussere Ebene. Wer den
Member-Inhalt direkt dekodiert, liest gzip-Magic als Text und verliert
jede Zeile dieser Dateien — real 68.975 von 404.825 (17 %).

- .gz-Member werden ein zweites Mal entpackt, die Zahl wird gemeldet
- --limit-lines greift jetzt auch im Tarball-Zweig
- Beispiele unlesbarer Zeilen zeigen den Dateinamen

scripts/raw-log-collect.py

@@ -117,11 +117,20 @@ def collect_from_dir(root: str):
                 yield os.path.join(dirpath, fn)
 
 
-def collect_from_tarball(tar_path: str):
+def collect_from_tarball(tar_path: str, stats: Counter | None = None):
     """Rohlogs aus einem Archiv-Tarball (Pfade unter raw/).
 
     Die Tarballs des Backfills enthalten raw/<absoluter-pfad-ohne-fuehrenden-slash>.
+
+    FALLE (real, 17 % Datenverlust): die `.gz`-Dateien IM Tarball sind selbst
+    noch gzip-komprimiert. `tarfile` entpackt nur die aeussere Ebene — wer den
+    Member-Inhalt direkt dekodiert, liest das gzip-Magic (\x1f\x8b) als Text und
+    verliert jede Zeile dieser Dateien. Gemessen am x14-Tarball: 68.975 von
+    404.825 Zeilen (17 %) fielen so aus, verteilt auf genau die `.gz`-Member
+    (access.log.2/.3/.4.gz, error.log.2/.3/.4.gz). Deshalb: Member mit `.gz`
+    ein zweites Mal entpacken.
     """
+    gz_ok = gz_fail = 0
     with tarfile.open(tar_path, "r:gz") as tf:
         for member in tf.getmembers():
             if not member.isfile():
@@ -136,8 +145,22 @@ def collect_from_tarball(tar_path: str):
             if fh is None:
                 continue
             logpath = "/" + name.split("raw/", 1)[1] if "raw/" in name else name
-            for raw in fh:
-                yield logpath, raw.decode("utf-8", "replace").rstrip("\n")
+            data = fh.read()
+            text = None
+            if name.endswith(".gz"):
+                try:
+                    text = gzip.decompress(data).decode("utf-8", "replace")
+                    gz_ok += 1
+                except Exception:                              # noqa: BLE001
+                    gz_fail += 1
+                    text = data.decode("utf-8", "replace")
+            else:
+                text = data.decode("utf-8", "replace")
+            for raw in text.splitlines():
+                yield logpath, raw
+    if stats is not None:
+        stats["gz_member_ok"] += gz_ok
+        stats["gz_member_fail"] += gz_fail
 
 
 def build_labels(node: str, logpath: str, imported: str) -> dict:
@@ -243,12 +266,17 @@ def main() -> int:
         # Tarballs zuletzt (sie liegen historisch VOR den Live-Dateien).
         for tb in args.include_tarball:
             print(f"[collect] Tarball: {tb}")
-            for logpath, raw in collect_from_tarball(tb):
+            for logpath, raw in collect_from_tarball(tb, stats):
+                if args.limit_lines and n_ok >= args.limit_lines:
+                    break
                 if not raw.strip():
                     continue
                 ts = parse_ts(raw)
                 if ts is None:
                     n_nots += 1
+                    if len(examples) < 5:
+                        examples.append(f"[{os.path.basename(logpath)}] "
+                                        f"{raw[:110]}")
                     continue
                 if since_ns and ts < since_ns:
                     n_skip += 1
@@ -267,6 +295,11 @@ def main() -> int:
 
     print(f"[collect] {n_ok} Zeilen geschrieben  (access={stats['access']} "
           f"error={stats['error']})")
+    if stats["gz_member_ok"] or stats["gz_member_fail"]:
+        print(f"[collect] gzip-Member im Tarball entpackt: "
+              f"{stats['gz_member_ok']}"
+              + (f", FEHLGESCHLAGEN: {stats['gz_member_fail']}"
+                 if stats["gz_member_fail"] else ""))
     if n_nots:
         print(f"[collect] WARNUNG: {n_nots} Zeilen OHNE erkennbaren Zeitstempel "
               f"wurden uebersprungen (kein stilles Verwerfen):")