diff --git a/backend/app/api/v1/admin_jobs.py b/backend/app/api/v1/admin_jobs.py new file mode 100644 index 00000000..e1c36ed2 --- /dev/null +++ b/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 diff --git a/backend/app/api/v1/admin_scrape.py b/backend/app/api/v1/admin_scrape.py index eb0faac4..f3777cc6 100644 --- a/backend/app/api/v1/admin_scrape.py +++ b/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} diff --git a/backend/app/main.py b/backend/app/main.py index 60cfbce1..541875af 100644 --- a/backend/app/main.py +++ b/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"]) diff --git a/backend/app/models/job_settings.py b/backend/app/models/job_settings.py new file mode 100644 index 00000000..76365043 --- /dev/null +++ b/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) diff --git a/backend/app/schemas/job_settings.py b/backend/app/schemas/job_settings.py new file mode 100644 index 00000000..38a8f183 --- /dev/null +++ b/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 diff --git a/backend/app/services/job_settings.py b/backend/app/services/job_settings.py new file mode 100644 index 00000000..66b0922e --- /dev/null +++ b/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] diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index ff52418d..f14367a1 100644 --- a/backend/app/workers/celery_app.py +++ b/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) diff --git a/backend/app/workers/tasks/nspd_geo.py b/backend/app/workers/tasks/nspd_geo.py index 4862e113..aa62a394 100644 --- a/backend/app/workers/tasks/nspd_geo.py +++ b/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) diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 6b577205..2abf2718 100644 --- a/docker-compose.prod.yml +++ b/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.