#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""验证 shipment_pipeline 的「降级保护」真的生效 —— 不碰生产数据，用临时文件跑。

断言：
  ① 历史有轨迹、本轮查不到  → 必须沿用历史（routes 保留 + stale=True），绝不降级成 no_data
  ② 历史无轨迹、本轮查不到  → 如实写 no_data（不伪造）
  ③ 历史有轨迹、本轮查到了  → 用新数据覆盖（正常更新）
  ④ 手机号校验失败且无历史  → 单独标 phone_mismatch
"""
import json, io, os, sys, tempfile, importlib.util

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
spec = importlib.util.spec_from_file_location('sp', os.path.join(ROOT, '脚本', 'shipment_pipeline.py'))
sp = importlib.util.module_from_spec(spec); spec.loader.exec_module(sp)

TMP = tempfile.mkdtemp(prefix='sptest_')
FAKE = os.path.join(TMP, 'shipment_status.json')

GOOD_ROUTES = [{'acceptTime': '2026-09-28 15:37:06', 'opCode': '80', 'remark': '已签收', 'acceptAddress': '宁波市'}]
prev = {'updated_at': 'x', 'total': 4, 'orders': [
    {'orderId': 'A1', 'trackingNo': 'SFAAA', 'shipment': {'mailNo': 'SFAAA', 'status': 'delivered',
        'statusLabel': '已签收', 'routes': GOOD_ROUTES, 'routeCount': 1}},
    {'orderId': 'A2', 'trackingNo': 'SFBBB', 'shipment': {'mailNo': 'SFBBB', 'status': 'no_data',
        'statusLabel': '暂无数据', 'routes': [], 'routeCount': 0}},
    {'orderId': 'A3', 'trackingNo': 'SFCCC', 'shipment': {'mailNo': 'SFCCC', 'status': 'delivered',
        'statusLabel': '已签收', 'routes': GOOD_ROUTES, 'routeCount': 1}},
    {'orderId': 'A4', 'trackingNo': 'SFDDD', 'shipment': {'mailNo': 'SFDDD', 'status': 'no_data',
        'statusLabel': '暂无数据', 'routes': [], 'routeCount': 0}},
]}
io.open(FAKE, 'w', encoding='utf-8').write(json.dumps(prev, ensure_ascii=False))
sp.OUTPUT = FAKE

def mk(oid, tn, phone='13800000000'):
    return {'单据编号': oid, '物流单号': tn, '收货人联系电话#': phone, '创建时间': '2026-09-21',
            '购货单位#': '测试客户', '创建者': '测试', '创建者邮箱': 't@b.com'}

# 伪造顺丰返回：SFA/SFB 查不到；SFC 查到；SFD 手机号校验失败
def fake_query(tns, tok, phone=None):
    tn = tns[0]
    if tn == 'SFCCC':
        return {'success': True, 'msgData': {'routeResps': [{'mailNo': tn, 'routes': GOOD_ROUTES}]}}
    if tn == 'SFDDD':
        return {'success': True, 'msgData': {'routeResps': [{'mailNo': tn, 'routes': [], 'reasonRemark': '手机号校验失败'}]}}
    return {'success': True, 'msgData': {'routeResps': [{'mailNo': tn, 'routes': [], 'reasonRemark': '暂无路由信息'}]}}

sp.sf_query_routes = fake_query
sp.attach_erp_ship_dates = lambda out: out          # 跳过 ERP 补挂，聚焦本次逻辑

out = sp.process_orders([mk('A1','SFAAA'), mk('A2','SFBBB'), mk('A3','SFCCC'), mk('A4','SFDDD')], 'fake-token')
by = {o['orderId']: o for o in out}

def chk(name, cond, detail=''):
    print('  %s %-46s %s' % ('✅' if cond else '❌', name, detail))
    return cond

print('=== 降级保护验证 ===')
s1 = by['A1']['shipment']
ok1 = chk('① 历史已签收 + 本轮查不到 → 沿用历史', s1.get('status') == 'delivered' and len(s1.get('routes') or []) == 1 and s1.get('stale'),
          'status=%s routes=%d stale=%s' % (s1.get('status'), len(s1.get('routes') or []), s1.get('stale')))
s2 = by['A2']['shipment']
ok2 = chk('② 历史无轨迹 + 本轮查不到 → 如实 no_data', s2.get('status') == 'no_data' and not s2.get('routes'),
          'status=%s routes=%d' % (s2.get('status'), len(s2.get('routes') or [])))
s3 = by['A3']['shipment']
ok3 = chk('③ 历史已签收 + 本轮查到 → 用新数据', s3.get('status') == 'delivered' and len(s3.get('routes') or []) == 1 and not s3.get('stale'),
          'status=%s stale=%s' % (s3.get('status'), s3.get('stale')))
s4 = by['A4']['shipment']
ok4 = chk('④ 手机号校验失败 + 无历史 → 单独标记', s4.get('status') == 'phone_mismatch',
          'status=%s label=%s' % (s4.get('status'), s4.get('statusLabel')))

print()
print('  生产文件未被触碰:', sp.OUTPUT != os.path.join(ROOT, '数据', 'shipment_status.json'))
print('  结论:', '✅ 全部通过' if all([ok1, ok2, ok3, ok4]) else '❌ 有断言失败')
sys.exit(0 if all([ok1, ok2, ok3, ok4]) else 1)