#!/usr/bin/env python3 """Transkriptor-Failed-Bulk-Retry via CDN-Download + Groq Whisper (2026-06-06). Kontext: 806 Failed-Files bei Transkriptor (453 unique ISB = Insufficient Balance). Audio liegt noch auf deren CDN (signierte mp3-URLs aus /tmp/tk-details.jsonl). Statt Transkriptor-Quota (6000 min < 26k min Bedarf): Groq whisper-large-v3-turbo ($0.04/h). Umgeht yt-dlp-Bot-Block UND Transkriptor-URL-Mode-Bug (KAR-571). Targets: - matched YouTube-Videos ohne Transcript in ingested-DB -> direkt ingest (raw JSON + DB) - generic (episode/audio) ISB-Fails aelter als 2026-06-05 -> unmatched-Dir (Matching spaeter) (juengere macht die Nightly-Podcast-Pipeline heute Nacht selbst mit frischer Quota) Usage: python3 tk-bulk-groq-retry.py [--limit N] [--dry-run] Output: /root/aria/state/transkriptor-harvest/bulk-log.jsonl (ein JSON-Event pro File) """ from __future__ import annotations import argparse import json import re import sqlite3 import subprocess import sys import tempfile import time import urllib.request import urllib.error from datetime import datetime, timezone from pathlib import Path DB = "/root/aria/brain/youtube/.state.sqlite" RAW_BASE = Path("/root/aria/brain/youtube/raw") HARVEST = Path("/root/aria/state/transkriptor-harvest") UNMATCHED_DIR = HARVEST / "unmatched" LOG_PATH = HARVEST / "bulk-log.jsonl" DETAILS = "/tmp/tk-details.jsonl" GROQ_CAP_BYTES = 24 * 1024 * 1024 GROQ_URL = "https://api.groq.com/openai/v1/audio/transcriptions" GENERIC_CUTOFF_MS = 1780617600000 # 2026-06-05 00:00 UTC — juengere generic skippen def load_groq_key() -> str: for line in open("/root/aria/.env"): if line.startswith("GROQ_API_KEY="): return line.strip().split("=", 1)[1].strip().strip('"').strip("'") raise RuntimeError("GROQ_API_KEY nicht gefunden") def slug(t: str) -> str: return re.sub(r"[^A-Za-z0-9]+", "_", t or "").strip("_").lower() def log_event(ev: dict) -> None: ev["ts"] = datetime.now(timezone.utc).isoformat() with LOG_PATH.open("a") as f: f.write(json.dumps(ev, ensure_ascii=False) + "\n") def download(url: str, dst: Path) -> bool: try: req = urllib.request.Request(url, headers={"User-Agent": "aria-bulk-retry/1.0"}) with urllib.request.urlopen(req, timeout=600) as r, dst.open("wb") as f: while True: chunk = r.read(1 << 20) if not chunk: break f.write(chunk) return dst.stat().st_size > 10_000 except Exception as e: log_event({"event": "download_fail", "url": url[:120], "error": str(e)[:200]}) return False def split_mp3(mp3: Path, dst_dir: Path) -> list[Path]: """Segment-Copy in <24MB Chunks (kein Re-Encode, schnell).""" size = mp3.stat().st_size if size <= GROQ_CAP_BYTES: return [mp3] # Dauer + Bitrate messen -> Segmentlaenge mit 15% Sicherheitsmarge try: out = subprocess.check_output( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", str(mp3)], timeout=60) duration = float(out.decode().strip()) except Exception: return [] bytes_per_sec = size / max(duration, 1) seg_seconds = int(GROQ_CAP_BYTES / bytes_per_sec * 0.85) pattern = str(dst_dir / "chunk_%03d.mp3") r = subprocess.run( ["ffmpeg", "-y", "-i", str(mp3), "-f", "segment", "-segment_time", str(seg_seconds), "-c", "copy", pattern], capture_output=True, timeout=900) if r.returncode != 0: return [] chunks = sorted(dst_dir.glob("chunk_*.mp3")) if any(c.stat().st_size > GROQ_CAP_BYTES for c in chunks): return [] return chunks def groq_transcribe(path: Path, api_key: str, max_retries: int = 6) -> str | None: """Multipart-Upload an Groq, language auto-detect, 429-Backoff.""" audio = path.read_bytes() boundary = "----AriaBulkRetry20260606" body = ( f"--{boundary}\r\n" f'Content-Disposition: form-data; name="model"\r\n\r\n' f"whisper-large-v3-turbo\r\n" f"--{boundary}\r\n" f'Content-Disposition: form-data; name="file"; filename="{path.name}"\r\n' f"Content-Type: audio/mpeg\r\n\r\n" ).encode() + audio + f"\r\n--{boundary}--\r\n".encode() for attempt in range(max_retries): req = urllib.request.Request( GROQ_URL, data=body, method="POST", headers={"Authorization": f"Bearer {api_key}", "Content-Type": f"multipart/form-data; boundary={boundary}", "User-Agent": "aria-voice/1.0 (curl-compatible)"}) try: with urllib.request.urlopen(req, timeout=600) as resp: return json.loads(resp.read()).get("text", "").strip() except urllib.error.HTTPError as e: if e.code == 429: retry_after = int(e.headers.get("retry-after", "0") or 0) wait = max(retry_after, 15 * (attempt + 1)) log_event({"event": "groq_429", "file": path.name, "wait": wait}) time.sleep(wait) continue log_event({"event": "groq_http_error", "file": path.name, "code": e.code, "body": e.read()[:200].decode(errors="replace")}) if e.code >= 500: time.sleep(10 * (attempt + 1)) continue return None except Exception as e: # noqa: BLE001 log_event({"event": "groq_error", "file": path.name, "error": str(e)[:200]}) time.sleep(10 * (attempt + 1)) return None def build_targets() -> list[dict]: con = sqlite3.connect(DB) con.row_factory = sqlite3.Row ing = list(con.execute( "SELECT video_id, channel, title, url, raw_path, transcript_chars FROM ingested")) by_slug = {} for r in ing: s = slug(r["title"]) if s and s not in by_slug: by_slug[s] = r con.close() details = [json.loads(l) for l in open(DETAILS)] # Dedupe by Slug (bzw. order_id fuer generic), neueste zuerst (frischeste CDN-URL) seen: dict[str, dict] = {} for d in sorted(details, key=lambda x: -int(x.get("_list_created_at") or 0)): if d.get("fail_code") != "ISB" or not d.get("mp3_file"): continue fn = d.get("_list_file_name") or d.get("file_name") or "" s = slug(fn) generic = s in ("audio", "episode", "") or len(s) < 8 key = d["order_id"] if generic else s if key in seen: continue row = None if not generic: for k, r in by_slug.items(): if k.startswith(s[:40]): row = r break if row is not None and row["transcript_chars"]: continue # schon transkribiert (anderer Weg) if generic and int(d.get("_list_created_at") or 0) >= GENERIC_CUTOFF_MS: continue # Nightly-Pipeline macht die selbst seen[key] = { "order_id": d["order_id"], "file_name": fn, "mp3": d["mp3_file"], "minutes": d.get("_list_minutes") or 0, "generic": generic, "video_id": row["video_id"] if row else None, "title": row["title"] if row else None, "url": row["url"] if row else None, "channel": row["channel"] if row else None, "raw_path": row["raw_path"] if row else None, } # Kurze zuerst (schneller Fortschritt), Monster-Tutorials ans Ende return sorted(seen.values(), key=lambda t: t["minutes"]) def ingest_matched(t: dict, text: str) -> None: con = sqlite3.connect(DB) raw_path = Path(t["raw_path"]) if t["raw_path"] else \ RAW_BASE / (slug(t["channel"] or "unknown")[:40] or "unknown") / f"{t['video_id']}.json" raw_path.parent.mkdir(parents=True, exist_ok=True) data = {} if raw_path.exists(): try: data = json.loads(raw_path.read_text()) except json.JSONDecodeError: data = {} data.setdefault("video_id", t["video_id"]) data.setdefault("title", t["title"]) data.setdefault("url", t["url"]) data["transcript"] = text data["transcript_chars"] = len(text) data["transcript_filled_at"] = datetime.now(timezone.utc).isoformat() data["transcript_filled_by"] = "tk-bulk-groq-retry-20260606" raw_path.write_text(json.dumps(data, indent=2, ensure_ascii=False)) con.execute("UPDATE ingested SET transcript_chars=?, raw_path=? WHERE video_id=?", (len(text), str(raw_path), t["video_id"])) con.execute("DELETE FROM triaged WHERE video_id=? AND verdict='skip' " "AND one_line_reason LIKE '%empty_transcript%'", (t["video_id"],)) con.commit() con.close() def save_unmatched(t: dict, text: str) -> None: out = UNMATCHED_DIR / f"{t['order_id']}.json" out.write_text(json.dumps({ "order_id": t["order_id"], "file_name": t["file_name"], "minutes": t["minutes"], "transcript": text, "transcript_chars": len(text), "harvested_at": datetime.now(timezone.utc).isoformat(), "source": "tk-bulk-groq-retry", }, indent=2, ensure_ascii=False)) def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--limit", type=int, default=0) ap.add_argument("--dry-run", action="store_true") args = ap.parse_args() UNMATCHED_DIR.mkdir(parents=True, exist_ok=True) api_key = load_groq_key() targets = build_targets() # Resume: bereits erfolgreich verarbeitete order_ids skippen done_ids = set() if LOG_PATH.exists(): for line in LOG_PATH.open(): try: ev = json.loads(line) if ev.get("event") == "ok": done_ids.add(ev.get("order_id")) except json.JSONDecodeError: continue if done_ids: before = len(targets) targets = [t for t in targets if t["order_id"] not in done_ids] print(f"[bulk] resume: {before - len(targets)} bereits erledigt, skip", flush=True) if args.limit: targets = targets[: args.limit] total_min = sum(t["minutes"] for t in targets) print(f"[bulk] targets={len(targets)} total_minutes={total_min} " f"(~${total_min / 60 * 0.04:.2f} Groq)", flush=True) if args.dry_run: for t in targets[:30]: print(f" {t['minutes']:>4d}min {'GEN' if t['generic'] else 'YT '} {t['file_name'][:60]}") return 0 done = fail = 0 for n, t in enumerate(targets, 1): label = t["file_name"][:55] with tempfile.TemporaryDirectory(prefix="tk-bulk-") as tmp: tmp_dir = Path(tmp) mp3 = tmp_dir / "audio.mp3" if not download(t["mp3"], mp3): fail += 1 log_event({"event": "fail", "stage": "download", "order_id": t["order_id"], "file": label}) print(f"[{n}/{len(targets)}] DL-FAIL {label}", flush=True) continue chunks = split_mp3(mp3, tmp_dir) if not chunks: fail += 1 log_event({"event": "fail", "stage": "split", "order_id": t["order_id"], "file": label}) print(f"[{n}/{len(targets)}] SPLIT-FAIL {label}", flush=True) continue parts = [] ok = True for c in chunks: txt = groq_transcribe(c, api_key) if txt is None: ok = False break parts.append(txt) time.sleep(2) # sanftes Pacing if not ok: fail += 1 log_event({"event": "fail", "stage": "groq", "order_id": t["order_id"], "file": label, "chunks_done": len(parts), "chunks": len(chunks)}) print(f"[{n}/{len(targets)}] GROQ-FAIL {label}", flush=True) continue text = "\n".join(parts).strip() if len(text) < 100: fail += 1 log_event({"event": "fail", "stage": "empty", "order_id": t["order_id"], "file": label, "chars": len(text)}) print(f"[{n}/{len(targets)}] EMPTY {label}", flush=True) continue if t["video_id"]: ingest_matched(t, text) dest = t["video_id"] else: save_unmatched(t, text) dest = "unmatched" done += 1 log_event({"event": "ok", "order_id": t["order_id"], "file": label, "dest": dest, "chars": len(text), "minutes": t["minutes"], "chunks": len(chunks)}) print(f"[{n}/{len(targets)}] OK {label} -> {dest} ({len(text)} chars)", flush=True) print(f"[bulk] FERTIG done={done} fail={fail} von {len(targets)}", flush=True) return 0 if __name__ == "__main__": sys.exit(main())