api/batch_jobs/batch_job_status.py

#!/usr/bin/env python3
"""
Persist batch job progress for priority-state pipelines (captions, analyze, catalog).

Writes JSON under ``data/cache/batch_jobs/`` and syncs to
``bronze.youtube_batch_job_runs`` (migration 073) for the live API dashboard.

Used by ``run_priority_states_last_n.sh`` and ``backfill_jurisdiction_transcripts.py``.
View in React: ``/data-explorer/batch-jobs`` (``GET /api/batch-jobs/stream`` SSE).
"""

from __future__ import annotations

import argparse
import functools
import json
import os
import re
import sys
import time
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional

_REPO_ROOT = Path(__file__).resolve().parents[3]
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))
_DEFAULT_JOBS_DIR = _REPO_ROOT / "data" / "cache" / "batch_jobs"
_INDEX_NAME = "index.json"
_SLUG_RE = re.compile(r"[^a-zA-Z0-9._-]+")


def _utc_now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def batch_inactivity_seconds() -> float:
    """Mark ``running`` batches cancelled after this many seconds without activity."""
    raw = os.getenv("BATCH_JOB_INACTIVITY_SECONDS", "3600")
    try:
        return max(60.0, float(raw))
    except (TypeError, ValueError):
        return 3600.0


def config_state_codes(cfg: Optional[Dict[str, Any]]) -> List[str]:
    raw = (cfg or {}).get("states") or []
    if isinstance(raw, str):
        return [s.strip().upper() for s in raw.split(",") if s.strip()]
    if isinstance(raw, list):
        return [str(s).strip().upper() for s in raw if str(s).strip()]
    return []


def _parse_utc_iso(iso: str) -> Optional[datetime]:
    text = (iso or "").strip()
    if not text:
        return None
    try:
        return datetime.fromisoformat(text.replace("Z", "+00:00"))
    except ValueError:
        return None


def last_batch_activity_at(job: BatchJob) -> datetime:
    """
    Latest timestamp that indicates real pipeline work (jurisdiction / video progress).

    Does not use ``job.updated_at`` (dashboard sync metadata). When nothing is recorded,
    falls back to ``started_at`` so idle ``running`` rows can stale-cancel instead of
    looking falsely active.
    """
    best: Optional[datetime] = None
    for j in job.jurisdictions:
        for iso in (
            j.updated_at,
            j.current_video_started_at,
            j.finished_at,
            j.started_at,
        ):
            dt = _parse_utc_iso(iso or "")
            if dt and (best is None or dt > best):
                best = dt
        for v in j.videos or []:
            dt = _parse_utc_iso(v.finished_at or "")
            if dt and (best is None or dt > best):
                best = dt
    if best is not None:
        return best
    dt = _parse_utc_iso(job.started_at or "")
    if dt:
        return dt
    return datetime.fromtimestamp(0, tz=timezone.utc)


def latest_dashboard_activity_at(jobs: List[BatchJob]) -> str:
    """Latest jurisdiction/video progress across batches (not API ``generated_at``)."""
    best: Optional[datetime] = None
    for job in jobs:
        try:
            dt = last_batch_activity_at(job)
            if dt.year > 1971 and (best is None or dt > best):
                best = dt
        except Exception:
            pass
    return best.isoformat() if best is not None and best.year > 1971 else ""


def state_progress_for_job(job: BatchJob) -> Dict[str, Any]:
    """
    Per-batch state coverage from jurisdiction rows.

    * started — at least one jurisdiction in that state is running or finished
    * completed — every jurisdiction row for that state is finished (none pending/running)
    """
    planned = config_state_codes(job.config)
    by_state: Dict[str, Dict[str, int]] = {}
    for j in job.jurisdictions:
        st = (j.state_code or "").strip().upper()
        if not st:
            continue
        bucket = by_state.setdefault(st, {"pending": 0, "running": 0, "done": 0})
        if j.status == "pending":
            bucket["pending"] += 1
        elif j.status == "running":
            bucket["running"] += 1
        else:
            bucket["done"] += 1

    universe = sorted(set(planned) | set(by_state))
    started_codes = [
        st
        for st in universe
        if by_state.get(st, {}).get("running", 0) + by_state.get(st, {}).get("done", 0) > 0
    ]
    completed_codes = [
        st
        for st in universe
        if by_state.get(st)
        and by_state[st].get("pending", 0) == 0
        and by_state[st].get("running", 0) == 0
        and by_state[st].get("done", 0) > 0
    ]
    return {
        "states_planned": len(universe),
        "states_started": len(started_codes),
        "states_completed": len(completed_codes),
        "states_started_codes": started_codes,
        "states_completed_codes": completed_codes,
    }


def _maybe_stale_cancel_batch(job: BatchJob) -> bool:
    """Cancel ``running`` batches with no activity for ``batch_inactivity_seconds()``."""
    if (job.status or "").lower() != "running":
        return False
    inactive = batch_inactivity_seconds()
    now = datetime.now(timezone.utc)
    if (now - last_batch_activity_at(job)).total_seconds() <= inactive:
        return False
    finished = _utc_now_iso()
    job.status = "cancelled"
    job.finished_at = finished
    cfg = dict(job.config or {})
    cfg["stale_cancel_reason"] = f"no activity for {int(inactive)}s"
    job.config = cfg
    for j in job.jurisdictions:
        if j.status == "running":
            j.status = "failed"
            j.exit_code = -1
            j.finished_at = finished
            j.updated_at = finished
            j.current_video_id = ""
            j.current_video_title = ""
            j.current_video_started_at = ""
    return True


def persist_batch_job(job: BatchJob) -> None:
    """Write lifecycle changes to Postgres and the JSON cache file when present."""
    job.updated_at = _utc_now_iso()
    _recompute_summary(job)
    try:
        from api.batch_jobs.batch_job_db import sync_batch_job_to_db

        sync_batch_job_to_db(job)
    except Exception:
        pass
    try:
        store = BatchJobStore(job.batch_id)
        if store.path.is_file():
            tmp = store.path.with_suffix(".json.tmp")
            tmp.write_text(
                json.dumps(job.to_dict(), indent=2, ensure_ascii=False) + "\n",
                encoding="utf-8",
            )
            tmp.replace(store.path)
            _update_index(store.root, job)
    except Exception:
        pass


def duration_seconds_from_catalog_minutes(duration_minutes: object) -> Optional[float]:
    """Convert ``bronze_event_youtube.duration_minutes`` to seconds for batch UI."""
    if duration_minutes is None:
        return None
    try:
        sec = float(duration_minutes) * 60.0
    except (TypeError, ValueError):
        return None
    if sec <= 0 or sec > 86400 * 48:  # cap absurd values (~48h)
        return None
    return round(sec, 1)


def _slug(s: str, max_len: int = 48) -> str:
    out = _SLUG_RE.sub("-", (s or "").strip()).strip("-")
    return (out[:max_len] if out else "batch")


def new_batch_id(step: str) -> str:
    ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    return f"{_slug(step, 24)}-{ts}"


def jobs_dir() -> Path:
    raw = (os.getenv("BATCH_JOBS_DIR") or "").strip()
    return Path(raw).resolve() if raw else _DEFAULT_JOBS_DIR


def _batch_database_url() -> str:
    return (
        os.getenv("NEON_DATABASE_URL_DEV", "").strip()
        or os.getenv("NEON_DATABASE_URL", "").strip()
        or os.getenv("DATABASE_URL", "").strip()
    )


def _batch_plan_cache_key(states: List[str], *, round_robin: bool) -> str:
    st_list = [s.strip().upper() for s in states if (s or "").strip()]
    return f"{','.join(st_list)}|{int(bool(round_robin))}"


def fetch_batch_plan_jurisdictions(
    states: List[str],
    *,
    round_robin: bool = True,
    database_url: Optional[str] = None,
) -> List[JurisdictionRun]:
    """
    Canonical jurisdictions for the batch plan (``pending``), aligned with
    ``intermediate.int_jurisdictions`` — not legacy ``c-AL-*`` ids on bronze video rows.
    """
    from core_lib.jurisdictions.jurisdiction_id import (
        _PREFIXED_USPS_GEOID_RE,
        _SLUG_GEOID_RE,
        _TYPED_JURISDICTION_ID_RE,
        ensure_canonical_jurisdiction_id,
    )

    st_list = [s.strip().upper() for s in states if (s or "").strip()]
    if not st_list:
        return []
    url = (database_url or "").strip() or _batch_database_url()
    if not url:
        return []
    try:
        import psycopg2
        from psycopg2.extras import RealDictCursor
    except ImportError:
        return []

    rows: list = []
    try:
        with psycopg2.connect(url) as conn, conn.cursor(
            cursor_factory=RealDictCursor
        ) as cur:
            cur.execute(
                """
                SELECT
                    j.state_code,
                    j.jurisdiction_id,
                    j.name AS jurisdiction_name,
                    j.jurisdiction_type::text AS jurisdiction_type
                FROM intermediate.int_jurisdictions j
                WHERE j.state_code = ANY(%s)
                  AND j.jurisdiction_id IS NOT NULL
                  AND BTRIM(j.jurisdiction_id) <> ''
                  AND EXISTS (
                      SELECT 1
                      FROM bronze.bronze_event_youtube y
                      WHERE y.state_code = j.state_code
                        AND (
                          y.jurisdiction_id = j.jurisdiction_id
                          OR y.jurisdiction_id = 'c-' || j.state_code || '-' || j.geoid
                          OR y.jurisdiction_id = 'municipality_' || j.geoid
                          OR y.jurisdiction_id LIKE '%\\_' || j.geoid ESCAPE '\\'
                        )
                  )
                ORDER BY j.state_code, j.name, j.jurisdiction_id
                """,
                (st_list,),
            )
            rows = list(cur.fetchall())
    except Exception:
        rows = []

    if not rows:
        try:
            with psycopg2.connect(url) as conn, conn.cursor(
                cursor_factory=RealDictCursor
            ) as cur:
                cur.execute(
                    """
                    SELECT DISTINCT ON (state_code, jurisdiction_id)
                           state_code,
                           jurisdiction_id,
                           jurisdiction_name,
                           NULL::text AS jurisdiction_type
                    FROM bronze.bronze_event_youtube
                    WHERE state_code = ANY(%s)
                      AND jurisdiction_id IS NOT NULL
                      AND BTRIM(jurisdiction_id) <> ''
                    ORDER BY state_code, jurisdiction_id, jurisdiction_name
                    """,
                    (st_list,),
                )
                rows = list(cur.fetchall())
        except Exception:
            return []

    if round_robin:
        from collections import defaultdict

        buckets: Dict[str, list] = defaultdict(list)
        for r in rows:
            buckets[str(r["state_code"]).upper()].append(r)
        active = [s for s in st_list if buckets.get(s)]
        max_len = max((len(buckets[s]) for s in active), default=0)
        ordered: list = []
        for i in range(max_len):
            for s in active:
                if i < len(buckets[s]):
                    ordered.append(buckets[s][i])
        rows = ordered

    out: List[JurisdictionRun] = []
    seen_keys: set[str] = set()
    for r in rows:
        raw_id = str(r["jurisdiction_id"]).strip()
        name = str(r.get("jurisdiction_name") or "").strip()
        if not name or name == raw_id or _looks_like_jurisdiction_id_label(name):
            name = ""
        jtype = str(r.get("jurisdiction_type") or "").strip()
        jid = ensure_canonical_jurisdiction_id(
            raw_id,
            name=name or None,
            jurisdiction_type=jtype or None,
            database_url=url,
        ) or raw_id
        if _PREFIXED_USPS_GEOID_RE.match(jid):
            continue
        if not (
            (_SLUG_GEOID_RE.match(jid) and not _TYPED_JURISDICTION_ID_RE.match(jid))
            or _TYPED_JURISDICTION_ID_RE.match(jid)
        ):
            continue
        key = _plan_jurisdiction_key(jid)
        if not key or key in seen_keys:
            continue
        seen_keys.add(key)
        row = JurisdictionRun(
            state_code=str(r["state_code"]).upper(),
            jurisdiction_id=jid,
            jurisdiction_name=name,
            jurisdiction_type=jtype,
            status="pending",
        )
        normalize_jurisdiction_run(row, database_url=url)
        out.append(row)

    return out


@functools.lru_cache(maxsize=16)
def _fetch_batch_plan_cached(cache_key: str, database_url: str) -> tuple:
    """Cacheable wrapper; returns tuple of row dicts for ``lru_cache``."""
    parts = cache_key.rsplit("|", 1)
    states = [s for s in parts[0].split(",") if s]
    round_robin = bool(int(parts[1]))
    runs = fetch_batch_plan_jurisdictions(
        states, round_robin=round_robin, database_url=database_url or None
    )
    return tuple(
        (
            r.state_code,
            r.jurisdiction_id,
            r.jurisdiction_name,
            r.jurisdiction_type,
            r.status,
        )
        for r in runs
    )


def fetch_batch_plan_jurisdictions_cached(
    states: List[str],
    *,
    round_robin: bool = True,
    database_url: Optional[str] = None,
) -> List[JurisdictionRun]:
    """Same as ``fetch_batch_plan_jurisdictions`` but deduped per process (dashboard load)."""
    url = (database_url or "").strip() or _batch_database_url()
    key = _batch_plan_cache_key(states, round_robin=round_robin)
    if not key or key == "|0":
        return []
    rows = _fetch_batch_plan_cached(key, url or "")
    return [
        JurisdictionRun(
            state_code=row[0],
            jurisdiction_id=row[1],
            jurisdiction_name=row[2],
            jurisdiction_type=row[3],
            status=row[4],
        )
        for row in rows
    ]


_STATUS_RANK = {
    "completed": 5,
    "failed": 4,
    "running": 3,
    "cancelled": 2,
    "pending": 1,
}


def _plan_jurisdiction_key(jurisdiction_id: str) -> str:
    """Match plan rows to batch rows even when id format differs (``municipality_*`` vs ``slug_geoid``)."""
    from core_lib.jurisdictions.jurisdiction_id import (
        parse_jurisdiction_id,
        resolve_canonical_jurisdiction_id,
    )

    jid = (jurisdiction_id or "").strip()
    if not jid:
        return ""
    canonical = resolve_canonical_jurisdiction_id(jid).strip().lower()
    _jt, geoid, _slug = parse_jurisdiction_id(canonical or jid)
    if geoid:
        return f"geoid:{geoid}"
    return canonical or jid.lower()


def _looks_like_jurisdiction_id_label(text: str) -> bool:
    """True when ``text`` is an id string, not a place name (e.g. ``c-AL-01021``)."""
    from core_lib.jurisdictions.jurisdiction_id import (
        _PREFIXED_USPS_GEOID_RE,
        _SLUG_GEOID_RE,
        _TYPED_JURISDICTION_ID_RE,
    )

    t = (text or "").strip()
    if not t:
        return False
    if _PREFIXED_USPS_GEOID_RE.match(t):
        return True
    if _TYPED_JURISDICTION_ID_RE.match(t):
        return True
    if _SLUG_GEOID_RE.match(t) and "_" in t and not re.search(r"\s", t):
        return True
    return False


def _lookup_jurisdiction_name_from_db(
    jurisdiction_id: str,
    *,
    database_url: Optional[str] = None,
) -> str:
    """Place name from ``int_jurisdictions`` for a canonical or legacy id."""
    from core_lib.jurisdictions.jurisdiction_id import parse_jurisdiction_id

    jid = (jurisdiction_id or "").strip()
    if not jid:
        return ""
    url = (database_url or "").strip() or _batch_database_url()
    if not url:
        return ""
    _jt, geoid, _slug = parse_jurisdiction_id(jid)
    try:
        import psycopg2
    except ImportError:
        return ""
    try:
        with psycopg2.connect(url) as conn, conn.cursor() as cur:
            cur.execute(
                """
                SELECT name FROM intermediate.int_jurisdictions
                WHERE jurisdiction_id = %s
                LIMIT 1
                """,
                (jid,),
            )
            row = cur.fetchone()
            if row and row[0]:
                return str(row[0]).strip()
            if geoid:
                cur.execute(
                    """
                    SELECT name FROM intermediate.int_jurisdictions
                    WHERE geoid = %s
                    ORDER BY jurisdiction_id
                    LIMIT 1
                    """,
                    (geoid,),
                )
                row = cur.fetchone()
                if row and row[0]:
                    return str(row[0]).strip()
    except Exception:
        return ""
    return ""


def normalize_jurisdiction_run(
    j: JurisdictionRun,
    *,
    database_url: Optional[str] = None,
) -> JurisdictionRun:
    """Canonical ``jurisdiction_id`` and human ``jurisdiction_name`` for dashboard rows."""
    from core_lib.jurisdictions.jurisdiction_id import (
        _PREFIXED_USPS_GEOID_RE,
        _SLUG_GEOID_RE,
        _TYPED_JURISDICTION_ID_RE,
        ensure_canonical_jurisdiction_id,
    )

    raw_id = (j.jurisdiction_id or "").strip()
    if not raw_id:
        return j
    name = (j.jurisdiction_name or "").strip()
    if (
        _SLUG_GEOID_RE.match(raw_id)
        and not _TYPED_JURISDICTION_ID_RE.match(raw_id)
        and name
        and not _looks_like_jurisdiction_id_label(name)
        and name != raw_id
    ):
        return j
    if not name or name == raw_id or _looks_like_jurisdiction_id_label(name):
        name = ""

    jid = ensure_canonical_jurisdiction_id(
        raw_id,
        name=name or None,
        jurisdiction_type=(j.jurisdiction_type or None),
        database_url=database_url,
    ) or raw_id
    if _PREFIXED_USPS_GEOID_RE.match(jid):
        jid = ensure_canonical_jurisdiction_id(
            raw_id,
            name=None,
            jurisdiction_type=(j.jurisdiction_type or "county"),
            database_url=database_url,
        ) or jid

    j.jurisdiction_id = jid
    if not name or _looks_like_jurisdiction_id_label(name):
        name = _lookup_jurisdiction_name_from_db(jid, database_url=database_url)
    if name and not _looks_like_jurisdiction_id_label(name):
        j.jurisdiction_name = name
    return j


def _upgrade_jurisdiction_from_plan(
    existing: JurisdictionRun, placeholder: JurisdictionRun
) -> None:
    """Apply canonical id and display name from the fresh plan row."""
    from core_lib.jurisdictions.jurisdiction_id import (
        _SLUG_GEOID_RE,
        _TYPED_JURISDICTION_ID_RE,
    )

    def _is_canonical_slug(jid: str) -> bool:
        return bool(_SLUG_GEOID_RE.match(jid) and not _TYPED_JURISDICTION_ID_RE.match(jid))

    pid = (placeholder.jurisdiction_id or "").strip()
    eid = (existing.jurisdiction_id or "").strip()
    if pid and (_is_canonical_slug(pid) or not _is_canonical_slug(eid)):
        existing.jurisdiction_id = pid
    pname = (placeholder.jurisdiction_name or "").strip()
    if pname and not _looks_like_jurisdiction_id_label(pname):
        existing.jurisdiction_name = pname
    if placeholder.jurisdiction_type and not (existing.jurisdiction_type or "").strip():
        existing.jurisdiction_type = placeholder.jurisdiction_type


def normalize_batch_job_jurisdictions(
    job: BatchJob,
    *,
    database_url: Optional[str] = None,
) -> bool:
    """Normalize all jurisdiction rows in place. True when ids/names need a DB persist."""
    from core_lib.jurisdictions.jurisdiction_id import _PREFIXED_USPS_GEOID_RE

    changed = False
    url = database_url
    for j in job.jurisdictions:
        before_id = (j.jurisdiction_id or "").strip()
        before_name = (j.jurisdiction_name or "").strip()
        normalize_jurisdiction_run(j, database_url=url)
        after_id = (j.jurisdiction_id or "").strip()
        after_name = (j.jurisdiction_name or "").strip()
        if before_id != after_id:
            changed = True
        elif _PREFIXED_USPS_GEOID_RE.match(before_id) and not _PREFIXED_USPS_GEOID_RE.match(
            after_id
        ):
            changed = True
        elif _looks_like_jurisdiction_id_label(before_name) and not _looks_like_jurisdiction_id_label(
            after_name
        ):
            changed = True
    return changed


def _prefer_jurisdiction_run(
    current: JurisdictionRun, other: JurisdictionRun
) -> JurisdictionRun:
    """Keep the row with the most advanced status (never downgrade completed → pending)."""
    cur_rank = _STATUS_RANK.get((current.status or "").lower(), 0)
    other_rank = _STATUS_RANK.get((other.status or "").lower(), 0)
    if other_rank > cur_rank:
        return other
    if other_rank < cur_rank:
        return current
    if (other.videos or other.stats) and not (current.videos or current.stats):
        return other
    return current


def needs_plan_expand(job: BatchJob) -> bool:
    """
  True when the dashboard should merge the full state plan into this batch.

  Skip for finished batches that already have their jurisdiction rows (avoids
  re-merging hundreds of pending placeholders on every API read).
    """
    cfg = job.config or {}
    if cfg.get("seed_plan") is False:
        return False
    if not cfg.get("states"):
        return False
    status = (job.status or "").lower()
    if status == "running":
        return True
    total = int(cfg.get("total_jurisdictions") or 0) or len(job.jurisdictions)
    if status in ("completed", "cancelled", "failed") and job.jurisdictions:
        if total > 0 and len(job.jurisdictions) >= total:
            return False
        if status in ("completed", "cancelled") and len(job.jurisdictions) > 0:
            return False
    return True


def expand_batch_job_plan(
    job: BatchJob,
    plan: Optional[List[JurisdictionRun]] = None,
) -> None:
    """Fill in ``pending`` rows for jurisdictions not started yet in this batch."""
    cfg = job.config or {}
    if cfg.get("seed_plan") is False:
        return
    states = cfg.get("states") or []
    if not states:
        return
    rr = cfg.get("round_robin")
    if rr is None:
        rr = True
    if plan is None:
        plan = fetch_batch_plan_jurisdictions_cached(states, round_robin=bool(rr))
    if not plan:
        return

    by_key: Dict[str, JurisdictionRun] = {}
    for j in job.jurisdictions:
        key = _plan_jurisdiction_key(j.jurisdiction_id)
        if not key:
            continue
        if key in by_key:
            by_key[key] = _prefer_jurisdiction_run(by_key[key], j)
        else:
            by_key[key] = j

    merged: List[JurisdictionRun] = []
    matched: set[str] = set()
    for placeholder in plan:
        key = _plan_jurisdiction_key(placeholder.jurisdiction_id)
        existing = by_key.get(key) if key else None
        if existing is not None:
            _upgrade_jurisdiction_from_plan(existing, placeholder)
            normalize_jurisdiction_run(existing)
            merged.append(existing)
            matched.add(key)
        else:
            normalize_jurisdiction_run(placeholder)
            merged.append(placeholder)
            if key:
                matched.add(key)

    for key, j in by_key.items():
        if key in matched:
            continue
        from core_lib.jurisdictions.jurisdiction_id import _PREFIXED_USPS_GEOID_RE

        if _PREFIXED_USPS_GEOID_RE.match((j.jurisdiction_id or "").strip()):
            continue
        normalize_jurisdiction_run(j)
        merged.append(j)

    job.jurisdictions = merged
    if not int(cfg.get("total_jurisdictions") or 0):
        job.config["total_jurisdictions"] = len(merged)


@dataclass
class VideoResult:
    video_id: str
    title: str = ""
    status: str = "pending"  # ok | fail | tombstoned | empty | rate_limit | skipped
    error: str = ""
    transcript_source: str = ""
    finished_at: str = ""
    duration_seconds: Optional[float] = None


@dataclass
class JurisdictionRun:
    state_code: str
    jurisdiction_id: str
    jurisdiction_name: str = ""
    jurisdiction_type: str = ""
    status: str = "pending"  # pending | running | completed | failed
    started_at: str = ""
    updated_at: str = ""
    finished_at: str = ""
    elapsed_seconds: float = 0.0
    exit_code: int = 0
    stats: Dict[str, int] = field(default_factory=dict)
    videos: List[VideoResult] = field(default_factory=list)
    file_counts: Dict[str, int] = field(
        default_factory=lambda: {"transcripts": 0, "analysis": 0, "reports": 0}
    )
    current_video_id: str = ""
    current_video_title: str = ""
    current_video_started_at: str = ""


@dataclass
class BatchJob:
    batch_id: str
    step: str
    status: str = "running"  # running | completed | failed | cancelled
    started_at: str = ""
    updated_at: str = ""
    finished_at: str = ""
    config: Dict[str, Any] = field(default_factory=dict)
    summary: Dict[str, Any] = field(default_factory=dict)
    jurisdictions: List[JurisdictionRun] = field(default_factory=list)

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> BatchJob:
        jurs = []
        for j in data.get("jurisdictions") or []:
            videos = []
            for v in j.get("videos") or []:
                vd = dict(v)
                if "duration_seconds" not in vd:
                    vd["duration_seconds"] = None
                elif vd["duration_seconds"] is not None:
                    try:
                        vd["duration_seconds"] = float(vd["duration_seconds"])
                    except (TypeError, ValueError):
                        vd["duration_seconds"] = None
                videos.append(VideoResult(**vd))
            jurs.append(
                JurisdictionRun(
                    state_code=j.get("state_code", ""),
                    jurisdiction_id=j.get("jurisdiction_id", ""),
                    jurisdiction_name=j.get("jurisdiction_name", ""),
                    jurisdiction_type=str(j.get("jurisdiction_type") or ""),
                    status=j.get("status", "pending"),
                    started_at=j.get("started_at", ""),
                    updated_at=j.get("updated_at", ""),
                    finished_at=j.get("finished_at", ""),
                    elapsed_seconds=float(j.get("elapsed_seconds") or 0),
                    exit_code=int(j.get("exit_code") or 0),
                    stats=dict(j.get("stats") or {}),
                    videos=videos,
                    file_counts=dict(
                        j.get("file_counts")
                        or {"transcripts": 0, "analysis": 0, "reports": 0}
                    ),
                    current_video_id=j.get("current_video_id", ""),
                    current_video_title=j.get("current_video_title", ""),
                    current_video_started_at=j.get("current_video_started_at", ""),
                )
            )
        return cls(
            batch_id=data["batch_id"],
            step=data.get("step", ""),
            status=data.get("status", "running"),
            started_at=data.get("started_at", ""),
            updated_at=data.get("updated_at", ""),
            finished_at=data.get("finished_at", ""),
            config=dict(data.get("config") or {}),
            summary=dict(data.get("summary") or {}),
            jurisdictions=jurs,
        )


_VIDEO_COUNT_KEYS = ("ok", "fail", "tombstoned", "empty", "rate_limit")

_STATUS_TO_STAT_KEY: Dict[str, str] = {
    "ok": "ok",
    "fail": "fail",
    "tombstoned": "tombstoned",
    "empty": "empty",
    "rate_limit": "rate_limit",
}


def _disk_file_count(fc: Dict[str, int], disk_key: str, legacy_key: str) -> int:
    """Prefer explicit ``*_disk`` keys; fall back to legacy ``transcripts`` / ``analysis`` keys."""
    if disk_key in fc:
        return int(fc.get(disk_key) or 0)
    return int(fc.get(legacy_key) or 0)


def transcript_seconds_from_job_videos(job: BatchJob) -> float:
    """Sum ``duration_seconds`` for batch video rows with status ``ok``."""
    total = 0.0
    for j in job.jurisdictions:
        for v in j.videos or []:
            if (v.status or "").strip().lower() != "ok":
                continue
            if v.duration_seconds is None:
                continue
            total += float(v.duration_seconds)
    return total


def _jurisdiction_video_counts(j: JurisdictionRun) -> Dict[str, int]:
    """Count finished videos for summary totals.

    While a jurisdiction is running, ``record_video`` appends to ``j.videos`` but
    ``jurisdiction_finish`` has not written final ``j.stats`` yet — prefer the video
    list so the dashboard does not lag behind jurisdiction success counts.
    """
    if j.videos:
        counts = {k: 0 for k in _VIDEO_COUNT_KEYS}
        for v in j.videos:
            key = _STATUS_TO_STAT_KEY.get((v.status or "").strip().lower())
            if key:
                counts[key] += 1
        return counts
    return {k: int((j.stats or {}).get(k) or 0) for k in _VIDEO_COUNT_KEYS}


def _running_file_clock(
    j: JurisdictionRun, now: datetime
) -> tuple[Optional[float], str, str, str]:
    """
  Seconds on the in-flight video for a running jurisdiction.

  Uses explicit ``current_video_started_at`` when set; otherwise time since the
  last completed video in this run, or since the jurisdiction started.
    """
    if j.status != "running":
        return None, "", "", ""

    if j.current_video_started_at:
        try:
            start = datetime.fromisoformat(
                j.current_video_started_at.replace("Z", "+00:00")
            )
            secs = max(0.0, (now - start).total_seconds())
            return (
                secs,
                j.current_video_id or "",
                (j.current_video_title or "")[:120],
                j.current_video_started_at,
            )
        except ValueError:
            pass

    start_iso = ""
    video_id = ""
    title = ""
    if j.videos:
        last = j.videos[-1]
        if last.finished_at:
            start_iso = last.finished_at
            video_id = last.video_id
            title = (last.title or "")[:120]
    if not start_iso and j.started_at:
        start_iso = j.started_at
    if not start_iso:
        return None, video_id, title, ""

    try:
        start = datetime.fromisoformat(start_iso.replace("Z", "+00:00"))
        secs = max(0.0, (now - start).total_seconds())
        return secs, video_id, title, start_iso
    except ValueError:
        return None, video_id, title, start_iso


def apply_batch_lifecycle(job: BatchJob) -> bool:
    """Stale-cancel inactive runs, then auto-finish when the plan is done. Returns True if status changed."""
    before = (job.status or "").lower()
    _maybe_stale_cancel_batch(job)
    _maybe_auto_finish_batch(job)
    return before != (job.status or "").lower()


def _maybe_auto_finish_batch(job: BatchJob) -> None:
    """
    Mark the batch completed when every target jurisdiction is done.

    Progress cards use jurisdiction counts; ``job.status`` only flipped on ``finish_batch``
    before. Single-jurisdiction tests (or a killed shell after the last backfill) left
    ``running`` at 100%.
    """
    if job.status != "running":
        return
    total_j = int(job.config.get("total_jurisdictions") or 0) or len(job.jurisdictions)
    if total_j <= 0:
        return
    if any(j.status == "running" for j in job.jurisdictions):
        return
    processed = sum(1 for j in job.jurisdictions if j.status in ("completed", "failed"))
    if processed < total_j:
        return
    job.status = "completed"
    if not job.finished_at:
        job.finished_at = _utc_now_iso()


def _recompute_summary(job: BatchJob) -> None:
    if (job.status or "").lower() == "running":
        _maybe_stale_cancel_batch(job)
    total_j = int(job.config.get("total_jurisdictions") or 0) or len(job.jurisdictions)
    processed = sum(1 for j in job.jurisdictions if j.status in ("completed", "failed"))
    success = sum(1 for j in job.jurisdictions if j.status == "completed" and j.exit_code == 0)
    failed = sum(1 for j in job.jurisdictions if j.status == "failed" or j.exit_code != 0)
    remaining = max(0, total_j - processed)

    v_ok = v_fail = v_tomb = v_empty = v_rl = 0
    for j in job.jurisdictions:
        counts = _jurisdiction_video_counts(j)
        v_ok += counts["ok"]
        v_fail += counts["fail"]
        v_tomb += counts["tombstoned"]
        v_empty += counts["empty"]
        v_rl += counts["rate_limit"]

    elapsed = 0.0
    if job.started_at:
        try:
            start = datetime.fromisoformat(job.started_at.replace("Z", "+00:00"))
            end = (
                datetime.fromisoformat(job.finished_at.replace("Z", "+00:00"))
                if job.finished_at
                else datetime.now(timezone.utc)
            )
            elapsed = max(0.0, (end - start).total_seconds())
        except ValueError:
            elapsed = 0.0

    eta_seconds: Optional[float] = None
    if processed > 0 and remaining > 0 and elapsed > 0:
        eta_seconds = (elapsed / processed) * remaining

    disk = {"transcripts": 0, "analysis": 0, "reports": 0}
    recent = {"analysis": 0, "reports": 0}
    bronze_download_rows = 0
    for j in job.jurisdictions:
        fc = j.file_counts or {}
        disk["transcripts"] += _disk_file_count(
            fc, "transcripts_disk", "transcripts"
        )
        disk["analysis"] += _disk_file_count(fc, "analysis_disk", "analysis")
        disk["reports"] += _disk_file_count(fc, "reports_disk", "reports")
        recent["analysis"] += int(fc.get("analysis_recent") or 0)
        recent["reports"] += int(fc.get("reports_recent") or 0)
        bronze_download_rows += int(fc.get("bronze_download_rows") or 0)

    files_processed = v_ok + v_fail + v_tomb + v_empty + v_rl
    videos_attempted = files_processed
    avg_seconds_per_file: Optional[float] = None
    if files_processed > 0 and elapsed > 0:
        avg_seconds_per_file = round(elapsed / files_processed, 1)

    current_file_seconds: Optional[float] = None
    current_video_id = ""
    current_video_title = ""
    current_jurisdiction_id = ""
    current_video_started_at = ""
    now = datetime.now(timezone.utc)
    for j in job.jurisdictions:
        secs, vid, title, started_iso = _running_file_clock(j, now)
        if secs is None:
            continue
        if current_file_seconds is None or secs >= current_file_seconds:
            current_file_seconds = secs
            current_video_id = vid
            current_video_title = title
            current_jurisdiction_id = j.jurisdiction_id
            current_video_started_at = started_iso

    job.summary = {
        "total_jurisdictions": total_j,
        "processed_jurisdictions": processed,
        "success_jurisdictions": success,
        "failed_jurisdictions": failed,
        "remaining_jurisdictions": remaining,
        "elapsed_seconds": round(elapsed, 1),
        "eta_seconds": round(eta_seconds, 1) if eta_seconds is not None else None,
        "videos_ok": v_ok,
        "videos_fail": v_fail,
        "videos_tombstoned": v_tomb,
        "videos_empty": v_empty,
        "videos_rate_limit": v_rl,
        # Batch-scoped caption outcomes (same source as Videos OK / fail cards).
        "files_transcripts": v_ok,
        "files_processed": files_processed,
        "videos_attempted": videos_attempted,
        # Policy cache on disk (all JSON under each jurisdiction folder, not this run only).
        "files_transcripts_disk": disk["transcripts"],
        "files_analysis": disk["analysis"],
        "files_reports": disk["reports"],
        # Rolling 24h throughput: analyses summarised / reports generated whose
        # files were last written within the past day (mtime-based).
        "files_analysis_recent": recent["analysis"],
        "files_reports_recent": recent["reports"],
        # Bronze rows with transcript_download_at (lifetime per jurisdiction).
        "bronze_download_rows": bronze_download_rows,
        "transcript_seconds": round(transcript_seconds_from_job_videos(job), 1),
        "avg_seconds_per_file": avg_seconds_per_file,
        "current_file_seconds": (
            round(current_file_seconds, 1)
            if current_file_seconds is not None
            else None
        ),
        "current_video_id": current_video_id,
        "current_video_title": current_video_title,
        "current_jurisdiction_id": current_jurisdiction_id,
        "current_video_started_at": current_video_started_at,
        **state_progress_for_job(job),
    }
    _maybe_auto_finish_batch(job)


class BatchJobStore:
    def __init__(self, batch_id: str, *, jobs_root: Optional[Path] = None) -> None:
        self.batch_id = batch_id
        self.root = jobs_root or jobs_dir()
        self.root.mkdir(parents=True, exist_ok=True)
        self.path = self.root / f"{batch_id}.json"

    def load(self) -> BatchJob:
        if not self.path.is_file():
            raise FileNotFoundError(self.path)
        data = json.loads(self.path.read_text(encoding="utf-8"))
        return BatchJob.from_dict(data)

    def save(self, job: BatchJob) -> None:
        job.updated_at = _utc_now_iso()
        _recompute_summary(job)
        tmp = self.path.with_suffix(".json.tmp")
        tmp.write_text(
            json.dumps(job.to_dict(), indent=2, ensure_ascii=False) + "\n",
            encoding="utf-8",
        )
        tmp.replace(self.path)
        _update_index(self.root, job)
        try:
            from api.batch_jobs.batch_job_db import sync_batch_job_to_db

            sync_batch_job_to_db(job)
        except Exception:
            pass

    def start_batch(
        self,
        *,
        step: str,
        config: Optional[Dict[str, Any]] = None,
    ) -> BatchJob:
        now = _utc_now_iso()
        cfg = dict(config or {})
        job = BatchJob(
            batch_id=self.batch_id,
            step=step,
            status="running",
            started_at=now,
            updated_at=now,
            config=cfg,
        )
        states = cfg.get("states") or []
        if states and cfg.get("seed_plan") is not False:
            plan = fetch_batch_plan_jurisdictions(
                states,
                round_robin=bool(cfg.get("round_robin", True)),
            )
            if plan:
                job.jurisdictions = plan
                if not int(cfg.get("total_jurisdictions") or 0):
                    cfg["total_jurisdictions"] = len(plan)
                    job.config = cfg
        self.save(job)
        return job

    def finish_batch(self, *, status: str = "completed") -> BatchJob:
        job = self.load()
        job.status = status
        job.finished_at = _utc_now_iso()
        self.save(job)
        return job

    def jurisdiction_start(
        self,
        *,
        state_code: str,
        jurisdiction_id: str,
        jurisdiction_name: str = "",
        pending_videos: int = 0,
    ) -> JurisdictionRun:
        from core_lib.jurisdictions.jurisdiction_id import resolve_canonical_jurisdiction_id

        jurisdiction_id = resolve_canonical_jurisdiction_id(
            jurisdiction_id.strip(),
            name=(jurisdiction_name or "").strip() or None,
        )
        job = self.load()
        now = _utc_now_iso()
        run = JurisdictionRun(
            state_code=state_code.upper(),
            jurisdiction_id=jurisdiction_id,
            jurisdiction_name=jurisdiction_name,
            status="running",
            started_at=now,
            updated_at=now,
        )
        if pending_videos:
            run.stats["pending"] = pending_videos
        start_key = _plan_jurisdiction_key(jurisdiction_id)
        job.jurisdictions = [
            j
            for j in job.jurisdictions
            if _plan_jurisdiction_key(j.jurisdiction_id) != start_key
        ]
        job.jurisdictions.append(run)
        self.save(job)
        return run

    def _find_jurisdiction(self, job: BatchJob, jurisdiction_id: str) -> JurisdictionRun:
        from core_lib.jurisdictions.jurisdiction_id import resolve_canonical_jurisdiction_id

        jid = resolve_canonical_jurisdiction_id((jurisdiction_id or "").strip()) or (
            jurisdiction_id or ""
        ).strip()
        key = _plan_jurisdiction_key(jid)
        for j in job.jurisdictions:
            if j.jurisdiction_id == jid:
                return j
            if key and _plan_jurisdiction_key(j.jurisdiction_id) == key:
                return j
        raise KeyError(jurisdiction_id)

    def video_start(
        self,
        *,
        jurisdiction_id: str,
        video_id: str,
        title: str = "",
    ) -> None:
        job = self.load()
        j = self._find_jurisdiction(job, jurisdiction_id)
        j.current_video_id = video_id
        j.current_video_title = (title or "")[:200]
        j.current_video_started_at = _utc_now_iso()
        j.updated_at = j.current_video_started_at
        self.save(job)

    def record_video(
        self,
        *,
        jurisdiction_id: str,
        video_id: str,
        status: str,
        title: str = "",
        error: str = "",
        transcript_source: str = "",
        duration_seconds: Optional[float] = None,
    ) -> None:
        job = self.load()
        j = self._find_jurisdiction(job, jurisdiction_id)
        if j.current_video_id == video_id:
            j.current_video_id = ""
            j.current_video_title = ""
            j.current_video_started_at = ""
        prior_status = ""
        for v in j.videos:
            if v.video_id == video_id:
                prior_status = (v.status or "").strip().lower()
                break
        j.videos = [v for v in j.videos if v.video_id != video_id]
        if prior_status:
            old_key = _STATUS_TO_STAT_KEY.get(prior_status)
            if old_key:
                j.stats[old_key] = max(0, int(j.stats.get(old_key) or 0) - 1)
        now = _utc_now_iso()
        j.videos.append(
            VideoResult(
                video_id=video_id,
                title=title[:200],
                status=status,
                error=(error or "")[:500],
                transcript_source=transcript_source,
                finished_at=now,
                duration_seconds=duration_seconds,
            )
        )
        new_key = _STATUS_TO_STAT_KEY.get((status or "").strip().lower())
        if new_key:
            j.stats[new_key] = int(j.stats.get(new_key) or 0) + 1
        j.updated_at = now
        self.save(job)

    def jurisdiction_finish(
        self,
        *,
        jurisdiction_id: str,
        exit_code: int,
        stats: Optional[Dict[str, int]] = None,
        file_counts: Optional[Dict[str, int]] = None,
    ) -> None:
        job = self.load()
        j = self._find_jurisdiction(job, jurisdiction_id)
        j.status = "completed" if exit_code == 0 else "failed"
        j.exit_code = exit_code
        j.finished_at = _utc_now_iso()
        j.updated_at = j.finished_at
        j.current_video_id = ""
        j.current_video_title = ""
        j.current_video_started_at = ""
        if stats:
            j.stats = {k: int(v) for k, v in stats.items()}
        if file_counts:
            j.file_counts = {k: int(v) for k, v in file_counts.items()}
        if j.started_at:
            try:
                start = datetime.fromisoformat(j.started_at.replace("Z", "+00:00"))
                end = datetime.fromisoformat(j.finished_at.replace("Z", "+00:00"))
                j.elapsed_seconds = max(0.0, (end - start).total_seconds())
            except ValueError:
                j.elapsed_seconds = 0.0
        self.save(job)

    def shell_jurisdiction_done(
        self,
        *,
        state_code: str,
        jurisdiction_id: str,
        jurisdiction_name: str,
        exit_code: int,
    ) -> None:
        """Mark jurisdiction when backfill did not write status (crash / missing --batch-id)."""
        try:
            job = self.load()
            self._find_jurisdiction(job, jurisdiction_id)
        except (FileNotFoundError, KeyError):
            self.jurisdiction_start(
                state_code=state_code,
                jurisdiction_id=jurisdiction_id,
                jurisdiction_name=jurisdiction_name,
            )
        self.jurisdiction_finish(
            jurisdiction_id=jurisdiction_id,
            exit_code=exit_code,
            stats={"shell_exit": exit_code},
        )


def _update_index(root: Path, job: BatchJob) -> None:
    index_path = root / _INDEX_NAME
    entries: List[Dict[str, Any]] = []
    if index_path.is_file():
        try:
            entries = json.loads(index_path.read_text(encoding="utf-8"))
        except json.JSONDecodeError:
            entries = []
    entries = [e for e in entries if e.get("batch_id") != job.batch_id]
    entries.insert(
        0,
        {
            "batch_id": job.batch_id,
            "step": job.step,
            "status": job.status,
            "started_at": job.started_at,
            "updated_at": job.updated_at,
            "summary": job.summary,
        },
    )
    entries = entries[:200]
    index_path.write_text(json.dumps(entries, indent=2) + "\n", encoding="utf-8")


def policy_disk_file_counts(scanned: Dict[str, int]) -> Dict[str, int]:
    """Map ``count_policy_files_for_jurisdiction`` keys to explicit on-disk counters."""
    return {
        "transcripts_disk": int(scanned.get("transcripts") or 0),
        "analysis_disk": int(scanned.get("analysis") or 0),
        "reports_disk": int(scanned.get("reports") or 0),
        # Rolling 24h throughput (mtime within the last day); 0 when the scanner
        # couldn't attribute per-jurisdiction stats (legacy fallback path).
        "analysis_recent": int(scanned.get("analysis_recent") or 0),
        "reports_recent": int(scanned.get("reports_recent") or 0),
    }


def count_policy_files_for_jurisdiction(
    cache_root: Path,
    *,
    state_code: str,
    jurisdiction_id: str,
) -> Dict[str, int]:
    """Count transcript/analysis/report files under policy cache for one jurisdiction."""
    from llm.gemini.policy_processing_status_report import (
        _DIR_ANALYSIS,
        _DIR_REPORTS,
        _DIR_TRANSCRIPTS,
        scan_jurisdiction_cache,
    )

    counts = {
        "transcripts": 0,
        "analysis": 0,
        "reports": 0,
        "analysis_recent": 0,
        "reports_recent": 0,
    }
    if not cache_root.is_dir():
        return counts
    by_jid = scan_jurisdiction_cache(cache_root)
    st = by_jid.get(jurisdiction_id)
    if st:
        counts["transcripts"] = st.transcripts
        counts["analysis"] = st.analysis_json
        counts["reports"] = st.reports_md
        counts["analysis_recent"] = st.analysis_recent
        counts["reports_recent"] = st.reports_recent
        return counts

    state = (state_code or "").upper()
    jid = jurisdiction_id.lower()
    for state_dir in cache_root.iterdir():
        if not state_dir.is_dir() or state_dir.name.upper() != state:
            continue
        for type_dir in state_dir.iterdir():
            if not type_dir.is_dir():
                continue
            for jdir in type_dir.iterdir():
                if jdir.name.lower() != jid and jid not in jdir.name.lower():
                    continue
                for ch in jdir.iterdir():
                    if not ch.is_dir():
                        continue
                    tx = ch / _DIR_TRANSCRIPTS
                    if tx.is_dir():
                        counts["transcripts"] += sum(
                            1 for p in tx.glob("*.json") if p.is_file()
                        )
                    an = ch / _DIR_ANALYSIS
                    if an.is_dir():
                        counts["analysis"] += sum(
                            1
                            for p in an.glob("*.json")
                            if p.is_file() and not p.name.startswith("_")
                        )
                    rp = ch / _DIR_REPORTS
                    if rp.is_dir():
                        counts["reports"] += sum(
                            1 for p in rp.glob("*.md") if p.is_file()
                        )
    return counts


def list_batches(*, limit: int = 50) -> List[BatchJob]:
    root = jobs_dir()
    jobs: List[BatchJob] = []
    index_path = root / _INDEX_NAME
    ids: List[str] = []
    if index_path.is_file():
        try:
            for row in json.loads(index_path.read_text(encoding="utf-8")):
                bid = row.get("batch_id")
                if bid:
                    ids.append(bid)
        except json.JSONDecodeError:
            pass
    if not ids:
        ids = sorted(
            (p.stem for p in root.glob("*.json") if p.name != _INDEX_NAME),
            reverse=True,
        )
    for bid in ids[:limit]:
        path = root / f"{bid}.json"
        if path.is_file():
            try:
                jobs.append(BatchJob.from_dict(json.loads(path.read_text(encoding="utf-8"))))
            except (json.JSONDecodeError, KeyError):
                continue
    return jobs


def _cli_start(args: argparse.Namespace) -> int:
    batch_id = (args.batch_id or "").strip() or new_batch_id(args.step)
    store = BatchJobStore(batch_id)
    config = {
        "states": [s.strip().upper() for s in (args.states or "").split(",") if s.strip()],
        "n": args.n,
        "delay": args.delay,
        "total_jurisdictions": args.total_jurisdictions,
        "transcript_source": args.transcript_source or "",
        "max_jurisdictions": args.max_jurisdictions or 0,
        "round_robin": bool(getattr(args, "round_robin", 1)),
        "seed_plan": not getattr(args, "no_seed_plan", False),
    }
    store.start_batch(step=args.step, config=config)
    print(batch_id)
    return 0


def _cli_jurisdiction_start(args: argparse.Namespace) -> int:
    store = BatchJobStore(args.batch_id)
    store.jurisdiction_start(
        state_code=args.state,
        jurisdiction_id=args.jurisdiction_id,
        jurisdiction_name=args.jurisdiction_name or "",
        pending_videos=args.pending_videos,
    )
    return 0


def _cli_jurisdiction_finish(args: argparse.Namespace) -> int:
    store = BatchJobStore(args.batch_id)
    stats = json.loads(args.stats) if args.stats else {}
    store.jurisdiction_finish(
        jurisdiction_id=args.jurisdiction_id,
        exit_code=args.exit_code,
        stats=stats,
    )
    return 0


def _cli_finish(args: argparse.Namespace) -> int:
    store = BatchJobStore(args.batch_id)
    store.finish_batch(status=args.status)
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    sub = parser.add_subparsers(dest="command", required=True)

    p_start = sub.add_parser("start", help="Create a new batch job file")
    p_start.add_argument("--step", required=True)
    p_start.add_argument("--batch-id", default="")
    p_start.add_argument("--states", default="")
    p_start.add_argument("--n", type=int, default=0)
    p_start.add_argument("--delay", type=float, default=0)
    p_start.add_argument("--total-jurisdictions", type=int, default=0)
    p_start.add_argument("--transcript-source", default="")
    p_start.add_argument("--max-jurisdictions", type=int, default=0)
    p_start.add_argument(
        "--round-robin",
        type=int,
        default=1,
        help="1 = interleave states in plan order (default), 0 = state-by-state",
    )
    p_start.add_argument(
        "--no-seed-plan",
        action="store_true",
        help="Do not pre-load all jurisdictions as pending rows",
    )
    p_start.set_defaults(func=_cli_start)

    p_js = sub.add_parser("jurisdiction-start")
    p_js.add_argument("--batch-id", required=True)
    p_js.add_argument("--state", required=True)
    p_js.add_argument("--jurisdiction-id", required=True)
    p_js.add_argument("--jurisdiction-name", default="")
    p_js.add_argument("--pending-videos", type=int, default=0)
    p_js.set_defaults(func=_cli_jurisdiction_start)

    p_jf = sub.add_parser("jurisdiction-finish")
    p_jf.add_argument("--batch-id", required=True)
    p_jf.add_argument("--jurisdiction-id", required=True)
    p_jf.add_argument("--exit-code", type=int, default=0)
    p_jf.add_argument("--stats", default="")
    p_jf.set_defaults(func=_cli_jurisdiction_finish)

    p_fin = sub.add_parser("finish")
    p_fin.add_argument("--batch-id", required=True)
    p_fin.add_argument("--status", default="completed")
    p_fin.set_defaults(func=_cli_finish)

    args = parser.parse_args()
    return int(args.func(args))


if __name__ == "__main__":
    raise SystemExit(main())