"""Load lazy-load DOM.RF pages (housing, housing_dev, realization, mortgage_rates, stat_series) into prod PG. Reads: data/raw/domrf_full/{realization,housing,housing_dev,mortgage_rates,stat_series}/* Writes: domrf_realization (new), domrf_xlsx_files (new), domrf_raw_endpoints (existing universal store, append-only). Connect via SSH tunnel: PG_HOST=host.docker.internal PG_PORT=15432 (default). """ import os, json, re, sys, hashlib, base64, subprocess, datetime HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.join(HERE, '..', 'raw', 'domrf_full') PG_HOST = os.environ.get('PG_HOST', 'host.docker.internal') PG_PORT = os.environ.get('PG_PORT', '15432') PG_USER = os.environ.get('PG_USER', 'gendesign') PG_DB = os.environ.get('PG_DB', 'gendesign') PG_PASS = os.environ.get('PGPASSWORD', '2J2SBPMKuS998fiwhtQqDhMI') SNAP = os.environ.get('SNAPSHOT_DATE', datetime.date.today().isoformat()) def psql(sql, capture=True): cmd = ['docker', 'run', '--rm', '-i', '-e', f'PGPASSWORD={PG_PASS}', 'postgres:16-alpine', 'psql', '-h', PG_HOST, '-p', PG_PORT, '-U', PG_USER, '-d', PG_DB, '-v', 'ON_ERROR_STOP=1', '--quiet'] res = subprocess.run(cmd, input=sql, capture_output=capture, text=True, encoding='utf-8') if res.returncode != 0: print('psql FAILED:', (res.stderr or '')[-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 upsert_chunked(rows, table, cols, conflict_cols, batch=500): """Insert rows in batches with ON CONFLICT DO UPDATE.""" if not rows: return 0 update = ', '.join(f'{c}=EXCLUDED.{c}' for c in cols if c not in conflict_cols) total = 0 for i in range(0, len(rows), batch): chunk = rows[i:i + batch] sql = (f"INSERT INTO {table} ({','.join(cols)}) VALUES " + ',\n'.join(chunk) + f"\nON CONFLICT ({','.join(conflict_cols)}) DO UPDATE SET {update};") psql(sql) total += len(chunk) return total # ── 1. REALIZATION (rpp/{total,housing,readyYear,developer}) ───────────────── RPP_FNAME_RE = re.compile( r'rpp_(total|housing|readyYear|developer)__' r'(?:typeSquare_(total|living)_)?' r'(?:regionCode_(\d+)_)?' r'repMonth_(\d+)_repYear_(\d+)' ) def load_realization(): sec_dir = os.path.join(ROOT, 'realization') if not os.path.isdir(sec_dir): print(' realization dir missing') return 0 cols = ['snapshot_date', 'endpoint_type', 'region_code', 'rep_year', 'rep_month', 'type_square', 'subject', 'subject_desc', 'share_total', 'total_square', 'open_square', 'sold_square', 'unsold_square', 'unopened_square', 'sold_perc', 'unsold_perc', 'unopened_perc', 'sold_amount', 'price_avg'] rows = [] seen = set() for fname in sorted(os.listdir(sec_dir)): if not fname.endswith('.json') or not fname.startswith('rpp_'): continue m = RPP_FNAME_RE.match(fname) if not m: continue ep_raw, ts_filename, rcode_filename, rmonth, ryear = m.groups() endpoint_type = {'total': 'total', 'housing': 'housing', 'readyYear': 'ready_year', 'developer': 'developer'}[ep_raw] region_code = int(rcode_filename) if rcode_filename else -1 try: doc = json.load(open(os.path.join(sec_dir, fname), encoding='utf-8')) except Exception as e: print(f' parse err {fname}: {e}') continue for it in doc.get('data') or []: ts = ts_filename or it.get('typeSquare') or 'total' subject = str(it.get('subject', '')) if not subject: continue key = (endpoint_type, region_code, ts, subject) if key in seen: continue seen.add(key) rows.append( f"({esc(SNAP)},{esc(endpoint_type)},{region_code}," f"{int(ryear)},{int(rmonth)},{esc(ts)},{esc(subject)}," f"{esc(it.get('subjectDesc'))},{num(it.get('shareTotal'))}," f"{num(it.get('totalSquare'))},{num(it.get('openSquare'))}," f"{num(it.get('soldSquare'))},{num(it.get('unsoldSquare'))}," f"{num(it.get('unopenedSquare'))},{num(it.get('soldPerc'))}," f"{num(it.get('unsoldPerc'))},{num(it.get('unopenedPerc'))}," f"{num(it.get('soldAmount'))},{num(it.get('priceAvg'))})" ) return upsert_chunked( rows, 'domrf_realization', cols, conflict_cols=['snapshot_date', 'endpoint_type', 'region_code', 'type_square', 'subject'], ) # ── 2. STAT_SERIES XLSX FILES ──────────────────────────────────────────────── def load_xlsx_files(): xlsx_dir = os.path.join(ROOT, 'stat_series', 'xlsx') if not os.path.isdir(xlsx_dir): print(' xlsx dir missing') return 0 base_url = 'https://xn--80az8a.xn--d1aqf.xn--p1ai/site/binaries/content/assets/domrf/xlsdashboard/' rows = [] cols = ['snapshot_date', 'filename', 'source_url', 'bytes', 'sha256', 'content'] for fname in sorted(os.listdir(xlsx_dir)): if not fname.endswith('.xlsx'): continue path = os.path.join(xlsx_dir, fname) with open(path, 'rb') as f: blob = f.read() sha = hashlib.sha256(blob).hexdigest() # PG bytea hex format: \xDEADBEEF hex_blob = '\\\\x' + blob.hex() rows.append( f"({esc(SNAP)},{esc(fname)},{esc(base_url + fname)}," f"{len(blob)},{esc(sha)},E{esc(hex_blob)})" ) return upsert_chunked( rows, 'domrf_xlsx_files', cols, conflict_cols=['snapshot_date', 'filename'], batch=5, # XLSX bytes are big — keep batches tiny to avoid huge SQL strings ) # ── 3. RAW JSONB store for SSR + housing per-region ────────────────────────── def load_raw_endpoints(): """Dump every captured JSON file from the 5 lazy pages into domrf_raw_endpoints (existing universal store). Keeps all data accessible for future normalization without re-scraping.""" cols = ['snapshot_date', 'section', 'endpoint', 'source_url', 'payload', 'payload_size'] rows = [] for section in ('housing', 'housing_dev', 'mortgage_rates', 'stat_series'): sec_dir = os.path.join(ROOT, section) if not os.path.isdir(sec_dir): continue for fname in sorted(os.listdir(sec_dir)): path = os.path.join(sec_dir, fname) if not os.path.isfile(path) or not fname.endswith('.json'): continue with open(path, encoding='utf-8') as f: text = f.read() # Validate JSON; skip if parse fails (rare). try: json.loads(text) except Exception: continue endpoint = fname[:-5] # drop .json rows.append( f"({esc(SNAP)},{esc(section)},{esc(endpoint)},NULL," f"{esc(text)}::jsonb,{len(text)})" ) return upsert_chunked( rows, 'domrf_raw_endpoints', cols, conflict_cols=['snapshot_date', 'section', 'endpoint'], batch=50, # payloads can be 100KB+ ) def main(): print(f'Snapshot date: {SNAP}') print(f'PG: {PG_USER}@{PG_HOST}:{PG_PORT}/{PG_DB}') schema = open(os.path.join(HERE, '33_schema_domrf_lazy.sql'), encoding='utf-8').read() psql(schema) print('schema applied') # Stamp snapshot in domrf_snapshots (FK target) psql(f"INSERT INTO domrf_snapshots(snapshot_date) VALUES ({esc(SNAP)}) " f"ON CONFLICT DO NOTHING;") n_real = load_realization() print(f' realization: {n_real} rows') n_xlsx = load_xlsx_files() print(f' xlsx_files: {n_xlsx} rows') n_raw = load_raw_endpoints() print(f' raw_endpoints: {n_raw} rows') if __name__ == '__main__': main()