#!/usr/bin/env python3 """ Build static HTML for batch job status (optional; primary UI is React). React app: Data explorer → Batch jobs (`/data-explorer/batch-jobs`) via `GET /api/batch-jobs`. Usage (repo root): .venv/bin/python packages/scrapers/src/scrapers/youtube/batch_job_dashboard.py --build .venv/bin/python packages/scrapers/src/scrapers/youtube/batch_job_dashboard.py --build --open .venv/bin/python packages/scrapers/src/scrapers/youtube/batch_job_dashboard.py --serve """ from __future__ import annotations import argparse import json import sys import webbrowser from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any, Dict, List, Literal VideoPayloadMode = Literal["none", "failed_only", "running", "all"] _FAILED_VIDEO_STATUSES = frozenset( { "fail", "failed", "tombstoned", "empty", "rate_limit", "error", } ) def _is_failed_video_status(status: str) -> bool: s = (status or "").strip().lower() if not s or s in ("ok", "pending", "skipped", "noop"): return False if s in _FAILED_VIDEO_STATUSES: return True return s != "ok" def slim_jurisdiction_videos( j: Dict[str, Any], *, video_mode: VideoPayloadMode, ) -> Dict[str, Any]: """Drop per-video rows when counts in ``stats`` are enough for the table.""" if video_mode == "all": return j out = dict(j) status = (out.get("status") or "").strip().lower() if video_mode == "running" and ( status == "running" or (out.get("current_video_id") or "").strip() ): return out videos = list(out.get("videos") or []) if video_mode == "failed_only": out["videos"] = [v for v in videos if _is_failed_video_status(v.get("status", ""))] else: out["videos"] = [] return out def slim_batch_dict( batch: Dict[str, Any], *, video_mode: VideoPayloadMode, ) -> Dict[str, Any]: out = dict(batch) jurs = batch.get("jurisdictions") or [] if not jurs: return out batch_running = (batch.get("status") or "").lower() == "running" mode: VideoPayloadMode = video_mode if video_mode == "running" and not batch_running: mode = "failed_only" out["jurisdictions"] = [ slim_jurisdiction_videos(j, video_mode=mode) if isinstance(j, dict) else j for j in jurs ] return out def _totals_from_batch_summaries(batches: List[Dict[str, Any]]) -> Dict[str, Any]: totals = { "batches": 0, "running": 0, "states": 0, "states_planned": 0, "states_started": 0, "states_completed": 0, "processed_jurisdictions": 0, "failed_jurisdictions": 0, "remaining_jurisdictions": 0, "videos_ok": 0, "videos_fail": 0, "videos_attempted": 0, "files_transcripts": 0, "files_transcripts_disk": 0, "transcript_hours": 0.0, "bronze_download_rows": 0, "files_analysis": 0, "files_reports": 0, "files_analysis_recent": 0, "files_reports_recent": 0, } planned_states: set[str] = set() started_states: set[str] = set() completed_states: set[str] = set() transcript_seconds = 0.0 from api.batch_jobs.batch_job_status import config_state_codes for d in batches: totals["batches"] += 1 if d.get("status") == "running": totals["running"] += 1 s = d.get("summary") or {} for st in config_state_codes(d.get("config") or {}): planned_states.add(st) for st in s.get("states_started_codes") or []: started_states.add(str(st).upper()) for st in s.get("states_completed_codes") or []: completed_states.add(str(st).upper()) totals["processed_jurisdictions"] += int(s.get("processed_jurisdictions") or 0) totals["failed_jurisdictions"] += int(s.get("failed_jurisdictions") or 0) totals["remaining_jurisdictions"] += int(s.get("remaining_jurisdictions") or 0) totals["videos_ok"] += int(s.get("videos_ok") or 0) totals["videos_fail"] += int(s.get("videos_fail") or 0) attempted = int(s.get("videos_attempted") or s.get("files_processed") or 0) if not attempted: attempted = ( int(s.get("videos_ok") or 0) + int(s.get("videos_fail") or 0) + int(s.get("videos_tombstoned") or 0) + int(s.get("videos_empty") or 0) + int(s.get("videos_rate_limit") or 0) ) totals["videos_attempted"] += attempted totals["files_transcripts"] += int(s.get("files_transcripts") or 0) totals["files_transcripts_disk"] += int(s.get("files_transcripts_disk") or 0) totals["bronze_download_rows"] += int(s.get("bronze_download_rows") or 0) totals["files_analysis"] += int(s.get("files_analysis") or 0) totals["files_reports"] += int(s.get("files_reports") or 0) totals["files_analysis_recent"] += int(s.get("files_analysis_recent") or 0) totals["files_reports_recent"] += int(s.get("files_reports_recent") or 0) transcript_seconds += float(s.get("transcript_seconds") or 0) totals["states_planned"] = len(planned_states) totals["states_started"] = len(started_states) totals["states_completed"] = len(completed_states) totals["states"] = totals["states_planned"] totals["transcript_hours"] = round(transcript_seconds / 3600.0, 2) return totals def _max_iso_timestamp(best: str, candidate: str) -> str: raw = (candidate or "").strip() if raw and (not best or raw > best): return raw return best def pipeline_activity_at_from_batches(batches: List[Dict[str, Any]]) -> str: """ Latest jurisdiction/video progress timestamp (not batch ``updated_at`` sync metadata). Includes ``summary.current_video_started_at`` (same clock as the Current file card) when the slim running snapshot has no jurisdiction rows. """ best = "" for b in batches: jurs = b.get("jurisdictions") or [] for j in jurs: if not isinstance(j, dict): continue for key in ( "updated_at", "current_video_started_at", "finished_at", "started_at", ): best = _max_iso_timestamp(best, j.get(key) or "") for v in j.get("videos") or []: if isinstance(v, dict): best = _max_iso_timestamp(best, v.get("finished_at") or "") summary = b.get("summary") or {} if isinstance(summary, dict): for key in ( "current_video_started_at", "current_jurisdiction_finished_at", ): best = _max_iso_timestamp(best, summary.get(key) or "") for key in ("finished_at", "started_at"): best = _max_iso_timestamp(best, b.get(key) or "") return best def build_dashboard_summary(*, limit: int = 30) -> Dict[str, Any]: """Fast path: batch list + totals from ``summary`` JSON only (no payload parse).""" import datetime as _dt import os use_db = os.getenv("BATCH_JOBS_USE_DB", "1").strip().lower() not in ( "0", "false", "no", "off", ) batches: List[Dict[str, Any]] = [] source = "files" sql_totals: Dict[str, Any] | None = None if use_db: try: from api.batch_jobs.batch_job_db import ( aggregate_dashboard_totals_from_db, list_batch_job_meta_from_db, running_batch_activity_from_db, ) # Totals first: it reaps stale ``running`` rows, so the meta read # below sees the corrected per-batch statuses in this same snapshot. reaped_totals = aggregate_dashboard_totals_from_db(limit=limit) batches = list_batch_job_meta_from_db(limit=limit) if batches: source = "database" sql_totals = reaped_totals running = running_batch_activity_from_db() if running: for b in batches: if b.get("batch_id") == running.get("batch_id"): b["jurisdictions"] = running.get("jurisdictions") or [] break except Exception: batches = [] if not batches: jobs = list_batches(limit=limit) batches = [ { "batch_id": j.batch_id, "step": j.step, "status": j.status, "started_at": j.started_at, "updated_at": j.updated_at, "finished_at": j.finished_at, "config": dict(j.config or {}), "summary": dict(j.summary or {}), "jurisdictions": [], } for j in jobs ] source = "files" totals = _totals_from_batch_summaries(batches) if sql_totals: totals.update( { k: sql_totals[k] for k in ( "batches", "running", "processed_jurisdictions", "failed_jurisdictions", "remaining_jurisdictions", "videos_ok", "videos_fail", "files_transcripts", "files_transcripts_disk", "bronze_download_rows", "files_analysis", "files_reports", "files_analysis_recent", "files_reports_recent", "files_analysis_errors_recent", "files_reports_errors_recent", "last_transcript_at", "last_analysis_at", "last_report_at", "transcript_hours", ) if k in sql_totals } ) last_activity = pipeline_activity_at_from_batches(batches) if not last_activity and sql_totals.get("last_activity_at"): last_activity = sql_totals["last_activity_at"] else: last_activity = pipeline_activity_at_from_batches(batches) stage_report: Dict[str, Any] = {"states": [], "rows": []} if use_db: from api.batch_jobs.batch_job_db import dashboard_stage_report stage_report = dashboard_stage_report() now = _dt.datetime.now(_dt.timezone.utc) return { "generated_at": now.isoformat(), "last_activity_at": last_activity, "totals": totals, "batches": batches, "stage_report": stage_report, "source": source, "detail": "summary", } def build_batch_state_jurisdictions(*, batch_id: str, state_code: str) -> List[Dict[str, Any]]: """ Slim jurisdiction rows for one state (no per-video arrays). Uses JSONB extraction for finished batches; single-state plan merge for running. """ from api.batch_jobs.batch_job_db import ( list_jurisdiction_rows_from_db, load_batch_job_from_db, ) from api.batch_jobs.batch_job_status import ( apply_batch_lifecycle, expand_batch_job_plan, fetch_batch_plan_jurisdictions_cached, needs_plan_expand, ) st = (state_code or "").strip().upper() if not st: return [] job = load_batch_job_from_db(batch_id) if not job: return [] status = (job.status or "").lower() if status == "running" and needs_plan_expand(job): rr = job.config.get("round_robin") if job.config else None if rr is None: rr = True plan = fetch_batch_plan_jurisdictions_cached([st], round_robin=bool(rr)) expand_batch_job_plan(job, plan=plan) apply_batch_lifecycle(job) full = job.to_dict() return [ slim_jurisdiction_videos(j, video_mode="none") for j in full.get("jurisdictions") or [] if (j.get("state_code") or "").upper() == st ] rows = list_jurisdiction_rows_from_db(batch_id, st) if rows: return rows apply_batch_lifecycle(job) full = job.to_dict() return [ slim_jurisdiction_videos(j, video_mode="none") for j in full.get("jurisdictions") or [] if (j.get("state_code") or "").upper() == st ] _REPO_ROOT = Path(__file__).resolve().parents[3] if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) from api.batch_jobs.batch_job_status import ( _REPO_ROOT, BatchJob, _recompute_summary, apply_batch_lifecycle, count_policy_files_for_jurisdiction, expand_batch_job_plan, jobs_dir, latest_dashboard_activity_at, list_batches, ) _DASHBOARD_NAME = "dashboard.html" _POLICY_CACHE = _REPO_ROOT / "data" / "cache" / "gemini_transcript_policy" def _fmt_duration(seconds: Any) -> str: if seconds is None: return "—" try: total = int(float(seconds)) except (TypeError, ValueError): return "—" if total < 60: return f"{total}s" m, s = divmod(total, 60) if m < 60: return f"{m}m {s}s" h, m = divmod(m, 60) return f"{h}h {m}m" def _enrich_file_counts(job: BatchJob) -> BatchJob: from api.batch_jobs.batch_job_status import policy_disk_file_counts for j in job.jurisdictions: scanned = count_policy_files_for_jurisdiction( _POLICY_CACHE, state_code=j.state_code, jurisdiction_id=j.jurisdiction_id, ) j.file_counts = dict(j.file_counts or {}) j.file_counts.update(policy_disk_file_counts(scanned)) return job def _aggregate_jobs( jobs: List[BatchJob], *, enrich_transcript_from_bronze: bool = False, ) -> Dict[str, Any]: batches: List[Dict[str, Any]] = [] totals = { "batches": 0, "running": 0, "states": 0, "states_planned": 0, "states_started": 0, "states_completed": 0, "processed_jurisdictions": 0, "failed_jurisdictions": 0, "remaining_jurisdictions": 0, "videos_ok": 0, "videos_fail": 0, "videos_attempted": 0, "files_transcripts": 0, "files_transcripts_disk": 0, "transcript_hours": 0.0, "bronze_download_rows": 0, "files_analysis": 0, "files_reports": 0, "files_analysis_recent": 0, "files_reports_recent": 0, } planned_states: set[str] = set() started_states: set[str] = set() completed_states: set[str] = set() transcript_seconds = 0.0 bronze_conn = None if enrich_transcript_from_bronze: try: from api.batch_jobs.batch_job_db import ( enrich_transcript_seconds_from_bronze, get_db_connection, ) bronze_conn = get_db_connection() except Exception: bronze_conn = None try: from api.batch_jobs.batch_job_status import ( _batch_plan_cache_key, _recompute_summary, config_state_codes, expand_batch_job_plan, fetch_batch_plan_jurisdictions_cached, needs_plan_expand, normalize_batch_job_jurisdictions, persist_batch_job, ) plan_by_states: Dict[str, list] = {} import logging _log = logging.getLogger(__name__) for job in jobs: try: status_before = job.status if needs_plan_expand(job): states = config_state_codes(job.config or {}) rr = job.config.get("round_robin") if job.config else None if rr is None: rr = True plan_key = _batch_plan_cache_key(states, round_robin=bool(rr)) if plan_key and plan_key not in plan_by_states: plan_by_states[plan_key] = fetch_batch_plan_jurisdictions_cached( states, round_robin=bool(rr) ) expand_batch_job_plan( job, plan=plan_by_states.get(plan_key) if plan_key else None ) if (job.status or "").lower() == "running": normalize_batch_job_jurisdictions(job) apply_batch_lifecycle(job) _recompute_summary(job) if status_before != job.status: try: persist_batch_job(job) except Exception as exc: _log.warning( "persist batch %s failed: %s", job.batch_id, exc ) if bronze_conn is not None: try: enrich_transcript_seconds_from_bronze(bronze_conn, job) except Exception: pass d = job.to_dict() except Exception as exc: _log.exception("batch %s skipped in dashboard: %s", job.batch_id, exc) d = job.to_dict() batches.append(d) s = d.get("summary") or {} totals["batches"] += 1 if d.get("status") == "running": totals["running"] += 1 for st in config_state_codes(job.config or {}): planned_states.add(st) for st in s.get("states_started_codes") or []: started_states.add(str(st).upper()) for st in s.get("states_completed_codes") or []: completed_states.add(str(st).upper()) totals["processed_jurisdictions"] += int(s.get("processed_jurisdictions") or 0) totals["failed_jurisdictions"] += int(s.get("failed_jurisdictions") or 0) totals["remaining_jurisdictions"] += int(s.get("remaining_jurisdictions") or 0) totals["videos_ok"] += int(s.get("videos_ok") or 0) totals["videos_fail"] += int(s.get("videos_fail") or 0) attempted = int(s.get("videos_attempted") or s.get("files_processed") or 0) if not attempted: attempted = ( int(s.get("videos_ok") or 0) + int(s.get("videos_fail") or 0) + int(s.get("videos_tombstoned") or 0) + int(s.get("videos_empty") or 0) + int(s.get("videos_rate_limit") or 0) ) totals["videos_attempted"] += attempted totals["files_transcripts"] += int(s.get("files_transcripts") or 0) totals["files_transcripts_disk"] += int(s.get("files_transcripts_disk") or 0) totals["bronze_download_rows"] += int(s.get("bronze_download_rows") or 0) totals["files_analysis"] += int(s.get("files_analysis") or 0) totals["files_reports"] += int(s.get("files_reports") or 0) totals["files_analysis_recent"] += int(s.get("files_analysis_recent") or 0) totals["files_reports_recent"] += int(s.get("files_reports_recent") or 0) transcript_seconds += float(s.get("transcript_seconds") or 0) finally: if bronze_conn is not None: try: bronze_conn.close() except Exception: pass totals["states_planned"] = len(planned_states) totals["states_started"] = len(started_states) totals["states_completed"] = len(completed_states) totals["states"] = totals["states_planned"] totals["transcript_hours"] = round(transcript_seconds / 3600.0, 2) import datetime as _dt from api.batch_jobs.batch_job_db import dashboard_stage_report now = _dt.datetime.now(_dt.timezone.utc) return { "generated_at": now.isoformat(), "last_activity_at": latest_dashboard_activity_at(jobs), "totals": totals, "batches": batches, "stage_report": dashboard_stage_report(), "source": "database", "detail": "full", } def _override_recent_counts_from_events(payload: Dict[str, Any]) -> None: """ Replace the batch-scoped disk-scan counters with the live per-event bronze stamps (migration 083): the 24h throughput, the all-time pipeline totals (so progress % is de-duplicated and current), and the per-step "ago" timestamps. Best-effort: leaves the disk-scan values in place if the DB/columns are missing. """ totals = payload.get("totals") if not isinstance(totals, dict): return try: from api.batch_jobs.batch_job_db import ( get_db_connection, policy_event_counts_24h, ) with get_db_connection() as conn: counts = policy_event_counts_24h(conn) totals["files_analysis_recent"] = int(counts.get("analysis") or 0) totals["files_reports_recent"] = int(counts.get("reports") or 0) totals["files_analysis_errors_recent"] = int(counts.get("analysis_errors") or 0) totals["files_reports_errors_recent"] = int(counts.get("reports_errors") or 0) transcripts_total = int(counts.get("transcripts_total") or 0) analysis_total = int(counts.get("analysis_total") or 0) reports_total = int(counts.get("reports_total") or 0) if transcripts_total > 0: totals["files_transcripts_disk"] = transcripts_total if analysis_total > 0: totals["files_analysis"] = analysis_total if reports_total > 0: totals["files_reports"] = reports_total totals["last_transcript_at"] = counts.get("last_transcript_at") or "" totals["last_analysis_at"] = counts.get("last_analysis_at") or "" totals["last_report_at"] = counts.get("last_report_at") or "" except Exception as exc: import logging logging.getLogger(__name__).debug( "recent-event count override skipped: %s", exc ) def build_dashboard_data( *, refresh_files: bool = False, enrich_bronze: bool = True, enrich_bronze_only_running: bool = False, batch_limit: int = 25, video_mode: VideoPayloadMode = "running", ) -> Dict[str, Any]: """ Build dashboard payload. Prefer Postgres (``bronze.youtube_batch_job_runs``); fall back to JSON files under ``data/cache/batch_jobs/``. ``refresh_files`` scans the policy cache on disk (slow). By default transcript counts come from ``bronze.bronze_event_youtube`` when ``enrich_bronze`` is true. """ import os use_db = os.getenv("BATCH_JOBS_USE_DB", "1").strip().lower() not in ( "0", "false", "no", "off", ) if use_db: try: from api.batch_jobs.batch_job_db import ( enrich_jobs_from_bronze, list_batch_jobs_from_db, sync_json_batches_to_db, ) jobs = list_batch_jobs_from_db(limit=batch_limit) if not jobs: sync_json_batches_to_db(limit=batch_limit) jobs = list_batch_jobs_from_db(limit=batch_limit) if jobs: if enrich_bronze: enrich_jobs_from_bronze( jobs, only_running_jurisdictions=True, enrich_disk=False, ) elif refresh_files: for job in jobs: _enrich_file_counts(job) payload = _aggregate_jobs( jobs, enrich_transcript_from_bronze=enrich_bronze ) _override_recent_counts_from_events(payload) if video_mode != "all": payload["batches"] = [ slim_batch_dict(b, video_mode=video_mode) for b in payload.get("batches") or [] ] payload["detail"] = "full" return payload except Exception as exc: import logging logging.getLogger(__name__).warning( "batch dashboard DB read failed, using JSON files: %s", exc ) jobs = list_batches(limit=batch_limit) if refresh_files: for job in jobs: _enrich_file_counts(job) elif enrich_bronze: try: from api.batch_jobs.batch_job_db import enrich_jobs_from_bronze enrich_jobs_from_bronze(jobs) except Exception: pass payload = _aggregate_jobs(jobs, enrich_transcript_from_bronze=enrich_bronze) payload["source"] = "files" if video_mode != "all": payload["batches"] = [ slim_batch_dict(b, video_mode=video_mode) for b in payload.get("batches") or [] ] payload["detail"] = "full" return payload def render_html(payload: Dict[str, Any]) -> str: data_json = json.dumps(payload, ensure_ascii=False) return f"""