gendesign/backend/app/workers/tasks/nspd_geo.py

"""Bulk NSPD geo backfill — resume-friendly Celery task.

Архитектура (см. data/sql/77_nspd_geo_jobs.sql):
    nspd_geo_jobs    — журнал заданий (один job = одна bulk-операция)
    nspd_geo_targets — список cad-номеров со status (pending/done/failed)

Resume:
    1. При worker_ready signal: ищем jobs со status='running' и stale heartbeat
       (>10 мин), переключаем их статус на 'resume_needed' + re-enqueue task.
    2. process_nspd_geo_job(job_id) сразу читает SELECT WHERE status='pending'
       — идемпотентно продолжает с того места, где остановился (UPSERT в
       cad_quarters_geom / cad_buildings гарантирует no dup'ов).

Запуск (3 пути):
    1. UI: POST /api/v1/admin/scrape/geo (создаёт job, наполняет targets, enqueue task)
    2. CLI: python -m app.workers.tasks.nspd_geo --enqueue-region 66
    3. Beat: можно добавить cron-расписание в celery_app.py
"""

from __future__ import annotations

import json
import logging
from typing import Any

from celery.exceptions import SoftTimeLimitExceeded
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.db import SessionLocal
from app.services.cadastre.bulk_harvest import _derive_quarter_cad, _safe_int, _safe_numeric
from app.services.scrapers.nspd_lite import (
    NspdLiteError,
    NspdLiteWafError,
    fetch_via_rosreestr2coord,
)
from app.workers.celery_app import celery_app

logger = logging.getLogger(__name__)

# Heartbeat каждые N items (компромисс: чаще = выше DB write, реже = stale-window больше)
HEARTBEAT_EVERY = 5
# WAF backoff (seconds) после первого 403/429
WAF_BACKOFF_BASE_S = 30

# Zombie-thresholds для cleanup_zombies (issue #1215).
# WAF-ветка процесса спит до WAF_BACKOFF_BASE_S * 8 = 240s (4 мин) внутри одной итерации —
# heartbeat за это время НЕ обновляется. Старый порог 2 мин ре-enqueue'ил живые WAF-jobs →
# два worker'а параллельно долбили забаненный NSPD-сервис, углубляя бан + затирали счётчики.
# Поднимаем running-порог выше max WAF-backoff (6 мин > 4 мин). Для 'paused' (consecutive_waf>=8)
# нужен большой cooldown, иначе минута beat сразу аннулирует WAF-паузу.
_ZOMBIE_RUNNING_THRESHOLD = "6 minutes"
_ZOMBIE_PAUSED_THRESHOLD = "30 minutes"
# Issue #1655: jobs, застрявшие в 'queued' после ре-enqueue (worker_ready /
# cleanup_zombies ставят 'queued' + apply_async), но чьё broker-сообщение
# потерялось (flush брокера, crash до pickup, purge очереди) — раньше никто
# не ре-reap'ил. cleanup_zombies матчил только 'running'/'paused'.
# Матчим только 'queued' c НЕ-NULL stale heartbeat_at: свежесозданный job
# (enqueue_geo_job) имеет heartbeat_at IS NULL (нет дефолта в схеме 77_) — его
# initial apply_async ещё в полёте, ре-reap'ить рано. Порог = running-порог
# (6 мин): сообщение либо взято воркером (→ status станет 'running' через claim),
# либо потеряно — 6 мин достаточно, чтобы отличить потерю от задержки очереди.
_ZOMBIE_QUEUED_THRESHOLD = "6 minutes"


# ── Helpers для job/target lifecycle ────────────────────────────────────────


def _log(
    db: Session, job_id: int, level: str, stage: str, message: str, cad_num: str | None = None
) -> None:
    """Запись в nspd_geo_log (виден через v_scrape_log_unified в admin UI)."""
    try:
        db.execute(
            text(
                """
                INSERT INTO nspd_geo_log (job_id, level, stage, cad_num, message)
                VALUES (:jid, :lvl, :stg, :cad, :msg)
                """
            ),
            {"jid": job_id, "lvl": level, "stg": stage, "cad": cad_num, "msg": message[:1000]},
        )
        db.commit()
    except Exception as e:
        logger.warning("nspd_geo_log write failed for job=%s: %s", job_id, e)
        try:
            db.rollback()
        except Exception:
            pass


def _start_job(db: Session, job_id: int) -> bool:
    """Атомарный claim job'а: переводит в 'running' ТОЛЬКО если он ещё не 'running'.

    Issue #1621: дублирующие task-сообщения на один job_id (worker_ready resume +
    cleanup_zombies beat одновременно, либо overlap контейнеров при rolling-redeploy,
    либо два prefork-child при --concurrency>1) раньше оба безусловно делали
    UPDATE→'running' и оба входили в while-loop → двойные WAF-хиты + затёртые счётчики.

    Теперь claim атомарен: `WHERE job_id=:id AND status <> 'running'` + RETURNING.
    Если строка не вернулась — кто-то уже держит claim, второй worker должен выйти.
    Returns True если claim получен, False если job уже 'running' (или не найден).
    """
    claimed = db.execute(
        text(
            """
            UPDATE nspd_geo_jobs
               SET status = 'running',
                   started_at = COALESCE(started_at, NOW()),
                   heartbeat_at = NOW(),
                   updated_at = NOW()
             WHERE job_id = :id
               AND status <> 'running'
            RETURNING job_id
            """
        ),
        {"id": job_id},
    ).scalar_one_or_none()
    db.commit()
    if claimed is None:
        return False
    _log(db, job_id, "info", "start", "job started/resumed")
    return True


def _heartbeat(db: Session, job_id: int, **counts: int) -> None:
    sets = ["heartbeat_at = NOW()", "updated_at = NOW()"]
    params: dict[str, Any] = {"id": job_id}
    for k, v in counts.items():
        sets.append(f"{k} = :{k}")
        params[k] = v
    try:
        db.execute(
            text(f"UPDATE nspd_geo_jobs SET {', '.join(sets)} WHERE job_id = :id"),
            params,
        )
        db.commit()
    except Exception as e:
        logger.warning("heartbeat failed for job=%s: %s", job_id, e)
        try:
            db.rollback()
        except Exception:
            pass


def _finish_job(db: Session, job_id: int, status: str, error: str | None = None) -> None:
    db.execute(
        text(
            """
            UPDATE nspd_geo_jobs
               SET status = :st, finished_at = NOW(), error = :err, updated_at = NOW()
             WHERE job_id = :id
            """
        ),
        {"id": job_id, "st": status, "err": error},
    )
    db.commit()


# ── Persistence для разных видов targets ────────────────────────────────────


def _save_quarter(db: Session, payload: dict, cad_num: str) -> int:
    """Парсит NSPD-response для квартала и UPSERT'ит в cad_quarters_geom."""
    feats = (payload.get("data") or {}).get("features") or []
    if not feats:
        return 0
    f = feats[0]
    geom_geojson = f.get("geometry")
    if not geom_geojson:
        return 0
    db.execute(
        text(
            """
            INSERT INTO cad_quarters_geom (cad_number, geom, raw_props, fetched_at)
            VALUES (:cad,
                    -- cad_quarters_geom.geom = geometry(MultiPolygon,4326) NOT NULL (migration 58).
                    -- NSPD может вернуть одноконтурный Polygon → ST_Multi coerce'ит в MultiPolygon,
                    -- иначе "Geometry type (Polygon) does not match column type (MultiPolygon)".
                    -- БЕЗ ST_Transform: rosreestr2coord отдаёт GeoJSON уже в WGS84
                    -- (EPSG:4326, degrees) для всех area_type — только ST_SetSRID.
                    -- ST_Transform из 3857 трактовал бы градусы как метры Mercator →
                    -- коллапс к Null Island (issue #1336).
                    ST_Multi(ST_SetSRID(ST_GeomFromGeoJSON(:g), 4326)),
                    CAST(:props AS jsonb), NOW())
            ON CONFLICT (cad_number) DO UPDATE
               SET geom = EXCLUDED.geom,
                   raw_props = EXCLUDED.raw_props,
                   fetched_at = NOW()
            """
        ),
        {
            "cad": cad_num,
            "g": json.dumps(geom_geojson, ensure_ascii=False),
            "props": json.dumps(f.get("properties") or {}, ensure_ascii=False),
        },
    )
    return 1


def _save_building(db: Session, payload: dict, cad_num: str) -> int:
    """Парсит NSPD-response для здания и UPSERT'ит в cad_buildings (wide schema, migration 92).

    cad_buildings (migration 92) — широкая схема:
      - quarter_cad_number (НЕ quarter_cad_num),
      - floors INT (NSPD может отдать "5-7" → _safe_int → NULL, иначе InvalidTextRepresentation),
      - source TEXT NOT NULL (нет дефолта) → обязателен,
      - geom GEOMETRY(Geometry, 4326) → полиморфный, ST_Multi НЕ нужен (mirror upsert_building).
    """
    feats = (payload.get("data") or {}).get("features") or []
    if not feats:
        return 0
    n = 0
    for f in feats:
        geom_geojson = f.get("geometry")
        opts = (f.get("properties") or {}).get("options") or {}
        if not geom_geojson:
            continue
        # Только этого здания (NSPD может вернуть здания всего квартала)
        if opts.get("cad_num") != cad_num:
            continue

        quarter_cad = opts.get("quarter_cad_number") or _derive_quarter_cad(cad_num)
        db.execute(
            text(
                """
                INSERT INTO cad_buildings (
                    cad_num, quarter_cad_number, geom, purpose, building_name,
                    readable_address, area, floors, raw_props, source
                ) VALUES (
                    :cad, :qcad,
                    -- БЕЗ ST_Transform: rosreestr2coord отдаёт GeoJSON уже в WGS84
                    -- (EPSG:4326, degrees) для всех area_type — только ST_SetSRID.
                    -- ST_Transform из 3857 трактовал бы градусы как метры Mercator →
                    -- коллапс к Null Island (issue #1336).
                    ST_SetSRID(ST_GeomFromGeoJSON(:g), 4326),
                    :purpose, :name, :addr, :area, :floors, CAST(:props AS jsonb), :source
                )
                ON CONFLICT (cad_num) DO UPDATE
                   SET quarter_cad_number = EXCLUDED.quarter_cad_number,
                       geom = EXCLUDED.geom,
                       purpose = EXCLUDED.purpose,
                       building_name = EXCLUDED.building_name,
                       readable_address = EXCLUDED.readable_address,
                       area = EXCLUDED.area,
                       floors = EXCLUDED.floors,
                       raw_props = EXCLUDED.raw_props,
                       source = EXCLUDED.source
                """
            ),
            {
                "cad": cad_num,
                "qcad": quarter_cad,
                "g": json.dumps(geom_geojson, ensure_ascii=False),
                "purpose": opts.get("purpose"),
                "name": opts.get("building_name") or opts.get("name"),
                "addr": opts.get("readable_address"),
                "area": _safe_numeric(opts.get("area")),
                "floors": _safe_int(opts.get("floors")),
                "props": json.dumps(opts, ensure_ascii=False),
                "source": "nspd",
            },
        )
        n += 1
    return n


def _save_parcel(db: Session, payload: dict, cad_num: str) -> int:
    """Парсит NSPD-response для земельного участка и UPSERT'ит в cad_parcels_geom.

    NB: rosreestr2coord для area_type=1 (parcel) отдаёт GeoJSON уже в WGS84
    (EPSG:4326, degrees). Просто ST_SetSRID — без ST_Transform, иначе геометрия
    сжимается до точки в районе (0,0) на острове Null. Quarters/buildings идут
    через тот же rosreestr2coord path и тоже WGS84 — им ST_Transform тоже НЕ нужен
    (issue #1336: раньше ошибочно трансформировались из 3857 → Null Island).
    """
    feats = (payload.get("data") or {}).get("features") or []
    if not feats:
        return 0
    f = feats[0]
    geom_geojson = f.get("geometry")
    if not geom_geojson:
        return 0
    db.execute(
        text(
            """
            INSERT INTO cad_parcels_geom (cad_num, geom, raw_props, fetched_at)
            VALUES (:cad,
                    -- cad_parcels.geom = geometry(MultiPolygon,4326) (migration 93;
                    -- cad_parcels_geom теперь back-compat VIEW над cad_parcels).
                    -- NSPD/rosreestr2coord отдаёт одноконтурный Polygon → ST_Multi coerce'ит
                    -- в MultiPolygon, иначе "Geometry type (Polygon) does not match column
                    -- type (MultiPolygon)". БЕЗ ST_Transform: parcel GeoJSON уже WGS84
                    -- (см. docstring) — только ST_SetSRID.
                    ST_Multi(ST_SetSRID(ST_GeomFromGeoJSON(:g), 4326)),
                    CAST(:props AS jsonb), NOW())
            ON CONFLICT (cad_num) DO UPDATE
               SET geom = EXCLUDED.geom,
                   raw_props = EXCLUDED.raw_props,
                   fetched_at = NOW()
            """
        ),
        {
            "cad": cad_num,
            "g": json.dumps(geom_geojson, ensure_ascii=False),
            "props": json.dumps(f.get("properties") or {}, ensure_ascii=False),
        },
    )
    return 1


def _persist_target(db: Session, thematic_id: int, payload: dict, cad_num: str) -> tuple[int, str]:
    """Returns (n_features_saved, table_name). Полиморфно по thematic_id."""
    if thematic_id == 1:
        return _save_parcel(db, payload, cad_num), "cad_parcels_geom"
    if thematic_id == 2:
        return _save_quarter(db, payload, cad_num), "cad_quarters_geom"
    if thematic_id == 5:
        return _save_building(db, payload, cad_num), "cad_buildings"
    return 0, ""


# ── Helper для создания job + targets ──────────────────────────────────────


def enqueue_geo_job(
    *,
    name: str | None,
    job_kind: str,
    source_kind: str,
    source_params: dict[str, Any],
    cad_nums_with_thematic: list[tuple[str, int]],
    rate_ms: int = 600,
    triggered_by: str = "manual",
) -> int:
    """Создаёт job + targets и возвращает job_id. Не enqueue'ит task сам.
    Caller должен сделать `process_nspd_geo_job.apply_async(args=[job_id])`.
    """
    import json as _json

    db = SessionLocal()
    try:
        job_id = db.execute(
            text(
                """
                INSERT INTO nspd_geo_jobs (
                    name, job_kind, source_kind, source_params,
                    rate_ms, triggered_by, status, targets_total
                ) VALUES (:n, :k, :sk, CAST(:sp AS jsonb), :r, :tb, 'queued', :tt)
                RETURNING job_id
                """
            ),
            {
                "n": name,
                "k": job_kind,
                "sk": source_kind,
                "sp": _json.dumps(source_params, ensure_ascii=False),
                "r": rate_ms,
                "tb": triggered_by,
                "tt": len(cad_nums_with_thematic),
            },
        ).scalar_one()

        # Bulk INSERT targets — parametrized to prevent SQL injection
        if cad_nums_with_thematic:
            db.execute(
                text(
                    "INSERT INTO nspd_geo_targets (job_id, cad_num, thematic_id, status) "
                    "VALUES (:job_id, :cad_num, :thematic_id, 'pending') "
                    "ON CONFLICT (job_id, cad_num, thematic_id) DO NOTHING"
                ),
                [
                    {"job_id": job_id, "cad_num": c, "thematic_id": t}
                    for c, t in cad_nums_with_thematic
                ],
            )
        db.commit()
        return int(job_id)
    finally:
        db.close()


# ── Main task ───────────────────────────────────────────────────────────────


@celery_app.task(
    bind=True,
    name="tasks.nspd_geo.process_job",
    max_retries=0,  # resume через worker_ready, не через retry
    soft_time_limit=3600 * 6,  # 6 часов max — крупные jobs
)
def process_nspd_geo_job(self: Any, job_id: int) -> dict[str, Any]:
    """Идёт по nspd_geo_targets WHERE job_id=:id AND status='pending'.

    Heartbeat каждые HEARTBEAT_EVERY items, persistent counters в job-row.
    Idempotent: уже-сохранённые UPSERT'ятся, status target → 'done'.
    """
    import time

    db = SessionLocal()
    try:
        # Загружаем параметры job'а
        job = (
            db.execute(
                text("SELECT * FROM nspd_geo_jobs WHERE job_id = :id"),
                {"id": job_id},
            )
            .mappings()
            .first()
        )
        if not job:
            return {"error": "job_not_found", "job_id": job_id}

        # Атомарный claim (issue #1621): если job уже 'running', значит другое
        # task-сообщение на тот же job_id уже в работе — выходим, не дублируя
        # WAF-хиты и не затирая счётчики параллельным циклом.
        if not _start_job(db, job_id):
            logger.info(
                "process_nspd_geo_job: job=%s already claimed (running) — skipping duplicate",
                job_id,
            )
            return {"job_id": job_id, "skipped": True, "reason": "already_running"}
        # Если в job-строке нет rate_ms — берём глобальный дефолт из job_settings.
        # Это позволяет менять дефолт через /admin/jobs/settings без перезапуска.
        if job["rate_ms"]:
            rate_ms = int(job["rate_ms"])
        else:
            from app.services.job_settings import get_setting_value

            rate_ms = int(get_setting_value("nspd_geo", "rate_ms", 600, db=db))

        # Текущие счётчики (для resume — продолжаем с тех значений)
        done = int(job["targets_done"] or 0)
        failed = int(job["targets_failed"] or 0)
        skipped = int(job["targets_skipped"] or 0)
        n_requests = int(job["requests_count"] or 0)
        n_waf = int(job["waf_blocked_count"] or 0)
        consecutive_waf = 0

        logger.info("process_nspd_geo_job start: job=%s rate=%dms", job_id, rate_ms)

        while True:
            target = (
                db.execute(
                    text(
                        """
                    SELECT target_id, cad_num, thematic_id, attempts
                      FROM nspd_geo_targets
                     WHERE job_id = :id AND status = 'pending'
                     ORDER BY target_id
                     LIMIT 1
                    """
                    ),
                    {"id": job_id},
                )
                .mappings()
                .first()
            )
            if not target:
                break

            tid = target["thematic_id"]
            cad = target["cad_num"]

            try:
                # NSPD geo fetch теперь только через rosreestr2coord lib (v5+).
                # Старая ветка с urllib-fetch (fetch_geoportal) удалена — community
                # пакет нам надёжнее, авторы обновляют под WAF.
                payload = {
                    "data": {
                        "features": [fetch_via_rosreestr2coord(cad, area_type=tid, timeout=15)]
                    }
                }
                payload["data"]["features"] = [f for f in payload["data"]["features"] if f]
                # rosreestr2coord v5 убрал internal delay-параметр → rate limit
                # делаем здесь, чтобы не словить WAF за высокую частоту.
                if rate_ms > 0:
                    time.sleep(rate_ms / 1000.0)
                n_requests += 1
                consecutive_waf = 0

                n_feats, saved_to = _persist_target(db, tid, payload, cad)
                db.execute(
                    text(
                        """
                        UPDATE nspd_geo_targets
                           SET status = 'done', result_features = :f,
                               saved_to_table = :s, fetched_at = NOW(),
                               attempts = attempts + 1
                         WHERE target_id = :tid
                        """
                    ),
                    {"f": n_feats, "s": saved_to or None, "tid": target["target_id"]},
                )
                db.commit()
                done += 1

            except NspdLiteWafError as e:
                consecutive_waf += 1
                n_waf += 1
                wait = WAF_BACKOFF_BASE_S * min(consecutive_waf, 8)
                logger.warning(
                    "WAF on %s (consecutive=%d), backoff %ds: %s",
                    cad,
                    consecutive_waf,
                    wait,
                    e,
                )
                # НЕ помечаем failed — остаётся pending для следующей итерации
                _heartbeat(db, job_id, requests_count=n_requests, waf_blocked_count=n_waf)
                _log(
                    db,
                    job_id,
                    "warn",
                    "waf",
                    f"WAF/429 (consecutive={consecutive_waf}), backoff {wait}s",
                    cad_num=cad,
                )
                time.sleep(wait)
                if consecutive_waf >= 8:
                    # 8 подряд WAF — пауза job'а до следующего worker_ready
                    _finish_job(
                        db,
                        job_id,
                        "paused",
                        error=f"WAF blocked {consecutive_waf} consecutive",
                    )
                    _log(
                        db,
                        job_id,
                        "error",
                        "paused",
                        f"job paused after {consecutive_waf} consecutive WAF blocks",
                    )
                    return {"job_id": job_id, "paused": True, "reason": "waf"}
                continue

            except SoftTimeLimitExceeded:
                # Celery soft_time_limit=6h истёк — graceful shutdown ДО generic except.
                # Issue #1235: `except (NspdLiteError, Exception)` ниже проглатывал бы это
                # исключение (SoftTimeLimitExceeded унаследован от Exception) → таска
                # продолжала бы цикл неограниченно дольше 6ч, занимая слот geo-очереди.
                # Контракт: коммитим финальный heartbeat + статусы targets, помечаем job
                # 'paused' (auto-resume через cleanup_zombies/worker_ready), re-raise чтобы
                # Celery освободил воркер и зарегистрировал лимит.
                logger.warning(
                    "process_nspd_geo_job soft time limit exceeded: job=%s done=%d failed=%d",
                    job_id,
                    done,
                    failed,
                )
                try:
                    _heartbeat(
                        db,
                        job_id,
                        targets_done=done,
                        targets_failed=failed,
                        targets_skipped=skipped,
                        requests_count=n_requests,
                        waf_blocked_count=n_waf,
                    )
                    _finish_job(
                        db,
                        job_id,
                        "paused",
                        error=f"soft_time_limit exceeded after {done + failed} targets",
                    )
                    _log(
                        db,
                        job_id,
                        "warn",
                        "paused",
                        f"job paused: Celery soft_time_limit (6h) reached "
                        f"after {done} done / {failed} failed",
                    )
                except Exception as flush_err:
                    # Финальный flush не должен скрыть SoftTimeLimitExceeded — лог + re-raise.
                    logger.exception(
                        "process_nspd_geo_job: failed to flush state on soft-limit: %s",
                        flush_err,
                    )
                raise

            except (NspdLiteError, Exception) as e:
                logger.warning("fetch failed for %s: %s", cad, e)
                _log(db, job_id, "warn", "fetch_fail", str(e)[:300], cad_num=cad)
                db.execute(
                    text(
                        """
                        UPDATE nspd_geo_targets
                           SET status = CASE WHEN attempts >= 2 THEN 'failed' ELSE 'pending' END,
                               attempts = attempts + 1,
                               error_msg = :em
                         WHERE target_id = :tid
                        """
                    ),
                    {"em": str(e)[:500], "tid": target["target_id"]},
                )
                db.commit()
                if int(target["attempts"]) >= 2:
                    failed += 1

            if (done + failed + skipped) % HEARTBEAT_EVERY == 0:
                _heartbeat(
                    db,
                    job_id,
                    targets_done=done,
                    targets_failed=failed,
                    targets_skipped=skipped,
                    requests_count=n_requests,
                    waf_blocked_count=n_waf,
                )

        _heartbeat(
            db,
            job_id,
            targets_done=done,
            targets_failed=failed,
            targets_skipped=skipped,
            requests_count=n_requests,
            waf_blocked_count=n_waf,
        )
        # done даже при failed targets — UI покажет failed_count отдельно
        _finish_job(db, job_id, "done")
        _log(
            db,
            job_id,
            "info",
            "done",
            f"done: {done} ok / {failed} failed / {n_requests} req / {n_waf} WAF blocks",
        )
        logger.info(
            "process_nspd_geo_job done: job=%s done=%d failed=%d req=%d waf=%d",
            job_id,
            done,
            failed,
            n_requests,
            n_waf,
        )
        return {
            "job_id": job_id,
            "done": done,
            "failed": failed,
            "skipped": skipped,
            "requests": n_requests,
            "waf_blocked": n_waf,
        }
    except SoftTimeLimitExceeded:
        # Inner handler уже коммитнул heartbeat + статус 'paused' + лог.
        # Не перетираем 'paused' на 'failed' через outer fallback — просто пробрасываем,
        # чтобы Celery освободил worker и зарегистрировал срабатывание soft_time_limit.
        raise
    except Exception as e:
        logger.exception("process_nspd_geo_job crashed: job=%s: %s", job_id, e)
        try:
            _finish_job(db, job_id, "failed", error=f"{type(e).__name__}: {e}")
        except Exception:
            pass
        raise
    finally:
        db.close()


@celery_app.task(name="tasks.nspd_geo.cleanup_zombies")
def cleanup_zombies() -> dict[str, Any]:
    """Periodic zombie cleanup — runs every minute via beat schedule.

    Re-enqueues stale nspd_geo_jobs split by status (issue #1215):
      - status='running' с heartbeat старше _ZOMBIE_RUNNING_THRESHOLD (6 мин) →
        выше max WAF-backoff (240s), не ре-enqueue'ит живые WAF-паузы внутри цикла.
      - status='paused' с heartbeat старше _ZOMBIE_PAUSED_THRESHOLD (30 мин) →
        WAF-пауза (consecutive_waf>=8) живёт минимум 30 мин, иначе минутный beat
        сразу аннулировал бы защиту и worker снова долбил бы забаненный сервис.
      - status='queued' c НЕ-NULL heartbeat старше _ZOMBIE_QUEUED_THRESHOLD (6 мин)
        → ре-enqueue'нутый job, чьё broker-сообщение потерялось (issue #1655).
        heartbeat_at IS NOT NULL отсекает свежесозданные jobs (initial apply_async
        ещё в полёте) — их heartbeat ставится только при первом _start_job.

    Idempotent: если зомби нет, ничего не делает. Активный job с свежим heartbeat
    не матчит WHERE-clause.
    """
    db = SessionLocal()
    resumed: list[int] = []
    try:
        rows = (
            db.execute(
                text(
                    """
                    UPDATE nspd_geo_jobs
                       SET status = 'queued',
                           error = COALESCE(error, 'auto-resume by beat cleanup')
                     WHERE (
                               status = 'running'
                               AND heartbeat_at
                                   < NOW() - CAST(:running_threshold AS interval)
                           )
                        OR (
                               status = 'paused'
                               AND heartbeat_at
                                   < NOW() - CAST(:paused_threshold AS interval)
                           )
                        OR (
                               -- issue #1655: ре-enqueue'нутый job застрял в 'queued'
                               -- (потерянное broker-сообщение). heartbeat_at IS NOT NULL
                               -- отсекает свежесозданные jobs с ещё-в-полёте apply_async.
                               status = 'queued'
                               AND heartbeat_at IS NOT NULL
                               AND heartbeat_at
                                   < NOW() - CAST(:queued_threshold AS interval)
                           )
                    RETURNING job_id
                    """
                ),
                {
                    "running_threshold": _ZOMBIE_RUNNING_THRESHOLD,
                    "paused_threshold": _ZOMBIE_PAUSED_THRESHOLD,
                    "queued_threshold": _ZOMBIE_QUEUED_THRESHOLD,
                },
            )
            .mappings()
            .all()
        )
        db.commit()
        resumed = [int(r["job_id"]) for r in rows]
        for jid in resumed:
            try:
                process_nspd_geo_job.apply_async(args=[jid], queue="geo")
                logger.info("cleanup_zombies: re-enqueued job=%s", jid)
            except Exception as e:
                logger.warning("cleanup_zombies: enqueue job=%s failed: %s", jid, e)
        if resumed:
            logger.info("cleanup_zombies: resumed %d zombie job(s): %s", len(resumed), resumed)
    except Exception as e:
        logger.exception("cleanup_zombies failed: %s", e)
        try:
            db.rollback()
        except Exception:
            pass
    finally:
        db.close()
    return {"resumed_count": len(resumed), "resumed_job_ids": resumed}