feat(jobs): centralized job_settings API + DB-driven beat schedule

- JobSetting ORM model (JSONB extra_config) + Pydantic schemas
- GET/PUT /api/v1/admin/jobs/settings + /{job_type} (X-Admin-Token auth)
- celery_app.py builds beat_schedule from job_settings DB (env fallback
  retained for safety on first deploy / DB unreachable)
- nspd_geo task reads rate_ms from job_settings when per-job row has
  no override
- enqueue/resume geo jobs route to queue_name from job_settings
- Worker container: --queues=celery,scrape_kn,geo (one container,
  three named queues — kn sweep no longer blocks nspd_geo)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/app/api/v1/admin_jobs.py

@@ -0,0 +1,89 @@
+"""Admin API для централизованных настроек Celery-job'ов.
+
+GET  /api/v1/admin/jobs/settings         — список всех job_settings
+GET  /api/v1/admin/jobs/settings/{type}  — одна строка по job_type
+PUT  /api/v1/admin/jobs/settings/{type}  — PATCH-style update
+
+Auth: X-Admin-Token header (тот же токен что у admin_scrape endpoints).
+"""
+
+from __future__ import annotations
+
+from typing import Annotated, Any
+
+from fastapi import APIRouter, Depends, Header, HTTPException
+from sqlalchemy.orm import Session
+
+from app.core.config import settings
+from app.core.db import get_db
+from app.schemas.job_settings import JobSettingRead, JobSettingUpdate
+
+router = APIRouter()
+
+
+def _check_token(x_admin_token: str | None) -> None:
+    if not settings.scrape_admin_token:
+        raise HTTPException(
+            status_code=503,
+            detail="admin jobs disabled — set SCRAPE_ADMIN_TOKEN",
+        )
+    if x_admin_token != settings.scrape_admin_token:
+        raise HTTPException(status_code=401, detail="invalid admin token")
+
+
+@router.get("/settings", response_model=list[JobSettingRead])
+def list_job_settings(
+    db: Annotated[Session, Depends(get_db)],
+    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
+) -> Any:
+    """Список всех job_settings (все job_type)."""
+    _check_token(x_admin_token)
+    from app.services.job_settings import get_all
+
+    return get_all(db)
+
+
+@router.get("/settings/{job_type}", response_model=JobSettingRead)
+def get_job_setting(
+    job_type: str,
+    db: Annotated[Session, Depends(get_db)],
+    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
+) -> Any:
+    """Настройки одного job_type. Возвращает fallback defaults если строки нет в БД."""
+    _check_token(x_admin_token)
+    from app.services.job_settings import get_one
+
+    return get_one(job_type, db)
+
+
+@router.put("/settings/{job_type}", response_model=JobSettingRead)
+def update_job_setting(
+    job_type: str,
+    payload: JobSettingUpdate,
+    db: Annotated[Session, Depends(get_db)],
+    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
+) -> Any:
+    """PATCH-style update: передавать только изменяемые поля.
+
+    После изменения cron_schedule нужен restart beat-контейнера.
+    Остальные поля (rate_ms, extra_config, enabled) применяются динамически.
+
+    cron_schedule='' устанавливает NULL (manual-only режим).
+    rate_ms=0 устанавливает NULL (no throttle).
+    """
+    _check_token(x_admin_token)
+    from app.services.job_settings import update
+
+    result = update(
+        job_type,
+        db,
+        enabled=payload.enabled,
+        queue_name=payload.queue_name,
+        cron_schedule=payload.cron_schedule,
+        rate_ms=payload.rate_ms,
+        max_retries=payload.max_retries,
+        max_concurrency=payload.max_concurrency,
+        extra_config=payload.extra_config,
+        updated_by="admin-ui",
+    )
+    return result

backend/app/api/v1/admin_scrape.py

@@ -682,7 +682,10 @@ def enqueue_geo_job(
         rate_ms=payload.rate_ms,
         triggered_by="manual",
     )
-    process_nspd_geo_job.apply_async(args=[job_id])
+    from app.services.job_settings import get_setting_value
+
+    geo_queue = get_setting_value("nspd_geo", "queue_name", "geo")
+    process_nspd_geo_job.apply_async(args=[job_id], queue=geo_queue)
     return {
         "job_id": job_id,
         "targets_total": len(cad_with_thematic),
@@ -775,6 +778,7 @@ def resume_geo_job(
 ) -> dict[str, Any]:
     """Re-enqueue paused/failed job. Resume idempotent через pending targets."""
     _check_token(x_admin_token)
+    from app.services.job_settings import get_setting_value
     from app.workers.tasks.nspd_geo import process_nspd_geo_job
 
     db.execute(
@@ -782,7 +786,8 @@ def resume_geo_job(
         {"id": job_id},
     )
     db.commit()
-    process_nspd_geo_job.apply_async(args=[job_id])
+    geo_queue = get_setting_value("nspd_geo", "queue_name", "geo")
+    process_nspd_geo_job.apply_async(args=[job_id], queue=geo_queue)
     return {"job_id": job_id, "resumed": True}
 
 

backend/app/main.py

@@ -5,7 +5,7 @@ import sentry_sdk
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
-from app.api.v1 import admin_leads, admin_scrape, analytics, concepts, parcels, photos
+from app.api.v1 import admin_jobs, admin_leads, admin_scrape, analytics, concepts, parcels, photos
 from app.core.config import settings
 
 
@@ -35,6 +35,7 @@ app.include_router(concepts.router, prefix="/api/v1/concepts", tags=["concepts"]
 app.include_router(parcels.router, prefix="/api/v1/parcels", tags=["parcels"])
 app.include_router(analytics.router, prefix="/api/v1/analytics", tags=["analytics"])
 app.include_router(admin_scrape.router, prefix="/api/v1/admin/scrape", tags=["admin"])
+app.include_router(admin_jobs.router, prefix="/api/v1/admin/jobs", tags=["admin"])
 app.include_router(admin_leads.router, prefix="/api/v1/admin/leads", tags=["admin"])
 app.include_router(photos.router, prefix="/api/v1/photos", tags=["photos"])
 

backend/app/models/job_settings.py

@@ -0,0 +1,25 @@
+"""SQLAlchemy ORM-модель для централизованной таблицы настроек Celery-job'ов."""
+
+from datetime import datetime
+
+from sqlalchemy import Boolean, DateTime, Integer, SmallInteger, Text
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.db import Base
+
+
+class JobSetting(Base):
+    __tablename__ = "job_settings"
+
+    job_type: Mapped[str] = mapped_column(Text, primary_key=True)
+    enabled: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
+    queue_name: Mapped[str] = mapped_column(Text, default="celery", nullable=False)
+    cron_schedule: Mapped[str | None] = mapped_column(Text, nullable=True)
+    rate_ms: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    max_retries: Mapped[int] = mapped_column(SmallInteger, default=2, nullable=False)
+    max_concurrency: Mapped[int] = mapped_column(SmallInteger, default=1, nullable=False)
+    extra_config: Mapped[dict] = mapped_column(JSONB, default=dict, nullable=False)
+    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    updated_by: Mapped[str | None] = mapped_column(Text, nullable=True)
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)

backend/app/schemas/job_settings.py

@@ -0,0 +1,34 @@
+"""Pydantic v2 схемы для job_settings API."""
+
+from datetime import datetime
+
+from pydantic import BaseModel, Field
+
+
+class JobSettingRead(BaseModel):
+    job_type: str
+    enabled: bool
+    queue_name: str
+    cron_schedule: str | None
+    rate_ms: int | None
+    max_retries: int
+    max_concurrency: int
+    extra_config: dict
+    updated_at: datetime
+    updated_by: str | None
+    description: str | None
+
+    model_config = {"from_attributes": True}
+
+
+class JobSettingUpdate(BaseModel):
+    """PATCH-style: передавать только изменяемые поля. Непереданные — без изменений."""
+
+    enabled: bool | None = None
+    queue_name: str | None = None
+    # Пустая строка → NULL (manual-only режим).
+    cron_schedule: str | None = None
+    rate_ms: int | None = Field(None, ge=0)
+    max_retries: int | None = Field(None, ge=0, le=10)
+    max_concurrency: int | None = Field(None, ge=1, le=20)
+    extra_config: dict | None = None

backend/app/services/job_settings.py

@@ -0,0 +1,284 @@
+"""Сервис для чтения и обновления централизованных настроек 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 = :extra_config::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]

backend/app/workers/celery_app.py

@@ -1,4 +1,8 @@
-"""Celery app + beat schedule."""
+"""Celery app + beat schedule.
+
+Beat schedule строится из таблицы job_settings при старте beat-контейнера.
+При недоступности БД — fallback на env-based расписание (старое поведение).
+"""
 
 import logging
 
@@ -10,6 +14,15 @@ 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 '*'."""
@@ -44,64 +57,156 @@ celery_app = Celery(
 )
 celery_app.conf.timezone = "Europe/Moscow"
 
-# Расписание задаётся через SCRAPE_KN_CRON env var (см. app.core.config).
-# Каждый региональный sweep оборачивается случайной задержкой 0..SCRAPE_KN_JITTER_SECONDS
-# внутри самой задачи (см. tasks.scrape_kn) — чтобы не бить ровно в одну минуту.
-celery_app.conf.beat_schedule = {
-    f"kn-region-{rc}": {
-        "task": "tasks.scrape_kn.scrape_kn_region",
-        "schedule": _parse_cron(settings.scrape_kn_cron),
-        "args": [rc, None],
-    }
-    for rc in _default_regions()
-}
 
-# NSPD beat schedule УДАЛЁН 2026-05-11.
-# Старый Playwright-based scraper (tasks.scrape_nspd) сменён на bulk geo-fetcher
-# через rosreestr2coord — запускается вручную через UI /admin/scrape/geo.
-# Расписания по умолчанию нет (cron можно добавить через beat позже если надо).
+def _build_beat_schedule_from_db() -> dict:
+    """Строит beat_schedule из job_settings БД.
 
-# Refresh ekb_districts медианы — ежемесячно 5-го числа в 04:00 МСК
-# (после публикации новых rosreestr-кварталов и NSPD beat-cycle).
-# Лёгкая задача (1-2с на 8 районов), без locks.
-celery_app.conf.beat_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},
-}
+    Возвращает пустой словарь при любой ошибке — 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
 
 
-# Objective sync — наш ЕДИНЫЙ source of truth для Objective-данных.
-# Дёргает api.objctv.ru напрямую и пишет в PostgreSQL (objective_lots /
-# corpus_room_month / lots_history) минуя любые промежуточные SQLite.
-#
-# Динамическая конфигурация в БД (objective_sync_config single-row):
-#   - cron_schedule (читается ОДИН РАЗ при старте beat — после изменения
-#     требует `docker compose restart beat`)
-#   - groups_csv, use_ddu, use_dkp, period_months_back, inter_group_delay_s,
-#     rate_ms, retries — читаются task'ом ПРИ КАЖДОМ запуске, динамически.
-#
-# Если БД недоступна на старте — fallback на settings.objective_sync_cron.
-def _objective_cron() -> "crontab":
-    """Прочитать cron из БД с fallback на settings."""
+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 as e:
-        logger.warning("objective beat cron: fallback на settings (%s)", e)
+    except Exception:
         cron_str = settings.objective_sync_cron
-    return _parse_cron(cron_str)
+
+    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
 
 
-celery_app.conf.beat_schedule["objective-sync"] = {
-    "task": "tasks.scrape_objective.sync_all_groups",
-    "schedule": _objective_cron(),
-    "kwargs": {
-        "triggered_by": "beat",
-    },
-}
+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)
+
+    return schedule
+
+
+celery_app.conf.beat_schedule = _build_beat_schedule()
 
 
 @worker_ready.connect
@@ -222,7 +327,7 @@ def _resume_zombie_runs(sender=None, **_kwargs) -> None:
 
         for jid in geo_resume_jobs:
             try:
-                process_nspd_geo_job.apply_async(args=[jid])
+                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)

backend/app/workers/tasks/nspd_geo.py

@@ -308,7 +308,14 @@ def process_nspd_geo_job(self: Any, job_id: int) -> dict[str, Any]:
             return {"error": "job_not_found", "job_id": job_id}
 
         _start_job(db, job_id)
-        rate_ms = int(job["rate_ms"] or 600)
+        # Если в 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)

docker-compose.prod.yml

@@ -96,7 +96,7 @@ services:
       # контейнера задан settings.objective_anton_sqlite_path (default
       # /data/anton-sqlite/analysis.db).
       - /opt/gendesign/site-finder:/data/anton-sqlite:ro
-    command: ["celery", "-A", "app.workers.celery_app", "worker", "--loglevel=info", "--concurrency=2"]
+    command: ["celery", "-A", "app.workers.celery_app", "worker", "--loglevel=info", "--concurrency=2", "--queues=celery,scrape_kn,geo"]
 
   beat:
     # Lean backend-образ (без Chromium) — beat только триггерит таски в Redis.