gendesign/data/sql/11_load_domrf.py

"""Loads DOM.RF analytics JSON files from data/raw/domrf/ into Postgres.

Usage:
    PG_HOST=localhost PG_PORT=15432 PGPASSWORD=... python data/sql/11_load_domrf.py
    # or for local test:
    PGPASSWORD=test PG_PORT=6543 python data/sql/11_load_domrf.py

Files expected (snapshot from https://наш.дом.рф/portal-analytics/api/):
  data/raw/domrf/regions.json          — list of FO with regions inside
  data/raw/domrf/developers.json       — flat list of {id, name}
  data/raw/domrf/by-region.json        — per-region aggregates with room-count details
  data/raw/domrf/by-flat-area-RF.json  — RF area distribution with room-count details
  data/raw/domrf/by-developer-RF.json  — RF top developers with room-count details
  data/raw/domrf/by-room-count-RF.json — RF totals (sanity check, not stored)
"""
import json, os, sys, datetime, subprocess

HERE = os.path.dirname(os.path.abspath(__file__))
RAW = os.path.join(HERE, '..', 'raw', 'domrf')

PG_HOST = os.environ.get('PG_HOST', '127.0.0.1')
PG_PORT = os.environ.get('PG_PORT', '5432')
PG_USER = os.environ.get('PG_USER', 'gendesign')
PG_DB   = os.environ.get('PG_DB',   'gendesign')
PG_PASS = os.environ.get('PGPASSWORD', '')

SNAPSHOT_DATE = os.environ.get('SNAPSHOT_DATE', datetime.date.today().isoformat())


def psql(sql_or_stdin, stdin_text=None):
    """Run psql via docker. Pipe SQL via stdin."""
    cmd = [
        'docker', 'run', '--rm', '-i',
        '-e', f'PGPASSWORD={PG_PASS}',
        'postgres:16-alpine',
        'psql', '-h', 'host.docker.internal', '-p', PG_PORT, '-U', PG_USER, '-d', PG_DB,
        '-v', 'ON_ERROR_STOP=1', '--quiet',
    ]
    if stdin_text is None:
        stdin_text = sql_or_stdin
    res = subprocess.run(cmd, input=stdin_text, capture_output=True, text=True, encoding='utf-8')
    if res.returncode != 0:
        print('psql FAILED:', res.stderr, file=sys.stderr)
        raise SystemExit(res.returncode)
    return res.stdout


def sql_escape(s):
    return s.replace("'", "''")


def main():
    raw_files = {
        'regions':         os.path.join(RAW, 'regions.json'),
        'developers':      os.path.join(RAW, 'developers.json'),
        'by-region':       os.path.join(RAW, 'by-region.json'),
        'by-flat-area-RF': os.path.join(RAW, 'by-flat-area-RF.json'),
        'by-developer-RF': os.path.join(RAW, 'by-developer-RF.json'),
    }
    for k, p in raw_files.items():
        if not os.path.exists(p):
            print(f'MISSING: {p}', file=sys.stderr); raise SystemExit(2)

    # 0. Apply schema (idempotent CREATE TABLE IF NOT EXISTS)
    schema = open(os.path.join(HERE, '10_schema_domrf.sql'), encoding='utf-8').read()
    psql(schema)
    print('schema applied')

    # 1. Snapshot row
    psql(f"""
        INSERT INTO domrf_snapshots (snapshot_date, notes)
        VALUES ('{SNAPSHOT_DATE}', 'naash.dom.rf portal-analytics, captured via chrome-devtools MCP')
        ON CONFLICT (snapshot_date) DO UPDATE SET fetched_at = NOW();
    """)

    # 2. Regions dictionary
    regions_data = json.load(open(raw_files['regions'], encoding='utf-8'))
    parts = []
    for fo in regions_data:
        for r in fo['regions']:
            parts.append(f"({r['id']}, '{sql_escape(r['name'])}', '{sql_escape(fo['name'])}')")
    if parts:
        psql(f"""
            INSERT INTO domrf_regions (region_id, region_name, federal_district)
            VALUES {','.join(parts)}
            ON CONFLICT (region_id) DO UPDATE SET
                region_name = EXCLUDED.region_name,
                federal_district = EXCLUDED.federal_district;
        """)
    print(f'regions: {sum(len(fo["regions"]) for fo in regions_data)}')

    # 3. Developers dictionary
    devs = json.load(open(raw_files['developers'], encoding='utf-8'))
    # batch insert in chunks of 500 to keep statement size sane
    BATCH = 500
    for i in range(0, len(devs), BATCH):
        chunk = devs[i:i+BATCH]
        parts = [f"('{sql_escape(d['id'])}', '{sql_escape(d['name'])}')" for d in chunk]
        psql(f"""
            INSERT INTO domrf_developers (developer_id, developer_name)
            VALUES {','.join(parts)}
            ON CONFLICT (developer_id) DO UPDATE SET developer_name = EXCLUDED.developer_name;
        """)
    print(f'developers: {len(devs)}')

    # 4. By-region aggregates: insert TOTAL row + per-room rows for each region
    byregion = json.load(open(raw_files['by-region'], encoding='utf-8'))
    rows = []
    for r in byregion:
        rid = r['id']
        # TOTAL row
        rows.append((rid, SNAPSHOT_DATE, 'TOTAL', r['flatCount'], r['area'], None))
        for d in r.get('details', []):
            rows.append((rid, SNAPSHOT_DATE, d['roomCountType'], d['flatCount'], d['area'], d.get('percent')))
    parts = [f"({rid}, '{sd}', '{rt}', {fc}, {a}, {'NULL' if p is None else p})"
             for (rid, sd, rt, fc, a, p) in rows]
    for i in range(0, len(parts), BATCH):
        psql(f"""
            INSERT INTO domrf_region_aggregates (region_id, snapshot_date, room_count_type, flat_count, area_sqm, percent)
            VALUES {','.join(parts[i:i+BATCH])}
            ON CONFLICT (region_id, snapshot_date, room_count_type) DO UPDATE SET
                flat_count = EXCLUDED.flat_count,
                area_sqm = EXCLUDED.area_sqm,
                percent = EXCLUDED.percent;
        """)
    print(f'region_aggregates rows: {len(rows)}')

    # 5. Flat-area distribution (RF-wide)
    byarea = json.load(open(raw_files['by-flat-area-RF'], encoding='utf-8'))
    rows = []
    for bucket in byarea:
        bid = bucket['id']
        rows.append((bid, 'TOTAL', bucket['flatCount'], bucket['area'], bucket.get('percent')))
        for d in bucket.get('details', []):
            rows.append((bid, d['roomCountType'], d['flatCount'], d['area'], d.get('percent')))
    parts = [f"('{SNAPSHOT_DATE}', 0, '{sql_escape(bid)}', '{rt}', {fc}, {a}, {'NULL' if p is None else p})"
             for (bid, rt, fc, a, p) in rows]
    psql(f"""
        INSERT INTO domrf_flat_area_distribution (snapshot_date, region_id, area_bucket, room_count_type, flat_count, area_sqm, percent)
        VALUES {','.join(parts)}
        ON CONFLICT (snapshot_date, region_id, area_bucket, room_count_type) DO UPDATE SET
            flat_count = EXCLUDED.flat_count,
            area_sqm = EXCLUDED.area_sqm,
            percent = EXCLUDED.percent;
    """)
    print(f'flat_area_distribution rows: {len(rows)}')

    # 6. By-developer aggregates (RF-wide because ?regionId is ignored)
    bydev = json.load(open(raw_files['by-developer-RF'], encoding='utf-8'))
    rows = []
    for dev in bydev:
        did = dev['id']
        rows.append((did, 'TOTAL', dev['flatCount'], dev['area'], dev.get('percent')))
        for d in dev.get('details', []):
            rows.append((did, d['roomCountType'], d['flatCount'], d['area'], d.get('percent')))
    parts = [f"('{sql_escape(did)}', '{SNAPSHOT_DATE}', 0, '{rt}', {fc}, {a}, {'NULL' if p is None else p})"
             for (did, rt, fc, a, p) in rows]
    for i in range(0, len(parts), BATCH):
        psql(f"""
            INSERT INTO domrf_developer_aggregates (developer_id, snapshot_date, region_id, room_count_type, flat_count, area_sqm, percent)
            VALUES {','.join(parts[i:i+BATCH])}
            ON CONFLICT (developer_id, snapshot_date, region_id, room_count_type) DO UPDATE SET
                flat_count = EXCLUDED.flat_count,
                area_sqm = EXCLUDED.area_sqm,
                percent = EXCLUDED.percent;
        """)
    print(f'developer_aggregates rows: {len(rows)}')

    # Final summary
    out = psql("""
        SELECT
            (SELECT COUNT(*) FROM domrf_regions)              AS regions,
            (SELECT COUNT(*) FROM domrf_developers)           AS developers,
            (SELECT COUNT(*) FROM domrf_region_aggregates)    AS region_agg,
            (SELECT COUNT(*) FROM domrf_flat_area_distribution) AS area_dist,
            (SELECT COUNT(*) FROM domrf_developer_aggregates) AS dev_agg;
    """)
    print('--- summary ---')
    print(out)


if __name__ == '__main__':
    main()