298 lines
12 KiB
Python
298 lines
12 KiB
Python
"""Migrate Anton's domrf.db (SQLite, 23.5 MB) into prod Postgres.
|
|
|
|
Reads: C:\\Users\\user\\Downloads\\Telegram Desktop\\db\\db\\domrf.db
|
|
Writes: 13 new tables created by 43_anton_import.sql + merges developers.
|
|
|
|
Skipped tables (duplicates of our existing data):
|
|
• regions, federal_districts (we have domrf_regions/_federal_districts)
|
|
• mortgage_rates (empty in Anton's DB, we have domrf_mortgage_rates)
|
|
• escrow_banks (we have domrf_escrow_banks, 28 vs 27)
|
|
• rc_classes/dev_size/city_pop (in our domrf_sold_ready_breakdown)
|
|
• rc_dynamics_rf (in our domrf_sold_ready_dynamics)
|
|
• egrn_deals (we have raw rosreestr_deals — 4.97M rows, far richer)
|
|
• _meta, _sources, sqlite_sequence, launch_5y(empty)
|
|
|
|
Special handling:
|
|
• developers (8 618) — merge missing rows into domrf_developers (we have 2 734)
|
|
"""
|
|
import os, sys, sqlite3, subprocess, datetime
|
|
|
|
SQLITE_PATH = r"C:\Users\user\Downloads\Telegram Desktop\db\db\domrf.db"
|
|
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_val(v):
|
|
if v is None:
|
|
return 'NULL'
|
|
if isinstance(v, bool):
|
|
return 'TRUE' if v else 'FALSE'
|
|
if isinstance(v, (int, float)):
|
|
if isinstance(v, float) and v != v: # NaN
|
|
return 'NULL'
|
|
return str(v)
|
|
return "'" + str(v).replace("'", "''") + "'"
|
|
|
|
|
|
def to_pg_bool(v):
|
|
if v is None:
|
|
return 'NULL'
|
|
return 'TRUE' if v else 'FALSE'
|
|
|
|
|
|
def to_pg_date(v):
|
|
"""SQLite stores dates as TEXT. Try parse YYYY-MM-DD; else NULL."""
|
|
if v is None or v == '':
|
|
return 'NULL'
|
|
s = str(v)[:10]
|
|
try:
|
|
datetime.date.fromisoformat(s)
|
|
return f"'{s}'"
|
|
except Exception:
|
|
return 'NULL'
|
|
|
|
|
|
def chunked_insert(rows, table, cols, conflict_cols=None, batch=500, do_update=False):
|
|
if not rows:
|
|
return 0
|
|
total = 0
|
|
on_conflict = ''
|
|
if conflict_cols:
|
|
if do_update:
|
|
update = ', '.join(f'{c}=EXCLUDED.{c}' for c in cols if c not in conflict_cols)
|
|
on_conflict = f' ON CONFLICT ({",".join(conflict_cols)}) DO UPDATE SET {update}'
|
|
else:
|
|
on_conflict = f' ON CONFLICT ({",".join(conflict_cols)}) DO NOTHING'
|
|
for i in range(0, len(rows), batch):
|
|
chunk = rows[i:i + batch]
|
|
sql = (f"INSERT INTO {table} ({','.join(cols)}) VALUES "
|
|
+ ','.join(chunk) + on_conflict + ';')
|
|
psql(sql)
|
|
total += len(chunk)
|
|
return total
|
|
|
|
|
|
def migrate_kn_flats(con):
|
|
rows = []
|
|
for r in con.execute('SELECT id, ods_id, type, number, is_studio, total_area, '
|
|
'living_area, rooms, status, price_rub, price_per_m2, '
|
|
'floor, num_floors, obj_id, city, region_cd, obj_name '
|
|
'FROM flats'):
|
|
(fid, ods, ftype, fnum, studio, ta, la, rooms, status, pr, ppm,
|
|
fl, nfl, obj_id, city, region, obj_name) = r
|
|
rows.append(
|
|
f"({fid},{esc_val(ods)},{esc_val(ftype)},{esc_val(fnum)},"
|
|
f"{to_pg_bool(studio)},{esc_val(ta)},{esc_val(la)},{esc_val(rooms)},"
|
|
f"{esc_val(status)},{esc_val(pr)},{esc_val(ppm)},{esc_val(fl)},"
|
|
f"{esc_val(nfl)},{esc_val(obj_id)},{esc_val(city)},{esc_val(region)},"
|
|
f"{esc_val(obj_name)},DEFAULT)"
|
|
)
|
|
return chunked_insert(
|
|
rows, 'domrf_kn_flats',
|
|
['id','ods_id','flat_type','flat_number','is_studio','total_area','living_area',
|
|
'rooms','status','price_rub','price_per_m2','floor','num_floors','obj_id',
|
|
'city','region_cd','obj_name','snapshot_date'],
|
|
conflict_cols=['id'], do_update=True,
|
|
)
|
|
|
|
|
|
def migrate_kn_objects(con):
|
|
rows = []
|
|
for r in con.execute('SELECT obj_id, hobj_id, comm_name, addr, short_addr, region_cd, '
|
|
'dev_id, dev_name, dev_inn, floor_min, floor_max, flat_count, '
|
|
'square_living, ready_dt, problem_flag, site_status, green_house, '
|
|
'escrow, obj_class, wall_type, energy_eff, latitude, longitude, '
|
|
'obj_status, is_prinzip, is_ekb, is_problem FROM objects'):
|
|
(oid, hid, cn, addr, sa, rc, did, dn, dinn, fmn, fmx, fc, sl, rdt,
|
|
pf, ss, gh, esc, oc, wt, ee, lat, lon, ostat, ipr, iekb, iprob) = r
|
|
rows.append(
|
|
f"({oid},{esc_val(hid)},{esc_val(cn)},{esc_val(addr)},{esc_val(sa)},"
|
|
f"{esc_val(rc)},{esc_val(did)},{esc_val(dn)},{esc_val(dinn)},"
|
|
f"{esc_val(fmn)},{esc_val(fmx)},{esc_val(fc)},{esc_val(sl)},"
|
|
f"{to_pg_date(rdt)},{esc_val(pf)},{esc_val(ss)},"
|
|
f"{to_pg_bool(gh)},{to_pg_bool(esc)},{esc_val(oc)},{esc_val(wt)},"
|
|
f"{esc_val(ee)},{esc_val(lat)},{esc_val(lon)},{esc_val(ostat)},"
|
|
f"{to_pg_bool(ipr)},{to_pg_bool(iekb)},{to_pg_bool(iprob)},DEFAULT)"
|
|
)
|
|
return chunked_insert(
|
|
rows, 'domrf_kn_objects',
|
|
['obj_id','hobj_id','comm_name','addr','short_addr','region_cd','dev_id','dev_name',
|
|
'dev_inn','floor_min','floor_max','flat_count','square_living','ready_dt',
|
|
'problem_flag','site_status','green_house','escrow','obj_class','wall_type',
|
|
'energy_eff','latitude','longitude','obj_status','is_prinzip','is_ekb','is_problem',
|
|
'snapshot_date'],
|
|
conflict_cols=['obj_id'], do_update=True,
|
|
)
|
|
|
|
|
|
def migrate_simple(con, src, dst, cols_src, cols_dst, conflict_cols, do_update=True,
|
|
transforms=None):
|
|
"""Generic migration: SELECT cols from src → INSERT into dst."""
|
|
transforms = transforms or {}
|
|
rows = []
|
|
for r in con.execute(f'SELECT {",".join(cols_src)} FROM "{src}"'):
|
|
vals = []
|
|
for col_name, raw in zip(cols_src, r):
|
|
if col_name in transforms:
|
|
vals.append(transforms[col_name](raw))
|
|
else:
|
|
vals.append(esc_val(raw))
|
|
rows.append('(' + ','.join(vals) + ')')
|
|
return chunked_insert(rows, dst, cols_dst, conflict_cols=conflict_cols, do_update=do_update)
|
|
|
|
|
|
def merge_developers(con):
|
|
"""Anton has 8 618 developers (incl. archived); we have 2 734.
|
|
INSERT only IDs that don't exist in our table — no update on existing."""
|
|
# Pull existing developer_ids from prod via psql
|
|
existing_ids = set()
|
|
res = psql('SELECT developer_id FROM domrf_developers;')
|
|
for line in res.splitlines():
|
|
line = line.strip()
|
|
if line and not line.startswith('---') and 'developer_id' not in line and 'rows' not in line and '(' not in line:
|
|
existing_ids.add(line)
|
|
print(f' existing developers in prod: {len(existing_ids)}')
|
|
|
|
new_rows = []
|
|
skipped = 0
|
|
cur = con.execute('SELECT id, name FROM developers WHERE id IS NOT NULL')
|
|
for did, name in cur:
|
|
if not did or not name:
|
|
continue
|
|
if str(did) in existing_ids:
|
|
skipped += 1
|
|
continue
|
|
new_rows.append(f"({esc_val(did)},{esc_val(name)})")
|
|
print(f' skipped (already in prod): {skipped}')
|
|
print(f' to INSERT (new from Anton): {len(new_rows)}')
|
|
|
|
return chunked_insert(
|
|
new_rows, 'domrf_developers', ['developer_id', 'developer_name'],
|
|
conflict_cols=['developer_id'], do_update=False,
|
|
)
|
|
|
|
|
|
def main():
|
|
con = sqlite3.connect(SQLITE_PATH)
|
|
|
|
print('--- domrf_kn_flats ---')
|
|
n = migrate_kn_flats(con)
|
|
print(f' loaded {n} flats')
|
|
|
|
print('--- domrf_kn_objects ---')
|
|
n = migrate_kn_objects(con)
|
|
print(f' loaded {n} objects')
|
|
|
|
print('--- cbr_mortgage_files ---')
|
|
n = migrate_simple(con, 'cbr_files', 'cbr_mortgage_files',
|
|
['filename', 'size_kb', 'description', 'url'],
|
|
['filename', 'size_kb', 'description', 'url'],
|
|
conflict_cols=['filename'])
|
|
print(f' loaded {n}')
|
|
|
|
print('--- cbr_mortgage_series ---')
|
|
n = migrate_simple(con, 'cbr_series', 'cbr_mortgage_series',
|
|
['series_id', 'region', 'period', 'value', 'title'],
|
|
['series_id', 'region', 'period', 'value', 'title'],
|
|
conflict_cols=['series_id', 'region', 'period'])
|
|
print(f' loaded {n}')
|
|
|
|
print('--- sber_ddf_metrics ---')
|
|
n = migrate_simple(con, 'ddf_metrics', 'sber_ddf_metrics',
|
|
['metric_key', 'region', 'value_2019', 'value_2025', 'unit', 'source'],
|
|
['metric_key', 'region', 'value_2019', 'value_2025', 'unit', 'source'],
|
|
conflict_cols=['metric_key'])
|
|
print(f' loaded {n}')
|
|
|
|
print('--- yandex_realty_zk ---')
|
|
n = migrate_simple(con, 'yandex_zk', 'yandex_realty_zk',
|
|
['yid', 'name', 'developer', 'developer_id', 'finished_obj', 'unfinished_obj',
|
|
'price_from', 'price_to', 'address', 'obj_class', 'latitude', 'longitude'],
|
|
['yid', 'name', 'developer', 'developer_id', 'finished_obj', 'unfinished_obj',
|
|
'price_from', 'price_to', 'address', 'obj_class', 'latitude', 'longitude'],
|
|
conflict_cols=['yid'])
|
|
print(f' loaded {n}')
|
|
|
|
print('--- yandex_realty_class_prices ---')
|
|
n = migrate_simple(con, 'yandex_class_prices', 'yandex_realty_class_prices',
|
|
['obj_class', 'total_sites', 'price_per_m2_min', 'price_per_m2_max'],
|
|
['obj_class', 'total_sites', 'price_per_m2_min', 'price_per_m2_max'],
|
|
conflict_cols=['obj_class'])
|
|
print(f' loaded {n}')
|
|
|
|
print('--- ekb_districts ---')
|
|
n = migrate_simple(con, 'ekb_districts', 'ekb_districts',
|
|
['name', 'population', 'zk_count', 'flat_count', 'area_m2',
|
|
'median_price_per_m2', 'mean_price_per_m2'],
|
|
['district_name', 'population', 'zk_count', 'flat_count', 'area_m2',
|
|
'median_price_per_m2', 'mean_price_per_m2'],
|
|
conflict_cols=['district_name'])
|
|
print(f' loaded {n}')
|
|
|
|
print('--- ekb_metro ---')
|
|
n = migrate_simple(con, 'ekb_metro', 'ekb_metro',
|
|
['station_name', 'zk_count', 'median_price_per_m2'],
|
|
['station_name', 'zk_count', 'median_price_per_m2'],
|
|
conflict_cols=['station_name'])
|
|
print(f' loaded {n}')
|
|
|
|
print('--- sv_cities ---')
|
|
n = migrate_simple(con, 'sv_cities', 'sv_cities',
|
|
['city_name', 'zk_count', 'flat_count', 'area_m2', 'median_price_per_m2'],
|
|
['city_name', 'zk_count', 'flat_count', 'area_m2', 'median_price_per_m2'],
|
|
conflict_cols=['city_name'])
|
|
print(f' loaded {n}')
|
|
|
|
print('--- sv_dev_sales ---')
|
|
n = migrate_simple(con, 'sv_dev_sales', 'sv_dev_sales',
|
|
['rank', 'dev_id', 'name', 'total_square', 'sold_perc', 'unsold_perc',
|
|
'unopened_perc', 'rep_period'],
|
|
['rank', 'dev_id', 'name', 'total_square', 'sold_perc', 'unsold_perc',
|
|
'unopened_perc', 'rep_period'],
|
|
conflict_cols=['dev_id', 'rep_period'])
|
|
print(f' loaded {n}')
|
|
|
|
print('--- domrf_dev_growth ---')
|
|
n = migrate_simple(con, 'dev_growth', 'domrf_dev_growth',
|
|
['dev_id', 'dev_name', 'area_2024', 'area_2025', 'delta_pct', 'region'],
|
|
['dev_id', 'dev_name', 'area_2024', 'area_2025', 'delta_pct', 'region'],
|
|
conflict_cols=['dev_name', 'region'])
|
|
print(f' loaded {n}')
|
|
|
|
print('--- rosreestr_rent_deals ---')
|
|
rows = []
|
|
for r in con.execute('SELECT quarter, obj_type, cad_quarter, area, rent_rub, '
|
|
'rent_per_m2, duration_months, region_cd FROM rent_deals'):
|
|
rows.append('(' + ','.join(esc_val(v) for v in r) + ')')
|
|
n = chunked_insert(
|
|
rows, 'rosreestr_rent_deals',
|
|
['quarter','obj_type','cad_quarter','area','rent_rub','rent_per_m2',
|
|
'duration_months','region_cd'],
|
|
conflict_cols=None, # no PK conflict — id is BIGSERIAL
|
|
)
|
|
print(f' loaded {n}')
|
|
|
|
print('--- merge developers ---')
|
|
n = merge_developers(con)
|
|
print(f' inserted {n} new developers')
|
|
|
|
con.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|