#!/usr/bin/env python3 """Read-only archive conversion. Output is a checksummed, source-qualified NDJSON snapshot. Run: python3 scripts/propertylab/prepare_research_import.py --sqlite ARCHIVE --output DIR No source records are deduplicated; row IDs identify displayed source observations. """ import argparse import contextlib import datetime as dt import hashlib import json import math import pathlib import re import sqlite3 TABLES = ['schemes', 'scheme_snapshot_values', 'sale_observations', 'monthly_observations', 'places', 'source_crosswalks'] def file_sha256(path): with path.open('rb') as handle: return hashlib.file_digest(handle, 'sha256').hexdigest() def digest(value): return hashlib.sha256(value.encode('utf-8')).hexdigest() def positive(value): if isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value) and value > 0: return value return None def month(value): return value if isinstance(value, str) and re.fullmatch(r'\d{4}-(0[1-9]|1[0-2])', value) else None def coordinates(row): lat, lng = row.get('lat'), row.get('lon', row.get('lng')) if not isinstance(lat, (int, float)) or not isinstance(lng, (int, float)) or not math.isfinite(lat + lng) or not -90 <= lat <= 90 or not -180 <= lng <= 180: raise ValueError('Invalid source coordinate') return {'latitude': lat, 'longitude': lng} def convert(source, output, collected_at): source = pathlib.Path(source).resolve() output = pathlib.Path(output).resolve() output.mkdir(parents=True, exist_ok=True) if (output / 'manifest.json').exists(): raise ValueError('Output already contains a completed manifest; use a fresh directory') with contextlib.ExitStack() as resources: archive_hash = file_sha256(source) snapshot = digest('propertylab-ingest-v1:' + archive_hash + ':' + collected_at) conn = sqlite3.connect(source.as_uri() + '?mode=ro', uri=True) resources.callback(conn.close) def records(dataset): for rid, payload in conn.execute('SELECT record_id,payload FROM records WHERE dataset=? ORDER BY record_id COLLATE BINARY', (dataset,)): yield rid, json.loads(payload) def keyed(dataset, key): return {p[key]: (rid, p) for rid, p in records(dataset)} checks = keyed('history_name_checks', 'requested_scheme') directory = keyed('scheme_directory', 'scheme') histories = keyed('distinct_history_results', 'scheme') handles = {name: resources.enter_context((output / (name + '.ndjson')).open('w', encoding='utf-8')) for name in TABLES} counts = {name: 0 for name in TABLES} def emit(table, row): handles[table].write(json.dumps(row, ensure_ascii=False, allow_nan=False, separators=(',', ':')) + '\n') counts[table] += 1 def ref(dataset, rid, payload=None, kind='scheme_aggregate'): payload = payload or {} return {'snapshotId': snapshot, 'dataset': dataset, 'sourceRecordId': rid, 'observationKind': kind, 'sourceUrl': payload.get('source_url'), 'pageContentId': payload.get('page_content_id'), 'rowPosition': payload.get('row_position'), 'archiveSha256': archive_hash} def crosswalk(entity, kind, dataset, rid, status='source_identity'): emit('source_crosswalks', {'id': digest(snapshot + ':' + dataset + ':' + rid), 'snapshot_id': snapshot, 'entity_id': entity, 'entity_kind': kind, 'source_dataset': dataset, 'source_record_id': rid, 'match_status': status, 'audit': {'identity': 'source record identity, not unique transaction identity'}}) schemes = {} names = {} withheld = 0 for rid, p in records('property_map'): name = p['scheme']; sid = digest(name)[:16] if sid in names and names[sid] != name: raise ValueError('Scheme ID collision: ' + sid) names[sid] = name; schemes[name] = sid check = checks.get(name, ('', {}))[1] exact = check.get('result') == 'exact_name_match' and check.get('returned_scheme') == name status = 'exact_name_match' if exact else ('unverified_scheme_link' if check.get('result') == 'exact_name_match' else check.get('result', 'unverified')) if not exact: withheld += 1 history = histories.get(name, ('', {}))[1] if exact else {} address = next((x.get('address', {}) for x in history.get('structured_data', []) if x.get('@type') == 'Place'), {}) dr = directory.get(name, ('', {}))[1] emit('schemes', {'scheme_id': sid, 'display_name': name, 'name_hash': digest(name), 'category': p['cls'], 'state': address.get('addressRegion') or None, 'district': dr.get('district'), 'mukim': dr.get('mukim'), **coordinates(p), 'precision': dr.get('level') or 'scheme', 'source_ref': ref('property_map', rid)}) emit('scheme_snapshot_values', {'snapshot_id': snapshot, 'scheme_id': sid, 'median_rm': positive(p.get('median_rm')), 'reported_psf': positive(p.get('psf')), 'source_n': p.get('n'), 'history_status': status, 'source_ref': ref('property_map', rid)}) crosswalk(sid, 'scheme', 'property_map', rid, 'exact_name') periods = [] for rid, p in records('monthly_property_summaries'): if p['scheme'] not in schemes: continue period = month(p.get('month')) if not period: raise ValueError('Invalid monthly observation period: ' + rid) periods.append(period) emit('monthly_observations', {'observation_id': digest(snapshot + ':monthly_property_summaries:' + rid), 'snapshot_id': snapshot, 'scheme_id': schemes[p['scheme']], 'month': period, 'source_n': p.get('n'), 'price_rm': positive(p.get('median_rm')), 'reported_psf': positive(p.get('psf')), 'quality_flags': ['source_psf_basis_unknown', 'aggregate_count_window_unconfirmed'], 'source_ref': ref('monthly_property_summaries', rid, p, 'scheme_month')}) for rid, p in records('recent_sales_distinct_pages'): sid = schemes.get(p['scheme']) check = checks.get(p['scheme'], ('', {}))[1] exact = sid is not None and check.get('result') == 'exact_name_match' and check.get('returned_scheme') == p['scheme'] status = 'exact_name_match' if exact else ('scheme_not_in_map' if sid is None else 'unverified_scheme_link') flags = ['source_area_basis_unknown', 'displayed_row_not_unique_transaction'] if not exact: flags.append(status) for key in ['price_rm', 'area_sqm', 'psf_rm']: if positive(p.get(key)) is None: flags.append('invalid_' + key) if month(p.get('month')) is None: flags.append('invalid_month') emit('sale_observations', {'sale_row_id': digest(snapshot + ':recent_sales_distinct_pages:' + rid), 'snapshot_id': snapshot, 'scheme_id': sid, 'month': month(p.get('month')), 'area_sqm': positive(p.get('area_sqm')), 'area_basis': 'unknown', 'price_rm': positive(p.get('price_rm')), 'reported_psf': positive(p.get('psf_rm')), 'link_status': status, 'quality_flags': flags, 'source_ref': ref('recent_sales_distinct_pages', rid, p, 'displayed_sale_row')}) for dataset in ['amenities', 'rail_stations']: for rid, p in records(dataset): ident = digest(snapshot + ':' + dataset + ':' + rid) emit('places', {'place_id': ident, 'snapshot_id': snapshot, 'name': p['name'], 'category': 'rail' if dataset == 'rail_stations' else p['kind'], 'subcategory': p.get('kind') if dataset == 'rail_stations' else p.get('sub'), **coordinates(p), 'rail_status': p.get('status') if dataset == 'rail_stations' else None, 'source_ref': ref(dataset, rid, p, 'place')}) for f in handles.values(): f.close() files = {name: {'file': name + '.ndjson', 'count': counts[name], 'sha256': file_sha256(output / (name + '.ndjson'))} for name in TABLES} manifest = {'schemaVersion': 1, 'snapshotId': snapshot, 'collectedAt': collected_at, 'coverageStart': min(periods), 'coverageEnd': max(periods), 'archiveSha256': archive_hash, 'methodVersion': 'propertylab-ingest-v1', 'source': 'PropertyLab uploaded archive; upstream attribution and URLs retained in source references', 'historyLinksWithheld': withheld, 'sourceInventory': [{'dataset': d, 'records': n} for d, n in conn.execute('SELECT dataset,COUNT(*) FROM records GROUP BY dataset')], 'files': files} (output / 'manifest.json').write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding='utf-8') conn.close() return manifest if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--sqlite', required=True) parser.add_argument('--output', required=True) parser.add_argument('--collected-at', default='2026-09-07') args = parser.parse_args() dt.date.fromisoformat(args.collected_at) print(json.dumps(convert(args.sqlite, args.output, args.collected_at), indent=2))