Merge pull request 'feat(sf): cross-load ETL tradein→gendesign newbuilding_listings — вариант A (#976)' (#1185) from feat/sf-newbuilding-crossload into main
All checks were successful
Deploy Trade-In / changes (push) Successful in 7s
Deploy / changes (push) Successful in 6s
Deploy Trade-In / build-frontend (push) Successful in 27s
Deploy Trade-In / test (push) Successful in 32s
Deploy Trade-In / build-browser (push) Successful in 24s
Deploy / build-frontend (push) Successful in 29s
Deploy Trade-In / build-backend (push) Successful in 25s
Deploy Trade-In / deploy (push) Successful in 44s
Deploy / build-backend (push) Successful in 7m12s
Deploy / build-worker (push) Successful in 7m18s
Deploy / deploy (push) Successful in 1m47s
All checks were successful
Deploy Trade-In / changes (push) Successful in 7s
Deploy / changes (push) Successful in 6s
Deploy Trade-In / build-frontend (push) Successful in 27s
Deploy Trade-In / test (push) Successful in 32s
Deploy Trade-In / build-browser (push) Successful in 24s
Deploy / build-frontend (push) Successful in 29s
Deploy Trade-In / build-backend (push) Successful in 25s
Deploy Trade-In / deploy (push) Successful in 44s
Deploy / build-backend (push) Successful in 7m12s
Deploy / build-worker (push) Successful in 7m18s
Deploy / deploy (push) Successful in 1m47s
Reviewed-on: #1185
This commit is contained in:
commit
9b06377df8
10 changed files with 613 additions and 1 deletions
|
|
@ -318,6 +318,25 @@ jobs:
|
||||||
done
|
done
|
||||||
echo "All migrations applied."
|
echo "All migrations applied."
|
||||||
|
|
||||||
|
# Bootstrap gendesign_reader password from env (post-migration, #976).
|
||||||
|
# SQL migration 101_gendesign_reader_role.sql creates role passwordless;
|
||||||
|
# password lives only in /opt/gendesign/tradein-mvp/.env.runtime.
|
||||||
|
# .env.runtime already sourced above (set -a; source .env.runtime).
|
||||||
|
#
|
||||||
|
# ⚠️ psql variable substitution (:'pw') НЕ работает внутри -c
|
||||||
|
# (переменная доходит до сервера as literal → syntax error at ':').
|
||||||
|
# Решение: передаём SQL через stdin (< file), psql интерполирует
|
||||||
|
# :'pw' ВНЕ dollar-quoted блока. Детали: ops/db-bootstrap/set_gendesign_reader_password.sql.
|
||||||
|
if [ -n "${TRADEIN_READER_PASSWORD:-}" ]; then
|
||||||
|
echo "→ Applying gendesign_reader password from env"
|
||||||
|
docker compose -p gendesign-tradein -f docker-compose.prod.yml exec -T postgres \
|
||||||
|
psql -U "${TRADEIN_POSTGRES_USER:-tradein}" -d tradein -v ON_ERROR_STOP=on \
|
||||||
|
-v "pw=${TRADEIN_READER_PASSWORD}" \
|
||||||
|
< ops/db-bootstrap/set_gendesign_reader_password.sql
|
||||||
|
else
|
||||||
|
echo "WARNING: TRADEIN_READER_PASSWORD not set in .env.runtime — gendesign_reader без пароля, ETL #976 не сможет подключиться"
|
||||||
|
fi
|
||||||
|
|
||||||
# Retry backend lifespan hook AFTER migrations applied.
|
# Retry backend lifespan hook AFTER migrations applied.
|
||||||
# tradein-backend startup runs ensure_fdw_user_mapping which needs
|
# tradein-backend startup runs ensure_fdw_user_mapping which needs
|
||||||
# FOREIGN SERVER gendesign_remote (created by 060_postgres_fdw_extension.sql).
|
# FOREIGN SERVER gendesign_remote (created by 060_postgres_fdw_extension.sql).
|
||||||
|
|
|
||||||
|
|
@ -1034,6 +1034,30 @@ def resume_geo_job(
|
||||||
return {"job_id": job_id, "resumed": True}
|
return {"job_id": job_id, "resumed": True}
|
||||||
|
|
||||||
|
|
||||||
|
# ── Newbuilding cross-load ETL (#976) ────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/newbuilding-crossload")
|
||||||
|
def trigger_newbuilding_crossload() -> dict[str, Any]:
|
||||||
|
"""Ручной запуск cross-load ETL tradein.houses → newbuilding_listings (#976).
|
||||||
|
|
||||||
|
Обычно запускается ночью через beat (03:30 МСК).
|
||||||
|
Если TRADEIN_DATABASE_URL не задан в settings — возвращает 503.
|
||||||
|
"""
|
||||||
|
if not settings.tradein_database_url:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=503,
|
||||||
|
detail=(
|
||||||
|
"TRADEIN_DATABASE_URL не задан — cross-load ETL отключён. "
|
||||||
|
"Добавь переменную в backend/.env.runtime и перезапусти worker."
|
||||||
|
),
|
||||||
|
)
|
||||||
|
from app.workers.tasks.etl_newbuilding_crossload import etl_newbuilding_crossload
|
||||||
|
|
||||||
|
result = etl_newbuilding_crossload.apply_async()
|
||||||
|
return {"task_id": result.id, "queued_at": "now"}
|
||||||
|
|
||||||
|
|
||||||
# ── Unified scrape dashboard (поверх v_scrape_runs_unified / v_scrape_log_unified) ──
|
# ── Unified scrape dashboard (поверх v_scrape_runs_unified / v_scrape_log_unified) ──
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -61,7 +61,7 @@ class Settings(BaseSettings):
|
||||||
if not stripped:
|
if not stripped:
|
||||||
return []
|
return []
|
||||||
return [int(tok.strip()) for tok in stripped.split(",") if tok.strip()]
|
return [int(tok.strip()) for tok in stripped.split(",") if tok.strip()]
|
||||||
if isinstance(value, (list, tuple)):
|
if isinstance(value, list | tuple):
|
||||||
return [int(item) for item in value]
|
return [int(item) for item in value]
|
||||||
return value
|
return value
|
||||||
|
|
||||||
|
|
@ -172,6 +172,12 @@ class Settings(BaseSettings):
|
||||||
# OBJECTIVE_ANTON_SQLITE_PATH=C:/Users/user/source/repos/gendesign/sf_anton_snapshot.db
|
# OBJECTIVE_ANTON_SQLITE_PATH=C:/Users/user/source/repos/gendesign/sf_anton_snapshot.db
|
||||||
objective_anton_sqlite_path: str = "/data/anton-sqlite/analysis.db"
|
objective_anton_sqlite_path: str = "/data/anton-sqlite/analysis.db"
|
||||||
|
|
||||||
|
# Cross-load ETL tradein→gendesign (#976 950-E5).
|
||||||
|
# Прямой psycopg-коннект к tradein-postgres через gendesign_shared network.
|
||||||
|
# Пример: postgresql://gendesign_reader:<pw>@tradein-postgres:5432/tradein
|
||||||
|
# Если пусто — ETL отключён (warn-log, задача возвращает {"disabled": true}).
|
||||||
|
tradein_database_url: str = ""
|
||||||
|
|
||||||
# OpenRouteService API (https://openrouteservice.org/dev/#/signup).
|
# OpenRouteService API (https://openrouteservice.org/dev/#/signup).
|
||||||
# Free tier: 2000 запросов/день. Используется в /parcels/{cad}/isochrones.
|
# Free tier: 2000 запросов/день. Используется в /parcels/{cad}/isochrones.
|
||||||
# Если не задан — endpoint вернёт 503 с инструкцией по регистрации.
|
# Если не задан — endpoint вернёт 503 с инструкцией по регистрации.
|
||||||
|
|
|
||||||
248
backend/app/services/etl/newbuilding_crossload.py
Normal file
248
backend/app/services/etl/newbuilding_crossload.py
Normal file
|
|
@ -0,0 +1,248 @@
|
||||||
|
"""Cross-load ETL: tradein.houses → gendesign.newbuilding_listings (#976 950-E5).
|
||||||
|
|
||||||
|
Читает houses + house_sources из tradein-БД (gendesign_reader role),
|
||||||
|
UPSERT-ит в gendesign newbuilding_listings по (source, ext_house_id).
|
||||||
|
Соединение к tradein через psycopg v3 (прямой коннект, НЕ SQLAlchemy).
|
||||||
|
|
||||||
|
Запускается:
|
||||||
|
- ночью Celery beat (03:30 МСК, tasks/etl_newbuilding_crossload.py)
|
||||||
|
- вручную через POST /api/v1/admin/scrape/newbuilding-crossload
|
||||||
|
|
||||||
|
Если settings.tradein_database_url пусто — сразу возвращает {"disabled": True}.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
import logging
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
import psycopg
|
||||||
|
from sqlalchemy import text
|
||||||
|
from sqlalchemy.orm import Session
|
||||||
|
|
||||||
|
from app.core.config import settings
|
||||||
|
from app.core.db import SessionLocal
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
_BATCH_SIZE = 500
|
||||||
|
|
||||||
|
# ── SELECT из tradein ────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
_SOURCE_SQL = """
|
||||||
|
SELECT
|
||||||
|
h.source,
|
||||||
|
h.ext_house_id,
|
||||||
|
h.url,
|
||||||
|
h.slug,
|
||||||
|
h.address,
|
||||||
|
h.full_address,
|
||||||
|
h.lat,
|
||||||
|
h.lon,
|
||||||
|
h.house_class,
|
||||||
|
h.developer_name,
|
||||||
|
h.developer_key,
|
||||||
|
h.year_built,
|
||||||
|
h.total_floors,
|
||||||
|
h.total_units,
|
||||||
|
h.rating,
|
||||||
|
h.reviews_count,
|
||||||
|
h.houses_by_turn,
|
||||||
|
h.cian_internal_house_id,
|
||||||
|
h.raw_payload,
|
||||||
|
h.last_scraped_at,
|
||||||
|
-- yandex_jk_id из house_sources (LEFT JOIN — может отсутствовать)
|
||||||
|
hs_y.ext_id AS yandex_jk_id
|
||||||
|
FROM houses h
|
||||||
|
LEFT JOIN house_sources hs_y
|
||||||
|
ON hs_y.house_id = h.id
|
||||||
|
AND hs_y.ext_source = 'yandex'
|
||||||
|
WHERE h.lat IS NOT NULL
|
||||||
|
AND h.lon IS NOT NULL
|
||||||
|
"""
|
||||||
|
|
||||||
|
# ── UPSERT в gendesign ───────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
_UPSERT_SQL = text("""
|
||||||
|
INSERT INTO newbuilding_listings (
|
||||||
|
source, ext_house_id,
|
||||||
|
url, slug,
|
||||||
|
address, full_address,
|
||||||
|
lat, lon,
|
||||||
|
geom,
|
||||||
|
house_class,
|
||||||
|
developer_name, developer_key,
|
||||||
|
year_built,
|
||||||
|
total_floors, total_units,
|
||||||
|
rating, reviews_count,
|
||||||
|
houses_by_turn,
|
||||||
|
cian_internal_house_id,
|
||||||
|
yandex_jk_id,
|
||||||
|
raw_payload,
|
||||||
|
source_last_scraped_at,
|
||||||
|
crossloaded_at
|
||||||
|
)
|
||||||
|
VALUES (
|
||||||
|
:source, :ext_house_id,
|
||||||
|
:url, :slug,
|
||||||
|
:address, :full_address,
|
||||||
|
:lat, :lon,
|
||||||
|
CASE
|
||||||
|
WHEN :lat2 IS NOT NULL AND :lon2 IS NOT NULL
|
||||||
|
THEN ST_SetSRID(ST_MakePoint(CAST(:lon2 AS double precision),
|
||||||
|
CAST(:lat2 AS double precision)), 4326)
|
||||||
|
ELSE NULL
|
||||||
|
END,
|
||||||
|
:house_class,
|
||||||
|
:developer_name, :developer_key,
|
||||||
|
CAST(:year_built AS integer),
|
||||||
|
CAST(:total_floors AS integer), CAST(:total_units AS integer),
|
||||||
|
CAST(:rating AS numeric), CAST(:reviews_count AS integer),
|
||||||
|
CAST(:houses_by_turn AS jsonb),
|
||||||
|
CAST(:cian_internal_house_id AS bigint),
|
||||||
|
:yandex_jk_id,
|
||||||
|
CAST(:raw_payload AS jsonb),
|
||||||
|
CAST(:source_last_scraped_at AS timestamptz),
|
||||||
|
now()
|
||||||
|
)
|
||||||
|
ON CONFLICT (source, ext_house_id) DO UPDATE SET
|
||||||
|
url = EXCLUDED.url,
|
||||||
|
slug = EXCLUDED.slug,
|
||||||
|
address = EXCLUDED.address,
|
||||||
|
full_address = EXCLUDED.full_address,
|
||||||
|
lat = EXCLUDED.lat,
|
||||||
|
lon = EXCLUDED.lon,
|
||||||
|
geom = EXCLUDED.geom,
|
||||||
|
house_class = EXCLUDED.house_class,
|
||||||
|
developer_name = EXCLUDED.developer_name,
|
||||||
|
developer_key = EXCLUDED.developer_key,
|
||||||
|
year_built = EXCLUDED.year_built,
|
||||||
|
total_floors = EXCLUDED.total_floors,
|
||||||
|
total_units = EXCLUDED.total_units,
|
||||||
|
rating = EXCLUDED.rating,
|
||||||
|
reviews_count = EXCLUDED.reviews_count,
|
||||||
|
houses_by_turn = EXCLUDED.houses_by_turn,
|
||||||
|
cian_internal_house_id = COALESCE(
|
||||||
|
EXCLUDED.cian_internal_house_id,
|
||||||
|
newbuilding_listings.cian_internal_house_id),
|
||||||
|
yandex_jk_id = COALESCE(
|
||||||
|
EXCLUDED.yandex_jk_id,
|
||||||
|
newbuilding_listings.yandex_jk_id),
|
||||||
|
raw_payload = EXCLUDED.raw_payload,
|
||||||
|
source_last_scraped_at = EXCLUDED.source_last_scraped_at,
|
||||||
|
crossloaded_at = now()
|
||||||
|
""")
|
||||||
|
|
||||||
|
|
||||||
|
def _row_to_params(row: Any) -> dict[str, Any]:
|
||||||
|
"""Преобразовать строку из tradein-курсора в dict параметров для UPSERT.
|
||||||
|
|
||||||
|
Публичная функция — тестируется отдельно (без БД).
|
||||||
|
row — объект с атрибутами (psycopg RealDictRow / dict / named row).
|
||||||
|
"""
|
||||||
|
if hasattr(row, "_asdict"):
|
||||||
|
r: dict[str, Any] = row._asdict()
|
||||||
|
elif hasattr(row, "keys"):
|
||||||
|
r = dict(row)
|
||||||
|
else:
|
||||||
|
r = dict(row)
|
||||||
|
|
||||||
|
houses_by_turn = r.get("houses_by_turn")
|
||||||
|
raw_payload = r.get("raw_payload")
|
||||||
|
|
||||||
|
return {
|
||||||
|
"source": r["source"],
|
||||||
|
"ext_house_id": r["ext_house_id"],
|
||||||
|
"url": r.get("url"),
|
||||||
|
"slug": r.get("slug"),
|
||||||
|
"address": r.get("address"),
|
||||||
|
"full_address": r.get("full_address"),
|
||||||
|
"lat": r.get("lat"),
|
||||||
|
"lon": r.get("lon"),
|
||||||
|
# дублируем lat/lon для CASE-ветки geom (psycopg не позволяет один bind два раза)
|
||||||
|
"lat2": r.get("lat"),
|
||||||
|
"lon2": r.get("lon"),
|
||||||
|
"house_class": r.get("house_class"),
|
||||||
|
"developer_name": r.get("developer_name"),
|
||||||
|
"developer_key": r.get("developer_key"),
|
||||||
|
"year_built": r.get("year_built"),
|
||||||
|
"total_floors": r.get("total_floors"),
|
||||||
|
"total_units": r.get("total_units"),
|
||||||
|
"rating": str(r["rating"]) if r.get("rating") is not None else None,
|
||||||
|
"reviews_count": r.get("reviews_count"),
|
||||||
|
"houses_by_turn": json.dumps(houses_by_turn, ensure_ascii=False)
|
||||||
|
if houses_by_turn is not None
|
||||||
|
else None,
|
||||||
|
"cian_internal_house_id": r.get("cian_internal_house_id"),
|
||||||
|
"yandex_jk_id": r.get("yandex_jk_id"),
|
||||||
|
"raw_payload": json.dumps(raw_payload, ensure_ascii=False)
|
||||||
|
if raw_payload is not None
|
||||||
|
else None,
|
||||||
|
"source_last_scraped_at": r.get("last_scraped_at"),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def run_crossload(db: Session | None = None) -> dict[str, Any]:
|
||||||
|
"""Выполнить cross-load ETL tradein→gendesign.
|
||||||
|
|
||||||
|
Если settings.tradein_database_url пуст — возвращает {"disabled": True}, не коннектится.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict с ключами: source_rows, upserted, skipped — или {"disabled": True}.
|
||||||
|
"""
|
||||||
|
if not settings.tradein_database_url:
|
||||||
|
logger.warning("etl_newbuilding_crossload: TRADEIN_DATABASE_URL не задан — ETL отключён")
|
||||||
|
return {"disabled": True}
|
||||||
|
|
||||||
|
close_db = db is None
|
||||||
|
if db is None:
|
||||||
|
db = SessionLocal()
|
||||||
|
|
||||||
|
source_rows = 0
|
||||||
|
upserted = 0
|
||||||
|
skipped = 0
|
||||||
|
|
||||||
|
logger.info("etl_newbuilding_crossload: start (batch_size=%d)", _BATCH_SIZE)
|
||||||
|
|
||||||
|
try:
|
||||||
|
connstr = settings.tradein_database_url
|
||||||
|
with psycopg.connect(connstr, row_factory=psycopg.rows.dict_row) as src:
|
||||||
|
with src.cursor(name="crossload_cur") as cur: # server-side cursor
|
||||||
|
cur.execute(_SOURCE_SQL)
|
||||||
|
while True:
|
||||||
|
batch = cur.fetchmany(_BATCH_SIZE)
|
||||||
|
if not batch:
|
||||||
|
break
|
||||||
|
source_rows += len(batch)
|
||||||
|
params_list = [_row_to_params(r) for r in batch]
|
||||||
|
for params in params_list:
|
||||||
|
try:
|
||||||
|
with db.begin_nested():
|
||||||
|
db.execute(_UPSERT_SQL, params)
|
||||||
|
upserted += 1
|
||||||
|
except Exception as exc:
|
||||||
|
skipped += 1
|
||||||
|
logger.warning(
|
||||||
|
"etl_newbuilding_crossload: upsert failed "
|
||||||
|
"source=%s ext_id=%s: %s",
|
||||||
|
params.get("source"),
|
||||||
|
params.get("ext_house_id"),
|
||||||
|
exc,
|
||||||
|
)
|
||||||
|
db.commit()
|
||||||
|
except Exception:
|
||||||
|
if not close_db:
|
||||||
|
db.rollback()
|
||||||
|
raise
|
||||||
|
finally:
|
||||||
|
if close_db:
|
||||||
|
db.close()
|
||||||
|
|
||||||
|
logger.info(
|
||||||
|
"etl_newbuilding_crossload: done — source_rows=%d upserted=%d skipped=%d",
|
||||||
|
source_rows,
|
||||||
|
upserted,
|
||||||
|
skipped,
|
||||||
|
)
|
||||||
|
return {"source_rows": source_rows, "upserted": upserted, "skipped": skipped}
|
||||||
|
|
@ -381,6 +381,17 @@ def build_beat_schedule() -> dict:
|
||||||
"options": {"queue": "celery"},
|
"options": {"queue": "celery"},
|
||||||
}
|
}
|
||||||
|
|
||||||
|
# Cross-load ETL tradein→gendesign (#976 950-E5): tradein.houses → newbuilding_listings.
|
||||||
|
# Ночной запуск: 00:30 UTC = 03:30 МСК (Celery conf.timezone=Europe/Moscow → crontab в МСК).
|
||||||
|
# Не в job_settings (технический ETL, не требует конфигурации UI).
|
||||||
|
# Идемпотентен через ON CONFLICT (source, ext_house_id).
|
||||||
|
# Если TRADEIN_DATABASE_URL не задан → warn-log, {"disabled": True} без исключения.
|
||||||
|
schedule["newbuilding-crossload-nightly"] = {
|
||||||
|
"task": "tasks.etl_newbuilding_crossload.etl_newbuilding_crossload",
|
||||||
|
"schedule": _parse_cron("30 0 * * *"), # 00:30 UTC = 03:30 МСК
|
||||||
|
"options": {"queue": "celery"},
|
||||||
|
}
|
||||||
|
|
||||||
# ППТ/ПМТ harvest (#1085): WFS-фетч геопортала ЕКБ по bbox агломерации → planning_projects.
|
# ППТ/ПМТ harvest (#1085): WFS-фетч геопортала ЕКБ по bbox агломерации → planning_projects.
|
||||||
# 1-е число месяца 06:00 МСК — после opportunity (05:30). Документы планировки стабильны,
|
# 1-е число месяца 06:00 МСК — после opportunity (05:30). Документы планировки стабильны,
|
||||||
# ежемесячно достаточно. WFS BBOX (не grid-walk) — один запрос на слой.
|
# ежемесячно достаточно. WFS BBOX (не grid-walk) — один запрос на слой.
|
||||||
|
|
|
||||||
27
backend/app/workers/tasks/etl_newbuilding_crossload.py
Normal file
27
backend/app/workers/tasks/etl_newbuilding_crossload.py
Normal file
|
|
@ -0,0 +1,27 @@
|
||||||
|
"""Celery task: ночной cross-load ETL tradein.houses → gendesign.newbuilding_listings.
|
||||||
|
|
||||||
|
Расписание: 03:30 МСК (hardcoded в beat_schedule, как cbr/poi/supply-layers).
|
||||||
|
Идемпотентен: UPSERT ON CONFLICT (source, ext_house_id).
|
||||||
|
Если TRADEIN_DATABASE_URL не задан → warn-log, возвращает {"disabled": True}.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
|
|
||||||
|
from app.services.etl.newbuilding_crossload import run_crossload
|
||||||
|
from app.workers.celery_app import celery_app
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
@celery_app.task(name="tasks.etl_newbuilding_crossload.etl_newbuilding_crossload")
|
||||||
|
def etl_newbuilding_crossload() -> dict:
|
||||||
|
"""Cross-load tradein.houses → gendesign.newbuilding_listings (#976).
|
||||||
|
|
||||||
|
Вызывается beat'ом в 03:30 МСК и вручную через POST /api/v1/admin/scrape/newbuilding-crossload.
|
||||||
|
"""
|
||||||
|
logger.info("task etl_newbuilding_crossload: start")
|
||||||
|
result = run_crossload()
|
||||||
|
logger.info("task etl_newbuilding_crossload: done — %s", result)
|
||||||
|
return result
|
||||||
171
backend/tests/services/test_newbuilding_crossload.py
Normal file
171
backend/tests/services/test_newbuilding_crossload.py
Normal file
|
|
@ -0,0 +1,171 @@
|
||||||
|
"""Тесты cross-load ETL tradein→gendesign newbuilding_listings (#976).
|
||||||
|
|
||||||
|
Тесты без реальной БД:
|
||||||
|
(a) disabled-режим: пустой tradein_database_url → {"disabled": True}, psycopg не вызывается.
|
||||||
|
(b) маппинг строки source → insert-параметры (_row_to_params).
|
||||||
|
(c) UPSERT SQL содержит ON CONFLICT (source, ext_house_id).
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
from app.services.etl.newbuilding_crossload import (
|
||||||
|
_UPSERT_SQL,
|
||||||
|
_row_to_params,
|
||||||
|
run_crossload,
|
||||||
|
)
|
||||||
|
|
||||||
|
# ─── (a) Disabled-режим ──────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_crossload_disabled_when_url_empty():
|
||||||
|
"""Если tradein_database_url пусто — возвращает {"disabled": True}, коннекта нет."""
|
||||||
|
with patch("app.services.etl.newbuilding_crossload.settings") as mock_settings:
|
||||||
|
mock_settings.tradein_database_url = ""
|
||||||
|
with patch("psycopg.connect") as mock_connect:
|
||||||
|
result = run_crossload()
|
||||||
|
|
||||||
|
assert result == {"disabled": True}
|
||||||
|
mock_connect.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
# ─── (b) Маппинг _row_to_params ──────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
_SAMPLE_ROW: dict = {
|
||||||
|
"source": "cian",
|
||||||
|
"ext_house_id": "123456",
|
||||||
|
"url": "https://cian.ru/novostrojki/123456/",
|
||||||
|
"slug": "zhk-test",
|
||||||
|
"address": "Екатеринбург, ул. Тестовая, 1",
|
||||||
|
"full_address": "Свердловская область, Екатеринбург, ул. Тестовая, 1",
|
||||||
|
"lat": 56.838_926,
|
||||||
|
"lon": 60.605_702,
|
||||||
|
"house_class": "comfort",
|
||||||
|
"developer_name": "ТестСтрой",
|
||||||
|
"developer_key": "teststr_001",
|
||||||
|
"year_built": 2025,
|
||||||
|
"total_floors": 18,
|
||||||
|
"total_units": 360,
|
||||||
|
"rating": 4.7,
|
||||||
|
"reviews_count": 12,
|
||||||
|
"houses_by_turn": [{"turn": 1, "corps": ["1А"]}],
|
||||||
|
"cian_internal_house_id": 9876543,
|
||||||
|
"raw_payload": {"foo": "bar"},
|
||||||
|
"last_scraped_at": None,
|
||||||
|
"yandex_jk_id": "yandex-42",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def test_row_to_params_basic_fields():
|
||||||
|
"""Прямые поля копируются без изменений."""
|
||||||
|
p = _row_to_params(_SAMPLE_ROW)
|
||||||
|
|
||||||
|
assert p["source"] == "cian"
|
||||||
|
assert p["ext_house_id"] == "123456"
|
||||||
|
assert p["url"] == "https://cian.ru/novostrojki/123456/"
|
||||||
|
assert p["house_class"] == "comfort"
|
||||||
|
assert p["developer_name"] == "ТестСтрой"
|
||||||
|
assert p["developer_key"] == "teststr_001"
|
||||||
|
assert p["year_built"] == 2025
|
||||||
|
assert p["total_floors"] == 18
|
||||||
|
assert p["total_units"] == 360
|
||||||
|
assert p["reviews_count"] == 12
|
||||||
|
assert p["yandex_jk_id"] == "yandex-42"
|
||||||
|
assert p["cian_internal_house_id"] == 9876543
|
||||||
|
|
||||||
|
|
||||||
|
def test_row_to_params_geo_duplicated():
|
||||||
|
"""lat/lon продублированы в lat2/lon2 для CASE-ветки geom."""
|
||||||
|
p = _row_to_params(_SAMPLE_ROW)
|
||||||
|
assert p["lat"] == _SAMPLE_ROW["lat"]
|
||||||
|
assert p["lon"] == _SAMPLE_ROW["lon"]
|
||||||
|
assert p["lat2"] == _SAMPLE_ROW["lat"]
|
||||||
|
assert p["lon2"] == _SAMPLE_ROW["lon"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_row_to_params_jsonb_serialized():
|
||||||
|
"""houses_by_turn и raw_payload сериализованы в JSON-строки."""
|
||||||
|
p = _row_to_params(_SAMPLE_ROW)
|
||||||
|
# houses_by_turn
|
||||||
|
hbt = json.loads(p["houses_by_turn"])
|
||||||
|
assert hbt == [{"turn": 1, "corps": ["1А"]}]
|
||||||
|
# raw_payload
|
||||||
|
rp = json.loads(p["raw_payload"])
|
||||||
|
assert rp == {"foo": "bar"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_row_to_params_none_jsonb():
|
||||||
|
"""Если houses_by_turn/raw_payload равны None — параметр None, не строка."""
|
||||||
|
row = dict(_SAMPLE_ROW)
|
||||||
|
row["houses_by_turn"] = None
|
||||||
|
row["raw_payload"] = None
|
||||||
|
p = _row_to_params(row)
|
||||||
|
assert p["houses_by_turn"] is None
|
||||||
|
assert p["raw_payload"] is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_row_to_params_rating_as_string():
|
||||||
|
"""rating приводится к строке для CAST(:rating AS numeric)."""
|
||||||
|
p = _row_to_params(_SAMPLE_ROW)
|
||||||
|
assert p["rating"] == "4.7"
|
||||||
|
|
||||||
|
|
||||||
|
def test_row_to_params_rating_none():
|
||||||
|
"""Если rating = None — параметр тоже None."""
|
||||||
|
row = dict(_SAMPLE_ROW)
|
||||||
|
row["rating"] = None
|
||||||
|
p = _row_to_params(row)
|
||||||
|
assert p["rating"] is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_row_to_params_source_last_scraped_at():
|
||||||
|
"""last_scraped_at из источника → source_last_scraped_at в параметрах."""
|
||||||
|
p = _row_to_params(_SAMPLE_ROW)
|
||||||
|
assert "source_last_scraped_at" in p
|
||||||
|
assert p["source_last_scraped_at"] is None # SAMPLE_ROW имеет None
|
||||||
|
|
||||||
|
|
||||||
|
# ─── (c) UPSERT SQL содержит ON CONFLICT ──────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def test_upsert_sql_contains_on_conflict():
|
||||||
|
"""UPSERT-запрос содержит ON CONFLICT (source, ext_house_id)."""
|
||||||
|
sql_text = str(_UPSERT_SQL)
|
||||||
|
assert "ON CONFLICT" in sql_text.upper()
|
||||||
|
assert "source" in sql_text
|
||||||
|
assert "ext_house_id" in sql_text
|
||||||
|
|
||||||
|
|
||||||
|
def test_upsert_sql_contains_do_update():
|
||||||
|
"""UPSERT-запрос обновляет строку при конфликте (DO UPDATE SET)."""
|
||||||
|
sql_text = str(_UPSERT_SQL)
|
||||||
|
assert "DO UPDATE SET" in sql_text.upper()
|
||||||
|
|
||||||
|
|
||||||
|
def test_upsert_sql_coalesce_external_ids():
|
||||||
|
"""yandex_jk_id и cian_internal_house_id используют COALESCE в DO UPDATE.
|
||||||
|
|
||||||
|
Источники cian/yandex пишут попеременно — важно не затирать чужой id NULL'ом.
|
||||||
|
Проверяем: в секции DO UPDATE каждая из двух колонок встречается после COALESCE,
|
||||||
|
а не как голое EXCLUDED.<col> (что затёрло бы чужой id NULL'ом).
|
||||||
|
"""
|
||||||
|
sql_text = str(_UPSERT_SQL)
|
||||||
|
sql_upper = sql_text.upper()
|
||||||
|
|
||||||
|
do_update_idx = sql_upper.index("DO UPDATE SET")
|
||||||
|
do_update_section = sql_upper[do_update_idx:]
|
||||||
|
|
||||||
|
# COALESCE должен присутствовать в секции DO UPDATE
|
||||||
|
assert "COALESCE" in do_update_section, "DO UPDATE должен содержать COALESCE"
|
||||||
|
|
||||||
|
# yandex_jk_id — строка вида "YANDEX_JK_ID = COALESCE("
|
||||||
|
assert (
|
||||||
|
"YANDEX_JK_ID = COALESCE(" in do_update_section
|
||||||
|
), "yandex_jk_id в DO UPDATE должен использовать COALESCE чтобы не затирать NULL'ом"
|
||||||
|
# cian_internal_house_id — строка вида "CIAN_INTERNAL_HOUSE_ID = COALESCE("
|
||||||
|
assert (
|
||||||
|
"CIAN_INTERNAL_HOUSE_ID = COALESCE(" in do_update_section
|
||||||
|
), "cian_internal_house_id в DO UPDATE должен использовать COALESCE чтобы не затирать NULL'ом"
|
||||||
|
|
@ -82,6 +82,10 @@ services:
|
||||||
timeout: 5s
|
timeout: 5s
|
||||||
retries: 5
|
retries: 5
|
||||||
start_period: 20s
|
start_period: 20s
|
||||||
|
# #976 cross-DB ETL tradein→gendesign: backend подключается к tradein-postgres
|
||||||
|
# через gendesign_shared network (tradein-postgres уже в этой сети).
|
||||||
|
# default — обязательно явно, иначе сервис выпадет из дефолтной сети.
|
||||||
|
networks: [default, shared]
|
||||||
|
|
||||||
frontend:
|
frontend:
|
||||||
image: ghcr.io/lekss361/gendesign-frontend:${IMAGE_TAG:-latest}
|
image: ghcr.io/lekss361/gendesign-frontend:${IMAGE_TAG:-latest}
|
||||||
|
|
@ -122,6 +126,10 @@ services:
|
||||||
# /data/anton-sqlite/analysis.db).
|
# /data/anton-sqlite/analysis.db).
|
||||||
- /opt/gendesign/site-finder:/data/anton-sqlite:ro
|
- /opt/gendesign/site-finder:/data/anton-sqlite:ro
|
||||||
command: ["celery", "-A", "app.workers.celery_app", "worker", "--loglevel=info", "--concurrency=8", "--queues=celery,scrape_kn,geo"]
|
command: ["celery", "-A", "app.workers.celery_app", "worker", "--loglevel=info", "--concurrency=8", "--queues=celery,scrape_kn,geo"]
|
||||||
|
# #976 cross-DB ETL tradein→gendesign: worker запускает etl_newbuilding_crossload task,
|
||||||
|
# которому нужен прямой TCP-доступ к tradein-postgres через gendesign_shared.
|
||||||
|
# default — обязательно явно, иначе сервис выпадет из дефолтной сети.
|
||||||
|
networks: [default, shared]
|
||||||
|
|
||||||
beat:
|
beat:
|
||||||
# Lean backend-образ (без Chromium) — beat только триггерит таски в Redis.
|
# Lean backend-образ (без Chromium) — beat только триггерит таски в Redis.
|
||||||
|
|
|
||||||
44
tradein-mvp/backend/data/sql/101_gendesign_reader_role.sql
Normal file
44
tradein-mvp/backend/data/sql/101_gendesign_reader_role.sql
Normal file
|
|
@ -0,0 +1,44 @@
|
||||||
|
-- 101_gendesign_reader_role.sql
|
||||||
|
-- Reader-роль для cross-load ETL gendesign → tradein (#976 950-E5).
|
||||||
|
-- Зеркало паттерна gendesign/data/sql/100_tradein_fdw_role.sql.
|
||||||
|
--
|
||||||
|
-- Роль gendesign_reader создаётся без пароля — пароль устанавливается
|
||||||
|
-- deploy-tradein.yml из env TRADEIN_READER_PASSWORD (bootstrap-шаг после миграций).
|
||||||
|
-- Никогда не embed password в SQL.
|
||||||
|
--
|
||||||
|
-- Доступ: ТОЛЬКО SELECT на houses, house_sources, houses_price_dynamics,
|
||||||
|
-- house_reliability_checks, house_reviews (источники cross-load ETL #976).
|
||||||
|
--
|
||||||
|
-- Идемпотентно: DO $$ IF NOT EXISTS — безопасно при повторном применении.
|
||||||
|
-- AUTO-APPLIED на прод при деплое через deploy-tradein.yml/_schema_migrations.
|
||||||
|
|
||||||
|
BEGIN;
|
||||||
|
|
||||||
|
DO $$
|
||||||
|
BEGIN
|
||||||
|
IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'gendesign_reader') THEN
|
||||||
|
CREATE ROLE gendesign_reader LOGIN;
|
||||||
|
END IF;
|
||||||
|
END$$;
|
||||||
|
|
||||||
|
COMMENT ON ROLE gendesign_reader IS
|
||||||
|
'Cross-load ETL reader from gendesign-backend (#976). '
|
||||||
|
'Password set by .forgejo/workflows/deploy-tradein.yml from env TRADEIN_READER_PASSWORD '
|
||||||
|
'post-migration step. Never embed password in SQL migrations.';
|
||||||
|
|
||||||
|
-- Defense-in-depth: явный REVOKE-периметр (любые PUBLIC-гранты игнорируются).
|
||||||
|
REVOKE ALL ON ALL TABLES IN SCHEMA public FROM gendesign_reader;
|
||||||
|
REVOKE ALL ON ALL SEQUENCES IN SCHEMA public FROM gendesign_reader;
|
||||||
|
REVOKE ALL ON ALL FUNCTIONS IN SCHEMA public FROM gendesign_reader;
|
||||||
|
|
||||||
|
GRANT CONNECT ON DATABASE tradein TO gendesign_reader;
|
||||||
|
GRANT USAGE ON SCHEMA public TO gendesign_reader;
|
||||||
|
|
||||||
|
-- SELECT только на таблицах-источниках ETL #976 (не весь public).
|
||||||
|
GRANT SELECT ON houses TO gendesign_reader;
|
||||||
|
GRANT SELECT ON house_sources TO gendesign_reader;
|
||||||
|
GRANT SELECT ON houses_price_dynamics TO gendesign_reader;
|
||||||
|
GRANT SELECT ON house_reliability_checks TO gendesign_reader;
|
||||||
|
GRANT SELECT ON house_reviews TO gendesign_reader;
|
||||||
|
|
||||||
|
COMMIT;
|
||||||
|
|
@ -0,0 +1,54 @@
|
||||||
|
-- Set gendesign_reader password from env.
|
||||||
|
-- Applied by .forgejo/workflows/deploy-tradein.yml after migrations:
|
||||||
|
-- psql -v pw="$TRADEIN_READER_PASSWORD" -f ops/db-bootstrap/set_gendesign_reader_password.sql
|
||||||
|
-- cwd на деплое = /opt/gendesign/tradein-mvp (путь относительный).
|
||||||
|
--
|
||||||
|
-- Idempotent: ALTER если роль существует, NOTICE и продолжает если нет
|
||||||
|
-- (migration 101_gendesign_reader_role.sql может ещё не примениться на первом деплое).
|
||||||
|
-- Пароль НИКОГДА не хранится в этом файле или в git — только имя переменной.
|
||||||
|
--
|
||||||
|
-- Format %L экранирует пароль как SQL string literal — безопасно даже с кавычками.
|
||||||
|
--
|
||||||
|
-- psql variable substitution (:'pw') НЕ интерполируется внутри dollar-quoted
|
||||||
|
-- блока ($$...$$) — это правило psql, не bug. Поэтому password передаём в DO
|
||||||
|
-- через сессионный GUC (set_config), который psql интерполирует ВНЕ dollar
|
||||||
|
-- quote, и читаем внутри через current_setting().
|
||||||
|
-- Reference incident: deploy 2026-05-24 (post-merge PR #503) упал на
|
||||||
|
-- "syntax error at or near ':'" — :'pw' дошёл до сервера as literal вместо
|
||||||
|
-- интерполяции при использовании -c вместо stdin + файл.
|
||||||
|
--
|
||||||
|
-- Источник переменной pw: TRADEIN_READER_PASSWORD из tradein-mvp/.env.runtime.
|
||||||
|
-- Передаётся через: psql -v "pw=${TRADEIN_READER_PASSWORD}"
|
||||||
|
--
|
||||||
|
-- ⚠️ `set_config(name, value, is_local) -> text` ВОЗВРАЩАЕТ установленное
|
||||||
|
-- значение. Без `\o /dev/null` psql напечатал бы пароль на stdout → leak в
|
||||||
|
-- Forgejo Actions deploy logs (retained, visible всем с repo read access).
|
||||||
|
-- Поэтому оборачиваем оба set_config вызова в `\o /dev/null` / `\o` brackets
|
||||||
|
-- — output mutes только для этих строк, NOTICE-сообщения из DO block
|
||||||
|
-- (idempotency signal) остаются видимыми.
|
||||||
|
--
|
||||||
|
-- Rollback path: НЕ revert этого файла (вернёт сломанный :'pw' внутри $$).
|
||||||
|
-- Корректный rollback — unset TRADEIN_READER_PASSWORD в
|
||||||
|
-- /opt/gendesign/tradein-mvp/.env.runtime на VPS, deploy-tradein.yml
|
||||||
|
-- тогда пропустит этот шаг полностью.
|
||||||
|
|
||||||
|
\o /dev/null
|
||||||
|
SELECT set_config('app.reader_pw', :'pw', false);
|
||||||
|
\o
|
||||||
|
|
||||||
|
DO $$
|
||||||
|
BEGIN
|
||||||
|
IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'gendesign_reader') THEN
|
||||||
|
EXECUTE format('ALTER ROLE gendesign_reader WITH PASSWORD %L', current_setting('app.reader_pw'));
|
||||||
|
RAISE NOTICE 'gendesign_reader password set';
|
||||||
|
ELSE
|
||||||
|
RAISE NOTICE 'gendesign_reader role missing — migration 101_gendesign_reader_role.sql not applied yet';
|
||||||
|
END IF;
|
||||||
|
END $$;
|
||||||
|
|
||||||
|
-- Clear GUC after use (defense-in-depth — не оставляем password в session state
|
||||||
|
-- даже на short connection). Same \o trick — set_config return value is empty
|
||||||
|
-- string here, но всё равно лишний row в stdout не нужен.
|
||||||
|
\o /dev/null
|
||||||
|
SELECT set_config('app.reader_pw', '', false);
|
||||||
|
\o
|
||||||
Loading…
Add table
Reference in a new issue