"""Read order-linked warehouse and shipping facts into an explicit private snapshot. Requires the optional PyMySQL package. This command never writes to the source DB. """ import argparse import json from pathlib import Path import sys sys.path.insert(0, str(Path(__file__).resolve().parents[1])) def export_tracking(connection, orders): output = {} for order in orders: key = (str(order.get('tenant_id') or ''), str(order.get('source_order_id') or '')) if not all(value.isdecimal() for value in key) or not order.get('order_no'): raise ValueError('Source snapshot requires numeric tenant_id/source_order_id and order_no') if key in output: raise ValueError('Source snapshot contains duplicate order identities') output[key] = {key_name: order[key_name] for key_name in ('tenant_id', 'source_order_id', 'order_no')} output[key]['tracking_json'] = [] ids = sorted({key[1] for key in output}) for start in range(0, len(ids), 100): batch = ids[start:start + 100] placeholders = ','.join('%s' for _ in batch) # Every relation uses both source order identity and customer identity. queries = [ (f'''SELECT o.tenant_id,o.id AS order_id, i.id AS fact_id, i.inbounded_at AS actual_at,i.updated_at AS observed_at FROM baiyue_order.ord_order o JOIN baiyue_firstleg.fl_inbound_order i ON i.order_id=o.id AND i.tenant_id=o.tenant_id WHERE o.deleted=0 AND i.deleted=0 AND i.inbounded_at IS NOT NULL AND o.id IN ({placeholders}) ORDER BY i.id''', 'inbound'), (f'''SELECT DISTINCT o.tenant_id,o.id AS order_id,i.id AS fact_id, i.frozen_at AS actual_at,i.frozen_at AS observed_at, p.container_booking_id,b.container_no FROM baiyue_order.ord_order o JOIN baiyue_firstleg.fl_loading_plan_item i ON i.order_id=o.id AND i.account_id=o.tenant_id AND i.deleted=0 JOIN baiyue_firstleg.fl_loading_plan p ON p.id=i.loading_plan_id AND p.plan_status='COMPLETED' JOIN baiyue_firstleg.fl_container_booking b ON b.id=p.container_booking_id WHERE o.deleted=0 AND i.frozen_at IS NOT NULL AND o.id IN ({placeholders}) ORDER BY i.id''', 'loading'), (f'''SELECT DISTINCT o.tenant_id,o.id AS order_id,n.id AS fact_id,n.standard_node, n.actual_at,n.estimated_at,n.observed_at,n.created_at,n.location_name, n.container_booking_id,b.container_no FROM baiyue_order.ord_order o JOIN baiyue_firstleg.fl_loading_plan_item i ON i.order_id=o.id AND i.account_id=o.tenant_id AND i.deleted=0 AND i.frozen_at IS NOT NULL JOIN baiyue_firstleg.fl_loading_plan p ON p.id=i.loading_plan_id AND p.plan_status='COMPLETED' JOIN baiyue_firstleg.fl_container_booking b ON b.id=p.container_booking_id JOIN baiyue_firstleg.fl_carrier_schedule_node_fact n ON n.container_booking_id=b.id WHERE o.deleted=0 AND o.id IN ({placeholders}) ORDER BY n.id''', 'shipping'), ] for sql, source in queries: with connection.cursor() as cursor: cursor.execute(sql, batch) for row in cursor.fetchall(): key = (str(row['tenant_id']), str(row['order_id'])) if key not in output: continue fact = {'event_id': source + ':' + str(row['fact_id']), 'code': {'inbound': 'DOMESTIC_RECEIVED', 'loading': 'LOADING_COMPLETED'}.get(source) or row['standard_node'], 'shipment_key': str(row.get('container_booking_id') or ''), 'shipment_no': row.get('container_no') or '', 'place': row.get('location_name') or ''} for field in ('actual_at', 'estimated_at', 'observed_at', 'created_at'): value = row.get(field) fact[field] = value.isoformat(sep=' ') if value else None output[key]['tracking_json'].append(fact) return list(output.values()) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--database-env', type=Path, required=True) parser.add_argument('--orders-snapshot', type=Path, required=True) parser.add_argument('--output', type=Path, required=True) args = parser.parse_args() if args.output.exists(): parser.error('Output already exists; choose a new private snapshot path') if not args.output.parent.is_dir(): parser.error('Output parent directory must exist') if not args.orders_snapshot.is_file() or args.orders_snapshot.stat().st_size > 10 * 1024 * 1024: parser.error('Orders snapshot must be an existing JSON file smaller than 10 MB') orders = json.loads(args.orders_snapshot.read_text(encoding='utf-8-sig')) if isinstance(orders, dict): orders = orders.get('orders') if not isinstance(orders, list) or not orders or len(orders) > 3000: parser.error('Orders snapshot must contain 1 to 3000 orders') import pymysql from dotenv import dotenv_values config = dotenv_values(args.database_env, interpolate=False, encoding='utf-8-sig') required = ('CHONGCHONG_DB_HOST', 'CHONGCHONG_DB_PORT', 'CHONGCHONG_DB_USERNAME', 'CHONGCHONG_DB_PASSWORD') if any(not config.get(key) for key in required): parser.error('Database env is missing connection settings') connection = pymysql.connect(host=config[required[0]], port=int(config[required[1]]), user=config[required[2]], password=config[required[3]], cursorclass=pymysql.cursors.DictCursor, connect_timeout=10, read_timeout=30, write_timeout=10, autocommit=False) try: with connection.cursor() as cursor: cursor.execute('SET SESSION TRANSACTION READ ONLY') cursor.execute('START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY') records = export_tracking(connection, orders) finally: connection.rollback() connection.close() args.output.write_text(json.dumps({'orders': records}, ensure_ascii=False, indent=2), encoding='utf-8') print(json.dumps({'orders': len(records), 'orders_with_tracking': sum(bool(row['tracking_json']) for row in records), 'facts': sum(len(row['tracking_json']) for row in records)})) if __name__ == '__main__': main()