gendesign/backend/app/workers/tasks/objective_etl.py

"""LEGACY/BOOTSTRAP Celery task: ETL Антоновского SQLite → наша PG.

⚠️ В обычной работе НЕ нужен — наш `tasks.scrape_objective.sync_all_groups`
тянет данные с api.objctv.ru напрямую в PostgreSQL (см. celery_app.py beat).

Этот ETL остаётся как ОДНОРАЗОВЫЙ инструмент для:
  - bootstrap PG из Антоновского SQLite (например при первом запуске,
    до того как наш beat отработал)
  - аварийного восстановления если наш sync долго не работал, а Антоновский
    /sf/ продолжал тянуть свою SQLite

Триггерится только вручную через POST /api/v1/admin/scrape/objective (UI).
Beat-расписания НЕТ.

Файл-источник: settings.objective_anton_sqlite_path
(на проде смонтирован bind-mount-ом docker-compose из /opt/gendesign/site-finder).

Логирует прогресс в objective_scrape_runs с group_name='anton-etl'.
"""

from __future__ import annotations

import logging
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.config import settings
from app.core.db import SessionLocal
from app.services.objective_etl import get_sqlite_info, run_etl
from app.workers.celery_app import celery_app

logger = logging.getLogger(__name__)


def _start_run(db: Session, triggered_by: str) -> int:
    row = db.execute(
        text(
            """
            INSERT INTO objective_scrape_runs (group_name, triggered_by, status)
            VALUES (:gn, :tb, 'running')
            RETURNING run_id
            """
        ),
        {"gn": "anton-etl", "tb": triggered_by},
    ).scalar_one()
    db.commit()
    return int(row)


def _heartbeat(db: Session, run_id: int, **counts: int) -> None:
    sets = ["heartbeat_at = NOW()"]
    params: dict[str, Any] = {"rid": run_id}
    for k, v in counts.items():
        sets.append(f"{k} = :{k}")
        params[k] = v
    try:
        db.execute(
            text(f"UPDATE objective_scrape_runs SET {', '.join(sets)} WHERE run_id = :rid"),
            params,
        )
        db.commit()
    except Exception as e:
        logger.warning("heartbeat failed for run=%s: %s", run_id, e)
        try:
            db.rollback()
        except Exception:
            pass


def _finish_run(
    db: Session, run_id: int, *, status: str, error: str | None = None, **counts: int
) -> None:
    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(
    bind=True,
    name="tasks.objective_etl.import_anton_objective",
    max_retries=0,  # ETL идемпотентный, но при сбое лучше человек посмотрит
)
def import_anton_objective(
    self: Any,
    triggered_by: str = "manual",
    sqlite_path: str | None = None,
) -> dict[str, Any]:
    """Полный ETL из Антоновского SQLite в нашу PG.

    Args:
        triggered_by: 'manual' | 'beat' | 'api' — для журнала
        sqlite_path: переопределить settings.objective_anton_sqlite_path

    Возвращает counts по таблицам + run_id.
    """
    sqlite_path_eff = sqlite_path or settings.objective_anton_sqlite_path
    db = SessionLocal()
    run_id: int | None = None
    try:
        run_id = _start_run(db, triggered_by)
        logger.info("import_anton_objective started, run_id=%s, sqlite=%s", run_id, sqlite_path_eff)

        # Преобразуем sqlalchemy-style db_url в обычный postgresql://
        # (psycopg.connect принимает postgresql://, не postgresql+psycopg://).
        db_url = settings.database_url.replace("postgresql+psycopg://", "postgresql://")

        # Прогресс-callback пишет в БД счётчики + heartbeat
        # Stages: mapping_start/done, crm_start/progress/done, lots_start/progress/done
        def _cb(stage: str, n: int) -> None:
            kw: dict[str, int] = {}
            if stage.startswith("mapping"):
                kw["rows_history"] = n  # переиспользуем поле; mapping count
            elif stage.startswith("crm"):
                kw["rows_corpus_room"] = n
            elif stage.startswith("lots"):
                kw["rows_lots"] = n
            _heartbeat(db, run_id, **kw)

        result = run_etl(
            sqlite_path_eff,
            db_url,
            batch_size=2000,
            progress_cb=_cb,
        )

        _finish_run(
            db,
            run_id,
            status="done",
            error=None,
            rows_lots=result["lots"],
            rows_corpus_room=result["corp_room_month"],
            rows_history=result["mappings"],  # переиспользуем поле под mappings
            reports_ok=3,
            reports_failed=0,
        )
        logger.info("import_anton_objective done, run_id=%s, result=%s", run_id, result)
        return {"run_id": run_id, "triggered_by": triggered_by, **result}

    except FileNotFoundError as e:
        logger.error("SQLite not found: %s", e)
        if run_id:
            try:
                _finish_run(db, run_id, status="failed", error=f"SQLite не найден: {e}")
            except Exception:
                pass
        # Дополним результат полезной диагностикой
        info = get_sqlite_info(sqlite_path_eff)
        return {"run_id": run_id, "error": "sqlite_not_found", "sqlite_info": info}

    except Exception as e:
        logger.exception("import_anton_objective failed: %s", e)
        if run_id:
            try:
                _finish_run(db, run_id, status="failed", error=f"{type(e).__name__}: {e}")
            except Exception:
                pass
        raise
    finally:
        db.close()