222 lines
8.4 KiB
Python
222 lines
8.4 KiB
Python
"""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()
|