#!/usr/bin/env python3
"""
Standalone AI media-description worker for Hydrogen.

Consumes the `work.media_to_describe` queue OUT-OF-BAND: for each queued media
it calls the vision model DIRECTLY (never through Hydrogen's synchronous
`/describe-ai` endpoint, which times out on slow models), then persists the
result through the admin API and removes the row from the queue.

Per media, one run does:
  1. GET  {APP_URL}/admin/media/{hex}/base64?format=jpg   -> JPEG data URI
  2. POST {AI_SERVER_BASE_URL}/v1/chat/completions        -> model JSON
  3. POST {APP_URL}/admin/media/describe                  -> core enrichment
  4. POST {APP_URL}/admin/media/{hex}/enrichment          -> poi + person_count
  5. DELETE FROM work.media_to_describe                   -> on success, and on
                                                             media-specific
                                                             failures (4xx, bad
                                                             image, unreadable
                                                             output). Kept only on
                                                             transient infra
                                                             errors (connect /
                                                             timeout / 5xx) for a
                                                             later retry.

Why this split: the model call is the slow part and lives here, off the HTTP
request path. Hydrogen only ever does fast, transactional writes.

Multi-environment: `work.media_to_describe` is SHARED (dev / preprod / prod all
point at the same `work` DB), so every row is stamped with the enqueuing
instance's `WORK_QUEUE_ENV`. This worker claims ONLY rows matching its own
`WORK_QUEUE_ENV` — run one worker per environment, each with the matching key,
so it never touches media whose file / API live on another instance.

Config is read from the Hydrogen `.env` (same keys the app uses); every value
can be overridden by a real environment variable of the same name. Nothing
Hydrogen-specific is imported — this file is fully standalone.

Dependencies:
    pip install PyMySQL requests

Usage:
    python describe_worker.py                 # process one batch
    python describe_worker.py --batch 50      # cap the batch size
    python describe_worker.py --dry-run       # do everything except the writes
    python describe_worker.py --env /path/.env
    python describe_worker.py --log /path/worker.log   # self-log to a file (no
                                              # shell redirection needed)
    python describe_worker.py --verbose       # empty-queue heartbeats + tracebacks

Logging: each line is "[time] [LEVEL] msg". Empty-queue runs are SILENT by
default (so a per-minute cron doesn't flood the log) — pass --verbose for a
heartbeat. The --log file is size-rotated (DESCRIBE_WORKER_LOG_MAX_BYTES,
default 5 MB -> keeps one .1 backup) so it stays bounded.

Cron — safe to fire every minute: the OS advisory lock makes a run exit
immediately if the previous one is still going, and it is released
automatically if that run crashes (no stale lock ever wedges the worker):
    * * * * * /usr/bin/python3 /path/to/bin/describe_worker.py >> /var/log/hydrogen-describe.log 2>&1

Windows Task Scheduler equivalent: schedule every minute with
"Do not start a new instance" as a second line of defense (the lock is the
primary guarantee).
"""

from __future__ import annotations

import argparse
import json
import os
import re
import sys
import time
import traceback
import uuid
from pathlib import Path

try:
    import pymysql
    import requests
except ImportError as exc:  # pragma: no cover
    sys.stderr.write(f"Missing dependency: {exc}. Run: pip install PyMySQL requests\n")
    sys.exit(2)


# --------------------------------------------------------------------------
# Moderation bitmask — kept in lock-step with MediaDescriptionProposal.php.
# --------------------------------------------------------------------------
FLAG_ILLEGAL = 1
FLAG_VIOLENT = 2
FLAG_SEXUAL = 4
FLAG_SELFIE = 8
FLAG_SCREENSHOT = 16
FLAG_AI = 32

SIGNAL_KEYS = {
    "is_illegal": FLAG_ILLEGAL,
    "is_violent": FLAG_VIOLENT,
    "is_sexual": FLAG_SEXUAL,
    "is_selfie": FLAG_SELFIE,
    "is_screenshot": FLAG_SCREENSHOT,
    "is_ai": FLAG_AI,
}


# --------------------------------------------------------------------------
# Config
# --------------------------------------------------------------------------
def load_env(env_path: Path) -> dict[str, str]:
    """Parse a KEY=VALUE .env file (ignores comments / blank lines / `export`)."""
    values: dict[str, str] = {}
    if not env_path.is_file():
        return values
    for line in env_path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if line == "" or line.startswith("#"):
            continue
        if line.startswith("export "):
            line = line[len("export "):]
        if "=" not in line:
            continue
        key, _, val = line.partition("=")
        key = key.strip()
        val = val.strip().strip('"').strip("'")
        if key:
            values[key] = val
    return values


class Config:
    def __init__(self, env: dict[str, str]):
        def get(key: str, default: str | None = None) -> str:
            # A real environment variable always wins over the .env file.
            return os.environ.get(key, env.get(key, default if default is not None else ""))

        # Queue database (work)
        self.work_host = get("WORK_DB_HOST")
        self.work_port = int(get("WORK_DB_PORT", "3306"))
        self.work_name = get("WORK_DB_NAME", "work")
        self.work_user = get("WORK_DB_USER")
        self.work_pass = get("WORK_DB_PASSWORD")

        # NB: the worker no longer touches the `ai` (prompts) or `hxa_bo`
        # (media_exif) databases — the prompt is now RENDERED server-side
        # (topics + EXIF substituted) and fetched over HTTP per media via
        # GET /admin/media/{hex}/describe-prompt. Only the WORK queue DB, the
        # Hydrogen admin API and the model server are needed.

        # Vision model server (OpenAI-compatible)
        self.ai_base_url = get("AI_SERVER_BASE_URL", "http://localhost:1234").rstrip("/")
        self.model = get("AI_DESCRIBE_MODEL", "mistralai/ministral-3-3b")
        self.max_tokens = int(get("AI_DESCRIBE_MAX_OUTPUT_TOKENS", "4096"))
        self.ai_timeout = int(get("AI_DESCRIBE_TIMEOUT_SECONDS", "300"))
        self.prompt_id = get("AI_DESCRIBE_PROMPT_ID", "media.identification")

        # Hydrogen admin API
        self.api_base = get("APP_URL", "http://localhost").rstrip("/")
        self.admin_token = get("ADMIN_API_TOKEN")

        # Environment discriminator. `work.media_to_describe` is SHARED across
        # environments; this worker claims ONLY the rows its own instance
        # enqueued (matched against the app's WORK_QUEUE_ENV). Must be the SAME
        # value the Hydrogen instance writes at enqueue time.
        self.queue_env = get("WORK_QUEUE_ENV", "hydrogen")

        # Worker knobs (worker-only, harmless defaults if absent from .env)
        self.batch = int(get("DESCRIBE_WORKER_BATCH", "20"))
        self.http_timeout = int(get("DESCRIBE_WORKER_HTTP_TIMEOUT", "30"))
        self.log_path = get("DESCRIBE_WORKER_LOG", "")
        # Rotate the log file once it reaches this size (bytes); 0 disables it.
        self.log_max_bytes = int(get("DESCRIBE_WORKER_LOG_MAX_BYTES", str(5 * 1024 * 1024)))
        self.lock_path = get(
            "DESCRIBE_WORKER_LOCK",
            str(Path(os.environ.get("TEMP", "/tmp")) / "hydrogen-describe-worker.lock"),
        )

    def require(self) -> None:
        missing = [
            name
            for name, value in {
                "WORK_DB_HOST": self.work_host,
                "WORK_DB_USER": self.work_user,
                "APP_URL": self.api_base,
                "ADMIN_API_TOKEN": self.admin_token,
                "WORK_QUEUE_ENV": self.queue_env,
            }.items()
            if value == ""
        ]
        if missing:
            raise SystemExit(f"Missing required config: {', '.join(missing)}")


# --------------------------------------------------------------------------
# Single-instance lock — an OS ADVISORY lock (fcntl.flock on POSIX,
# msvcrt.locking on Windows). Designed for a tight cron cadence (e.g. every
# minute): a second invocation while one is still running exits at once.
#
# Crucially the OS releases the lock automatically when the holder dies —
# including a hard kill / crash / power loss — so there is NO stale-lock that
# would wedge the worker forever (the failure mode of an O_EXCL lock file).
# --------------------------------------------------------------------------
class InstanceLock:
    def __init__(self, path: str):
        self.path = path
        self.fh = None  # a file object we keep open for the lock's lifetime

    def __enter__(self) -> "InstanceLock":
        # O_RDWR|O_CREAT: create if absent, never truncate a peer's file, and
        # no append semantics (so seek/write land where we expect).
        fd = os.open(self.path, os.O_RDWR | os.O_CREAT, 0o644)
        self.fh = os.fdopen(fd, "r+b")
        try:
            if os.name == "nt":
                import msvcrt
                self.fh.seek(0)
                msvcrt.locking(self.fh.fileno(), msvcrt.LK_NBLCK, 1)
            else:
                import fcntl
                fcntl.flock(self.fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        except OSError:
            self.fh.close()
            self.fh = None
            raise SystemExit(f"Another instance is already running (lock {self.path}); exiting.")

        # Record the live PID for observability (best-effort, never fatal).
        try:
            self.fh.seek(0)
            self.fh.truncate()
            self.fh.write(str(os.getpid()).encode())
            self.fh.flush()
        except OSError:
            pass
        return self

    def __exit__(self, *_exc) -> None:
        if self.fh is None:
            return
        try:
            if os.name == "nt":
                import msvcrt
                self.fh.seek(0)
                msvcrt.locking(self.fh.fileno(), msvcrt.LK_UNLCK, 1)
            else:
                import fcntl
                fcntl.flock(self.fh.fileno(), fcntl.LOCK_UN)
        except OSError:
            pass  # the OS releases it on close/exit regardless
        self.fh.close()


# --------------------------------------------------------------------------
# JSON repair — ports the tolerant parsing of MediaDescriptionProposal.php.
# --------------------------------------------------------------------------
def strip_code_fence(content: str) -> str:
    trimmed = content.strip()
    if trimmed.startswith("```"):
        trimmed = re.sub(r"^```[a-zA-Z0-9]*\s*\n?", "", trimmed)
        trimmed = re.sub(r"\n?```\s*$", "", trimmed)
    return trimmed.strip()


def escape_control_chars_in_strings(s: str) -> str:
    """Escape raw control chars a small model leaves inside a JSON string."""
    out: list[str] = []
    in_string = False
    escaped = False
    mapping = {"\n": "\\n", "\r": "\\r", "\t": "\\t", "\b": "\\b", "\f": "\\f"}
    for ch in s:
        if not in_string:
            if ch == '"':
                in_string = True
            out.append(ch)
            continue
        if escaped:
            out.append(ch)
            escaped = False
            continue
        if ch == "\\":
            out.append(ch)
            escaped = True
            continue
        if ch == '"':
            in_string = False
            out.append(ch)
            continue
        if ord(ch) < 0x20:
            out.append(mapping.get(ch, f"\\u{ord(ch):04x}"))
            continue
        out.append(ch)
    return "".join(out)


def repair_missing_commas(s: str) -> str:
    pattern = r'("(?:[^"\\]|\\.)*"|\d|\]|\})(\s*\n\s*)("(?:[^"\\]|\\.)*"\s*:)'
    return re.sub(pattern, r"\1,\2\3", s)


def parse_model_json(content: str) -> dict:
    text = strip_code_fence(content)
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass
    repaired = repair_missing_commas(escape_control_chars_in_strings(text))
    return json.loads(repaired)  # let a still-broken doc raise to the caller


# --------------------------------------------------------------------------
# Model output -> API payloads
# --------------------------------------------------------------------------
def _num(value, default: float = 0.0) -> float:
    return float(value) if isinstance(value, (int, float)) else default


def fold_flag(doc: dict) -> int:
    flag = 0
    for key, bit in SIGNAL_KEYS.items():
        signal = doc.get(key)
        if isinstance(signal, dict) and bool(signal.get("status", False)):
            flag |= bit
    return flag


def weighted_list(raw) -> list[dict]:
    """Normalise a list of strings or {name, probability} into {name, probability}."""
    out: list[dict] = []
    seen: set[str] = set()
    if not isinstance(raw, list):
        return out
    for item in raw:
        if isinstance(item, str):
            name, prob = item.strip(), 1.0
        elif isinstance(item, dict) and isinstance(item.get("name"), str):
            name, prob = item["name"].strip(), _num(item.get("probability"))
        else:
            continue
        if name == "" or name in seen:
            continue
        seen.add(name)
        out.append({"name": name, "probability": prob})
    return out


def describe_payload(media_hex: str, doc: dict) -> dict:
    themes = [t.strip() for t in doc.get("themes", []) if isinstance(t, str) and t.strip()]
    return {
        "id": str(uuid.UUID(hex=media_hex)),  # dashed UUID
        "flag": fold_flag(doc),
        "focus": list(dict.fromkeys(themes)),  # unique, order-preserving
        "title": doc.get("title") if isinstance(doc.get("title"), str) else None,
        "meta_title": doc.get("meta_title") if isinstance(doc.get("meta_title"), str) else None,
        "meta_description": doc.get("meta_description") if isinstance(doc.get("meta_description"), str) else None,
        "description": doc.get("description") if isinstance(doc.get("description"), str) else "",
        "objects": weighted_list(doc.get("objects")),
    }


def enrichment_payload(doc: dict) -> dict | None:
    body: dict = {}
    pois = weighted_list(doc.get("poi"))
    if pois:
        body["poi"] = pois
    pc = doc.get("person_count")
    if isinstance(pc, bool):
        pc = None
    if isinstance(pc, int) and pc >= 0:
        body["person_count"] = pc
    return body or None


# --------------------------------------------------------------------------
# I/O
# --------------------------------------------------------------------------
class HttpError(RuntimeError):
    """A non-2xx HTTP response, carrying the status so the caller can decide
    whether the media is a poison message (4xx → drop) or the failure is a
    transient server issue (5xx → keep for retry)."""

    def __init__(self, where: str, status: int, body: str):
        self.status = status
        super().__init__(f"{where} -> {status}: {body[:500]}")


def is_retryable(exc: BaseException) -> bool:
    """True when the failure is infrastructural and worth retrying (so the media
    stays in the queue): connection refused/reset, timeout, or a 5xx from the
    API / model server. Everything else (4xx, unreadable model output, missing
    media/file) is media-specific — retrying won't help, so it is dropped."""
    if isinstance(exc, (requests.exceptions.ConnectionError, requests.exceptions.Timeout)):
        return True
    if isinstance(exc, HttpError):
        return 500 <= exc.status < 600
    return False


def fetch_rendered_prompt(cfg: Config, media_hex: str) -> str:
    """Fetches the prompt already RENDERED for this media (the ai.prompts
    template with {{topics}} and {{exif}} substituted) from the Hydrogen admin
    API. Single source of truth: the worker never re-implements the substitution
    nor touches the ai / hxa_bo databases."""
    resp = requests.get(
        f"{cfg.api_base}/admin/media/{media_hex}/describe-prompt",
        headers={"Authorization": f"Bearer {cfg.admin_token}"},
        timeout=cfg.http_timeout,
    )
    if resp.status_code != 200:
        raise HttpError("describe-prompt endpoint", resp.status_code, resp.text)
    prompt = resp.json().get("prompt")
    if not isinstance(prompt, str) or prompt.strip() == "":
        raise RuntimeError(f"describe-prompt returned no prompt for {media_hex}")
    return prompt


def claim_batch(cfg: Config) -> list[str]:
    conn = pymysql.connect(
        host=cfg.work_host, port=cfg.work_port, user=cfg.work_user,
        password=cfg.work_pass, database=cfg.work_name, charset="utf8mb4",
    )
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT LOWER(HEX(`media_id`)) FROM `media_to_describe` "
                "WHERE `environment` = %s "
                "ORDER BY `created_at` ASC, `media_id` ASC LIMIT %s",
                (cfg.queue_env, cfg.batch),
            )
            return [r[0] for r in cur.fetchall()]
    finally:
        conn.close()


def dequeue(cfg: Config, media_hex: str) -> None:
    conn = pymysql.connect(
        host=cfg.work_host, port=cfg.work_port, user=cfg.work_user,
        password=cfg.work_pass, database=cfg.work_name, charset="utf8mb4",
    )
    try:
        with conn.cursor() as cur:
            cur.execute(
                "DELETE FROM `media_to_describe` WHERE `media_id` = UNHEX(%s) AND `environment` = %s",
                (media_hex, cfg.queue_env),
            )
        conn.commit()
    finally:
        conn.close()


def get_image_data_uri(cfg: Config, media_hex: str) -> str:
    resp = requests.get(
        f"{cfg.api_base}/admin/media/{media_hex}/base64",
        params={"format": "jpg"},
        headers={"Authorization": f"Bearer {cfg.admin_token}"},
        timeout=cfg.http_timeout,
    )
    if resp.status_code != 200:
        raise HttpError("base64 endpoint", resp.status_code, resp.text)
    data = resp.json()
    image = data.get("image")
    if not image:
        raise RuntimeError(f"base64 endpoint returned no image for {media_hex}")
    # The model server rejects non-JPEG data URIs ("'url' field must be a base64
    # encoded image"). If this environment's API predates the ?format=jpg option
    # it silently returns WebP — surface that as an actionable error instead of a
    # cryptic downstream 400.
    if not image.startswith("data:image/jpeg"):
        mime = image.split(";", 1)[0] if image.startswith("data:") else "unknown"
        raise RuntimeError(
            f"base64 endpoint returned {mime}, not JPEG — the model rejects it. "
            "Deploy the ?format=jpg support (GetMediaBase64Action) to this "
            "environment's API."
        )
    return image


def call_model(cfg: Config, prompt: str, image_data_uri: str) -> str:
    payload = {
        "model": cfg.model,
        "messages": [
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt},
                    {"type": "image_url", "image_url": {"url": image_data_uri}},
                ],
            }
        ],
        "max_tokens": cfg.max_tokens,
    }
    resp = requests.post(
        f"{cfg.ai_base_url}/v1/chat/completions",
        json=payload,
        headers={"Content-Type": "application/json", "Accept": "application/json"},
        timeout=cfg.ai_timeout,
    )
    if resp.status_code != 200:
        # Surface the server's own message (e.g. "model not loaded", a bad image
        # rejection, …) instead of a generic HTTPError that hides the reason.
        raise HttpError("AI /v1/chat/completions", resp.status_code, resp.text)
    data = resp.json()
    choices = data.get("choices") or []
    if not choices:
        raise RuntimeError("model response had no choices")
    content = (choices[0].get("message") or {}).get("content")
    if not isinstance(content, str) or content.strip() == "":
        raise RuntimeError("model returned empty content (a reasoning model? disable thinking)")
    return content


def post_describe(cfg: Config, payload: dict) -> None:
    resp = requests.post(
        f"{cfg.api_base}/admin/media/describe",
        json=payload,
        headers={"Authorization": f"Bearer {cfg.admin_token}", "Content-Type": "application/json"},
        timeout=cfg.http_timeout,
    )
    if resp.status_code != 200:
        raise HttpError("/admin/media/describe", resp.status_code, resp.text)


def post_enrichment(cfg: Config, media_hex: str, body: dict) -> None:
    resp = requests.post(
        f"{cfg.api_base}/admin/media/{media_hex}/enrichment",
        json=body,
        headers={"Authorization": f"Bearer {cfg.admin_token}", "Content-Type": "application/json"},
        timeout=cfg.http_timeout,
    )
    if resp.status_code != 200:
        raise HttpError(f"/admin/media/{media_hex}/enrichment", resp.status_code, resp.text)


# --------------------------------------------------------------------------
# Orchestration
# --------------------------------------------------------------------------
# Optional file sink so the worker can be launched directly (py script ...)
# without any shell redirection — set via --log or DESCRIBE_WORKER_LOG.
_LOG_FH = None


def log(msg: str, level: str = "INFO") -> None:
    line = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] [{level:<5}] {msg}"
    print(line, flush=True)
    if _LOG_FH is not None:
        _LOG_FH.write(line + "\n")
        _LOG_FH.flush()


def _rotate_log(path: Path, max_bytes: int) -> None:
    """Single-backup size rotation: once the log reaches max_bytes it becomes
    <log>.1 (replacing any previous .1) and a fresh file starts. Keeps the log
    bounded for a worker that fires every minute. No-op when max_bytes <= 0."""
    if max_bytes <= 0 or not path.exists():
        return
    try:
        if path.stat().st_size < max_bytes:
            return
        backup = path.with_name(path.name + ".1")
        if backup.exists():
            backup.unlink()
        path.rename(backup)
    except OSError:
        pass  # rotation is best-effort; never block the run on it


def process_one(cfg: Config, media_hex: str, dry_run: bool) -> dict:
    """Runs the full pipeline for ONE media and returns a summary of what was
    produced. Raises on any failure; the caller logs and times the outcome."""
    # The prompt is rendered per media server-side ({{topics}} + {{exif}}), so
    # we fetch it fresh for each media rather than reusing a shared template.
    prompt = fetch_rendered_prompt(cfg, media_hex)
    image = get_image_data_uri(cfg, media_hex)

    content = call_model(cfg, prompt, image)
    doc = parse_model_json(content)

    core = describe_payload(media_hex, doc)
    extra = enrichment_payload(doc)

    if not dry_run:
        post_describe(cfg, core)
        if extra is not None:
            post_enrichment(cfg, media_hex, extra)

    return {
        "flag": core["flag"],
        "focus": len(core["focus"]),
        "objects": len(core["objects"]),
        "poi": len((extra or {}).get("poi", [])),
        "person_count": (extra or {}).get("person_count"),
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Hydrogen out-of-band AI describe worker.")
    parser.add_argument("--env", default=str(Path(__file__).resolve().parent.parent / ".env"),
                        help="Path to the Hydrogen .env (default: repo .env).")
    parser.add_argument("--batch", type=int, default=None, help="Override batch size.")
    parser.add_argument("--dry-run", action="store_true", help="Do everything except the writes.")
    parser.add_argument("--log", default=None,
                        help="Append run output to this file (also printed to stdout). "
                             "Lets you launch the script directly, with no shell redirection. "
                             "Falls back to DESCRIBE_WORKER_LOG from the env.")
    parser.add_argument("--verbose", action="store_true",
                        help="Log empty-queue heartbeats and full tracebacks on unexpected errors.")
    args = parser.parse_args()

    cfg = Config(load_env(Path(args.env)))
    if args.batch is not None:
        cfg.batch = args.batch
    cfg.require()

    # File logging: --log wins over DESCRIBE_WORKER_LOG. Rotated then opened in
    # append so cron runs accumulate; failure to open is non-fatal (stdout works).
    log_path = args.log or cfg.log_path
    if log_path:
        global _LOG_FH
        try:
            p = Path(log_path)
            p.parent.mkdir(parents=True, exist_ok=True)
            _rotate_log(p, cfg.log_max_bytes)
            _LOG_FH = open(log_path, "a", encoding="utf-8")
        except OSError as exc:
            print(f"WARNING: cannot open log file {log_path}: {exc}", flush=True)

    with InstanceLock(cfg.lock_path):
        media_ids = claim_batch(cfg)
        if not media_ids:
            # Silent by default: at one run/minute an "empty" line every time
            # would flood the log. --verbose keeps a heartbeat when wanted.
            if args.verbose:
                log(f"Queue empty (env={cfg.queue_env}); nothing to do.")
            return 0

        mode = "DRY-RUN" if args.dry_run else "LIVE"
        log(f"Claimed {len(media_ids)} media | env={cfg.queue_env} "
            f"api={cfg.api_base} model={cfg.model} mode={mode}")

        ok = 0
        failed = 0
        run_start = time.monotonic()
        for media_hex in media_ids:
            started = time.monotonic()
            try:
                r = process_one(cfg, media_hex, args.dry_run)
                dt = time.monotonic() - started
                tag = "DRY-RUN" if args.dry_run else "OK"
                if not args.dry_run:
                    dequeue(cfg, media_hex)
                log(f"{tag} {media_hex} in {dt:.1f}s | flag={r['flag']} "
                    f"focus={r['focus']} objects={r['objects']} poi={r['poi']} "
                    f"person_count={r['person_count']}")
                ok += 1
            except Exception as exc:  # one bad media must not stop the batch
                dt = time.monotonic() - started
                failed += 1
                # Drop the media from the queue UNLESS the failure is transient
                # infra (connection/timeout/5xx) — those are kept for retry so a
                # passing outage doesn't wipe a whole batch. A media-specific
                # failure (4xx, bad image, unreadable output) is removed: retry
                # would just re-fail every minute. Re-inject later if needed via
                # POST /admin/jobs/describe-queue/requeue.
                retryable = is_retryable(exc)
                disposition = "kept for retry" if retryable else "removed from queue"
                if not args.dry_run and not retryable:
                    try:
                        dequeue(cfg, media_hex)
                    except Exception as del_exc:
                        disposition = f"REMOVE FAILED ({type(del_exc).__name__})"
                log(f"{media_hex} in {dt:.1f}s ({disposition}): "
                    f"{type(exc).__name__}: {exc}", level="ERROR")
                # Dump a traceback only for unexpected errors, and only in verbose.
                if args.verbose and not isinstance(exc, (RuntimeError, requests.exceptions.RequestException)):
                    log(traceback.format_exc().rstrip(), level="ERROR")

        total = time.monotonic() - run_start
        log(f"Run done in {total:.1f}s: {ok} ok, {failed} failed, {len(media_ids)} total.")
        return 0 if failed == 0 else 1


if __name__ == "__main__":
    sys.exit(main())