7eb686bd7d
Closes #105 (Phase 3 geocoding — отдельный issue). Phase 4: 'ekburg-permits-monthly' crontab(0 2 1 * *) → tasks.refresh_all Phase 5: /analyze returns engineering.recent_permits_in_quarter + permits_summary (rns/rve counts/area/units, by_developer top-3) Filter: quarter prefix (3-segment cadastral match). Spatial 500m radius — после Phase 3 geocoding. Tests: test_analyze_recent_permits.py (3 mock), comment updates на inline_weights/market_price (shifted execute ordering).
248 lines
10 KiB
Python
248 lines
10 KiB
Python
"""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() -> 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"},
|
||
}
|
||
|
||
# #105 Phase 4: ЕКБ РНС/РВЭ — ежемесячно 1-го числа в 05:00 МСК (02:00 UTC)
|
||
schedule["ekburg-permits-monthly"] = {
|
||
"task": "tasks.ekburg_permits_sync.refresh_all",
|
||
"schedule": _parse_cron("0 2 1 * *"),
|
||
"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
|