gendesign/backend/app/workers/celery_app.py

"""Celery app + beat schedule.

Beat schedule строится из таблицы job_settings при старте beat-контейнера.
При недоступности БД — fallback на env-based расписание (старое поведение).
"""

import logging

from celery import Celery
from celery.schedules import crontab
from celery.signals import worker_process_init, worker_ready

from app.core.config import settings

logger = logging.getLogger(__name__)

# Словарь job_type → celery task name (полное dotted name).
# Используется при построении beat_schedule из job_settings.
TASK_MAP: dict[str, str] = {
    "scrape_kn": "tasks.scrape_kn.scrape_kn_region",
    "objective_sync": "tasks.scrape_objective.sync_all_groups",
    # nspd_geo и objective_etl не имеют cron — запускаются вручную.
    # При необходимости можно добавить сюда.
}


def _parse_cron(spec: str) -> crontab:
    """Parse 'M H DoM Mon DoW' crontab string into Celery crontab. Empty fields default to '*'."""
    parts = spec.strip().split()
    if len(parts) != 5:
        raise ValueError(f"crontab spec must have 5 fields, got: {spec!r}")
    minute, hour, dom, month, dow = parts
    return crontab(
        minute=minute,
        hour=hour,
        day_of_month=dom,
        month_of_year=month,
        day_of_week=dow,
    )


def _default_regions() -> list[int]:
    return [int(x.strip()) for x in settings.scrape_kn_default_regions.split(",") if x.strip()]


celery_app = Celery(
    "gendesign",
    broker=settings.redis_url,
    backend=settings.redis_url,
    include=[
        "app.workers.tasks.scrape_kn",
        "app.workers.tasks.refresh_analytics",
        "app.workers.tasks.scrape_objective",
        "app.workers.tasks.objective_etl",
        "app.workers.tasks.nspd_geo",
        "app.workers.tasks.nspd_sync",
        "app.workers.tasks.poi_sync",
        "app.workers.tasks.noise_sync",
        "app.workers.tasks.pzz_sync",
    ],
)
celery_app.conf.timezone = "Europe/Moscow"


def _build_beat_schedule_from_db() -> dict:
    """Строит beat_schedule из job_settings БД.

    Возвращает пустой словарь при любой ошибке — caller добавит fallback.
    """
    try:
        from app.core.db import SessionLocal
        from app.services.job_settings import get_all_safe

        rows = get_all_safe(SessionLocal)
    except Exception as e:
        logger.warning("build_beat_schedule: не удалось прочитать job_settings: %s", e)
        return {}

    schedule: dict = {}
    for row in rows:
        jt = row["job_type"]
        if not row.get("enabled"):
            continue
        cron_str = row.get("cron_schedule")
        if not cron_str:
            # NULL cron = только ручной запуск, не добавляем в beat
            continue
        if jt not in TASK_MAP:
            logger.warning(
                "build_beat_schedule: job_type '%s' не имеет TASK_MAP-записи — пропускаем", jt
            )
            continue
        try:
            cron = _parse_cron(cron_str)
        except Exception as e:
            logger.warning(
                "build_beat_schedule: невалидный cron '%s' для '%s': %s", cron_str, jt, e
            )
            continue

        queue = row.get("queue_name", "celery") or "celery"

        if jt == "scrape_kn":
            # scrape_kn запускается отдельной задачей на каждый регион
            extra = row.get("extra_config") or {}
            regions: list[int] = extra.get("default_regions") or _default_regions()
            for rc in regions:
                schedule[f"kn-region-{rc}"] = {
                    "task": TASK_MAP[jt],
                    "schedule": cron,
                    "args": [rc, None],
                    "options": {"queue": queue},
                }
            logger.info(
                "build_beat_schedule: scrape_kn cron=%r regions=%s queue=%s",
                cron_str,
                regions,
                queue,
            )
        else:
            kwargs: dict = {}
            if jt == "objective_sync":
                kwargs = {"triggered_by": "beat"}
            schedule[jt] = {
                "task": TASK_MAP[jt],
                "schedule": cron,
                "kwargs": kwargs,
                "options": {"queue": queue},
            }
            logger.info("build_beat_schedule: %s cron=%r queue=%s", jt, cron_str, queue)

    return schedule


def _build_beat_schedule_fallback() -> dict:
    """Fallback-расписание из env vars — используется если БД недоступна на старте."""
    logger.warning("beat_schedule: строим fallback из env vars")
    schedule: dict = {}

    # scrape_kn
    try:
        for rc in _default_regions():
            schedule[f"kn-region-{rc}"] = {
                "task": "tasks.scrape_kn.scrape_kn_region",
                "schedule": _parse_cron(settings.scrape_kn_cron),
                "args": [rc, None],
            }
    except Exception as e:
        logger.warning("beat_schedule fallback: scrape_kn failed: %s", e)

    # objective_sync
    try:
        from app.core.db import SessionLocal
        from app.services.objective_sync_config import get_cron_schedule_safe

        cron_str = get_cron_schedule_safe(SessionLocal)
    except Exception:
        cron_str = settings.objective_sync_cron

    try:
        schedule["objective-sync"] = {
            "task": "tasks.scrape_objective.sync_all_groups",
            "schedule": _parse_cron(cron_str),
            "kwargs": {"triggered_by": "beat"},
        }
    except Exception as e:
        logger.warning("beat_schedule fallback: objective_sync failed: %s", e)

    # refresh ekb-districts medians — ежемесячно 5-го числа в 04:00 МСК
    try:
        schedule["refresh-ekb-districts-medians"] = {
            "task": "tasks.refresh_analytics.refresh_ekb_districts_medians",
            "schedule": _parse_cron("0 4 5 * *"),
            "kwargs": {"window_months": 24, "min_deals": 50},
        }
    except Exception as e:
        logger.warning("beat_schedule fallback: refresh_analytics failed: %s", e)

    return schedule


def _build_beat_schedule() -> dict:
    """Строит beat_schedule: сначала из DB, при неудаче — fallback на env.

    Всегда добавляет refresh-ekb-districts-medians (нет в job_settings).
    """
    schedule: dict = {}
    try:
        schedule = _build_beat_schedule_from_db()
    except Exception as e:
        logger.error("_build_beat_schedule: неожиданная ошибка: %s — fallback", e)

    if not schedule:
        # БД пустая или недоступна — полный fallback
        return _build_beat_schedule_fallback()

    # Всегда добавляем refresh_analytics (этот job нет в job_settings,
    # он технический и не требует конфигурации через UI).
    if "refresh-ekb-districts-medians" not in schedule:
        try:
            schedule["refresh-ekb-districts-medians"] = {
                "task": "tasks.refresh_analytics.refresh_ekb_districts_medians",
                "schedule": _parse_cron("0 4 5 * *"),
                "kwargs": {"window_months": 24, "min_deals": 50},
                "options": {"queue": "celery"},
            }
        except Exception as e:
            logger.warning("beat_schedule: refresh_analytics failed: %s", e)

    # Zombie cleanup: каждую минуту проверяет nspd_geo_jobs с heartbeat > 2 мин
    # и re-enqueue'ит их. Замена worker_ready signal handler'а — он не fires
    # стабильно в нашем setup, beat-task надёжнее.
    schedule["nspd-geo-zombie-cleanup"] = {
        "task": "tasks.nspd_geo.cleanup_zombies",
        "schedule": _parse_cron("* * * * *"),
        "options": {"queue": "celery"},
    }

    # OSM POI sync — еженедельно в понедельник в 03:00 МСК
    schedule["poi-sync-weekly"] = {
        "task": "tasks.poi_sync.sync_osm_poi_ekb",
        "schedule": _parse_cron("0 3 * * mon"),
        "options": {"queue": "celery"},
    }

    # OSM noise sources sync — еженедельно в понедельник в 03:30 МСК
    # (через 30 мин после poi-sync, чтобы не нагружать Overpass одновременно)
    schedule["noise-sync-weekly"] = {
        "task": "tasks.noise_sync.sync_osm_noise_sources_ekb",
        "schedule": _parse_cron("30 3 * * mon"),
        "options": {"queue": "celery"},
    }

    # NSPD quarter dump refresh — понедельник 04:00 МСК.
    # Fanout'ит harvest_quarter.apply_async для кварталов старше 90 дней или
    # отсутствующих в nspd_quarter_dumps. batch_size=50 — ~5 мин при 600ms rate.
    # Запускается через час после poi-sync чтобы не создавать конкуренцию.
    schedule["nspd-harvest-stale-quarters"] = {
        "task": "tasks.nspd_sync.harvest_stale_quarters",
        "schedule": _parse_cron("0 4 * * mon"),
        "kwargs": {"region_code": 66, "max_age_days": 90, "batch_size": 50},
        "options": {"queue": "celery"},
    }

    return schedule


celery_app.conf.beat_schedule = _build_beat_schedule()


@worker_process_init.connect
def _reset_db_connections(sender=None, **_kwargs) -> None:
    """Dispose SQLAlchemy engine in each prefork child process.

    Background: Celery prefork pool fork()'s child processes from the master
    that already imported SQLAlchemy + psycopg. The child inherits open TCP
    sockets to PostgreSQL, but psycopg connections cannot be safely shared
    across processes after fork — first use raises:
        psycopg.ProgrammingError: can't change 'autocommit' now:
        connection in transaction status INTRANS

    Fix: dispose the engine in each child. SQLAlchemy will lazily open fresh
    connections per-child. Standard Celery + SQLA recipe.
    """
    try:
        from app.core.db import engine

        engine.dispose()
        logger.info("worker_process_init: SQLAlchemy engine disposed for fork")
    except Exception as e:
        logger.warning("worker_process_init: engine dispose failed: %s", e)


@worker_ready.connect
def _resume_zombie_runs(sender=None, **_kwargs) -> None:
    """When a worker finishes booting (after redeploy/restart), find any sweep
    that was 'running' or 'paused' and re-enqueue a resume task. Each resume
    creates a fresh run_id linked via resumed_from_run_id; the original row is
    marked 'zombie' so the audit trail is preserved.

    No time threshold: by definition, on worker_ready ANY 'running' row is a
    zombie because there is no active worker. Previously we required heartbeat
    >5 min, which skipped jobs interrupted seconds before redeploy — they
    stayed in 'running' status forever and required manual cancel/resume.
    """
    logger.info("worker_ready: resume scan starting")
    from sqlalchemy import text

    # Persistent breadcrumb: write to nspd_geo_log so we can confirm via DB
    # query whether worker_ready actually fired (independent of container logs).
    # NULL job_id to avoid FK violation (FK is to nspd_geo_jobs.job_id which
    # has no 0 row).
    try:
        from app.core.db import SessionLocal

        _db = SessionLocal()
        try:
            _db.execute(
                text(
                    "INSERT INTO nspd_geo_log (job_id, level, stage, message) "
                    "VALUES (NULL, 'info', 'worker_ready', 'worker_ready signal fired')"
                )
            )
            _db.commit()
        except Exception:
            _db.rollback()
            raise
        finally:
            _db.close()
    except Exception as _e:
        logger.warning("worker_ready: breadcrumb insert failed: %s", _e)

    from app.core.db import SessionLocal

    # ── kn_scrape_runs resume ──
    db = SessionLocal()
    ids: list[int] = []
    try:
        rows = (
            db.execute(
                text(
                    """
                    SELECT run_id
                      FROM kn_scrape_runs
                     WHERE status = 'running'
                       AND objects_snapshot IS NOT NULL
                     ORDER BY started_at ASC
                     LIMIT 20
                    """
                )
            )
            .mappings()
            .all()
        )
        if rows:
            ids = [int(r["run_id"]) for r in rows]
            # Помечаем найденные как 'zombie' одним апдейтом — resume создаст новые
            # run_id со ссылкой resumed_from_run_id.
            db.execute(
                text(
                    """
                    UPDATE kn_scrape_runs
                       SET status = 'zombie',
                           finished_at = NOW(),
                           error = COALESCE(error,
                               'auto-zombie at worker_ready, resume scheduled')
                     WHERE run_id = ANY(:ids)
                    """
                ),
                {"ids": ids},
            )
            db.commit()
        else:
            logger.info("worker_ready: нет stale kn runs для resume")
    except Exception as e:
        logger.exception("worker_ready kn resume scan failed: %s", e)
        try:
            db.rollback()
        except Exception:
            pass
    finally:
        db.close()

    # Enqueue resume tasks. Late import to avoid circular at module load.
    from app.workers.tasks.scrape_kn import resume_kn_run

    for rid in ids:
        try:
            resume_kn_run.apply_async(args=[rid])
            logger.info("worker_ready: resume_kn_run enqueued for run=%s", rid)
        except Exception as e:
            logger.warning("worker_ready: failed to enqueue resume for run=%s: %s", rid, e)

    # NSPD-runs (старый Playwright-scraper) resume УДАЛЁН 2026-05-11. Скрапер
    # снят с эксплуатации в пользу bulk geo-fetcher (nspd_geo). История
    # старых runs сохранена в nspd_scrape_runs — никаких side-effects.

    # NSPD geo-jobs: bulk-fetcher с собственной resume-логикой через
    # nspd_geo_jobs / nspd_geo_targets. Resume любых 'running' / 'paused' jobs
    # — на worker_ready по определению нет активных воркеров, всё running ==
    # zombie. Раньше требовали heartbeat >10мин, что пропускало jobs убитых
    # за минуту до редеплоя и оставляло их вечно висеть.
    db = SessionLocal()
    geo_resume_jobs: list[int] = []
    try:
        rows = (
            db.execute(
                text(
                    """
                    UPDATE nspd_geo_jobs
                       SET status = 'queued',
                           error = COALESCE(error, 'auto-resume at worker_ready')
                     WHERE status IN ('running', 'paused')
                    RETURNING job_id
                    """
                )
            )
            .mappings()
            .all()
        )
        db.commit()
        geo_resume_jobs = [int(r["job_id"]) for r in rows]
        for jid in geo_resume_jobs:
            logger.info("worker_ready: NSPD geo job=%s — resume scheduled", jid)
    except Exception as e:
        logger.warning("worker_ready nspd_geo resume scan failed: %s", e)
        try:
            db.rollback()
        except Exception:
            pass
    finally:
        db.close()

    if geo_resume_jobs:
        from app.workers.tasks.nspd_geo import process_nspd_geo_job

        for jid in geo_resume_jobs:
            try:
                process_nspd_geo_job.apply_async(args=[jid], queue="geo")
                logger.info("worker_ready: process_nspd_geo_job enqueued job=%s", jid)
            except Exception as e:
                logger.warning("worker_ready: failed to enqueue geo resume job=%s: %s", jid, e)
    logger.info("worker_ready: resume scan finished (geo_jobs=%d)", len(geo_resume_jobs))

    # Sanity check: nspd_quarter_dumps table must exist (migration 88).
    # Logs critical error but does NOT crash the worker — table may be absent
    # in dev/staging before migration is applied.
    try:
        from app.core.db import SessionLocal as _SessionLocal

        _db = _SessionLocal()
        try:
            _db.execute(text("SELECT 1 FROM nspd_quarter_dumps LIMIT 0"))
            logger.info("worker_ready: nspd_quarter_dumps table OK")
        except Exception as _te:
            logger.critical(
                "worker_ready: nspd_quarter_dumps table missing or inaccessible — "
                "apply migration 88_nspd_quarter_dumps.sql before using harvest_quarter. "
                "Error: %s",
                _te,
            )
        finally:
            _db.close()
    except Exception as _e:
        logger.warning("worker_ready: nspd_quarter_dumps sanity check failed: %s", _e)