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

"""Celery task для регулярного sync с api.objctv.ru.

Стратегия:
    Раз в неделю (по умолчанию пн 05:00 МСК) тянем 2 канонических отчёта
    по группе из settings.objective_default_group:
    - Сводные/Корпуса (за последний полный месяц) → objective_corpus_room_month
    - Поквартирные/Лоты (без дат — текущая шахматка) → objective_lots + history

    NB: На текущем тарифе Объектива «Сводные/Лоты» и «Поквартирные/Корпуса»
    отвечают HTTP 500 (проверено data/sql/69b_objective_minimal.py). Когда тариф
    расширят — добавить обратно в `jobs`.

    Все payload-ы пишутся в objective_raw_reports (jsonb-колонка) и сразу же
    парсятся в нормализованный слой через data/sql/70_parse_objective_raw.py.
    Если парсинг упал — raw остаётся, можно re-parse через
    `python data/sql/70_parse_objective_raw.py --latest-run`.
"""

from __future__ import annotations

import json
import logging
from datetime import date, timedelta
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.scrapers.objective import (
    ObjectiveAPIError,
    ObjectiveAuthError,
    ObjectiveClient,
)
from app.workers.celery_app import celery_app

logger = logging.getLogger(__name__)


def _start_run(db: Session, group_name: str, 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": group_name, "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
    db.execute(
        text(f"UPDATE objective_scrape_runs SET {', '.join(sets)} WHERE run_id = :rid"),
        params,
    )
    db.commit()


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()


def _save_raw(
    db: Session,
    run_id: int,
    *,
    report_section: str,
    report_type: str,
    report_name: str,
    group_name: str,
    complex_name: str | None,
    start_date: date | None,
    end_date: date | None,
    use_ddu: bool,
    use_dkp: bool,
    payload: Any,
) -> int:
    body = json.dumps(payload, ensure_ascii=False)
    row = db.execute(
        text(
            """
            INSERT INTO objective_raw_reports
                (run_id, page, report_section, report_type, report_name,
                 group_name, complex_name, start_date, end_date,
                 use_ddu, use_dkp, api_version, payload, payload_size)
            VALUES (:run_id, 'Отчеты', :section, :type, :name,
                    :group_name, :complex_name, :start_date, :end_date,
                    :use_ddu, :use_dkp, 'v2', CAST(:payload AS jsonb), :size)
            RETURNING raw_id
            """
        ),
        {
            "run_id": run_id,
            "section": report_section,
            "type": report_type,
            "name": report_name,
            "group_name": group_name,
            "complex_name": complex_name,
            "start_date": start_date,
            "end_date": end_date,
            "use_ddu": use_ddu,
            "use_dkp": use_dkp,
            "payload": body,
            "size": len(body.encode("utf-8")),
        },
    ).scalar_one()
    db.commit()
    return int(row)


@celery_app.task(
    bind=True,
    name="tasks.scrape_objective.sync_objective_group",
    max_retries=2,
)
def sync_objective_group(
    self: Any,
    group_name: str | None = None,
    triggered_by: str = "beat",
) -> dict[str, Any]:
    """Sync 4 канонических отчёта Объектив для одной группы (территории).

    Идемпотентность: каждый прогон создаёт новые objective_raw_reports
    (history-style). Нормализация (parse → objective_oks/lots) — отдельная
    задача после изучения реальной схемы payload через smoke-script.
    """
    if not settings.objective_api_key:
        logger.warning("sync_objective_group skipped: OBJECTIVE_API_KEY не задан")
        return {"skipped": True, "reason": "no_api_key"}

    group = group_name or settings.objective_default_group
    end_date = date.today().replace(day=1)
    start_date = (end_date - timedelta(days=1)).replace(day=1)

    db = SessionLocal()
    run_id: int | None = None
    try:
        run_id = _start_run(db, group, triggered_by)
        client = ObjectiveClient()

        reports_ok = 0
        reports_failed = 0
        n_requests = 0
        rows_corpus_room = 0
        rows_lots = 0
        rows_history = 0

        # Импортируем парсер и автоматически нормализуем после save_raw.
        # Сделано как late-import чтобы scrape_objective.py подгружался без
        # data/sql при unit-тестах celery_app.
        from importlib import util as _importlib_util
        from pathlib import Path as _Path

        _parser_path = (
            _Path(__file__).resolve().parents[3] / "data" / "sql" / "70_parse_objective_raw.py"
        )
        _spec = _importlib_util.spec_from_file_location("objective_parser_70", _parser_path)
        if _spec is None or _spec.loader is None:
            raise RuntimeError(f"Не удалось загрузить парсер по пути {_parser_path}")
        parser_mod = _importlib_util.module_from_spec(_spec)
        _spec.loader.exec_module(parser_mod)

        # На нашем тарифе доступны только Сводные/Корпуса и Поквартирные/Лоты.
        # «Сводные/Лоты» и «Поквартирные/Корпуса» возвращают 500 (проверено).
        jobs: list[tuple[str, str, dict, str, str, str]] = [
            (
                "corp_sum",
                "report_corpuses_summary",
                {
                    "group_name": group,
                    "start_date": start_date,
                    "end_date": end_date,
                    "use_ddu": True,
                    "use_dkp": True,
                },
                "Объединенные данные",
                "Сводные",
                "Корпуса",
            ),
            (
                "lots_pf",
                "report_lots_per_flat",
                {"group_name": group, "use_ddu": True, "use_dkp": True},
                "Объединенные данные",
                "Поквартирные",
                "Лоты",
            ),
        ]

        try:
            for kind, fn_name, params, section, rtype, rname in jobs:
                try:
                    method = getattr(client, fn_name)
                    payload = method(**params)
                    n_requests += 1
                    raw_id = _save_raw(
                        db,
                        run_id,
                        report_section=section,
                        report_type=rtype,
                        report_name=rname,
                        group_name=group,
                        complex_name=None,
                        start_date=params.get("start_date"),
                        end_date=params.get("end_date"),
                        use_ddu=params.get("use_ddu", True),
                        use_dkp=params.get("use_dkp", False),
                        payload=payload,
                    )
                    reports_ok += 1

                    # Inline-нормализация в objective_corpus_room_month / lots / history.
                    snap = date.today()
                    try:
                        if kind == "corp_sum":
                            n = parser_mod.parse_corp_sum(payload, group, raw_id, db, dry_run=False)
                            rows_corpus_room += n
                            db.execute(
                                text(
                                    "UPDATE objective_raw_reports "
                                    "   SET rows_extracted = :n WHERE raw_id = :rid"
                                ),
                                {"n": n, "rid": raw_id},
                            )
                        elif kind == "lots_pf":
                            n_lots, n_hist = parser_mod.parse_lots_pf(
                                payload, raw_id, snap, db, dry_run=False
                            )
                            rows_lots += n_lots
                            rows_history += n_hist
                            db.execute(
                                text(
                                    "UPDATE objective_raw_reports "
                                    "   SET rows_extracted = :n WHERE raw_id = :rid"
                                ),
                                {"n": n_lots, "rid": raw_id},
                            )
                        db.commit()
                    except Exception as parse_err:
                        # Парсинг упал — raw уже сохранён, можно re-parse позже.
                        db.rollback()
                        logger.exception(
                            "sync_objective_group: parser failed for %s/%s/%s raw_id=%s: %s",
                            section,
                            rtype,
                            rname,
                            raw_id,
                            parse_err,
                        )

                    _heartbeat(
                        db,
                        run_id,
                        reports_ok=reports_ok,
                        reports_failed=reports_failed,
                        requests_count=n_requests,
                        rows_corpus_room=rows_corpus_room,
                        rows_lots=rows_lots,
                        rows_history=rows_history,
                    )
                except (ObjectiveAuthError, ObjectiveAPIError) as e:
                    reports_failed += 1
                    logger.warning(
                        "sync_objective_group: %s/%s/%s failed: %s", section, rtype, rname, e
                    )
        finally:
            client.close()

        _finish_run(
            db,
            run_id,
            status="done" if reports_failed == 0 else "failed",
            error=None if reports_failed == 0 else f"{reports_failed} reports failed",
            reports_ok=reports_ok,
            reports_failed=reports_failed,
            requests_count=n_requests,
            rows_corpus_room=rows_corpus_room,
            rows_lots=rows_lots,
            rows_history=rows_history,
        )

        return {
            "run_id": run_id,
            "group_name": group,
            "reports_ok": reports_ok,
            "reports_failed": reports_failed,
            "requests": n_requests,
            "rows_corpus_room": rows_corpus_room,
            "rows_lots": rows_lots,
            "rows_history": rows_history,
            "start_date": start_date.isoformat(),
            "end_date": end_date.isoformat(),
        }
    except Exception as 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()