gendesign/backend/app/services/job_settings.py

"""Сервис для чтения и обновления централизованных настроек Celery-job'ов.

Таблица job_settings содержит по одной строке на job_type:
  'scrape_kn' | 'nspd_geo' | 'objective_etl' | 'objective_sync'

Используется:
  - workers.celery_app: строит beat_schedule при старте (один раз)
  - api.v1.admin_jobs: GET/PUT /admin/jobs/settings
  - workers.tasks.*: читают rate_ms / extra_config при каждом запуске

Безопасный fallback: если строка отсутствует или БД недоступна — возвращает
hardcoded defaults чтобы не ломать старый код.
"""

from __future__ import annotations

import logging
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import text

logger = logging.getLogger(__name__)

# ── Fallback defaults (до инициализации БД / отсутствующие строки) ────────────

_DEFAULTS: dict[str, dict[str, Any]] = {
    "scrape_kn": {
        "enabled": True,
        "queue_name": "scrape_kn",
        "cron_schedule": "15 4 * * mon",
        "rate_ms": None,
        "max_retries": 2,
        "max_concurrency": 1,
        "extra_config": {"default_regions": [66]},
        "description": "Скрейпинг КН (fallback default)",
    },
    "nspd_geo": {
        "enabled": True,
        "queue_name": "geo",
        "cron_schedule": None,
        "rate_ms": 600,
        "max_retries": 2,
        "max_concurrency": 1,
        "extra_config": {},
        "description": "NSPD geo backfill (fallback default)",
    },
    "objective_etl": {
        "enabled": True,
        "queue_name": "celery",
        "cron_schedule": None,
        "rate_ms": None,
        "max_retries": 2,
        "max_concurrency": 1,
        "extra_config": {"sqlite_path": "/data/anton-sqlite/analysis.db"},
        "description": "ETL SQLite → PG (fallback default)",
    },
    "objective_sync": {
        "enabled": True,
        "queue_name": "celery",
        "cron_schedule": "0 6 * * tue",
        "rate_ms": None,
        "max_retries": 2,
        "max_concurrency": 1,
        "extra_config": {
            "groups_csv": "Свердловская область,Челябинск,Тюмень,Пермь",
            "use_ddu": True,
            "use_dkp": True,
            "period_months_back": 1,
            "inter_group_delay_s": 30,
            "rate_ms": 3000,
        },
        "description": "Objective sync (fallback default)",
    },
}

_FALLBACK_UPDATED_AT = datetime(2026, 1, 1, tzinfo=UTC)


def _row_to_dict(r: Any) -> dict[str, Any]:
    """Маппинг ORM / sqlalchemy row → dict."""
    return {
        "job_type": r["job_type"],
        "enabled": bool(r["enabled"]),
        "queue_name": r["queue_name"],
        "cron_schedule": r["cron_schedule"],
        "rate_ms": r["rate_ms"],
        "max_retries": int(r["max_retries"]),
        "max_concurrency": int(r["max_concurrency"]),
        "extra_config": r["extra_config"] or {},
        "updated_at": r["updated_at"],
        "updated_by": r["updated_by"],
        "description": r["description"],
    }


def _fallback(job_type: str) -> dict[str, Any]:
    """Возвращает fallback dict для job_type; логирует warning."""
    logger.warning("job_settings: строка '%s' не найдена — используем hardcoded defaults", job_type)
    base = _DEFAULTS.get(
        job_type,
        {
            "enabled": True,
            "queue_name": "celery",
            "cron_schedule": None,
            "rate_ms": None,
            "max_retries": 2,
            "max_concurrency": 1,
            "extra_config": {},
            "description": f"Unknown job_type '{job_type}' — fallback",
        },
    )
    return {
        "job_type": job_type,
        **base,
        "updated_at": _FALLBACK_UPDATED_AT,
        "updated_by": None,
    }


# ── Public API ────────────────────────────────────────────────────────────────


def get_all(db) -> list[dict[str, Any]]:
    """Вернуть все строки job_settings. При ошибке БД — fallback на _DEFAULTS."""
    try:
        rows = (
            db.execute(
                text(
                    """
                    SELECT job_type, enabled, queue_name, cron_schedule, rate_ms,
                           max_retries, max_concurrency, extra_config,
                           updated_at, updated_by, description
                      FROM job_settings
                     ORDER BY job_type
                    """
                )
            )
            .mappings()
            .all()
        )
    except Exception as e:
        logger.warning("get_all job_settings: БД недоступна — fallback. %s", e)
        return [_fallback(jt) for jt in _DEFAULTS]

    if not rows:
        logger.warning("get_all job_settings: таблица пустая — fallback")
        return [_fallback(jt) for jt in _DEFAULTS]

    return [_row_to_dict(r) for r in rows]


def get_one(job_type: str, db) -> dict[str, Any]:
    """Вернуть одну строку по job_type. При отсутствии — fallback с warning."""
    try:
        row = (
            db.execute(
                text(
                    """
                    SELECT job_type, enabled, queue_name, cron_schedule, rate_ms,
                           max_retries, max_concurrency, extra_config,
                           updated_at, updated_by, description
                      FROM job_settings
                     WHERE job_type = :jt
                    """
                ),
                {"jt": job_type},
            )
            .mappings()
            .first()
        )
    except Exception as e:
        logger.warning("get_one job_settings '%s': БД недоступна — fallback. %s", job_type, e)
        return _fallback(job_type)

    if row is None:
        return _fallback(job_type)

    return _row_to_dict(row)


def update(
    job_type: str,
    db,
    *,
    enabled: bool | None = None,
    queue_name: str | None = None,
    cron_schedule: str | None = None,
    rate_ms: int | None = None,
    max_retries: int | None = None,
    max_concurrency: int | None = None,
    extra_config: dict | None = None,
    updated_by: str | None = None,
) -> dict[str, Any]:
    """PATCH-style update — обновляет только переданные поля.

    cron_schedule='' → NULL (manual-only). updated_at выставляется в NOW().
    """
    sets: list[str] = ["updated_at = NOW()"]
    params: dict[str, Any] = {"jt": job_type}

    for field, value in [
        ("enabled", enabled),
        ("queue_name", queue_name),
        ("max_retries", max_retries),
        ("max_concurrency", max_concurrency),
        ("updated_by", updated_by),
    ]:
        if value is not None:
            sets.append(f"{field} = :{field}")
            params[field] = value

    # cron_schedule: None = не трогать; '' = явный NULL (manual-only)
    if cron_schedule is not None:
        sets.append("cron_schedule = :cron_schedule")
        params["cron_schedule"] = cron_schedule if cron_schedule else None

    # rate_ms: None = не трогать; 0 = явно обнулить (no throttle)
    if rate_ms is not None:
        sets.append("rate_ms = :rate_ms")
        params["rate_ms"] = rate_ms if rate_ms > 0 else None

    if extra_config is not None:
        sets.append("extra_config = CAST(:extra_config AS jsonb)")
        import json

        params["extra_config"] = json.dumps(extra_config, ensure_ascii=False)

    db.execute(
        text(f"UPDATE job_settings SET {', '.join(sets)} WHERE job_type = :jt"),
        params,
    )
    db.commit()
    return get_one(job_type, db)


def get_setting_value(job_type: str, key: str, default: Any = None, db=None) -> Any:
    """Читает значение из top-level полей строки или из extra_config[key].

    Порядок поиска:
      1. Прямое поле строки (enabled, rate_ms, queue_name, ...)
      2. extra_config[key]
      3. default

    Если db=None — возвращает default из _DEFAULTS[job_type].extra_config[key].
    """
    top_level_fields = {
        "enabled",
        "queue_name",
        "cron_schedule",
        "rate_ms",
        "max_retries",
        "max_concurrency",
        "updated_at",
        "updated_by",
        "description",
    }
    if db is None:
        defaults = _DEFAULTS.get(job_type, {})
        if key in top_level_fields:
            return defaults.get(key, default)
        return (defaults.get("extra_config") or {}).get(key, default)

    row = get_one(job_type, db)
    if key in top_level_fields:
        return row.get(key, default)
    return (row.get("extra_config") or {}).get(key, default)


def get_all_safe(db_factory) -> list[dict[str, Any]]:
    """Безопасное чтение всех строк при старте Celery beat — открывает свою сессию.

    db_factory: callable () -> Session (обычно SessionLocal)
    При любой ошибке — fallback на hardcoded defaults.
    """
    try:
        db = db_factory()
        try:
            return get_all(db)
        finally:
            db.close()
    except Exception as e:
        logger.warning("get_all_safe job_settings: fallback на defaults: %s", e)
        return [_fallback(jt) for jt in _DEFAULTS]