408 lines
15 KiB
Python
408 lines
15 KiB
Python
"""Celery app + beat schedule.
|
||
|
||
Beat schedule строится из таблицы job_settings при старте beat-контейнера.
|
||
При недоступности БД — fallback на env-based расписание (старое поведение).
|
||
"""
|
||
|
||
import logging
|
||
|
||
from celery import Celery
|
||
from celery.schedules import crontab
|
||
from celery.signals import worker_process_init, worker_ready
|
||
|
||
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()]
|
||
|
||
|
||
celery_app = Celery(
|
||
"gendesign",
|
||
broker=settings.redis_url,
|
||
backend=settings.redis_url,
|
||
include=[
|
||
"app.workers.tasks.scrape_kn",
|
||
"app.workers.tasks.refresh_analytics",
|
||
"app.workers.tasks.scrape_objective",
|
||
"app.workers.tasks.objective_etl",
|
||
"app.workers.tasks.nspd_geo",
|
||
"app.workers.tasks.poi_sync",
|
||
"app.workers.tasks.noise_sync",
|
||
"app.workers.tasks.pzz_sync",
|
||
],
|
||
)
|
||
celery_app.conf.timezone = "Europe/Moscow"
|
||
|
||
|
||
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"},
|
||
}
|
||
|
||
return schedule
|
||
|
||
|
||
celery_app.conf.beat_schedule = _build_beat_schedule()
|
||
|
||
|
||
@worker_process_init.connect
|
||
def _reset_db_connections(sender=None, **_kwargs) -> None:
|
||
"""Dispose SQLAlchemy engine in each prefork child process.
|
||
|
||
Background: Celery prefork pool fork()'s child processes from the master
|
||
that already imported SQLAlchemy + psycopg. The child inherits open TCP
|
||
sockets to PostgreSQL, but psycopg connections cannot be safely shared
|
||
across processes after fork — first use raises:
|
||
psycopg.ProgrammingError: can't change 'autocommit' now:
|
||
connection in transaction status INTRANS
|
||
|
||
Fix: dispose the engine in each child. SQLAlchemy will lazily open fresh
|
||
connections per-child. Standard Celery + SQLA recipe.
|
||
"""
|
||
try:
|
||
from app.core.db import engine
|
||
|
||
engine.dispose()
|
||
logger.info("worker_process_init: SQLAlchemy engine disposed for fork")
|
||
except Exception as e:
|
||
logger.warning("worker_process_init: engine dispose failed: %s", e)
|
||
|
||
|
||
@worker_ready.connect
|
||
def _resume_zombie_runs(sender=None, **_kwargs) -> None:
|
||
"""When a worker finishes booting (after redeploy/restart), find any sweep
|
||
that was 'running' or 'paused' and re-enqueue a resume task. Each resume
|
||
creates a fresh run_id linked via resumed_from_run_id; the original row is
|
||
marked 'zombie' so the audit trail is preserved.
|
||
|
||
No time threshold: by definition, on worker_ready ANY 'running' row is a
|
||
zombie because there is no active worker. Previously we required heartbeat
|
||
>5 min, which skipped jobs interrupted seconds before redeploy — they
|
||
stayed in 'running' status forever and required manual cancel/resume.
|
||
"""
|
||
logger.info("worker_ready: resume scan starting")
|
||
from sqlalchemy import text
|
||
|
||
# Persistent breadcrumb: write to nspd_geo_log so we can confirm via DB
|
||
# query whether worker_ready actually fired (independent of container logs).
|
||
# NULL job_id to avoid FK violation (FK is to nspd_geo_jobs.job_id which
|
||
# has no 0 row).
|
||
try:
|
||
from app.core.db import SessionLocal
|
||
|
||
_db = SessionLocal()
|
||
try:
|
||
_db.execute(
|
||
text(
|
||
"INSERT INTO nspd_geo_log (job_id, level, stage, message) "
|
||
"VALUES (NULL, 'info', 'worker_ready', 'worker_ready signal fired')"
|
||
)
|
||
)
|
||
_db.commit()
|
||
finally:
|
||
_db.close()
|
||
except Exception as _e:
|
||
logger.warning("worker_ready: breadcrumb insert failed: %s", _e)
|
||
|
||
from app.core.db import SessionLocal
|
||
|
||
# ── kn_scrape_runs resume ──
|
||
db = SessionLocal()
|
||
ids: list[int] = []
|
||
try:
|
||
rows = (
|
||
db.execute(
|
||
text(
|
||
"""
|
||
SELECT run_id
|
||
FROM kn_scrape_runs
|
||
WHERE status = 'running'
|
||
AND objects_snapshot IS NOT NULL
|
||
ORDER BY started_at ASC
|
||
LIMIT 20
|
||
"""
|
||
)
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
if rows:
|
||
ids = [int(r["run_id"]) for r in rows]
|
||
# Помечаем найденные как 'zombie' одним апдейтом — resume создаст новые
|
||
# run_id со ссылкой resumed_from_run_id.
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE kn_scrape_runs
|
||
SET status = 'zombie',
|
||
finished_at = NOW(),
|
||
error = COALESCE(error,
|
||
'auto-zombie at worker_ready, resume scheduled')
|
||
WHERE run_id = ANY(:ids)
|
||
"""
|
||
),
|
||
{"ids": ids},
|
||
)
|
||
db.commit()
|
||
else:
|
||
logger.info("worker_ready: нет stale kn runs для resume")
|
||
except Exception as e:
|
||
logger.exception("worker_ready kn resume scan failed: %s", e)
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
db.close()
|
||
|
||
# Enqueue resume tasks. Late import to avoid circular at module load.
|
||
from app.workers.tasks.scrape_kn import resume_kn_run
|
||
|
||
for rid in ids:
|
||
try:
|
||
resume_kn_run.apply_async(args=[rid])
|
||
logger.info("worker_ready: resume_kn_run enqueued for run=%s", rid)
|
||
except Exception as e:
|
||
logger.warning("worker_ready: failed to enqueue resume for run=%s: %s", rid, e)
|
||
|
||
# NSPD-runs (старый Playwright-scraper) resume УДАЛЁН 2026-05-11. Скрапер
|
||
# снят с эксплуатации в пользу bulk geo-fetcher (nspd_geo). История
|
||
# старых runs сохранена в nspd_scrape_runs — никаких side-effects.
|
||
|
||
# NSPD geo-jobs: bulk-fetcher с собственной resume-логикой через
|
||
# nspd_geo_jobs / nspd_geo_targets. Resume любых 'running' / 'paused' jobs
|
||
# — на worker_ready по определению нет активных воркеров, всё running ==
|
||
# zombie. Раньше требовали heartbeat >10мин, что пропускало jobs убитых
|
||
# за минуту до редеплоя и оставляло их вечно висеть.
|
||
db = SessionLocal()
|
||
geo_resume_jobs: list[int] = []
|
||
try:
|
||
rows = (
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE nspd_geo_jobs
|
||
SET status = 'queued',
|
||
error = COALESCE(error, 'auto-resume at worker_ready')
|
||
WHERE status IN ('running', 'paused')
|
||
RETURNING job_id
|
||
"""
|
||
)
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
db.commit()
|
||
geo_resume_jobs = [int(r["job_id"]) for r in rows]
|
||
for jid in geo_resume_jobs:
|
||
logger.info("worker_ready: NSPD geo job=%s — resume scheduled", jid)
|
||
except Exception as e:
|
||
logger.warning("worker_ready nspd_geo resume scan failed: %s", e)
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
db.close()
|
||
|
||
if geo_resume_jobs:
|
||
from app.workers.tasks.nspd_geo import process_nspd_geo_job
|
||
|
||
for jid in geo_resume_jobs:
|
||
try:
|
||
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)
|
||
logger.info("worker_ready: resume scan finished (geo_jobs=%d)", len(geo_resume_jobs))
|