diff --git a/.gitignore b/.gitignore index 40941af3..dc1d2ab1 100644 --- a/.gitignore +++ b/.gitignore @@ -38,6 +38,10 @@ data/raw/prinzip_crm.sqlite data/raw/domrf_photos/ # Objective API smoke dumps (ходят в backend/data/raw когда скрипт запускают из backend/) backend/data/raw/ + +# Antonov SQLite snapshot для ETL (340 MB) +sf_anton_snapshot.db +sf_anton_snapshot.db-* *.csv *.csv.zip !backend/db/init/*.sql diff --git a/backend/app/api/v1/admin_scrape.py b/backend/app/api/v1/admin_scrape.py index 652afa2a..4edcc5e4 100644 --- a/backend/app/api/v1/admin_scrape.py +++ b/backend/app/api/v1/admin_scrape.py @@ -566,6 +566,241 @@ def nspd_coverage( } +# ── Objective ETL (Антоновский SQLite → PG) ───────────────────────────────── + + +class TriggerObjectiveEtlRequest(BaseModel): + sqlite_path: str | None = Field( + default=None, + description="Override settings.objective_anton_sqlite_path (abs file path)", + ) + + +@router.post("/objective") +def trigger_objective_etl( + payload: TriggerObjectiveEtlRequest, + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, +) -> dict[str, Any]: + """Запустить ETL Антоновского SQLite → наша PG. + + SQLite-файл должен быть доступен внутри Celery-worker контейнера: + - на проде: смонтирован bind-mount-ом из /opt/gendesign/site-finder/ + - локально: путь передать через payload.sqlite_path + """ + _check_token(x_admin_token) + from app.workers.tasks.objective_etl import import_anton_objective + + result = import_anton_objective.apply_async( + kwargs={ + "triggered_by": "manual", + "sqlite_path": payload.sqlite_path, + }, + ) + return { + "task_id": result.id, + "sqlite_path": payload.sqlite_path or settings.objective_anton_sqlite_path, + } + + +@router.get("/objective/runs") +def list_objective_runs( + db: Annotated[Session, Depends(get_db)], + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + limit: int = 20, +) -> list[dict[str, Any]]: + _check_token(x_admin_token) + rows = ( + db.execute( + text( + """ + SELECT run_id, started_at, finished_at, heartbeat_at, + group_name, reports_ok, reports_failed, + rows_corpus_room, rows_lots, rows_history, + requests_count, status, error, triggered_by + FROM objective_scrape_runs + ORDER BY started_at DESC + LIMIT :lim + """ + ), + {"lim": min(limit, 100)}, + ) + .mappings() + .all() + ) + return [ + { + "run_id": r["run_id"], + "started_at": r["started_at"].isoformat() if r["started_at"] else None, + "finished_at": r["finished_at"].isoformat() if r["finished_at"] else None, + "heartbeat_at": r["heartbeat_at"].isoformat() if r["heartbeat_at"] else None, + "group_name": r["group_name"], + "reports_ok": r["reports_ok"], + "reports_failed": r["reports_failed"], + "rows_corpus_room": r["rows_corpus_room"], + "rows_lots": r["rows_lots"], + # Поле в Schema 68 v2 называется rows_history — мы переиспользуем его + # под количество загруженных mapping'ов из jk_objective_match. + "rows_mappings": r["rows_history"], + "requests_count": r["requests_count"], + "status": r["status"], + "error": r["error"], + "triggered_by": r["triggered_by"], + } + for r in rows + ] + + +class TriggerObjectiveSyncRequest(BaseModel): + """Ручной триггер нашего sync_all_groups (платный API Объектива).""" + + groups: list[str] | None = Field( + default=None, + description="Override settings.objective_sync_groups (csv). " + "Если None — берёт дефолт из settings.", + max_length=20, + ) + + +@router.post("/objective/sync-our") +def trigger_objective_sync_our( + payload: TriggerObjectiveSyncRequest, + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, +) -> dict[str, Any]: + """Запустить НАШ sync_all_groups (тратит limits Объектива). + + В отличие от ETL (POST /objective) — это вызовы api.objctv.ru напрямую, + с парсингом payload через app.workers.tasks.scrape_objective. + Полезно когда нужно расширить покрытие за пределы Антоновского ЕКБ + (Свердл.обл целиком + Челябинск + Тюмень + Пермь). + """ + _check_token(x_admin_token) + if not settings.objective_api_key: + raise HTTPException( + status_code=503, + detail="OBJECTIVE_API_KEY не задан (settings.objective_api_key пуст)", + ) + from app.workers.tasks.scrape_objective import sync_all_groups + + groups_eff = payload.groups or [ + g.strip() for g in settings.objective_sync_groups.split(",") if g.strip() + ] + result = sync_all_groups.apply_async( + kwargs={ + "groups": payload.groups, # None → task возьмёт дефолт + "triggered_by": "manual", + }, + ) + return { + "task_id": result.id, + "groups_planned": groups_eff, + "estimate_minutes": len(groups_eff) + * 3, # ~3 мин на группу (corp_sum + lots_pf + 30s pause) + } + + +class ObjectiveConfigUpdateRequest(BaseModel): + """PATCH-style — обновляет только переданные поля.""" + + cron_schedule: str | None = None + groups_csv: str | None = None + use_ddu: bool | None = None + use_dkp: bool | None = None + period_months_back: int | None = Field(default=None, ge=1, le=24) + inter_group_delay_s: int | None = Field(default=None, ge=0, le=300) + rate_ms: int | None = Field(default=None, ge=0, le=10000) + retries: int | None = Field(default=None, ge=0, le=5) + + +@router.get("/objective/config") +def get_objective_config( + db: Annotated[Session, Depends(get_db)], + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, +) -> dict[str, Any]: + """Текущая динамическая конфигурация Objective sync (single row).""" + _check_token(x_admin_token) + from app.services.objective_sync_config import get_config + + return get_config(db).to_dict() + + +@router.put("/objective/config") +def update_objective_config( + payload: ObjectiveConfigUpdateRequest, + db: Annotated[Session, Depends(get_db)], + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, +) -> dict[str, Any]: + """PATCH-style update. После изменения cron_schedule нужен restart beat, + остальные поля применяются на следующем sync-вызове автоматически.""" + _check_token(x_admin_token) + from app.services.objective_sync_config import update_config + + cfg = update_config( + db, + cron_schedule=payload.cron_schedule, + groups_csv=payload.groups_csv, + use_ddu=payload.use_ddu, + use_dkp=payload.use_dkp, + period_months_back=payload.period_months_back, + inter_group_delay_s=payload.inter_group_delay_s, + rate_ms=payload.rate_ms, + retries=payload.retries, + updated_by="admin-ui", + ) + return { + **cfg.to_dict(), + "_note": ( + "cron_schedule applied только после restart beat-контейнера; " + "остальные параметры подхватятся при следующем запуске sync_all_groups" + ), + } + + +@router.get("/objective/coverage") +def objective_coverage( + db: Annotated[Session, Depends(get_db)], + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, +) -> dict[str, Any]: + """Что у нас в БД сейчас + что лежит в SQLite Антона (size, last modified). + + Помогает понять стоит ли запускать ETL: если SQLite старше последнего + успешного run'а — данные не изменятся. + """ + _check_token(x_admin_token) + from app.services.objective_etl import get_sqlite_info + + pg_row = ( + db.execute( + text( + """ + SELECT + (SELECT COUNT(*) FROM objective_lots) AS lots, + (SELECT COUNT(*) FROM objective_corpus_room_month) AS crm, + (SELECT COUNT(*) FROM objective_complex_mapping) AS mappings, + (SELECT MAX(snapshot_date) FROM objective_lots) AS last_lot_snapshot, + (SELECT MAX(finished_at) FROM objective_scrape_runs + WHERE status = 'done' + AND triggered_by IN ('manual','beat','api')) AS last_run_at + """ + ) + ) + .mappings() + .first() + ) + sqlite_info = get_sqlite_info(settings.objective_anton_sqlite_path) + return { + "pg": { + "lots": int(pg_row["lots"] or 0), + "corp_room_month": int(pg_row["crm"] or 0), + "mappings": int(pg_row["mappings"] or 0), + "last_lot_snapshot": pg_row["last_lot_snapshot"].isoformat() + if pg_row["last_lot_snapshot"] + else None, + "last_run_at": pg_row["last_run_at"].isoformat() if pg_row["last_run_at"] else None, + }, + "sqlite": sqlite_info, + } + + # ── DomRF (kn) — runs list ─────────────────────────────────────────────────── diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 130728bd..8a26953c 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -55,11 +55,35 @@ class Settings(BaseSettings): # Группа = «настройка территории» внутри Объектива. Чаще всего совпадает с # городом/областью. Доступные группы определяются тарифом ключа. objective_default_group: str = "Екатеринбург" + # Список групп для еженедельного auto-sync (csv). Эмпирически проверено + # 2026-05-10 что эти группы доступны на нашем тарифе: + # - "Свердловская область" → 1626 corp_sum/3д (= ЕКБ + спутники + регион) + # - "Челябинск" → 355 + # - "Тюмень" → 985 + # - "Пермь" → 423 + # Стратегия Variant B (включить наш sync параллельно Антоновому ETL): + # Антон тянет только Екб (303k lots). Наш sync даёт +30k Свердл.обл + # за пределами Екб + ~700k квартир УрФО (Челябинск, Тюмень, Пермь). + # Итог в PG: ~1М квартир уральского + западносибирского рынка. + # Между группами task делает паузу _OBJECTIVE_INTER_GROUP_DELAY = 30с + # чтобы не упереться в rate-limit Объектива. + objective_sync_groups: str = "Свердловская область,Челябинск,Тюмень,Пермь" # Токен Bearer короткоживущий (предположительно 30 мин — уточнить). # Кешируем в Redis с этим TTL (с запасом 5 мин). objective_token_ttl_seconds: int = 25 * 60 - # Расписание sync — еженедельно по понедельникам в 05:00 МСК (после kn-sweep). - objective_sync_cron: str = "0 5 * * mon" + # Расписание sync — еженедельно по вторникам в 06:00 МСК. + # ПОЧЕМУ ВТОРНИК (а не понедельник как раньше): Антон тянет в понедельник, + # к утру вторника его SQLite уже обновлён — наш sync захватывает свежий + # snapshot за тот же отчётный период. + objective_sync_cron: str = "0 6 * * tue" + + # ETL Антоновского /sf/api/* SQLite → нашу PG (objective_* таблицы). + # На проде монтируется bind-mount-ом docker-compose: + # /opt/gendesign/site-finder/analysis.db -> /data/anton-sqlite/analysis.db + # Триггерится вручную через POST /api/v1/admin/scrape/objective. + # Для локальной разработки — указать абсолютный путь к скачанному snapshot: + # OBJECTIVE_ANTON_SQLITE_PATH=C:/Users/user/source/repos/gendesign/sf_anton_snapshot.db + objective_anton_sqlite_path: str = "/data/anton-sqlite/analysis.db" settings = Settings() diff --git a/backend/app/services/objective_etl.py b/backend/app/services/objective_etl.py new file mode 100644 index 00000000..045f5aad --- /dev/null +++ b/backend/app/services/objective_etl.py @@ -0,0 +1,537 @@ +"""ETL: SQLite Антона (/sf/) → наша PG (objective_* таблицы). + +Reusable библиотека вокруг логики `data/sql/71_etl_anton_sqlite_to_pg.py`. +Используется Celery-task `tasks.objective_etl.import_anton_objective` и +admin-endpoint `POST /api/v1/admin/scrape/objective`. + +Файл-источник по умолчанию: settings.objective_anton_sqlite_path +(на проде смонтирован bind-mount-ом docker-compose из /opt/gendesign/site-finder). + +Стратегия: + - lots: INSERT … ON CONFLICT (objective_lot_id) DO UPDATE + - crm: INSERT … ON CONFLICT (report_month, group_name, project_name, + corpus_name, room_bucket) DO UPDATE + - mapping: дедупликация по (objective_complex_name, objective_group), + потом INSERT … ON CONFLICT DO UPDATE + +Heartbeat / прогресс (для UI) выставляется через callback `progress_cb`, +который Celery-task передаёт со UPDATE objective_scrape_runs. +""" + +from __future__ import annotations + +import logging +import re +import sqlite3 +from collections.abc import Callable +from datetime import date +from pathlib import Path +from typing import Any + +import psycopg2 +import psycopg2.extras + +logger = logging.getLogger(__name__) + +GROUP_NAME = "Екатеринбург" # У Антона в SQLite нет колонки group_name + + +# ── helpers ───────────────────────────────────────────────────────────────── + + +def _parse_date(s) -> date | None: + if s is None or s == "": + return None + try: + return date.fromisoformat(str(s)[:10]) + except (ValueError, TypeError): + return None + + +def _parse_month_to_first_day(s) -> date | None: + if not s: + return None + try: + y, m = str(s).split("-") + return date(int(y), int(m), 1) + except (ValueError, TypeError): + return None + + +def _normalize_rooms(s) -> int | None: + if not s: + return None + sl = str(s).strip().lower() + if "студ" in sl: + return 0 + m = re.match(r"^(\d+)", sl) + if m: + return min(int(m.group(1)), 5) + return None + + +def _bool_from_yes(v) -> bool | None: + if v is None or v == "": + return None + s = str(v).strip().lower() + if s in ("да", "true", "1", "yes"): + return True + if s in ("нет", "false", "0", "no"): + return False + return None + + +def _site_id_to_obj_id(s) -> int | None: + """'jk:49854' → 49854. 'parcel:...' → None.""" + if not s: + return None + s = str(s) + if s.startswith("jk:"): + try: + return int(s[3:]) + except ValueError: + return None + return None + + +def _to_int(v) -> int | None: + if v is None or v == "": + return None + try: + return int(v) + except (TypeError, ValueError): + return None + + +def _to_num(v) -> float | None: + if v is None or v == "": + return None + try: + return float(v) + except (TypeError, ValueError): + return None + + +# ── row mappings ──────────────────────────────────────────────────────────── + + +def _row_to_lot(r: sqlite3.Row, snapshot_date: date) -> tuple: + return ( + r["lot_id"], + r["project_id"], + r["project"], + r["developer"], + r["city"], + r["district"], + r["corpus"], + r["address"], + r["obj_class"], + r["section"], + _to_int(r["floor"]), + r["lot_num"], + r["room_kind"], + _parse_date(r["sales_start"]), + _parse_date(r["plan_date"]), + _parse_date(r["fact_date"]), + _to_int(r["readiness_pct"]), + r["construction_stage"], + r["finish_type"], + r["status"], + _bool_from_yes(r["sold"]), + r["rooms_dev"], + r["rooms_pd"], + r["rooms_obj"], + _normalize_rooms(r["rooms_obj"]), + _to_num(r["area_dev"]), + _to_num(r["area_pd"]), + _to_num(r["budget_rub"]), + _to_num(r["price_per_m2"]), + _to_num(r["offer_price"]), + _to_num(r["delta_price_rub"]), + _to_num(r["delta_price_pct"]), + r["price_method"], + _parse_date(r["price_set_date"]), + _parse_date(r["price_actual_date"]), + _parse_date(r["contract_date"]), + _parse_date(r["register_date"]), + r["deal_type"], + r["buyer_type"], + r["register_num"], + r["encumbrance"], + r["bank"], + _parse_date(r["encumbrance_start"]), + _parse_date(r["egrn_actual_date"]), + snapshot_date, + ) + + +_LOT_INSERT = """ +INSERT INTO objective_lots ( + objective_lot_id, objective_project_id, project_name, developer, city, + district, corpus_name, address, class, section, floor, lot_number, + premise_kind, sales_start_date, plan_completion_date, actual_completion_date, + readiness_pct, construction_stage, finishing, status, is_sold, + rooms_dev_site, rooms_pd, rooms_objective, rooms_int, + area_dev_site, area_pd, + price_calculated_total_rub, price_per_m2_rub, price_offer_total_rub, + price_delta_rub, price_delta_pct, pricing_method, price_set_date, + price_actual_date, contract_date, registration_date, contract_type, + buyer_type, registration_number, encumbrance_type, bank_name, + encumbrance_start_date, egrn_actual_date, snapshot_date +) VALUES %s +ON CONFLICT (objective_lot_id) DO UPDATE SET + project_name = EXCLUDED.project_name, + developer = EXCLUDED.developer, district = EXCLUDED.district, + corpus_name = EXCLUDED.corpus_name, address = EXCLUDED.address, + class = EXCLUDED.class, section = EXCLUDED.section, floor = EXCLUDED.floor, + lot_number = EXCLUDED.lot_number, premise_kind = EXCLUDED.premise_kind, + sales_start_date = EXCLUDED.sales_start_date, + plan_completion_date = EXCLUDED.plan_completion_date, + actual_completion_date = EXCLUDED.actual_completion_date, + readiness_pct = EXCLUDED.readiness_pct, + construction_stage = EXCLUDED.construction_stage, + finishing = EXCLUDED.finishing, + status = EXCLUDED.status, is_sold = EXCLUDED.is_sold, + rooms_dev_site = EXCLUDED.rooms_dev_site, + rooms_pd = EXCLUDED.rooms_pd, + rooms_objective = EXCLUDED.rooms_objective, + rooms_int = EXCLUDED.rooms_int, + area_dev_site = EXCLUDED.area_dev_site, area_pd = EXCLUDED.area_pd, + price_calculated_total_rub = EXCLUDED.price_calculated_total_rub, + price_per_m2_rub = EXCLUDED.price_per_m2_rub, + price_offer_total_rub = EXCLUDED.price_offer_total_rub, + price_delta_rub = EXCLUDED.price_delta_rub, + price_delta_pct = EXCLUDED.price_delta_pct, + pricing_method = EXCLUDED.pricing_method, + price_set_date = EXCLUDED.price_set_date, + price_actual_date = EXCLUDED.price_actual_date, + contract_date = EXCLUDED.contract_date, + registration_date = EXCLUDED.registration_date, + contract_type = EXCLUDED.contract_type, buyer_type = EXCLUDED.buyer_type, + registration_number = EXCLUDED.registration_number, + encumbrance_type = EXCLUDED.encumbrance_type, + bank_name = EXCLUDED.bank_name, + encumbrance_start_date = EXCLUDED.encumbrance_start_date, + egrn_actual_date = EXCLUDED.egrn_actual_date, + snapshot_date = EXCLUDED.snapshot_date, + updated_at = NOW() +""" + + +def _row_to_crm(r: sqlite3.Row) -> tuple: + return ( + _parse_month_to_first_day(r["month"]), + GROUP_NAME, + r["project"], + r["developer"], + r["district"], + r["obj_class"], + r["corpus"], + r["rooms_bucket"], + _normalize_rooms(r["rooms_bucket"]), + _parse_date(r["sales_start"]), + _parse_date(r["plan_date"]), + _parse_date(r["fact_date"]), + _to_int(r["months_in_sales"]), + _to_int(r["lots_pd"]), + _to_num(r["area_pd"]), + # ДДУ — у Антона нет + None, + None, + None, + None, + None, + None, + None, + # ДКП + None, + None, + None, + None, + None, + None, + None, + # Всего + _to_int(r["deals_total"]), + _to_int(r["deals_priced"]), + _to_num(r["sold_volume_m2"]), + _to_num(r["sold_volume_m2"]), + None, + _to_num(r["avg_price_m2"]), + _to_num(r["avg_area_m2"]), + # Offer + _to_int(r["stock_lots"]), + _to_num(r["stock_m2"]), + None, + _to_num(r["stock_avg_price_m2"]), + ) + + +_CRM_INSERT = """ +INSERT INTO objective_corpus_room_month ( + report_month, group_name, project_name, developer, district, class, + corpus_name, room_bucket, rooms_int, + sales_start_date, plan_completion_date, actual_completion_date, + months_in_realization, lots_pd_count, lots_pd_area, + deals_ddu_count, deals_ddu_count_priced, deals_ddu_vol_m2, + deals_ddu_vol_m2_priced, deals_ddu_sum_mln_rub, + deals_ddu_avg_price_thousand_rub_per_m2, deals_ddu_avg_area_m2, + deals_dkp_count, deals_dkp_count_priced, deals_dkp_vol_m2, + deals_dkp_vol_m2_priced, deals_dkp_sum_mln_rub, + deals_dkp_avg_price_thousand_rub_per_m2, deals_dkp_avg_area_m2, + deals_total_count, deals_total_count_priced, deals_total_vol_m2, + deals_total_vol_m2_priced, deals_total_sum_mln_rub, + deals_total_avg_price_thousand_rub_per_m2, deals_total_avg_area_m2, + offer_count, offer_area_m2, offer_sum_mln_rub, + offer_avg_price_thousand_rub_per_m2 +) VALUES %s +ON CONFLICT (report_month, group_name, project_name, corpus_name, room_bucket) +DO UPDATE SET + developer = EXCLUDED.developer, district = EXCLUDED.district, + class = EXCLUDED.class, rooms_int = EXCLUDED.rooms_int, + sales_start_date = EXCLUDED.sales_start_date, + plan_completion_date = EXCLUDED.plan_completion_date, + actual_completion_date = EXCLUDED.actual_completion_date, + months_in_realization = EXCLUDED.months_in_realization, + lots_pd_count = EXCLUDED.lots_pd_count, + lots_pd_area = EXCLUDED.lots_pd_area, + deals_total_count = EXCLUDED.deals_total_count, + deals_total_count_priced = EXCLUDED.deals_total_count_priced, + deals_total_vol_m2 = EXCLUDED.deals_total_vol_m2, + deals_total_vol_m2_priced = EXCLUDED.deals_total_vol_m2_priced, + deals_total_avg_price_thousand_rub_per_m2 = EXCLUDED.deals_total_avg_price_thousand_rub_per_m2, + deals_total_avg_area_m2 = EXCLUDED.deals_total_avg_area_m2, + offer_count = EXCLUDED.offer_count, + offer_area_m2 = EXCLUDED.offer_area_m2, + offer_avg_price_thousand_rub_per_m2 = EXCLUDED.offer_avg_price_thousand_rub_per_m2, + fetched_at = NOW() +""" + + +def _row_to_mapping(r: sqlite3.Row) -> tuple | None: + obj_id = _site_id_to_obj_id(r["site_id"]) + if obj_id is None: + return None + return ( + r["project"], + None, + GROUP_NAME, + obj_id, + False, + r["method"], + _to_num(r["score"]), + "imported_from_anton_sf", + ) + + +_MAP_INSERT = """ +INSERT INTO objective_complex_mapping ( + objective_complex_name, objective_project_id, objective_group, + domrf_obj_id, is_reviewed, match_method, match_score, note +) VALUES %s +ON CONFLICT (objective_complex_name, objective_group) DO UPDATE SET + domrf_obj_id = EXCLUDED.domrf_obj_id, + match_method = EXCLUDED.match_method, + match_score = EXCLUDED.match_score, + note = EXCLUDED.note, + updated_at = NOW() +""" + + +# ── orchestration ─────────────────────────────────────────────────────────── + + +def _etl_lots( + sl: sqlite3.Connection, pg, batch_size: int, progress_cb: Callable[[int], None] | None +) -> int: + snap = date.today() + cur = sl.execute("SELECT * FROM objective_lots") + n = 0 + batch: list[tuple] = [] + pg_cur = pg.cursor() + while True: + rows = cur.fetchmany(batch_size) + if not rows: + break + for r in rows: + batch.append(_row_to_lot(r, snap)) + if len(batch) >= batch_size: + psycopg2.extras.execute_values(pg_cur, _LOT_INSERT, batch, page_size=batch_size) + pg.commit() + n += len(batch) + batch.clear() + if progress_cb and n % (batch_size * 5) == 0: + progress_cb(n) + if batch: + psycopg2.extras.execute_values(pg_cur, _LOT_INSERT, batch, page_size=len(batch)) + pg.commit() + n += len(batch) + pg_cur.close() + if progress_cb: + progress_cb(n) + return n + + +def _etl_crm( + sl: sqlite3.Connection, pg, batch_size: int, progress_cb: Callable[[int], None] | None +) -> tuple[int, int]: + cur = sl.execute("SELECT * FROM objective_corp_month") + n = 0 + skipped = 0 + batch: list[tuple] = [] + pg_cur = pg.cursor() + while True: + rows = cur.fetchmany(batch_size) + if not rows: + break + for r in rows: + t = _row_to_crm(r) + if not t[0] or not t[2] or not t[6] or not t[7]: + skipped += 1 + continue + batch.append(t) + if len(batch) >= batch_size: + psycopg2.extras.execute_values(pg_cur, _CRM_INSERT, batch, page_size=batch_size) + pg.commit() + n += len(batch) + batch.clear() + if batch: + psycopg2.extras.execute_values(pg_cur, _CRM_INSERT, batch, page_size=len(batch)) + pg.commit() + n += len(batch) + pg_cur.close() + if progress_cb: + progress_cb(n) + return n, skipped + + +def _etl_mapping(sl: sqlite3.Connection, pg) -> tuple[int, int]: + cur = sl.execute("SELECT * FROM jk_objective_match ORDER BY score DESC NULLS LAST") + rows = cur.fetchall() + seen: set[tuple] = set() + batch: list[tuple] = [] + skipped_dups = 0 + for r in rows: + t = _row_to_mapping(r) + if t is None: + continue + key = (t[0], t[2]) # (complex_name, group) + if key in seen: + skipped_dups += 1 + continue + seen.add(key) + batch.append(t) + if not batch: + return 0, skipped_dups + pg_cur = pg.cursor() + psycopg2.extras.execute_values(pg_cur, _MAP_INSERT, batch, page_size=500) + pg.commit() + pg_cur.close() + return len(batch), skipped_dups + + +# ── public API ────────────────────────────────────────────────────────────── + + +def get_sqlite_info(sqlite_path: str | Path) -> dict[str, Any]: + """Метаданные SQLite-файла без полного запуска ETL. + + Используется в admin-coverage endpoint чтобы UI знал свежесть данных, + размер файла и сколько потенциально rows будет залито. + """ + p = Path(sqlite_path) + info: dict[str, Any] = { + "path": str(p), + "exists": p.exists(), + } + if not p.exists(): + return info + st = p.stat() + info["size_bytes"] = st.st_size + info["modified_at"] = st.st_mtime # epoch seconds + try: + c = sqlite3.connect(p) + info["lots"] = c.execute("SELECT COUNT(*) FROM objective_lots").fetchone()[0] + info["corp_room_month"] = c.execute("SELECT COUNT(*) FROM objective_corp_month").fetchone()[ + 0 + ] + info["mappings"] = c.execute("SELECT COUNT(*) FROM jk_objective_match").fetchone()[0] + c.close() + except sqlite3.Error as e: + info["error"] = str(e) + return info + + +def run_etl( + sqlite_path: str | Path, + db_url: str, + *, + batch_size: int = 2000, + progress_cb: Callable[[str, int], None] | None = None, +) -> dict[str, int]: + """Полный ETL: SQLite → PG. + + Args: + sqlite_path: путь к analysis.db (на проде — bind-mount /data/anton-sqlite/...) + db_url: postgresql://… connection string + batch_size: размер batch для execute_values (default 2000) + progress_cb: optional callback(stage: str, n_rows: int) для UI прогресса + + Returns: + dict с counts: {'lots', 'corp_room_month', 'mappings', 'crm_skipped', + 'mapping_dups_skipped'} + + Raises: + FileNotFoundError если SQLite не найден + psycopg2.Error при ошибках БД + """ + p = Path(sqlite_path) + if not p.exists(): + raise FileNotFoundError(f"SQLite не найден: {p}") + + logger.info("ETL Anton SQLite → PG: %s", p) + sl = sqlite3.connect(p) + sl.row_factory = sqlite3.Row + pg = psycopg2.connect(db_url) + try: + if progress_cb: + progress_cb("mapping_start", 0) + n_map, dups = _etl_mapping(sl, pg) + logger.info(" mapping: %d upserted, %d dups skipped", n_map, dups) + if progress_cb: + progress_cb("mapping_done", n_map) + + if progress_cb: + progress_cb("crm_start", 0) + n_crm, crm_skipped = _etl_crm( + sl, + pg, + batch_size, + lambda n: progress_cb("crm_progress", n) if progress_cb else None, + ) + logger.info(" crm: %d upserted, %d skipped", n_crm, crm_skipped) + if progress_cb: + progress_cb("crm_done", n_crm) + + if progress_cb: + progress_cb("lots_start", 0) + n_lots = _etl_lots( + sl, + pg, + batch_size, + lambda n: progress_cb("lots_progress", n) if progress_cb else None, + ) + logger.info(" lots: %d upserted", n_lots) + if progress_cb: + progress_cb("lots_done", n_lots) + finally: + sl.close() + pg.close() + + return { + "lots": n_lots, + "corp_room_month": n_crm, + "mappings": n_map, + "crm_skipped": crm_skipped, + "mapping_dups_skipped": dups, + } diff --git a/backend/app/services/objective_sync_config.py b/backend/app/services/objective_sync_config.py new file mode 100644 index 00000000..873284ca --- /dev/null +++ b/backend/app/services/objective_sync_config.py @@ -0,0 +1,178 @@ +"""Read/update Objective sync config (single-row БД-конфигурация). + +Используется: + - tasks.scrape_objective.sync_all_groups: читает groups/use_ddu/use_dkp/ + period_months_back/inter_group_delay_s/rate_ms/retries при каждом запуске + - workers.celery_app: читает cron_schedule при старте beat (один раз) + - api.v1.admin_scrape: GET/PUT /admin/scrape/objective/config + +Если таблица objective_sync_config недоступна (БД не поднята / таблицы нет / +ошибка) — все методы возвращают defaults на основе settings.objective_*. +Это обеспечивает безопасный fallback для cold-start и unit-тестов. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import Any + +from sqlalchemy import text + +from app.core.config import settings + +logger = logging.getLogger(__name__) + + +@dataclass +class ObjectiveSyncConfig: + cron_schedule: str + groups_csv: str + use_ddu: bool + use_dkp: bool + period_months_back: int + inter_group_delay_s: int + rate_ms: int + retries: int + updated_at: str | None = None + updated_by: str | None = None + + @property + def groups(self) -> list[str]: + return [g.strip() for g in self.groups_csv.split(",") if g.strip()] + + def to_dict(self) -> dict[str, Any]: + return { + "cron_schedule": self.cron_schedule, + "groups_csv": self.groups_csv, + "groups": self.groups, + "use_ddu": self.use_ddu, + "use_dkp": self.use_dkp, + "period_months_back": self.period_months_back, + "inter_group_delay_s": self.inter_group_delay_s, + "rate_ms": self.rate_ms, + "retries": self.retries, + "updated_at": self.updated_at, + "updated_by": self.updated_by, + } + + +def _defaults_from_settings() -> ObjectiveSyncConfig: + """Fallback когда БД-таблицы нет / ошибка.""" + return ObjectiveSyncConfig( + cron_schedule=settings.objective_sync_cron, + groups_csv=settings.objective_sync_groups, + use_ddu=True, + use_dkp=True, + period_months_back=1, + inter_group_delay_s=30, + rate_ms=3000, + retries=2, + ) + + +def get_config(db) -> ObjectiveSyncConfig: + """Прочитать config (single row id=1). Fallback на settings при ошибке.""" + try: + r = ( + db.execute( + text( + """ + SELECT cron_schedule, groups_csv, use_ddu, use_dkp, + period_months_back, inter_group_delay_s, rate_ms, retries, + updated_at, updated_by + FROM objective_sync_config WHERE id = 1 + """ + ) + ) + .mappings() + .first() + ) + except Exception as e: + logger.warning("get_config: fallback на settings (БД недоступна?): %s", e) + return _defaults_from_settings() + + if not r: + # Таблица есть но пустая — посеем default и вернём + try: + db.execute( + text("INSERT INTO objective_sync_config (id) VALUES (1) ON CONFLICT DO NOTHING") + ) + db.commit() + except Exception: + pass + return _defaults_from_settings() + + return ObjectiveSyncConfig( + cron_schedule=r["cron_schedule"], + groups_csv=r["groups_csv"], + use_ddu=bool(r["use_ddu"]), + use_dkp=bool(r["use_dkp"]), + period_months_back=int(r["period_months_back"]), + inter_group_delay_s=int(r["inter_group_delay_s"]), + rate_ms=int(r["rate_ms"]), + retries=int(r["retries"]), + updated_at=r["updated_at"].isoformat() if r["updated_at"] else None, + updated_by=r["updated_by"], + ) + + +def update_config( + db, + *, + cron_schedule: str | None = None, + groups_csv: str | None = None, + use_ddu: bool | None = None, + use_dkp: bool | None = None, + period_months_back: int | None = None, + inter_group_delay_s: int | None = None, + rate_ms: int | None = None, + retries: int | None = None, + updated_by: str | None = None, +) -> ObjectiveSyncConfig: + """PATCH-style update — обновляет только переданные поля.""" + sets: list[str] = ["updated_at = NOW()"] + params: dict[str, Any] = {} + for field, value in [ + ("cron_schedule", cron_schedule), + ("groups_csv", groups_csv), + ("use_ddu", use_ddu), + ("use_dkp", use_dkp), + ("period_months_back", period_months_back), + ("inter_group_delay_s", inter_group_delay_s), + ("rate_ms", rate_ms), + ("retries", retries), + ("updated_by", updated_by), + ]: + if value is not None: + sets.append(f"{field} = :{field}") + params[field] = value + if len(sets) == 1: + # Только updated_at — нечего обновлять + return get_config(db) + + # Гарантия что строка существует + db.execute(text("INSERT INTO objective_sync_config (id) VALUES (1) ON CONFLICT DO NOTHING")) + db.execute( + text(f"UPDATE objective_sync_config SET {', '.join(sets)} WHERE id = 1"), + params, + ) + db.commit() + return get_config(db) + + +def get_cron_schedule_safe(db_factory) -> str: + """Безопасное чтение cron при старте Celery beat — открывает свою сессию. + + db_factory: callable () -> Session (обычно SessionLocal) + """ + try: + db = db_factory() + try: + cfg = get_config(db) + return cfg.cron_schedule + finally: + db.close() + except Exception as e: + logger.warning("get_cron_schedule_safe: fallback на settings: %s", e) + return settings.objective_sync_cron diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index 065a84d7..ad346817 100644 --- a/backend/app/workers/celery_app.py +++ b/backend/app/workers/celery_app.py @@ -43,6 +43,7 @@ celery_app = Celery( "app.workers.tasks.scrape_nspd", "app.workers.tasks.refresh_analytics", "app.workers.tasks.scrape_objective", + "app.workers.tasks.objective_etl", ], ) celery_app.conf.timezone = "Europe/Moscow" @@ -86,13 +87,35 @@ celery_app.conf.beat_schedule["refresh-ekb-districts-medians"] = { "kwargs": {"window_months": 24, "min_deals": 50}, } -# Objective sync — еженедельно (объёмы умеренные, ключ платный = бережём limits). -# По умолчанию: пн 05:00 МСК. Если OBJECTIVE_API_KEY пуст — task сразу skipped. + +# Objective sync — наш ЕДИНЫЙ source of truth для Objective-данных. +# Дёргает api.objctv.ru напрямую и пишет в PostgreSQL (objective_lots / +# corpus_room_month / lots_history) минуя любые промежуточные SQLite. +# +# Динамическая конфигурация в БД (objective_sync_config single-row): +# - cron_schedule (читается ОДИН РАЗ при старте beat — после изменения +# требует `docker compose restart beat`) +# - groups_csv, use_ddu, use_dkp, period_months_back, inter_group_delay_s, +# rate_ms, retries — читаются task'ом ПРИ КАЖДОМ запуске, динамически. +# +# Если БД недоступна на старте — fallback на settings.objective_sync_cron. +def _objective_cron() -> "crontab": + """Прочитать cron из БД с fallback на settings.""" + 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 as e: + logger.warning("objective beat cron: fallback на settings (%s)", e) + cron_str = settings.objective_sync_cron + return _parse_cron(cron_str) + + celery_app.conf.beat_schedule["objective-sync"] = { - "task": "tasks.scrape_objective.sync_objective_group", - "schedule": _parse_cron(settings.objective_sync_cron), + "task": "tasks.scrape_objective.sync_all_groups", + "schedule": _objective_cron(), "kwargs": { - "group_name": settings.objective_default_group, "triggered_by": "beat", }, } diff --git a/backend/app/workers/tasks/objective_etl.py b/backend/app/workers/tasks/objective_etl.py new file mode 100644 index 00000000..6e9d12bd --- /dev/null +++ b/backend/app/workers/tasks/objective_etl.py @@ -0,0 +1,169 @@ +"""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 в psycopg2-style + # (psycopg2 не принимает '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() diff --git a/backend/app/workers/tasks/scrape_objective.py b/backend/app/workers/tasks/scrape_objective.py index 7b15cb46..7e03dd1f 100644 --- a/backend/app/workers/tasks/scrape_objective.py +++ b/backend/app/workers/tasks/scrape_objective.py @@ -143,26 +143,43 @@ def sync_objective_group( self: Any, group_name: str | None = None, triggered_by: str = "beat", + use_ddu: bool = True, + use_dkp: bool = True, + period_months_back: int = 1, + rate_ms: int = 3000, + retries: int = 2, ) -> dict[str, Any]: - """Sync 4 канонических отчёта Объектив для одной группы (территории). + """Sync канонические отчёты Объектив для одной группы (территории). + + Args: + group_name: имя группы (например 'Свердловская область'). None → settings default + triggered_by: 'beat'/'manual'/'api'/'beat-multi' (журнал) + use_ddu/use_dkp: какие сделки тянуть (передаётся в API) + period_months_back: для Сводных/Корпуса — сколько месяцев назад + захватывать в start_date (1=последний полный месяц) + rate_ms: пауза между HTTP-запросами в client (мс) + retries: ретраи на 429/5xx Идемпотентность: каждый прогон создаёт новые objective_raw_reports - (history-style). Нормализация (parse → objective_oks/lots) — отдельная - задача после изучения реальной схемы payload через smoke-script. + (history-style). Парсинг сразу через late-import парсера 70. """ 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 + # period_months_back=1 → последний полный месяц [start_date..end_date) + # period_months_back=N → N последних полных месяцев end_date = date.today().replace(day=1) - start_date = (end_date - timedelta(days=1)).replace(day=1) + start_date = end_date + for _ in range(max(period_months_back, 1)): + start_date = (start_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() + client = ObjectiveClient(rate_ms=rate_ms, retries=retries) reports_ok = 0 reports_failed = 0 @@ -196,8 +213,8 @@ def sync_objective_group( "group_name": group, "start_date": start_date, "end_date": end_date, - "use_ddu": True, - "use_dkp": True, + "use_ddu": use_ddu, + "use_dkp": use_dkp, }, "Объединенные данные", "Сводные", @@ -206,7 +223,7 @@ def sync_objective_group( ( "lots_pf", "report_lots_per_flat", - {"group_name": group, "use_ddu": True, "use_dkp": True}, + {"group_name": group, "use_ddu": use_ddu, "use_dkp": use_dkp}, "Объединенные данные", "Поквартирные", "Лоты", @@ -326,3 +343,141 @@ def sync_objective_group( raise finally: db.close() + + +# ── Multi-group wrapper для еженедельного beat ────────────────────────────── + + +@celery_app.task( + bind=True, + name="tasks.scrape_objective.sync_all_groups", + max_retries=0, +) +def sync_all_groups( + self: Any, + groups: list[str] | None = None, + triggered_by: str = "beat", + # Все остальные параметры — None = взять из БД-конфига objective_sync_config + use_ddu: bool | None = None, + use_dkp: bool | None = None, + period_months_back: int | None = None, + inter_group_delay_s: int | None = None, + rate_ms: int | None = None, + retries: int | None = None, +) -> dict[str, Any]: + """Перебирает группы из БД-конфига (или explicit override) и тянет каждую. + + Все None-параметры подменяются из таблицы objective_sync_config (PATCH-style). + Это позволяет менять настройки sync через админку без редеплоя. + + Args: + groups: explicit override (если None — берётся из БД config.groups_csv) + triggered_by: 'beat' | 'manual' | 'api' + use_ddu/use_dkp/period_months_back/inter_group_delay_s/rate_ms/retries: + None → читается из БД config + + Returns: + dict {'groups': [...], 'total_lots': N, 'total_corpus_room': N, + 'total_failed': N, 'config_used': {...}} + """ + import time as _time + + from app.services.objective_sync_config import get_config + + if not settings.objective_api_key: + logger.warning("sync_all_groups skipped: OBJECTIVE_API_KEY не задан") + return {"skipped": True, "reason": "no_api_key"} + + # Загружаем динамический config из БД (с fallback на settings) + cfg_db = SessionLocal() + try: + cfg = get_config(cfg_db) + finally: + cfg_db.close() + + # PATCH-merge: explicit args > БД config + eff_groups = groups if groups is not None else cfg.groups + eff_use_ddu = use_ddu if use_ddu is not None else cfg.use_ddu + eff_use_dkp = use_dkp if use_dkp is not None else cfg.use_dkp + eff_period = period_months_back if period_months_back is not None else cfg.period_months_back + eff_delay = inter_group_delay_s if inter_group_delay_s is not None else cfg.inter_group_delay_s + eff_rate_ms = rate_ms if rate_ms is not None else cfg.rate_ms + eff_retries = retries if retries is not None else cfg.retries + + if not eff_groups: + return {"skipped": True, "reason": "no_groups_configured"} + + logger.info( + "sync_all_groups starting for %d groups: %s " + "(use_ddu=%s use_dkp=%s period_months=%s inter_group=%ss rate=%sms retries=%s)", + len(eff_groups), + eff_groups, + eff_use_ddu, + eff_use_dkp, + eff_period, + eff_delay, + eff_rate_ms, + eff_retries, + ) + + results: list[dict[str, Any]] = [] + total_lots = 0 + total_crm = 0 + total_failed = 0 + + for idx, group in enumerate(eff_groups): + logger.info("[%d/%d] sync_objective_group(group=%r) START", idx + 1, len(eff_groups), group) + try: + res = sync_objective_group( + self, + group_name=group, + triggered_by=f"{triggered_by}-multi", + use_ddu=eff_use_ddu, + use_dkp=eff_use_dkp, + period_months_back=eff_period, + rate_ms=eff_rate_ms, + retries=eff_retries, + ) + if res.get("skipped"): + results.append({"group": group, "status": "skipped", "reason": res.get("reason")}) + else: + total_lots += int(res.get("rows_lots") or 0) + total_crm += int(res.get("rows_corpus_room") or 0) + if int(res.get("reports_failed") or 0) > 0: + total_failed += 1 + results.append({"group": group, "status": "done", **res}) + except Exception as e: + logger.exception( + "[%d/%d] sync for group=%r failed: %s", idx + 1, len(eff_groups), group, e + ) + total_failed += 1 + results.append( + {"group": group, "status": "failed", "error": f"{type(e).__name__}: {e}"} + ) + + if idx < len(eff_groups) - 1: + logger.info("inter-group sleep %ds", eff_delay) + _time.sleep(eff_delay) + + logger.info( + "sync_all_groups DONE: total_lots=%d total_crm=%d failed=%d", + total_lots, + total_crm, + total_failed, + ) + return { + "groups": results, + "total_lots": total_lots, + "total_corpus_room": total_crm, + "total_failed": total_failed, + "n_groups": len(eff_groups), + "config_used": { + "groups": eff_groups, + "use_ddu": eff_use_ddu, + "use_dkp": eff_use_dkp, + "period_months_back": eff_period, + "inter_group_delay_s": eff_delay, + "rate_ms": eff_rate_ms, + "retries": eff_retries, + }, + } diff --git a/data/sql/71_etl_anton_sqlite_to_pg.py b/data/sql/71_etl_anton_sqlite_to_pg.py new file mode 100644 index 00000000..cc9a889a --- /dev/null +++ b/data/sql/71_etl_anton_sqlite_to_pg.py @@ -0,0 +1,517 @@ +"""ETL: SQLite Антона (sf_anton_snapshot.db) → наша PG. + +Грузит 3 таблицы из его SQLite в нашу schema 68 v2: + - objective_lots (303 677 строк) + - objective_corp_month ( 19 738 строк) + - jk_objective_match ( 229 строк) + +Запуск: + cd backend + DATABASE_URL=postgresql://gendesign:PWD@localhost:15432/gendesign \\ + uv run python -X utf8 ../data/sql/71_etl_anton_sqlite_to_pg.py [--dry-run] [--limit N] + +Идемпотентно через ON CONFLICT DO UPDATE по UNIQUE-ключам. +Для повторной заливки: данные в таблицах перезатрутся (snapshot_date обновится). + +Источник: Антоновский /sf/api/* SQLite (~340 MB), скачанный в repo-root через scp: + scp gendesign:/opt/gendesign/site-finder/analysis.db sf_anton_snapshot.db + +Mapping полей задокументирован в _ROW_TO_LOT / _ROW_TO_CRM / _ROW_TO_MAPPING ниже. + +После прогона ETL рекомендуется DISABLE наш Celery objective-sync beat task +(Антон уже еженедельно тянет — не дублируем платный API). +""" + +from __future__ import annotations + +import argparse +import logging +import os +import re +import sqlite3 +import sys +from datetime import date, datetime +from pathlib import Path + +import psycopg2 +import psycopg2.extras + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") +logger = logging.getLogger("etl_anton") + +REPO_ROOT = Path(__file__).resolve().parent.parent.parent +SQLITE_DEFAULT = REPO_ROOT / "sf_anton_snapshot.db" + +GROUP_NAME = "Екатеринбург" # У Антона нет колонки group_name — у нас всё под Екб + +# ── helpers ───────────────────────────────────────────────────────────────── + + +def _parse_date(s: str | None) -> date | None: + """SQLite хранит как ISO-строку или None.""" + if s is None or s == "": + return None + try: + return date.fromisoformat(s[:10]) + except (ValueError, TypeError): + return None + + +def _parse_month_to_first_day(s: str | None) -> date | None: + """'2025-08' → date(2025, 8, 1).""" + if not s: + return None + try: + y, m = s.split("-") + return date(int(y), int(m), 1) + except (ValueError, TypeError): + return None + + +def _normalize_rooms(s: str | None) -> int | None: + """'студия'→0, '1'→1, …, '4+'→4 (как в наш парсер 70).""" + if not s: + return None + sl = s.strip().lower() + if "студ" in sl: + return 0 + m = re.match(r"^(\d+)", sl) + if m: + return min(int(m.group(1)), 5) + return None + + +def _bool_from_yes(v) -> bool | None: + if v is None or v == "": + return None + s = str(v).strip().lower() + if s in ("да", "true", "1", "yes"): + return True + if s in ("нет", "false", "0", "no"): + return False + return None + + +def _site_id_to_obj_id(s: str | None) -> int | None: + """'jk:49854' → 49854. 'parcel:...' → None.""" + if not s: + return None + if s.startswith("jk:"): + try: + return int(s[3:]) + except ValueError: + return None + return None + + +def _to_int(v) -> int | None: + if v is None or v == "": + return None + try: + return int(v) + except (TypeError, ValueError): + return None + + +def _to_num(v) -> float | None: + if v is None or v == "": + return None + try: + return float(v) + except (TypeError, ValueError): + return None + + +# ── row mappings ──────────────────────────────────────────────────────────── + + +def _row_to_lot(r: sqlite3.Row, snapshot_date: date) -> tuple: + """SQLite objective_lots row → tuple для INSERT в наш objective_lots.""" + return ( + r["lot_id"], # objective_lot_id + r["project_id"], # objective_project_id + r["project"], # project_name (NOT NULL) + r["developer"], + r["city"], + r["district"], + r["corpus"], # corpus_name + r["address"], + r["obj_class"], # class + r["section"], + _to_int(r["floor"]), + r["lot_num"], # lot_number + r["room_kind"], # premise_kind + _parse_date(r["sales_start"]), + _parse_date(r["plan_date"]), + _parse_date(r["fact_date"]), + _to_int(r["readiness_pct"]), + r["construction_stage"], + r["finish_type"], # finishing + r["status"], + _bool_from_yes(r["sold"]), # is_sold + r["rooms_dev"], # rooms_dev_site + r["rooms_pd"], + r["rooms_obj"], # rooms_objective + _normalize_rooms(r["rooms_obj"]), # rooms_int (0..5) + _to_num(r["area_dev"]), # area_dev_site + _to_num(r["area_pd"]), + _to_num(r["budget_rub"]), # price_calculated_total_rub + _to_num(r["price_per_m2"]), # price_per_m2_rub (Р/м², не тыс) + _to_num(r["offer_price"]), # price_offer_total_rub + _to_num(r["delta_price_rub"]), + _to_num(r["delta_price_pct"]), + r["price_method"], # pricing_method + _parse_date(r["price_set_date"]), + _parse_date(r["price_actual_date"]), + _parse_date(r["contract_date"]), + _parse_date(r["register_date"]), # registration_date + r["deal_type"], # contract_type + r["buyer_type"], + r["register_num"], # registration_number + r["encumbrance"], # encumbrance_type + r["bank"], # bank_name + _parse_date(r["encumbrance_start"]), # encumbrance_start_date + _parse_date(r["egrn_actual_date"]), + snapshot_date, # snapshot_date — день ETL + ) + + +_LOT_INSERT = """ +INSERT INTO objective_lots ( + objective_lot_id, objective_project_id, project_name, developer, city, + district, corpus_name, address, class, section, floor, lot_number, + premise_kind, sales_start_date, plan_completion_date, actual_completion_date, + readiness_pct, construction_stage, finishing, status, is_sold, + rooms_dev_site, rooms_pd, rooms_objective, rooms_int, + area_dev_site, area_pd, + price_calculated_total_rub, price_per_m2_rub, price_offer_total_rub, + price_delta_rub, price_delta_pct, pricing_method, price_set_date, + price_actual_date, contract_date, registration_date, contract_type, + buyer_type, registration_number, encumbrance_type, bank_name, + encumbrance_start_date, egrn_actual_date, snapshot_date +) VALUES %s +ON CONFLICT (objective_lot_id) DO UPDATE SET + project_name = EXCLUDED.project_name, + developer = EXCLUDED.developer, district = EXCLUDED.district, + corpus_name = EXCLUDED.corpus_name, address = EXCLUDED.address, + class = EXCLUDED.class, section = EXCLUDED.section, floor = EXCLUDED.floor, + lot_number = EXCLUDED.lot_number, premise_kind = EXCLUDED.premise_kind, + sales_start_date = EXCLUDED.sales_start_date, + plan_completion_date = EXCLUDED.plan_completion_date, + actual_completion_date = EXCLUDED.actual_completion_date, + readiness_pct = EXCLUDED.readiness_pct, + construction_stage = EXCLUDED.construction_stage, + finishing = EXCLUDED.finishing, + status = EXCLUDED.status, is_sold = EXCLUDED.is_sold, + rooms_dev_site = EXCLUDED.rooms_dev_site, + rooms_pd = EXCLUDED.rooms_pd, + rooms_objective = EXCLUDED.rooms_objective, + rooms_int = EXCLUDED.rooms_int, + area_dev_site = EXCLUDED.area_dev_site, area_pd = EXCLUDED.area_pd, + price_calculated_total_rub = EXCLUDED.price_calculated_total_rub, + price_per_m2_rub = EXCLUDED.price_per_m2_rub, + price_offer_total_rub = EXCLUDED.price_offer_total_rub, + price_delta_rub = EXCLUDED.price_delta_rub, + price_delta_pct = EXCLUDED.price_delta_pct, + pricing_method = EXCLUDED.pricing_method, + price_set_date = EXCLUDED.price_set_date, + price_actual_date = EXCLUDED.price_actual_date, + contract_date = EXCLUDED.contract_date, + registration_date = EXCLUDED.registration_date, + contract_type = EXCLUDED.contract_type, buyer_type = EXCLUDED.buyer_type, + registration_number = EXCLUDED.registration_number, + encumbrance_type = EXCLUDED.encumbrance_type, + bank_name = EXCLUDED.bank_name, + encumbrance_start_date = EXCLUDED.encumbrance_start_date, + egrn_actual_date = EXCLUDED.egrn_actual_date, + snapshot_date = EXCLUDED.snapshot_date, + updated_at = NOW() +""" + + +def _row_to_crm(r: sqlite3.Row) -> tuple: + """objective_corp_month → objective_corpus_room_month. + + Антон НЕ разделяет ДДУ vs ДКП vs всего — у нас только totals из его данных. + NULL'и в ddu/dkp колонках — Антон их не давал. + """ + return ( + _parse_month_to_first_day(r["month"]), # report_month + GROUP_NAME, # group_name + r["project"], # project_name + r["developer"], + r["district"], + r["obj_class"], # class + r["corpus"], # corpus_name + r["rooms_bucket"], # room_bucket + _normalize_rooms(r["rooms_bucket"]), # rooms_int + _parse_date(r["sales_start"]), + _parse_date(r["plan_date"]), + _parse_date(r["fact_date"]), + _to_int(r["months_in_sales"]), # months_in_realization + _to_int(r["lots_pd"]), # lots_pd_count + _to_num(r["area_pd"]), # lots_pd_area + # ДДУ — у Антона нет, NULL'им + None, None, None, None, None, None, None, + # ДКП — то же + None, None, None, None, None, None, None, + # Всего (Антон даёт только это) + _to_int(r["deals_total"]), # deals_total_count + _to_int(r["deals_priced"]), # deals_total_count_priced + _to_num(r["sold_volume_m2"]), # deals_total_vol_m2 + _to_num(r["sold_volume_m2"]), # deals_total_vol_m2_priced (= тот же) + None, # deals_total_sum_mln_rub — нет + _to_num(r["avg_price_m2"]), # deals_total_avg_price_thousand_rub_per_m2 (тыс.Р/м²) + _to_num(r["avg_area_m2"]), # deals_total_avg_area_m2 + # Offer + _to_int(r["stock_lots"]), # offer_count + _to_num(r["stock_m2"]), # offer_area_m2 + None, # offer_sum_mln_rub — нет + _to_num(r["stock_avg_price_m2"]), # offer_avg_price_thousand_rub_per_m2 + ) + + +_CRM_INSERT = """ +INSERT INTO objective_corpus_room_month ( + report_month, group_name, project_name, developer, district, class, + corpus_name, room_bucket, rooms_int, + sales_start_date, plan_completion_date, actual_completion_date, + months_in_realization, lots_pd_count, lots_pd_area, + deals_ddu_count, deals_ddu_count_priced, deals_ddu_vol_m2, + deals_ddu_vol_m2_priced, deals_ddu_sum_mln_rub, + deals_ddu_avg_price_thousand_rub_per_m2, deals_ddu_avg_area_m2, + deals_dkp_count, deals_dkp_count_priced, deals_dkp_vol_m2, + deals_dkp_vol_m2_priced, deals_dkp_sum_mln_rub, + deals_dkp_avg_price_thousand_rub_per_m2, deals_dkp_avg_area_m2, + deals_total_count, deals_total_count_priced, deals_total_vol_m2, + deals_total_vol_m2_priced, deals_total_sum_mln_rub, + deals_total_avg_price_thousand_rub_per_m2, deals_total_avg_area_m2, + offer_count, offer_area_m2, offer_sum_mln_rub, + offer_avg_price_thousand_rub_per_m2 +) VALUES %s +ON CONFLICT (report_month, group_name, project_name, corpus_name, room_bucket) +DO UPDATE SET + developer = EXCLUDED.developer, district = EXCLUDED.district, + class = EXCLUDED.class, rooms_int = EXCLUDED.rooms_int, + sales_start_date = EXCLUDED.sales_start_date, + plan_completion_date = EXCLUDED.plan_completion_date, + actual_completion_date = EXCLUDED.actual_completion_date, + months_in_realization = EXCLUDED.months_in_realization, + lots_pd_count = EXCLUDED.lots_pd_count, + lots_pd_area = EXCLUDED.lots_pd_area, + deals_total_count = EXCLUDED.deals_total_count, + deals_total_count_priced = EXCLUDED.deals_total_count_priced, + deals_total_vol_m2 = EXCLUDED.deals_total_vol_m2, + deals_total_vol_m2_priced = EXCLUDED.deals_total_vol_m2_priced, + deals_total_avg_price_thousand_rub_per_m2 = EXCLUDED.deals_total_avg_price_thousand_rub_per_m2, + deals_total_avg_area_m2 = EXCLUDED.deals_total_avg_area_m2, + offer_count = EXCLUDED.offer_count, + offer_area_m2 = EXCLUDED.offer_area_m2, + offer_avg_price_thousand_rub_per_m2 = EXCLUDED.offer_avg_price_thousand_rub_per_m2, + fetched_at = NOW() +""" + + +def _row_to_mapping(r: sqlite3.Row) -> tuple | None: + """jk_objective_match → objective_complex_mapping. + Возвращает None если site_id не jk: (parcel пропускаем — у нас нет parcel-таблицы). + """ + obj_id = _site_id_to_obj_id(r["site_id"]) + if obj_id is None: + return None + return ( + r["project"], # objective_complex_name + None, # objective_project_id (Антон не хранит) + GROUP_NAME, + obj_id, # domrf_obj_id + False, # is_reviewed + r["method"], # match_method ('fuzzy', etc.) + _to_num(r["score"]), # match_score + "imported_from_anton_sf", # note + ) + + +_MAP_INSERT = """ +INSERT INTO objective_complex_mapping ( + objective_complex_name, objective_project_id, objective_group, + domrf_obj_id, is_reviewed, match_method, match_score, note +) VALUES %s +ON CONFLICT (objective_complex_name, objective_group) DO UPDATE SET + domrf_obj_id = EXCLUDED.domrf_obj_id, + match_method = EXCLUDED.match_method, + match_score = EXCLUDED.match_score, + note = EXCLUDED.note, + updated_at = NOW() +""" + + +# ── orchestration ─────────────────────────────────────────────────────────── + + +def etl_lots(sl: sqlite3.Connection, pg, dry_run: bool, limit: int | None, + batch_size: int) -> int: + snap = date.today() + sql = "SELECT * FROM objective_lots" + if limit: + sql += f" LIMIT {limit}" + cur = sl.execute(sql) + n = 0 + batch: list[tuple] = [] + pg_cur = pg.cursor() if pg else None + while True: + rows = cur.fetchmany(batch_size) + if not rows: + break + for r in rows: + batch.append(_row_to_lot(r, snap)) + if len(batch) >= batch_size: + if not dry_run: + psycopg2.extras.execute_values(pg_cur, _LOT_INSERT, batch, page_size=batch_size) + pg.commit() + n += len(batch) + batch.clear() + if n % (batch_size * 5) == 0: + logger.info(" lots: %d upserted", n) + if batch: + if not dry_run: + psycopg2.extras.execute_values(pg_cur, _LOT_INSERT, batch, page_size=len(batch)) + pg.commit() + n += len(batch) + if pg_cur is not None: + pg_cur.close() + return n + + +def etl_crm(sl: sqlite3.Connection, pg, dry_run: bool, limit: int | None, + batch_size: int) -> int: + sql = "SELECT * FROM objective_corp_month" + if limit: + sql += f" LIMIT {limit}" + cur = sl.execute(sql) + n = 0 + skipped = 0 + batch: list[tuple] = [] + pg_cur = pg.cursor() if pg else None + while True: + rows = cur.fetchmany(batch_size) + if not rows: + break + for r in rows: + t = _row_to_crm(r) + # Skip строк где report_month/project/corpus/room_bucket NULL — они нарушат UNIQUE. + if not t[0] or not t[2] or not t[6] or not t[7]: + skipped += 1 + continue + batch.append(t) + if len(batch) >= batch_size: + if not dry_run: + psycopg2.extras.execute_values(pg_cur, _CRM_INSERT, batch, page_size=batch_size) + pg.commit() + n += len(batch) + batch.clear() + if batch: + if not dry_run: + psycopg2.extras.execute_values(pg_cur, _CRM_INSERT, batch, page_size=len(batch)) + pg.commit() + n += len(batch) + if pg_cur is not None: + pg_cur.close() + if skipped: + logger.warning(" corp_month: пропущено %d строк с пустыми обязательными полями", skipped) + return n + + +def etl_mapping(sl: sqlite3.Connection, pg, dry_run: bool) -> int: + # ORDER BY score DESC — при дублях берём mapping с наибольшим match score. + cur = sl.execute("SELECT * FROM jk_objective_match ORDER BY score DESC NULLS LAST") + rows = cur.fetchall() + # Дедупликация по (objective_complex_name, objective_group) — наш UNIQUE. + # У Антона разные site_id (jk:N) могут матчиться в один и тот же Objective project + # (fuzzy mapping для близких имён). Берём первый = с лучшим score. + seen: set[tuple[str, str]] = set() + batch: list[tuple] = [] + skipped_dups = 0 + for r in rows: + t = _row_to_mapping(r) + if t is None: + continue + key = (t[0], t[2]) # (complex_name, group) + if key in seen: + skipped_dups += 1 + continue + seen.add(key) + batch.append(t) + if skipped_dups: + logger.info(" mapping: дедуплицировано %d дублей (один Objective project ↔ несколько jk)", skipped_dups) + if not batch: + return 0 + if not dry_run: + pg_cur = pg.cursor() + psycopg2.extras.execute_values(pg_cur, _MAP_INSERT, batch, page_size=500) + pg.commit() + pg_cur.close() + return len(batch) + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--sqlite", default=str(SQLITE_DEFAULT), + help=f"Путь к SQLite (default: {SQLITE_DEFAULT})") + ap.add_argument("--db-url", default=os.environ.get("DATABASE_URL"), + help="postgres URL (или env DATABASE_URL). Через SSH-туннель: " + "postgresql://gendesign:PWD@localhost:15432/gendesign") + ap.add_argument("--dry-run", action="store_true", + help="Не писать в PG (только посчитать строки)") + ap.add_argument("--limit", type=int, default=None, + help="Ограничить кол-во строк (для пробного прогона)") + ap.add_argument("--batch-size", type=int, default=500, + help="execute_values batch size (default 500)") + ap.add_argument("--only", choices=("lots", "crm", "mapping"), default=None, + help="Грузить только одну таблицу") + args = ap.parse_args() + + sqlite_path = Path(args.sqlite) + if not sqlite_path.exists(): + logger.error("SQLite не найден: %s", sqlite_path) + return 1 + if not args.db_url and not args.dry_run: + logger.error("Нужен --db-url или env DATABASE_URL") + return 2 + + logger.info("ETL: %s → %s", sqlite_path.name, args.db_url or "(dry-run)") + + sl = sqlite3.connect(sqlite_path) + sl.row_factory = sqlite3.Row + + pg = None + if not args.dry_run: + pg = psycopg2.connect(args.db_url) + + t0 = datetime.now() + try: + if args.only in (None, "mapping"): + logger.info("[1/3] objective_complex_mapping (Антоновский match)…") + n_map = etl_mapping(sl, pg, args.dry_run) + logger.info(" → upserted %d mappings", n_map) + + if args.only in (None, "crm"): + logger.info("[2/3] objective_corpus_room_month (~19 738 строк)…") + n_crm = etl_crm(sl, pg, args.dry_run, args.limit, args.batch_size) + logger.info(" → upserted %d corp_month rows", n_crm) + + if args.only in (None, "lots"): + logger.info("[3/3] objective_lots (~303 677 строк)…") + n_lots = etl_lots(sl, pg, args.dry_run, args.limit, args.batch_size) + logger.info(" → upserted %d lots", n_lots) + finally: + sl.close() + if pg is not None: + pg.close() + + elapsed = (datetime.now() - t0).total_seconds() + logger.info("ETL DONE in %.1fs%s", elapsed, " (dry-run)" if args.dry_run else "") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/data/sql/72_objective_sync_config.sql b/data/sql/72_objective_sync_config.sql new file mode 100644 index 00000000..7489b924 --- /dev/null +++ b/data/sql/72_objective_sync_config.sql @@ -0,0 +1,46 @@ +-- Динамическая конфигурация Objective sync — single-row settings. +-- +-- Зачем нужна: +-- 1. В админке /admin/scrape/objective можно менять параметры запроса +-- (groups, use_ddu/dkp, period_months_back, inter_group_delay) без редеплоя. +-- 2. Cron-расписание тоже хранится здесь — Celery beat читает его при старте. +-- Изменение требует restart beat-контейнера (это указано в UI как hint). +-- +-- Pattern: single-row (id=1), UPSERT для безопасных обновлений. +-- Если строки нет — task/beat падают на settings.* env как fallback. + +CREATE TABLE IF NOT EXISTS objective_sync_config ( + id SMALLINT PRIMARY KEY DEFAULT 1 CHECK (id = 1), + -- Cron-расписание для Celery beat (5-полевой формат: 'minute hour dom month dow'). + -- Celery beat перечитывает это поле только при старте — после изменения + -- нужен `docker compose restart beat`. + cron_schedule TEXT NOT NULL DEFAULT '0 6 * * tue', + -- Группы для sync (csv). Эмпирически проверены на тарифе: + -- 'Свердловская область', 'Челябинск', 'Тюмень', 'Пермь', + -- 'Екатеринбург', 'Верхняя Пышма', 'Берёзовский', 'Нижний Тагил', + -- 'Каменск-Уральский', 'Первоуральск', 'Новоуральск', 'Сысерть'. + groups_csv TEXT NOT NULL DEFAULT 'Свердловская область,Челябинск,Тюмень,Пермь', + -- Параметры запроса /v2/Report/GetReport. + use_ddu BOOLEAN NOT NULL DEFAULT TRUE, + use_dkp BOOLEAN NOT NULL DEFAULT TRUE, + -- Сколько месяцев назад тянуть для Сводных отчётов (corp_sum). + -- 1 = только последний полный месяц (default). 12 = весь год. + -- Поквартирные/Лоты не имеют дат — всегда snapshot. + period_months_back SMALLINT NOT NULL DEFAULT 1 CHECK (period_months_back BETWEEN 1 AND 24), + -- Пауза между группами в одном task'е (секунды). Защита от 429. + inter_group_delay_s SMALLINT NOT NULL DEFAULT 30 CHECK (inter_group_delay_s BETWEEN 0 AND 300), + -- Пауза между HTTP-запросами в client (мс). Передаётся в ObjectiveClient(rate_ms=…). + rate_ms SMALLINT NOT NULL DEFAULT 3000 CHECK (rate_ms BETWEEN 0 AND 10000), + -- Сколько ретраев на 429/5xx. + retries SMALLINT NOT NULL DEFAULT 2 CHECK (retries BETWEEN 0 AND 5), + updated_at timestamptz NOT NULL DEFAULT NOW(), + updated_by TEXT +); + +COMMENT ON TABLE objective_sync_config IS + 'Single-row динамическая конфигурация Objective sync. Все task-ы и beat ' + 'читают отсюда. Если строки нет — fallback на settings.* env.'; + +-- Засеваем единственную строку с дефолтами (если ещё не создана). +INSERT INTO objective_sync_config (id) VALUES (1) +ON CONFLICT (id) DO NOTHING; diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 76b243ba..13237fb8 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -91,6 +91,11 @@ services: condition: service_started volumes: - ./data:/app/data # playwright_state.json + photo binaries + # Read-only bind-mount Антоновского /sf/ SQLite — для objective_etl task. + # На VPS файл лежит в /opt/gendesign/site-finder/analysis.db; путь внутри + # контейнера задан settings.objective_anton_sqlite_path (default + # /data/anton-sqlite/analysis.db). + - /opt/gendesign/site-finder:/data/anton-sqlite:ro command: ["celery", "-A", "app.workers.celery_app", "worker", "--loglevel=info", "--concurrency=2"] beat: diff --git a/frontend/src/app/admin/layout.tsx b/frontend/src/app/admin/layout.tsx index fb87372c..d97662c9 100644 --- a/frontend/src/app/admin/layout.tsx +++ b/frontend/src/app/admin/layout.tsx @@ -6,6 +6,7 @@ import { usePathname } from "next/navigation"; const TABS = [ { href: "/admin/scrape", label: "Скрапер DOM.РФ" }, { href: "/admin/scrape/nspd", label: "Скрапер NSPD" }, + { href: "/admin/scrape/objective", label: "ETL Objective" }, { href: "/admin/leads", label: "Лиды PRINZIP" }, ]; diff --git a/frontend/src/app/admin/scrape/objective/page.tsx b/frontend/src/app/admin/scrape/objective/page.tsx new file mode 100644 index 00000000..e7b90a58 --- /dev/null +++ b/frontend/src/app/admin/scrape/objective/page.tsx @@ -0,0 +1,754 @@ +"use client"; + +import { useState } from "react"; +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; + +import { apiFetch } from "@/lib/api"; + +interface ObjectiveRunRow { + run_id: number; + started_at: string | null; + finished_at: string | null; + heartbeat_at: string | null; + group_name: string; + reports_ok: number; + reports_failed: number; + rows_corpus_room: number; + rows_lots: number; + rows_mappings: number; + requests_count: number; + status: string; + error: string | null; + triggered_by: string; +} + +interface ObjectiveCoverage { + pg: { + lots: number; + corp_room_month: number; + mappings: number; + last_lot_snapshot: string | null; + last_run_at: string | null; + }; + sqlite: { + path: string; + exists: boolean; + size_bytes?: number; + modified_at?: number; // epoch seconds + lots?: number; + corp_room_month?: number; + mappings?: number; + error?: string; + }; +} + +interface ObjectiveTriggerResp { + task_id: string; + sqlite_path: string; +} + +interface ObjectiveSyncOurResp { + task_id: string; + groups_planned: string[]; + estimate_minutes: number; +} + +interface ObjectiveConfig { + cron_schedule: string; + groups_csv: string; + groups: string[]; + use_ddu: boolean; + use_dkp: boolean; + period_months_back: number; + inter_group_delay_s: number; + rate_ms: number; + retries: number; + updated_at: string | null; + updated_by: string | null; + _note?: string; +} + +function formatBytes(b: number | undefined): string { + if (!b) return "—"; + const mb = b / 1024 / 1024; + return mb >= 100 ? `${mb.toFixed(0)} MB` : `${mb.toFixed(1)} MB`; +} + +function formatEpoch(sec: number | undefined): string { + if (!sec) return "—"; + const d = new Date(sec * 1000); + return d.toLocaleString("ru-RU"); +} + +function formatIso(s: string | null): string { + if (!s) return "—"; + return new Date(s).toLocaleString("ru-RU"); +} + +function diffSecs(a: string | null, b: string | null): string { + if (!a || !b) return "—"; + const dt = (new Date(b).getTime() - new Date(a).getTime()) / 1000; + if (dt < 60) return `${dt.toFixed(1)}с`; + return `${(dt / 60).toFixed(1)} мин`; +} + +export default function ObjectiveEtlAdminPage() { + const [token, setToken] = useState(() => + typeof window === "undefined" + ? "" + : (localStorage.getItem("admin_token") ?? ""), + ); + const [overrideSqlitePath, setOverrideSqlitePath] = useState(""); + const [lastTaskId, setLastTaskId] = useState(null); + // Группы для нашего sync — csv. Пусто = backend возьмёт настройки из БД + const [syncGroupsCsv, setSyncGroupsCsv] = useState(""); + const [lastSyncTaskId, setLastSyncTaskId] = useState(null); + // Локальный state для формы настроек (заполняется из cfg при первом ответе) + const [cfgDraft, setCfgDraft] = useState>({}); + const [cfgSaved, setCfgSaved] = useState(false); + const queryClient = useQueryClient(); + + const saveToken = (v: string) => { + setToken(v); + localStorage.setItem("admin_token", v); + }; + + const coverage = useQuery({ + queryKey: ["objective-coverage", token], + queryFn: () => + apiFetch("/api/v1/admin/scrape/objective/coverage", { + headers: { "X-Admin-Token": token }, + }), + enabled: !!token, + refetchInterval: 10000, + }); + + const runs = useQuery({ + queryKey: ["objective-runs", token], + queryFn: () => + apiFetch( + "/api/v1/admin/scrape/objective/runs?limit=20", + { headers: { "X-Admin-Token": token } }, + ), + enabled: !!token, + refetchInterval: 5000, + }); + + const trigger = useMutation({ + mutationFn: () => + apiFetch("/api/v1/admin/scrape/objective", { + method: "POST", + headers: { + "X-Admin-Token": token, + "Content-Type": "application/json", + }, + body: JSON.stringify({ + sqlite_path: overrideSqlitePath || null, + }), + }), + onSuccess: (data) => { + setLastTaskId(data.task_id); + void queryClient.invalidateQueries({ queryKey: ["objective-runs"] }); + void queryClient.invalidateQueries({ queryKey: ["objective-coverage"] }); + }, + }); + + // Динамический config из БД + const cfg = useQuery({ + queryKey: ["objective-config", token], + queryFn: () => + apiFetch("/api/v1/admin/scrape/objective/config", { + headers: { "X-Admin-Token": token }, + }), + enabled: !!token, + refetchOnWindowFocus: false, + }); + + // При первом успешном fetch'е — заполняем draft + if (cfg.data && Object.keys(cfgDraft).length === 0) { + setCfgDraft(cfg.data); + } + + const saveCfg = useMutation({ + mutationFn: (patch: Partial) => + apiFetch("/api/v1/admin/scrape/objective/config", { + method: "PUT", + headers: { + "X-Admin-Token": token, + "Content-Type": "application/json", + }, + body: JSON.stringify(patch), + }), + onSuccess: (data) => { + setCfgDraft(data); + setCfgSaved(true); + window.setTimeout(() => setCfgSaved(false), 3000); + void queryClient.invalidateQueries({ queryKey: ["objective-config"] }); + }, + }); + + // Наш sync через api.objctv.ru (платный API → расширение покрытия УрФО) + const syncOur = useMutation({ + mutationFn: () => { + const groups = syncGroupsCsv + .split(",") + .map((g) => g.trim()) + .filter(Boolean); + return apiFetch( + "/api/v1/admin/scrape/objective/sync-our", + { + method: "POST", + headers: { + "X-Admin-Token": token, + "Content-Type": "application/json", + }, + body: JSON.stringify({ + groups: groups.length ? groups : null, + }), + }, + ); + }, + onSuccess: (data) => { + setLastSyncTaskId(data.task_id); + void queryClient.invalidateQueries({ queryKey: ["objective-runs"] }); + }, + }); + + // Подсказка про свежесть данных + const sqliteFresher = (() => { + const cov = coverage.data; + if (!cov?.sqlite.modified_at || !cov?.pg.last_run_at) return null; + const sqliteMs = cov.sqlite.modified_at * 1000; + const pgMs = new Date(cov.pg.last_run_at).getTime(); + return sqliteMs > pgMs; + })(); + + return ( +
+
+

+ Objective: sync с api.objctv.ru → наша PG +

+ + ← Назад к kn-scraper + +
+ +

+ Source of truth = эта PG. Наш Celery каждый вторник + 06:00 МСК тянет 4 группы (Свердл.обл + Челябинск + Тюмень + Пермь) с + api.objctv.ru напрямую в objective_lots / corpus_room_month / + lots_history. SQLite Антона (/sf/) — отдельный микросервис со своим + scoring-слоем, используется только для bootstrap-импорта (не нужен для + регулярной работы). +

+ + {/* Token */} +
+ + saveToken(e.target.value)} + placeholder="введи token, сохранится в localStorage" + className="mt-1 w-full rounded border border-gray-300 px-3 py-2 font-mono text-sm" + /> +
+ + {/* Coverage — текущее состояние нашей PG */} +
+

+ 📦 Состояние PG (источник истины) +

+ {coverage.isLoading && ( +

Загрузка…

+ )} + {coverage.data && ( +
+
+
objective_lots
+
+ {coverage.data.pg.lots.toLocaleString("ru-RU")} +
+
+
+
corp×room×month
+
+ {coverage.data.pg.corp_room_month.toLocaleString("ru-RU")} +
+
+
+
complex_mapping
+
+ {coverage.data.pg.mappings.toLocaleString("ru-RU")} +
+
+
+
+ last lot snapshot:{" "} + + {coverage.data.pg.last_lot_snapshot ?? "—"} + +
+
+ last successful sync:{" "} + + {formatIso(coverage.data.pg.last_run_at)} + +
+
+
+
+ SQLite Антона (legacy):{" "} + {coverage.data.sqlite.exists ? ( + + ✓ {formatBytes(coverage.data.sqlite.size_bytes)} + + ) : ( + ✗ не примонтирован + )} +
+
+ файл обновлён:{" "} + + {formatEpoch(coverage.data.sqlite.modified_at)} + +
+
+ lots в SQLite:{" "} + + {coverage.data.sqlite.lots?.toLocaleString("ru-RU") ?? "—"} + +
+
+
+ )} +
+ + {/* PRIMARY: Наш sync с api.objctv.ru → PG (source of truth) */} +
+

+ 🌐 Наш sync: api.objctv.ru → PostgreSQL +

+

+ Основной production pipeline. Дёргает api.objctv.ru напрямую и пишет в + objective_lots / corpus_room_month / lots_history в нашей PG. + Покрытие: 4 группы УрФО (Свердл.обл целиком + Челябинск + Тюмень + + Пермь). Каждая группа = 2 отчёта (corp_sum + lots_pf), пауза 30с между + группами. ~3 минуты на группу. Тратит лимиты тарифа Объектива. +

+

+ ⏰ Auto: расписание из настроек (см. блок «⚙️ Настройки» ниже, сейчас:{" "} + {cfg.data?.cron_schedule ?? "..."} + ). Кнопка ниже — ручной триггер для немедленного запуска. +

+
+ + setSyncGroupsCsv(e.target.value)} + placeholder={ + cfg.data?.groups_csv ?? "Свердл.обл,Челябинск,Тюмень,Пермь" + } + className="mt-1 w-full rounded border border-gray-300 px-3 py-2 font-mono text-xs" + /> +

+ Доступные на тарифе (probe 2026-05-10): Екатеринбург, Свердловская + область, Свердловск, Верхняя Пышма, Берёзовский, Сысерть, Нижний + Тагил, Каменск-Уральский, Первоуральск, Новоуральск, Челябинск, + Тюмень, Пермь. +

+
+ + {lastSyncTaskId && ( +

+ Запущена task {lastSyncTaskId}. + Прогресс — в таблице прогонов ниже (один run per группа). +

+ )} + {syncOur.error && ( +

+ Ошибка: {(syncOur.error as Error).message} +

+ )} +
+ + {/* ⚙️ Настройки (динамические, из БД objective_sync_config) */} +
+ + ⚙️ Настройки sync (расписание + параметры запроса) + +
+ {!cfg.data &&

Загрузка…

} + {cfg.data && ( + <> +
+
+ + + setCfgDraft({ + ...cfgDraft, + cron_schedule: e.target.value, + }) + } + placeholder="0 6 * * tue" + className="mt-1 w-full rounded border border-gray-300 px-3 py-2 font-mono text-xs" + /> +

+ Сейчас активно:{" "} + {cfg.data.cron_schedule} +

+
+ +
+ + + setCfgDraft({ ...cfgDraft, groups_csv: e.target.value }) + } + placeholder="Свердловская область,Челябинск,Тюмень,Пермь" + className="mt-1 w-full rounded border border-gray-300 px-3 py-2 font-mono text-xs" + /> +
+ +
+ + +
+ +
+ + +
+ +
+ + + setCfgDraft({ + ...cfgDraft, + period_months_back: Number(e.target.value), + }) + } + className="mt-1 w-full rounded border border-gray-300 px-3 py-2 font-mono" + /> +

+ 1 = только последний полный месяц. 12 = последний год. +

+
+ +
+ + + setCfgDraft({ + ...cfgDraft, + inter_group_delay_s: Number(e.target.value), + }) + } + className="mt-1 w-full rounded border border-gray-300 px-3 py-2 font-mono" + /> +

+ Пауза между группами (сек). Меньше → быстрее, но риск 429. +

+
+ +
+ + + setCfgDraft({ + ...cfgDraft, + rate_ms: Number(e.target.value), + }) + } + className="mt-1 w-full rounded border border-gray-300 px-3 py-2 font-mono" + /> +
+ +
+ + + setCfgDraft({ + ...cfgDraft, + retries: Number(e.target.value), + }) + } + className="mt-1 w-full rounded border border-gray-300 px-3 py-2 font-mono" + /> +
+
+ +
+ + + {cfgSaved && ( + + ✓ Сохранено. Параметры (кроме cron) применятся со следующим + sync. + + )} + {saveCfg.error && ( + + {(saveCfg.error as Error).message} + + )} +
+ + {cfg.data.updated_at && ( +

+ Последнее изменение: {formatIso(cfg.data.updated_at)}{" "} + {cfg.data.updated_by && `(by ${cfg.data.updated_by})`} +

+ )} + + )} +
+
+ + {/* SECONDARY: Bootstrap ETL Антоновского SQLite → PG */} +
+ + 🛠 Bootstrap ETL: SQLite Антона → PG{" "} + + (одноразовый импорт; в обычной работе не нужен — наш sync всё делает + напрямую) + + +
+

+ Используется только когда Антоновский /sf/ имеет данные, которых ещё + нет в нашей PG (например при первом запуске). После того как наш + sync отработал хотя бы 1 раз — этот ETL не нужен.{" "} + Это legacy-инструмент. +

+ + {sqliteFresher === true && ( +
+ ⚠️ SQLite Антона обновлён ПОСЛЕ последнего ETL — но это уже + неважно если наш sync настроен на cron. Используй кнопку только + если хочешь срочно подхватить Антоновские данные раньше нашего + следующего sync (вторник 06:00). +
+ )} + +
+ + setOverrideSqlitePath(e.target.value)} + placeholder="/data/anton-sqlite/analysis.db" + className="mt-1 w-full rounded border border-gray-300 px-3 py-2 font-mono text-xs" + /> +
+ + {lastTaskId && ( +

+ Запущена task {lastTaskId}. +

+ )} + {trigger.error && ( +

+ Ошибка: {(trigger.error as Error).message} +

+ )} +
+
+ + {/* Runs */} +
+
+ 📜 Последние прогоны (top 20, обновляется каждые 5с) +
+
+ + + + + + + + + + + + + + + + {runs.data?.map((r) => { + const heartbeatLag = r.heartbeat_at + ? (Date.now() - new Date(r.heartbeat_at).getTime()) / 1000 + : null; + const isStale = + r.status === "running" && + heartbeatLag !== null && + heartbeatLag > 60; + return ( + + + + + + + + + + + + ); + })} + {runs.data?.length === 0 && ( + + + + )} + +
run_idstarteddurationlotscorp×roommappingsstatustriggererror
{r.run_id}{formatIso(r.started_at)} + {diffSecs(r.started_at, r.finished_at ?? r.heartbeat_at)} + + {r.rows_lots?.toLocaleString("ru-RU") ?? "—"} + + {r.rows_corpus_room?.toLocaleString("ru-RU") ?? "—"} + + {r.rows_mappings?.toLocaleString("ru-RU") ?? "—"} + + + {isStale ? "stale" : r.status} + + + {r.triggered_by} + + {r.error ?? ""} +
+ Прогонов ещё не было +
+
+
+
+ ); +}