接入灰度订单关联物流节点并区分预计与实际轨迹

2026-10-30 10:13:11 +08:00
parent ce445d0fb8
commit 6a428bd468
10 changed files with 464 additions and 47 deletions
@@ -0,0 +1,113 @@
+"""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()