"""Loads scraped DOM.RF JSONs into normalized PG tables. Reads: data/raw/domrf_full/
/*.json Writes: domrf_launch_top, ..._obj_class, ..._monthly, domrf_sold_ready_index, ..._breakdown, ..._dynamics, domrf_project_finance, domrf_commissioning, domrf_mortgage_dashboard, domrf_mortgage_details. Existing tables (domrf_developers, domrf_regions, domrf_developer_aggregates, domrf_region_aggregates, domrf_developers_full, domrf_escrow_banks, domrf_guaranty_regions, domrf_housing_summary, domrf_sold_out, domrf_mortgage_rates, etc.) are NOT touched. """ import os, json, sys, subprocess, datetime HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.join(HERE, '..', 'raw', 'domrf_full') 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', '') SNAP = os.environ.get('SNAPSHOT_DATE', datetime.date.today().isoformat()) def psql(sql): 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'] res = subprocess.run(cmd, input=sql, capture_output=True, text=True, encoding='utf-8') if res.returncode != 0: print('psql FAILED:', res.stderr[-2000:], file=sys.stderr) raise SystemExit(res.returncode) return res.stdout def esc(s): if s is None: return 'NULL' return "'" + str(s).replace("'", "''") + "'" def num(v): if v is None or v == '-' or v == '': return 'NULL' try: return str(float(v)) except (ValueError, TypeError): return 'NULL' def integer(v): if v is None: return 'NULL' try: return str(int(float(v))) except (ValueError, TypeError): return 'NULL' def load_json(section, fname): p = os.path.join(ROOT, section, fname + '.json') if not os.path.exists(p): return None return json.load(open(p, encoding='utf-8')) def find_files(section, prefix): """Find files in section/ starting with prefix (e.g. 'launch_developers_app__').""" sec_dir = os.path.join(ROOT, section) if not os.path.isdir(sec_dir): return [] return [f for f in os.listdir(sec_dir) if f.startswith(prefix) and f.endswith('.json')] def best_size_file(section, prefix): """Pick the file with largest size= parameter (or only one).""" files = find_files(section, prefix) if not files: return None # Prefer size=1000 over size=15 files.sort(key=lambda f: 0 if 'size_1000' in f else 1) return files[0][:-5] # strip .json def main(): schema = open(os.path.join(HERE, '31_schema_domrf_full.sql'), encoding='utf-8').read() psql(schema) print('schema applied') # ── 1. LAUNCH ──────────────────────────────────────────────────────────── rep_year = 2026 parts = [] for dim, prefix_app, prefix_rnv in [ ('developer', 'launch_developers_app__', 'launch_developers_rnv__'), ('region', 'launch_regions_app__', 'launch_regions_rnv__'), ('area', 'launch_areas_app__', 'launch_areas_rnv__'), ('fo', 'launch_federal-districts_app__', 'launch_federal-districts_rnv__'), ]: for metric, prefix in [('app', prefix_app), ('rnv', prefix_rnv)]: fname = best_size_file('launch', prefix) if not fname: continue d = load_json('launch', fname) calc = d.get('calculationType', 'SQUARE') for v in d.get('values', []): eid = str(v.get('id', v.get('name', ''))) ename = v.get('name', eid) value = v.get('value') if value is None: continue parts.append(f"('{SNAP}', {rep_year}, {esc(calc)}, '{dim}', '{metric}', " f"{esc(eid)}, {esc(ename)}, {num(value)})") if parts: for i in range(0, len(parts), 500): psql(f""" INSERT INTO domrf_launch_top (snapshot_date, rep_year, calc_type, dim_type, metric_type, entity_id, entity_name, value) VALUES {','.join(parts[i:i+500])} ON CONFLICT (snapshot_date, rep_year, calc_type, dim_type, metric_type, entity_id) DO UPDATE SET entity_name = EXCLUDED.entity_name, value = EXCLUDED.value; """) print(f'launch_top: {len(parts)}') # obj_class d = load_json('launch', 'launch_obj-class_table__calculationType_SQUARE_repYear_2026') \ or load_json('launch', 'launch_obj-class_table') if d: calc = d.get('calculationType', 'SQUARE') parts = [f"('{SNAP}', {rep_year}, {esc(calc)}, {integer(o['objClassCd'])}, " f"{esc(o['objClassDesc'])}, {num(o.get('appValue'))}, {num(o.get('rnvValue'))})" for o in d.get('objClassTypes', [])] if parts: psql(f""" INSERT INTO domrf_launch_obj_class (snapshot_date, rep_year, calc_type, obj_class_cd, obj_class_desc, app_value, rnv_value) VALUES {','.join(parts)} ON CONFLICT (snapshot_date, rep_year, calc_type, obj_class_cd) DO UPDATE SET obj_class_desc = EXCLUDED.obj_class_desc, app_value = EXCLUDED.app_value, rnv_value = EXCLUDED.rnv_value; """) print(f'launch_obj_class: {len(parts)}') # monthly time series d = load_json('launch', 'charts') if d: calc = d.get('calculationType', 'SQUARE') parts = [] for y in d.get('repYears', []): yr = y['repYear'] for m in y.get('repMonth', []): parts.append(f"('{SNAP}', {esc(calc)}, {yr}, {integer(m['repMonth'])}, " f"{num(m.get('appValue'))}, {num(m.get('rnvValue'))})") if parts: for i in range(0, len(parts), 500): psql(f""" INSERT INTO domrf_launch_monthly (snapshot_date, calc_type, rep_year, rep_month, app_value, rnv_value) VALUES {','.join(parts[i:i+500])} ON CONFLICT (snapshot_date, calc_type, rep_year, rep_month) DO UPDATE SET app_value = EXCLUDED.app_value, rnv_value = EXCLUDED.rnv_value; """) print(f'launch_monthly: {len(parts)}') # ── 2. SOLD READY ──────────────────────────────────────────────────────── rep_year, rep_month = 2026, 3 d = load_json('sold_ready', 'ready-construction_index__repMonth_3_repYear_2026') if d: psql(f""" INSERT INTO domrf_sold_ready_index (snapshot_date, rep_year, rep_month, square_sum, sold_perc, sold_sum, sold_ready_perc, ready_perc) VALUES ('{SNAP}', {rep_year}, {rep_month}, {num(d.get('squareSumIndex'))}, {num(d.get('soldPercIndex'))}, {num(d.get('soldSumIndex'))}, {num(d.get('soldReadyPercIndex'))}, {num(d.get('readyPercIndex'))}) ON CONFLICT (snapshot_date, rep_year, rep_month) DO UPDATE SET square_sum = EXCLUDED.square_sum, sold_perc = EXCLUDED.sold_perc, sold_sum = EXCLUDED.sold_sum, sold_ready_perc = EXCLUDED.sold_ready_perc, ready_perc = EXCLUDED.ready_perc; """) print('sold_ready_index: 1') d = load_json('sold_ready', 'ready-construction_charts__repMonth_3_repYear_2026') if d: parts = [] seen = set() for chart in d.get('charts', []): ct = chart.get('chartType') for row in chart.get('data', []): key = row.get('key', '') if not key: continue pk = (ct, key) if pk in seen: continue seen.add(pk) extra = {k: v for k, v in row.items() if k not in {'key', 'squareSum', 'soldPerc', 'readyPerc', 'soldReadyPerc', 'foCd'}} extra_json = json.dumps(extra, ensure_ascii=False) if extra else None parts.append(f"('{SNAP}', {rep_year}, {rep_month}, {esc(ct)}, {esc(key)}, " f"{num(row.get('squareSum'))}, {num(row.get('soldPerc'))}, " f"{num(row.get('readyPerc'))}, {num(row.get('soldReadyPerc'))}, " f"{integer(row.get('foCd'))}, " f"{('NULL' if extra_json is None else esc(extra_json) + '::jsonb')})") if parts: for i in range(0, len(parts), 500): psql(f""" INSERT INTO domrf_sold_ready_breakdown (snapshot_date, rep_year, rep_month, chart_type, entity_key, square_sum, sold_perc, ready_perc, sold_ready_perc, fo_cd, extra) VALUES {','.join(parts[i:i+500])} ON CONFLICT (snapshot_date, rep_year, rep_month, chart_type, entity_key) DO UPDATE SET square_sum = EXCLUDED.square_sum, sold_perc = EXCLUDED.sold_perc, ready_perc = EXCLUDED.ready_perc, sold_ready_perc = EXCLUDED.sold_ready_perc, fo_cd = EXCLUDED.fo_cd, extra = EXCLUDED.extra; """) print(f'sold_ready_breakdown: {len(parts)}') d = load_json('sold_ready', 'ready-construction_dynamics__dynamicChartType_readyPercChart_2CsoldPercChart_2CsoldReadyPercChart_2CsquareSum') if d: parts = [] for chart in d.get('dynamicCharts', []): ct = chart.get('dynamicChartType') for y in chart.get('repYears', []): yr = y.get('repYear') for m in y.get('repMonths', []): parts.append(f"('{SNAP}', {esc(ct)}, {integer(yr)}, {integer(m.get('repMonth'))}, {num(m.get('value'))})") if parts: for i in range(0, len(parts), 500): psql(f""" INSERT INTO domrf_sold_ready_dynamics (snapshot_date, dynamic_chart_type, rep_year, rep_month, value) VALUES {','.join(parts[i:i+500])} ON CONFLICT (snapshot_date, dynamic_chart_type, rep_year, rep_month) DO UPDATE SET value = EXCLUDED.value; """) print(f'sold_ready_dynamics: {len(parts)}') # ── 3. PROJECT FINANCE (share_construction) ───────────────────────────── d = load_json('share_construction', 'dashboard') or load_json('share_construction', 'project_finance_dashboard__dateDay_26-04-2026') if d: rows = d.get('data', {}).get('data', []) if isinstance(d.get('data'), dict) else d.get('data', []) parts = [] for row in rows: extra = {k: v for k, v in row.items() if k not in { 'id','subjectType','subject','regionCd','foCd','foDesc', 'rnsCnt','devCnt','livSqAmt', 'guarantyEscrowRnsCnt','guarantyEscrowDevCnt','guarantyEscrowLivSqAmt', 'guarantyZosgRnsCnt','guarantyZosgDevCnt','guarantyZosgLivSqAmt', 'guarantyRnsCnt','guarantyDevCnt','guarantyLivSqAmt', 'nonguarantyRnsCnt','nonguarantyDevCnt','nonguarantyLivSqAmt'}} parts.append(( row.get('subjectType'), row.get('subject'), row.get('regionCd'), row.get('foCd'), row.get('foDesc'), row.get('rnsCnt'), row.get('devCnt'), row.get('livSqAmt'), row.get('guarantyEscrowRnsCnt'), row.get('guarantyEscrowDevCnt'), row.get('guarantyEscrowLivSqAmt'), row.get('guarantyZosgRnsCnt'), row.get('guarantyZosgDevCnt'), row.get('guarantyZosgLivSqAmt'), row.get('guarantyRnsCnt'), row.get('guarantyDevCnt'), row.get('guarantyLivSqAmt'), row.get('nonguarantyRnsCnt'), row.get('nonguarantyDevCnt'), row.get('nonguarantyLivSqAmt'), json.dumps(extra, ensure_ascii=False) if extra else None, )) if parts: # Clear today's data first (since PK is by subject_type+name, may have duplicates by region within fo) psql(f"DELETE FROM domrf_project_finance WHERE snapshot_date = '{SNAP}';") sql_parts = [] for p in parts: (st, sn, rc, fc, fd, rn, dn, ls, gers, gerd, gerls, gzrs, gzrd, gzrls, grns, grdv, grls, nrns, nrdv, nrls, ej) = p sql_parts.append( f"('{SNAP}', NULL, {esc(st)}, {esc(sn)}, {integer(rc)}, {integer(fc)}, {esc(fd)}, " f"{integer(rn)}, {integer(dn)}, {num(ls)}, " f"{integer(gers)}, {integer(gerd)}, {num(gerls)}, " f"{integer(gzrs)}, {integer(gzrd)}, {num(gzrls)}, " f"{integer(grns)}, {integer(grdv)}, {num(grls)}, " f"{integer(nrns)}, {integer(nrdv)}, {num(nrls)}, " f"{('NULL' if ej is None else esc(ej) + '::jsonb')})" ) for i in range(0, len(sql_parts), 200): psql(f""" INSERT INTO domrf_project_finance (snapshot_date, report_date, subject_type, subject_name, region_cd, fo_cd, fo_desc, rns_cnt, dev_cnt, liv_sq_amt, guaranty_escrow_rns_cnt, guaranty_escrow_dev_cnt, guaranty_escrow_liv_sq_amt, guaranty_zosg_rns_cnt, guaranty_zosg_dev_cnt, guaranty_zosg_liv_sq_amt, guaranty_rns_cnt, guaranty_dev_cnt, guaranty_liv_sq_amt, nonguaranty_rns_cnt, nonguaranty_dev_cnt, nonguaranty_liv_sq_amt, extra) VALUES {','.join(sql_parts[i:i+200])} ON CONFLICT DO NOTHING; """) print(f'project_finance: {len(parts)}') # ── 4. COMMISSIONING ─────────────────────────────────────────────────── d = load_json('commissioning', 'building_summary_data') if d: rows = d.get('data', []) parts = [] seen = set() for r in rows: pk = (r.get('regionId'), r.get('reportingPeriod')) if pk in seen: continue seen.add(pk) extra = {k: v for k, v in r.items() if k not in { 'regionId','regionName','regionType','reportingPeriod','year', 'accumulatedFactArea','accumulatedFactAreaMultifamily','accumulatedFactAreaPrivate', 'accumulatedFactAreaChange','accumulatedFactAreaChangeShare'}} parts.append(f"('{SNAP}', {integer(r.get('regionId'))}, {esc(r.get('regionName'))}, " f"{esc(r.get('regionType'))}, {esc(r.get('reportingPeriod'))}::date, " f"{integer(r.get('year'))}, " f"{num(r.get('accumulatedFactArea'))}, " f"{num(r.get('accumulatedFactAreaMultifamily'))}, " f"{num(r.get('accumulatedFactAreaPrivate'))}, " f"{num(r.get('accumulatedFactAreaChange'))}, " f"{num(r.get('accumulatedFactAreaChangeShare'))}, " f"{esc(json.dumps(extra, ensure_ascii=False)) + '::jsonb' if extra else 'NULL'})") if parts: psql(f""" INSERT INTO domrf_commissioning (snapshot_date, region_id, region_name, region_type, reporting_period, rep_year, accumulated_fact_area, accumulated_fact_area_multifamily, accumulated_fact_area_private, accumulated_fact_area_change, accumulated_fact_area_change_share, extra) VALUES {','.join(parts)} ON CONFLICT (snapshot_date, region_id, reporting_period) DO UPDATE SET region_name = EXCLUDED.region_name, accumulated_fact_area = EXCLUDED.accumulated_fact_area, accumulated_fact_area_multifamily = EXCLUDED.accumulated_fact_area_multifamily, accumulated_fact_area_private = EXCLUDED.accumulated_fact_area_private, accumulated_fact_area_change = EXCLUDED.accumulated_fact_area_change, accumulated_fact_area_change_share = EXCLUDED.accumulated_fact_area_change_share; """) print(f'commissioning: {len(parts)}') # ── 5. MORTGAGE STATS ────────────────────────────────────────────────── d = load_json('mortgage_stats', 'mortgage_dashboard_general') if d and d.get('data'): m = d['data'] psql(f""" INSERT INTO domrf_mortgage_dashboard (snapshot_date, total_credit_count, total_credit_count_delta_pct, primary_credit_count, primary_credit_count_delta_pct, secondary_credit_count, secondary_credit_count_delta_pct, total_credit_amount, total_credit_amount_delta_pct, primary_credit_amount, primary_credit_amount_delta_pct, secondary_credit_amount, secondary_credit_amount_delta_pct, total_credit_avg_rate, total_credit_avg_rate_delta, primary_credit_avg_rate, primary_credit_avg_rate_delta, secondary_credit_avg_rate, secondary_credit_avg_rate_delta) VALUES ('{SNAP}', {integer(m.get('totalCreditCount'))}, {num(m.get('totalCreditCountDeltaPct'))}, {integer(m.get('primaryCreditCount'))}, {num(m.get('primaryCreditCountDeltaPct'))}, {integer(m.get('secondaryCreditCount'))}, {num(m.get('secondaryCreditCountDeltaPct'))}, {num(m.get('totalCreditAmount'))}, {num(m.get('totalCreditAmountDeltaPct'))}, {num(m.get('primaryCreditAmount'))}, {num(m.get('primaryCreditAmountDeltaPct'))}, {num(m.get('secondaryCreditAmount'))}, {num(m.get('secondaryCreditAmountDeltaPct'))}, {num(m.get('totalCreditAvgRate'))}, {num(m.get('totalCreditAvgRateDelta'))}, {num(m.get('primaryCreditAvgRate'))}, {num(m.get('primaryCreditAvgRateDelta'))}, {num(m.get('secondaryCreditAvgRate'))}, {num(m.get('secondaryCreditAvgRateDelta'))}) ON CONFLICT (snapshot_date) DO UPDATE SET total_credit_count = EXCLUDED.total_credit_count, total_credit_amount = EXCLUDED.total_credit_amount, total_credit_avg_rate = EXCLUDED.total_credit_avg_rate; """) print('mortgage_dashboard: 1') d = load_json('mortgage_stats', 'mortgage_dashboard_details') if d and d.get('data'): rows = d['data'] parts = [f"('{SNAP}', {esc(r.get('currency'))}, {num(r.get('creditAmountAvg'))}, " f"{num(r.get('creditAmountAvgDeltaPct'))}, {num(r.get('creditAvgPeriod'))}, " f"{num(r.get('creditAvgPeriodDeltaPct'))}, {num(r.get('creditDebtsAmount'))}, " f"{num(r.get('creditDebtsAmountDeltaPct'))}, {num(r.get('creditDebtsOverduePercent'))}, " f"{num(r.get('creditDebtsOverduePercentDelta'))})" for r in rows] if parts: psql(f""" INSERT INTO domrf_mortgage_details (snapshot_date, currency, credit_amount_avg, credit_amount_avg_delta_pct, credit_avg_period, credit_avg_period_delta_pct, credit_debts_amount, credit_debts_amount_delta_pct, credit_debts_overdue_percent, credit_debts_overdue_percent_delta) VALUES {','.join(parts)} ON CONFLICT (snapshot_date, currency) DO UPDATE SET credit_amount_avg = EXCLUDED.credit_amount_avg, credit_avg_period = EXCLUDED.credit_avg_period, credit_debts_amount = EXCLUDED.credit_debts_amount; """) print(f'mortgage_details: {len(parts)}') print('\n--- summary ---') out = psql(f""" SELECT 'launch_top' tab, COUNT(*) FROM domrf_launch_top WHERE snapshot_date = '{SNAP}' UNION ALL SELECT 'launch_obj_class', COUNT(*) FROM domrf_launch_obj_class WHERE snapshot_date = '{SNAP}' UNION ALL SELECT 'launch_monthly', COUNT(*) FROM domrf_launch_monthly WHERE snapshot_date = '{SNAP}' UNION ALL SELECT 'sold_ready_index', COUNT(*) FROM domrf_sold_ready_index WHERE snapshot_date = '{SNAP}' UNION ALL SELECT 'sold_ready_breakdown', COUNT(*) FROM domrf_sold_ready_breakdown WHERE snapshot_date = '{SNAP}' UNION ALL SELECT 'sold_ready_dynamics', COUNT(*) FROM domrf_sold_ready_dynamics WHERE snapshot_date = '{SNAP}' UNION ALL SELECT 'project_finance', COUNT(*) FROM domrf_project_finance WHERE snapshot_date = '{SNAP}' UNION ALL SELECT 'commissioning', COUNT(*) FROM domrf_commissioning WHERE snapshot_date = '{SNAP}' UNION ALL SELECT 'mortgage_dashboard', COUNT(*) FROM domrf_mortgage_dashboard WHERE snapshot_date = '{SNAP}' UNION ALL SELECT 'mortgage_details', COUNT(*) FROM domrf_mortgage_details WHERE snapshot_date = '{SNAP}' ORDER BY 1; """) print(out) if __name__ == '__main__': main()