Merge pull request 'feat(sf): cross-load ETL tradein→gendesign newbuilding_listings — вариант A (#976)' (#1185) from feat/sf-newbuilding-crossload into main

Reviewed-on: #1185

.forgejo/workflows/deploy-tradein.yml

@@ -318,6 +318,25 @@ jobs:
             done
             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.
             # tradein-backend startup runs ensure_fdw_user_mapping which needs
             # FOREIGN SERVER gendesign_remote (created by 060_postgres_fdw_extension.sql).

backend/app/api/v1/admin_scrape.py

@@ -1034,6 +1034,30 @@ def resume_geo_job(
     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) ──
 
 

backend/app/core/config.py

@@ -61,7 +61,7 @@ class Settings(BaseSettings):
             if not stripped:
                 return []
             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 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: 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).
     # Free tier: 2000 запросов/день. Используется в /parcels/{cad}/isochrones.
     # Если не задан — endpoint вернёт 503 с инструкцией по регистрации.

backend/app/services/etl/newbuilding_crossload.py

@@ -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}

backend/app/workers/beat_schedule.py

@@ -381,6 +381,17 @@ def build_beat_schedule() -> dict:
         "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.
     # 1-е число месяца 06:00 МСК — после opportunity (05:30). Документы планировки стабильны,
     # ежемесячно достаточно. WFS BBOX (не grid-walk) — один запрос на слой.

backend/app/workers/tasks/etl_newbuilding_crossload.py

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

backend/tests/services/test_newbuilding_crossload.py

@@ -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'ом"

docker-compose.prod.yml

@@ -82,6 +82,10 @@ services:
       timeout: 5s
       retries: 5
       start_period: 20s
+    # #976 cross-DB ETL tradein→gendesign: backend подключается к tradein-postgres
+    # через gendesign_shared network (tradein-postgres уже в этой сети).
+    # default — обязательно явно, иначе сервис выпадет из дефолтной сети.
+    networks: [default, shared]
 
   frontend:
     image: ghcr.io/lekss361/gendesign-frontend:${IMAGE_TAG:-latest}
@@ -122,6 +126,10 @@ services:
       # /data/anton-sqlite/analysis.db).
       - /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"]
+    # #976 cross-DB ETL tradein→gendesign: worker запускает etl_newbuilding_crossload task,
+    # которому нужен прямой TCP-доступ к tradein-postgres через gendesign_shared.
+    # default — обязательно явно, иначе сервис выпадет из дефолтной сети.
+    networks: [default, shared]
 
   beat:
     # Lean backend-образ (без Chromium) — beat только триггерит таски в Redis.

tradein-mvp/backend/data/sql/101_gendesign_reader_role.sql

@@ -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;

tradein-mvp/ops/db-bootstrap/set_gendesign_reader_password.sql

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