gendesign/backend/app/services/objective_sync_config.py
lekss361 7c05d0a0d8 feat(objective): full sync pipeline + dynamic admin config
Objective API (api.objctv.ru) интегрирован как новый source of truth для
per-flat данных по новостройкам УрФО. Заменяет промежуточный Anton-SQLite
(legacy bootstrap-ETL остался как fallback в свёрнутом блоке UI).

Schema (data/sql/68_v2 — applied на проде, 6 таблиц):
- objective_lots          (UPSERT по lot_id; 303 677 rows)
- objective_corpus_room_month (long-формат месяц×корпус×room_bucket; 19 738 rows)
- objective_lots_history  (append-only weekly snapshots для elasticity)
- objective_complex_mapping (Objective ComplexName ↔ domrf_kn_objects.obj_id)
- objective_raw_reports   (jsonb страховка на смену схемы API)
- objective_scrape_runs   (журнал прогонов)
+ data/sql/72_objective_sync_config (single-row динамический конфиг)

Backend:
- services/scrapers/objective.py: ObjectiveClient — Bearer-токен (Redis +
  in-memory fallback), retry на 401/429/5xx, Retry-After header support
- services/objective_etl.py: ETL SQLite Антона → PG (legacy)
- services/objective_sync_config.py: read/update single-row config
- workers/tasks/scrape_objective.py:
  * sync_objective_group: 2 рабочих отчёта (corp_sum, lots_pf), inline-парсинг
  * sync_all_groups: wrapper, перебирает группы из БД-конфига с
    inter-group паузой; PATCH-merge explicit args > DB config
- workers/tasks/objective_etl.py: Celery task для legacy bootstrap
- workers/celery_app.py: beat читает cron из БД при старте (fallback на env)
- api/v1/admin_scrape.py: 5 новых endpoints для /objective/*

Frontend (frontend/src/app/admin/scrape/objective/page.tsx):
- PRIMARY blue блок «🌐 Наш sync» с input для override групп
- Collapsible «⚙️ Настройки» с формой (cron + 8 параметров) → PUT в БД
- Coverage-панель с PG counts + строка про SQLite Антона как legacy
- Collapsible «🛠 Bootstrap ETL» — legacy-инструмент

Beat schedule: вторник 06:00 МСК, ~10-15 мин на 4 группы (Свердл.обл +
Челябинск + Тюмень + Пермь = ~700K квартир УрФО). Расписание и параметры
меняются через админку без редеплоя (cron требует restart beat).

Эмпирические находки об API (probe 2026-05-10):
- 13 из 21 проверенных group_name доступны на тарифе (включая «Свердловская
  область», «Челябинск», «Тюмень», «Пермь», но НЕ «Свердловская обл» —
  формат имени строгий)
- ComplexName требует БЕЗ префикса «ЖК» и БЕЗ кавычек
- Поле «Банк» для всех 303k = NULL (тариф не отдаёт), но «Тип обременения»
  работает (36% строк = ипотека) → ipoteka_share возможен, банковская
  атрибуция — нет

Docker:
- bind-mount /opt/gendesign/site-finder:/data/anton-sqlite:ro в worker

GitHub backlog: добавлены #22-25 (recommend_mix v3 — 4 сабтаска по
улучшению алгоритма на основе фидбэка про POI / границы районов /
конкурентов / окно данных / success-driven mix).

Knowledge graph (memory/memory-gendesign.jsonl): обновлены entities
Objective_Integration_May07_2026, Schema_Objective_v2_May07,
Objective_API_Findings_May07, Module_Objective_Client + новый
Session_End_May07_2026.

TODO для прода: прописать OBJECTIVE_API_KEY=<key> в backend/.env +
docker compose restart worker beat.
2026-05-10 19:54:15 +03:00

178 lines
5.9 KiB
Python

"""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