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