#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 顺丰物流「按需刷新」服务 —— 供订单管理页「🔄 刷新物流」按钮 + 进页自动刷新调用。 设计要点 -------- 1. 只查未签收运单（当前 ~74 条），**逐条**查询（顺丰接口批量传多号时只校验第一个，禁止批量） → 实测 12 秒跑完。 2. **直接返回刷新结果给前端合并**：工作台走 CDN 时读的是 CDN 上那份 shipment_status.json， 光写本地文件前端是看不到的 → 返回体里带 orders[]，前端按 trackingNo 合并进内存后重渲染。 3. 同时写回本地 `数据/shipment_status.json`（原子替换），让 2 小时一次的 cb_sync.sh 把最新状态带上 CDN，惠及其它消费者（日报等）。 4. 并发保护：文件锁（两个销售同时点只跑一次）+ 冷却（默认 120s 内复用上次结果，force=1 跳过）。 5. ⚠️ 故意**不写** `~/.hermes/scripts/sf_watchdog_state.json`：那是有状态推送的记忆文件， 由 cron 的 watchdog 负责。若这里先写了状态，watchdog 会认为"已推过"而吞掉 「已签收 / 派送中」的飞书通知 —— 销售就收不到提醒了。 """ import os import sys import json import time import importlib.util BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) SHIP = os.path.join(BASE, '数据', 'shipment_status.json') LOCK = os.path.join(BASE, '数据', '.shipment_refresh.lock') LAST = os.path.join(BASE, '数据', '.shipment_refresh_last.json') WATCHDOG = os.path.expanduser('~/.hermes/scripts/sf_delivery_watchdog.py') COOLDOWN_SEC = 120 QUERY_GAP = 0.02 # 逐条查询间的极小间隔（顺丰单条往返约 0.18s；不批量是为了绕开 checkPhoneNo 只校验首单的坑） _wd = None def _watchdog(): """复用 watchdog 的顺丰查询与状态判定（opCode 优先 + 已派送成功≠派送中的铁律都在那边）""" global _wd if _wd is None: spec = importlib.util.spec_from_file_location('sf_watchdog', WATCHDOG) m = importlib.util.module_from_spec(spec) spec.loader.exec_module(m) _wd = m return _wd def _read_json(path, default=None): try: with open(path, encoding='utf-8') as f: return json.load(f) except Exception: return default def _write_json_atomic(path, obj): tmp = path + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump(obj, f, ensure_ascii=False) os.replace(tmp, path) def _load_last(): d = _read_json(LAST) or {} return d if isinstance(d, dict) else {} def summary(): """当前物流数据的新鲜度（不触发查询）""" ship = _read_json(SHIP) or {} orders = ship.get('orders') or [] pend = [o for o in orders if (o.get('trackingNo') or '').strip() and (o.get('shipment') or {}).get('status') != 'delivered'] last = _load_last() return { 'ok': True, 'updated_at': ship.get('updated_at') or '', 'total': len(orders), 'pending': len(pend), 'last_refresh_at': last.get('at') or '', 'last_elapsed': last.get('elapsed'), } class _Busy(Exception): pass class _FileLock(object): def __init__(self, path): self.path = path self.fh = None def __enter__(self): self.fh = open(self.path, 'w') try: import fcntl fcntl.flock(self.fh, fcntl.LOCK_EX | fcntl.LOCK_NB) except Exception: self.fh.close() self.fh = None raise _Busy() return self def __exit__(self, *a): try: import fcntl if self.fh: fcntl.flock(self.fh, fcntl.LOCK_UN) except Exception: pass try: if self.fh: self.fh.close() except Exception: pass def refresh(force=False): """刷新未签收运单的顺丰轨迹。返回 {ok, checked, ok_count, changed, orders[...], ...}""" last = _load_last() now = time.time() # 冷却：120 秒内复用上次结果（多个销售同时进页面不会把顺丰打爆） if not force and last.get('orders') and (now - float(last.get('ts') or 0)) < COOLDOWN_SEC: out = dict(last) out['ok'] = True out['cooldown'] = True out['age'] = round(now - float(last.get('ts') or 0), 1) return out try: with _FileLock(LOCK): return _refresh_locked(force) except _Busy: # 另一个请求正在查（约 12 秒），把上次结果给它，前端继续用旧数据即可 if last.get('orders'): out = dict(last) out['ok'] = True out['busy'] = True out['cooldown'] = True return out return {'ok': False, 'busy': True, 'error': '正在刷新中，请稍候重试'} def _ensure_plan_dates(orders): """兜底：任何写回 shipment_status.json 的路径都保证 ERP 预计发货日期字段在。 教训（2026-09-23）：顺丰 12 点 cron 的管道子进程在我改码前启动，写出无 planShipDate 的版本， watchdog 又把它写回 → 页面「预计发货」列全空。所有写回路径都要兜一层。 """ try: import sys as _s, os as _o _d = _o.path.dirname(_o.path.abspath(__file__)) if _d not in _s.path: _s.path.insert(0, _d) import shipment_pipeline as _sp _sp.attach_erp_ship_dates(orders) except Exception as _e: print('⚠️ 补挂 ERP 发货日期失败:', _e) def _refresh_locked(force): wd = _watchdog() ship = _read_json(SHIP) or {} orders = ship.get('orders') or [] pending = [o for o in orders if (o.get('trackingNo') or '').strip() and (o.get('shipment') or {}).get('status') != 'delivered'] t0 = time.time() token = wd.sf_get_token() if not token: return {'ok': False, 'error': '顺丰 token 获取失败'} results = {} # tn -> (fine, last_route, routes) n_ok = 0 for o in pending: tn = (o.get('trackingNo') or '').strip() try: raw = wd.sf_query_routes([tn], token, wd.phone4(o)) resp = (raw.get('msgData', {}).get('routeResps', []) or [{}])[0] if raw.get('success') else {} fine, lastr = wd.fine_status(resp.get('routes', []), resp.get('reasonRemark', '')) if fine not in ('no_data', 'unknown'): results[tn] = (fine, lastr, resp.get('routes', []) or []) n_ok += 1 except Exception: pass time.sleep(QUERY_GAP) changed = [] out_orders = [] tn2pending = {(o.get('trackingNo') or '').strip(): o for o in pending} for tn, (fine, lastr, routes) in results.items(): coarse, label = wd.COARSE.get(fine, (None, None)) if not coarse: continue o = tn2pending.get(tn) if not o: continue old = (o.get('shipment') or {}).get('status') or 'none' sh = o.get('shipment') or {} sh['mailNo'] = tn sh['status'] = coarse sh['statusLabel'] = label if lastr: sh['lastRoute'] = lastr if routes: sh['routes'] = routes sh['routeCount'] = len(routes) sh['fine'] = fine o['shipment'] = sh if coarse != old: changed.append({'trackingNo': tn, 'from': old, 'to': coarse, 'label': label, 'orderId': o.get('orderId') or '', 'customer': o.get('customer') or '', 'creator': o.get('creator') or '', 'remark': (lastr or {}).get('remark', '')[:60]}) out_orders.append({'trackingNo': tn, 'status': coarse, 'statusLabel': label, 'fine': fine, 'lastRoute': lastr, 'routeCount': sh.get('routeCount') or 0, 'orderId': o.get('orderId') or '', 'updatedAt': time.strftime('%Y-%m-%d %H:%M:%S')}) if results: _ensure_plan_dates(orders) ship['updated_at'] = time.strftime('%Y-%m-%d %H:%M:%S') _write_json_atomic(SHIP, ship) payload = { 'ok': True, 'updated_at': ship.get('updated_at') or '', 'checked': len(pending), 'ok_count': n_ok, 'no_route': len(pending) - n_ok, 'changed_count': len(changed), 'changed': changed, 'orders': out_orders, 'elapsed': round(time.time() - t0, 1), 'force': bool(force), 'at': time.strftime('%Y-%m-%d %H:%M:%S'), 'ts': time.time(), } try: _write_json_atomic(LAST, payload) except Exception: pass return payload if __name__ == '__main__': force = '--force' in sys.argv r = refresh(force=force) print(json.dumps({k: v for k, v in r.items() if k not in ('orders', 'changed')}, ensure_ascii=False, indent=2)) for c in (r.get('changed') or [])[:12]: print(' %-16s %s → %s %s' % (c['trackingNo'], c['from'], c['to'], c['remark']))