283 lines
12 KiB
Python
283 lines
12 KiB
Python
"""Load historical scrape into prod (realization + sold_ready).
|
|
|
|
Reads: data/raw/domrf_history/<section>/<YYYY-MM>/*.json
|
|
Writes: domrf_realization, domrf_sold_ready_index/breakdown/dynamics
|
|
+ a fresh row in domrf_snapshots per period.
|
|
|
|
SNAPSHOT_DATE for each period = last day of repMonth (so PK collisions are
|
|
avoided across multiple historical pulls done on the same calendar day).
|
|
"""
|
|
import os, json, re, sys, subprocess, calendar, datetime
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
ROOT = os.path.join(HERE, '..', 'raw', 'domrf_history')
|
|
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')
|
|
|
|
|
|
def psql(sql):
|
|
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=True, 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:
|
|
f = float(v)
|
|
if f != f:
|
|
return 'NULL'
|
|
return repr(f)
|
|
except (ValueError, TypeError):
|
|
return 'NULL'
|
|
|
|
|
|
def upsert_chunked(rows, table, cols, conflict_cols, batch=500):
|
|
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 "
|
|
+ ','.join(chunk)
|
|
+ (f" ON CONFLICT ({','.join(conflict_cols)}) DO UPDATE SET {update};"
|
|
if update else f" ON CONFLICT ({','.join(conflict_cols)}) DO NOTHING;"))
|
|
psql(sql)
|
|
total += len(chunk)
|
|
return total
|
|
|
|
|
|
# ── REALIZATION ─────────────────────────────────────────────────────────────
|
|
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(period_dir, snap):
|
|
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(period_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 0
|
|
try:
|
|
doc = json.load(open(os.path.join(period_dir, fname), encoding='utf-8'))
|
|
except Exception:
|
|
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'],
|
|
)
|
|
|
|
|
|
# ── SOLD_READY ──────────────────────────────────────────────────────────────
|
|
def load_sold_ready(period_dir, snap, year, month):
|
|
n_idx = n_brk = n_dyn = 0
|
|
# index file
|
|
for fname in os.listdir(period_dir):
|
|
path = os.path.join(period_dir, fname)
|
|
if not fname.startswith('ready-construction_index') or not fname.endswith('.json'):
|
|
continue
|
|
try:
|
|
d = json.load(open(path, encoding='utf-8'))
|
|
except Exception:
|
|
continue
|
|
row = (
|
|
f"({esc(snap)},{int(year)},{int(month)},"
|
|
f"{num(d.get('squareSumIndex'))},{num(d.get('soldPercIndex'))},"
|
|
f"{num(d.get('soldSumIndex'))},{num(d.get('soldReadyPercIndex'))},"
|
|
f"{num(d.get('readyPercIndex'))})"
|
|
)
|
|
n_idx = upsert_chunked(
|
|
[row], 'domrf_sold_ready_index',
|
|
['snapshot_date', 'rep_year', 'rep_month',
|
|
'square_sum', 'sold_perc', 'sold_sum', 'sold_ready_perc', 'ready_perc'],
|
|
conflict_cols=['snapshot_date', 'rep_year', 'rep_month'],
|
|
)
|
|
# charts → breakdown (shape: {"charts":[{"chartType":"...","data":[...]}]} )
|
|
# Each chart_type has exactly one discriminator field — extracted to its
|
|
# own typed column instead of dumped into JSONB.
|
|
chart_rows = []
|
|
seen_keys = set()
|
|
for fname in os.listdir(period_dir):
|
|
if not (fname.startswith('ready-construction_charts')
|
|
or fname.startswith('ready-construction_ready-year-charts')) or not fname.endswith('.json'):
|
|
continue
|
|
try:
|
|
d = json.load(open(os.path.join(period_dir, fname), encoding='utf-8'))
|
|
except Exception:
|
|
continue
|
|
# Format A: {"charts":[{"chartType":"foChart","data":[...]}, ...]}
|
|
for chart_obj in (d.get('charts') or []):
|
|
chart_type = chart_obj.get('chartType')
|
|
if not chart_type:
|
|
continue
|
|
for it in chart_obj.get('data') or []:
|
|
if not isinstance(it, dict):
|
|
continue
|
|
key = it.get('key') or it.get('id') or it.get('subject') or it.get('name') or ''
|
|
if not key:
|
|
continue
|
|
pk = (chart_type, str(key))
|
|
if pk in seen_keys:
|
|
continue
|
|
seen_keys.add(pk)
|
|
chart_rows.append(
|
|
f"({esc(snap)},{int(year)},{int(month)},{esc(chart_type)},{esc(str(key))},"
|
|
f"{num(it.get('squareSum'))},{num(it.get('soldPerc'))},"
|
|
f"{num(it.get('readyPerc'))},{num(it.get('soldReadyPerc'))},"
|
|
f"{esc(it.get('foCd'))},"
|
|
f"{esc(it.get('regionCd'))},"
|
|
f"{esc(it.get('devCalcGroupId'))},"
|
|
f"{esc(it.get('devGroupValue'))},"
|
|
f"{esc(it.get('cityPopulation'))},"
|
|
f"{esc(it.get('objClass'))},"
|
|
f"NULL)" # ready_year — only for readyYearChart format below
|
|
)
|
|
# Format B: {"readyYears":[{"readyYear":2030,...}]} → readyYearChart
|
|
for it in (d.get('readyYears') or []):
|
|
if not isinstance(it, dict):
|
|
continue
|
|
ry = it.get('readyYear')
|
|
if ry is None:
|
|
continue
|
|
key = str(ry)
|
|
pk = ('readyYearChart', key)
|
|
if pk in seen_keys:
|
|
continue
|
|
seen_keys.add(pk)
|
|
chart_rows.append(
|
|
f"({esc(snap)},{int(year)},{int(month)},{esc('readyYearChart')},{esc(key)},"
|
|
f"{num(it.get('squareSum'))},{num(it.get('soldPerc'))},"
|
|
f"{num(it.get('readyPerc'))},{num(it.get('soldReadyPerc'))},"
|
|
f"NULL,NULL,NULL,NULL,NULL,NULL,{int(ry)})"
|
|
)
|
|
n_brk = upsert_chunked(
|
|
chart_rows, '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', 'region_cd', 'dev_calc_group_id', 'dev_group_value',
|
|
'city_population', 'obj_class', 'ready_year'],
|
|
conflict_cols=['snapshot_date', 'rep_year', 'rep_month', 'chart_type', 'entity_key'],
|
|
)
|
|
# dynamics: {"dynamicCharts":[{"dynamicChartType":"soldPercChart",
|
|
# "repYears":[{"repYear":2020,"repMonths":[{"repMonth":1,"value":41.54}]}]}]}
|
|
dyn_rows = []
|
|
dyn_seen = set()
|
|
for fname in os.listdir(period_dir):
|
|
if not fname.startswith('ready-construction_dynamics') or not fname.endswith('.json'):
|
|
continue
|
|
try:
|
|
d = json.load(open(os.path.join(period_dir, fname), encoding='utf-8'))
|
|
except Exception:
|
|
continue
|
|
for chart_obj in (d.get('dynamicCharts') or []):
|
|
ch_type = chart_obj.get('dynamicChartType')
|
|
if not ch_type:
|
|
continue
|
|
for ry_obj in chart_obj.get('repYears') or []:
|
|
ry = ry_obj.get('repYear')
|
|
if ry is None:
|
|
continue
|
|
for pt in ry_obj.get('repMonths') or []:
|
|
rm = pt.get('repMonth')
|
|
if rm is None:
|
|
continue
|
|
pk = (ch_type, ry, rm)
|
|
if pk in dyn_seen:
|
|
continue
|
|
dyn_seen.add(pk)
|
|
dyn_rows.append(
|
|
f"({esc(snap)},{esc(ch_type)},{int(ry)},{int(rm)},{num(pt.get('value'))})"
|
|
)
|
|
n_dyn = upsert_chunked(
|
|
dyn_rows, 'domrf_sold_ready_dynamics',
|
|
['snapshot_date', 'dynamic_chart_type', 'rep_year', 'rep_month', 'value'],
|
|
conflict_cols=['snapshot_date', 'dynamic_chart_type', 'rep_year', 'rep_month'],
|
|
)
|
|
return n_idx, n_brk, n_dyn
|
|
|
|
|
|
def main():
|
|
if not os.path.isdir(ROOT):
|
|
print(f'{ROOT} not found, run 40_scrape_history.py first')
|
|
return
|
|
|
|
# Discover periods from realization subdirs
|
|
realization_root = os.path.join(ROOT, 'realization')
|
|
periods = sorted(
|
|
d for d in os.listdir(realization_root)
|
|
if re.match(r'^\d{4}-\d{2}$', d) and os.path.isdir(os.path.join(realization_root, d))
|
|
)
|
|
print(f'Periods to load: {periods}')
|
|
|
|
for period in periods:
|
|
year, month = map(int, period.split('-'))
|
|
last_day = calendar.monthrange(year, month)[1]
|
|
snap = f'{year:04d}-{month:02d}-{last_day:02d}'
|
|
|
|
# Ensure parent snapshot row
|
|
psql(f"INSERT INTO domrf_snapshots(snapshot_date) VALUES ({esc(snap)}) ON CONFLICT DO NOTHING;")
|
|
|
|
period_dir = os.path.join(realization_root, period)
|
|
n_real = load_realization(period_dir, snap)
|
|
|
|
sr_dir = os.path.join(ROOT, 'sold_ready', period)
|
|
n_idx = n_brk = n_dyn = 0
|
|
if os.path.isdir(sr_dir):
|
|
n_idx, n_brk, n_dyn = load_sold_ready(sr_dir, snap, year, month)
|
|
|
|
print(f'{period} (snap={snap}): realization={n_real} '
|
|
f'sr_index={n_idx} sr_breakdown={n_brk} sr_dynamics={n_dyn}')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|