gendesign/backend/app/workers/beat_schedule.py

"""Beat schedule builder for Celery.

Priority order:
1. DB-driven — читает таблицу job_settings при старте beat-контейнера.
2. Fallback — env-based расписание (SCRAPE_KN_CRON, OBJECTIVE_SYNC_CRON),
   используется если БД недоступна или таблица пустая.
3. Hardcoded entries — добавляются поверх DB-расписания (не в job_settings):
   - refresh-ekb-districts-medians (аналитика, ежемесячно 5-го в 04:00)
   - nspd-geo-zombie-cleanup (каждую минуту, замена нестабильного worker_ready)
   - poi-sync-weekly (OSM POI, понедельник 03:00)
   - noise-sync-weekly (OSM шум, понедельник 03:30)

Публичный API: build_beat_schedule() → dict.
"""

import logging

from celery.schedules import crontab

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()]


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

    Returns: (schedule, rows_seen).
    - rows_seen > 0  → DB reachable, есть конфигурация job_settings (даже если все disabled)
    - rows_seen == 0 → DB reachable, но таблица пустая (fresh install до seed'а)
    - **RAISES** при недоступности БД (table missing, connection refused и т.п.) —
      caller должен поймать и использовать env-based fallback как safety net.

    Используем `get_all` (strict, raises on failure), НЕ `get_all_safe` — нам важно
    отличить "DB unreachable" от "DB сказал ничего не schedule'ить".
    Без этого пара UPDATE job_settings SET enabled=false бессильна:
    `_build_beat_schedule_from_db` вернул бы {} как при DB error → caller fallback'ает
    на env-cron → отключённые jobs снова в schedule. См. инцидент 2026-05-24
    (WAF cooldown disable scrape_kn → fallback вернул его обратно).
    """
    from app.core.db import SessionLocal
    from app.services.job_settings import get_all

    db = SessionLocal()
    try:
        rows = get_all(db)
    finally:
        db.close()

    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, len(rows)


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 (job_settings) если DB reachable И rows > 0
    - Env-based fallback ТОЛЬКО когда DB unreachable ИЛИ table пустая (fresh install)
    - Пустой schedule с rows>0 (все jobs disabled намеренно) — RESPECT, без fallback

    Всегда добавляет hardcoded entries поверх (refresh-analytics, OSM POI/noise,
    nspd cleanup) — они не управляются через job_settings.
    """
    schedule: dict = {}
    rows_seen = 0
    db_reachable = True
    try:
        schedule, rows_seen = _build_beat_schedule_from_db()
    except Exception as e:
        logger.error(
            "build_beat_schedule: DB unreachable: %s — fallback на env vars (safety net)", e
        )
        db_reachable = False

    if not db_reachable or rows_seen == 0:
        # DB недоступна ИЛИ fresh install (job_settings пустая) → env-based safety net
        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 — DISABLED 2026-05-14 per Bug_NSPD_WMS_NotBulk
    # post-mortem (vault: fixes/Bug_NSPD_WMS_NotBulk_2026_May14.md).
    #
    # Action item #1: disable beat schedule до Sprint 2 fix (grid sampling
    # rewrite). harvest_quarter в текущей реализации пишет почти пустые dumps
    # из-за single-pixel WMS GetFeatureInfo bug. Запуск Mon 04:00 МСК потратит
    # rate-limit budget и заполнит nspd_quarter_dumps мусором.
    #
    # Task code остаётся в tasks/nspd_sync.py — re-enable после Sprint 2 grid
    # sampling rewrite (см. Bug_NSPD_WMS_NotBulk_2026_May14 → Sprint 2 fix-strategy).
    # До тех пор harvest_quarter можно вызывать вручную через admin endpoint.
    #
    # 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