bot-backend
51655b6aaf
All checks were successful
CI Trade-In / changes (pull_request) Successful in 21s
CI / changes (pull_request) Successful in 13s
CI Trade-In / backend-tests (pull_request) Has been skipped
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Successful in 2m9s
CI / backend-tests (pull_request) Successful in 15m30s
D5: расширен per-doc try/except в izyatie_ocr_ingest на весь body (fetch→OCR→extract→upsert→commit), не только fetch, + db.rollback() на сбое — один битый документ теперь skip'ается, а не убивает весь недельный батч. D6: etl_newbuilding_crossload обёрнут в тот же start/success/failure breadcrumb паттерн, что у sibling-тасков (objective_scrape_runs, group='newbuilding-crossload') — падения видны в operational-логе, не только Celery FAILURE + GlitchTip. На failure re-raise (Celery/GlitchTip срабатывают), breadcrumb-flush failure не маскирует оригинальное исключение. Без миграции (таблица есть). Refs #2445
118 lines
4.9 KiB
Python
118 lines
4.9 KiB
Python
"""Celery task: ночной cross-load ETL tradein.houses → gendesign.newbuilding_listings.
|
||
|
||
Расписание: 03:30 МСК (hardcoded в beat_schedule, как cbr/poi/supply-layers).
|
||
Идемпотентен: UPSERT ON CONFLICT (source, ext_house_id).
|
||
Если TRADEIN_DATABASE_URL не задан → warn-log, возвращает {"disabled": True}.
|
||
|
||
Breadcrumb (issue #2445 D6): прогон логируется в objective_scrape_runs (тот же
|
||
журнал, что используют соседние nightly/weekly-таски — см.
|
||
app/workers/tasks/objective_etl.py) с group_name='newbuilding-crossload', чтобы
|
||
сбой был виден оператору в admin-таблице логов, а не только в Celery FAILURE +
|
||
GlitchTip. Таблица переиспользуется как есть — без новой миграции; поля
|
||
rows_lots/rows_history/requests_count не про Objective-домен здесь, а про
|
||
source_rows/upserted/skipped cross-load'а (см. комментарии у _finish_run).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from typing import Any
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.db import SessionLocal
|
||
from app.services.etl.newbuilding_crossload import run_crossload
|
||
from app.workers.celery_app import celery_app
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_GROUP_NAME = "newbuilding-crossload"
|
||
|
||
|
||
def _start_run(db: Session, triggered_by: str) -> int:
|
||
"""Создать breadcrumb-строку в objective_scrape_runs, статус 'running'."""
|
||
row = db.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO objective_scrape_runs (group_name, triggered_by, status)
|
||
VALUES (:gn, :tb, 'running')
|
||
RETURNING run_id
|
||
"""
|
||
),
|
||
{"gn": _GROUP_NAME, "tb": triggered_by},
|
||
).scalar_one()
|
||
db.commit()
|
||
return int(row)
|
||
|
||
|
||
def _finish_run(
|
||
db: Session, run_id: int, *, status: str, error: str | None = None, **counts: int
|
||
) -> None:
|
||
"""Обновить breadcrumb-строку финальным статусом ('done' | 'failed') + счётчиками."""
|
||
sets = ["finished_at = NOW()", "status = :status", "error = :error"]
|
||
params: dict[str, Any] = {"rid": run_id, "status": status, "error": error}
|
||
for k, v in counts.items():
|
||
sets.append(f"{k} = :{k}")
|
||
params[k] = v
|
||
db.execute(
|
||
text(f"UPDATE objective_scrape_runs SET {', '.join(sets)} WHERE run_id = :rid"),
|
||
params,
|
||
)
|
||
db.commit()
|
||
|
||
|
||
@celery_app.task(name="tasks.etl_newbuilding_crossload.etl_newbuilding_crossload")
|
||
def etl_newbuilding_crossload(triggered_by: str = "beat") -> dict:
|
||
"""Cross-load tradein.houses → gendesign.newbuilding_listings (#976).
|
||
|
||
Вызывается beat'ом в 03:30 МСК и вручную через POST /api/v1/admin/scrape/newbuilding-crossload.
|
||
|
||
Args:
|
||
triggered_by: 'beat' | 'manual' | 'api' — для журнала objective_scrape_runs.
|
||
|
||
Breadcrumb-строка в objective_scrape_runs логирует start/success/failure —
|
||
на исключении из run_crossload() помечает run 'failed' и ре-рейзит, чтобы
|
||
Celery ушёл в FAILURE и GlitchTip продолжил ловить сбой (issue #2445 D6).
|
||
"""
|
||
db = SessionLocal()
|
||
run_id: int | None = None
|
||
try:
|
||
run_id = _start_run(db, triggered_by)
|
||
logger.info("task etl_newbuilding_crossload: start run_id=%s", run_id)
|
||
|
||
result = run_crossload(db=db)
|
||
|
||
if result.get("disabled"):
|
||
_finish_run(db, run_id, status="done", error="TRADEIN_DATABASE_URL not set")
|
||
logger.info("task etl_newbuilding_crossload: done (disabled) run_id=%s", run_id)
|
||
return {"run_id": run_id, **result}
|
||
|
||
_finish_run(
|
||
db,
|
||
run_id,
|
||
status="done",
|
||
error=None,
|
||
rows_history=result.get("source_rows", 0),
|
||
rows_lots=result.get("upserted", 0),
|
||
requests_count=result.get("skipped", 0),
|
||
)
|
||
logger.info("task etl_newbuilding_crossload: done run_id=%s — %s", run_id, result)
|
||
return {"run_id": run_id, **result}
|
||
|
||
except Exception as e:
|
||
logger.exception("task etl_newbuilding_crossload: failed run_id=%s: %s", run_id, e)
|
||
if run_id is not None:
|
||
try:
|
||
_finish_run(db, run_id, status="failed", error=f"{type(e).__name__}: {e}")
|
||
except Exception:
|
||
# Breadcrumb-flush не должен скрыть исходное исключение — Celery
|
||
# всё равно должен уйти в FAILURE + GlitchTip его поймает.
|
||
logger.exception(
|
||
"task etl_newbuilding_crossload: failed to write failure breadcrumb "
|
||
"for run_id=%s",
|
||
run_id,
|
||
)
|
||
raise
|
||
finally:
|
||
db.close()
|