#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""生成「工作台物流信息缺失清单」自包含 HTML —— 按原因分类 + 责任销售分组 + 合并发出关联。
数据源：数据/shipment_status.json（唯一授权来源）+ 数据/mcp_orders_input.json（补飞书备注）
输出：销售分析/工作台物流信息缺失清单_<日期>.html
"""
import json, io, os, sys, time, re
from collections import defaultdict
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.join(ROOT, '脚本'))
import brand_kit as BK
D = json.load(io.open(os.path.join(ROOT, '数据', 'shipment_status.json'), encoding='utf-8'))
O = D['orders']
try:
_M = json.load(io.open(os.path.join(ROOT, '数据', 'mcp_orders_input.json'), encoding='utf-8'))
REMARK = {r.get('单据编号'): ((r.get('备注内部') or ''), (r.get('备注客户') or '')) for r in _M}
except Exception:
REMARK = {}
TODAY = time.strftime('%Y%m%d')
OUT = os.path.join(ROOT, '销售分析', '工作台物流信息缺失清单_%s.html' % TODAY)
# ── 分类 ──
g_merge, g_notn, g_phone, g_noroute = [], [], [], []
for o in O:
tn = (o.get('trackingNo') or '').strip()
s = o.get('shipment') or {}
reason = str(s.get('reason') or '')
if not tn:
(g_merge if o.get('mergedInto') else g_notn).append(o)
elif s.get('status') == 'phone_mismatch' or '手机号校验失败' in reason:
g_phone.append(o)
elif not s.get('routes'):
g_noroute.append(o)
# ── 疑似合并（同客户 + 同发货日，且对方有运单号）—— 只提示，不自动关联 ──
_key = {}
for o in O:
if (o.get('trackingNo') or '').strip():
_key.setdefault((o.get('customer') or '', o.get('planShipDate') or ''), []).append(o)
g_suspect = []
for o in g_notn:
k = (o.get('customer') or '', o.get('planShipDate') or '')
if k in _key:
g_suspect.append((o, _key[k][0]))
_suspect_ids = set(x[0]['orderId'] for x in g_suspect)
g_notn = [o for o in g_notn if o['orderId'] not in _suspect_ids]
def esc(x):
return (str(x if x is not None else '')
.replace('&', '&').replace('<', '<').replace('>', '>').replace('"', '"'))
def rmk(o, n=64):
ri, rc = REMARK.get(o.get('orderId'), (o.get('remarkInternal') or '', o.get('remarkCustomer') or ''))
t = re.sub(r'\s+', ' ', (ri or rc or '')).strip()
return t[:n] + ('…' if len(t) > n else '')
def by_creator(lst):
d = defaultdict(list)
for o in lst:
d[o.get('creator') or '(无下单人)'].append(o)
return dict(sorted(d.items(), key=lambda kv: -len(kv[1])))
def rows(lst, action, extra=None):
out = []
for o in sorted(lst, key=lambda x: (x.get('creator') or '', x.get('orderId') or '')):
out.append(
'
'
'
%s
%s
%s
%s
'
'
%s
%s
%s
%s
'
'
%s
' % (
esc(o.get('orderId')), esc(o.get('customer')), esc(o.get('customer')),
esc(o.get('creator') or '—'), esc(o.get('dept') or '—'),
esc(o.get('recipient') or '—'), esc(o.get('phone') or '—'),
esc(o.get('trackingNo') or '—'),
(extra(o) if extra else ''), esc(action)))
return ''.join(out)
def section(title, icon, lst, action, color, note, extra=None):
if not lst:
return ''
grp = by_creator(lst)
grp_rows = ''.join(
'
👤 %s %d 单
%s'
% (esc(k), len(v), rows(v, action, extra)) for k, v in grp.items())
return (
''
'
' % (
esc(mi.get('orderId')), esc(mi.get('trackingNo') or ''),
esc(mi.get('statusLabel') or ''), esc(rmk(o))))
def _plain_extra(o):
return '
%s
' % (esc(rmk(o)) or '无备注')
def _suspect_extra(o):
b = next((p for a, p in g_suspect if a['orderId'] == o['orderId']), {}) or {}
return ('? 疑似并入 %s%s'
'
%s
' % (esc(b.get('orderId') or ''),
esc(b.get('trackingNo') or ''), esc(rmk(o))))
n_all = len(O)
total_bad = len(g_merge) + len(g_notn) + len(g_phone) + len(g_noroute) + len(g_suspect)
html = '''
工作台物流信息缺失清单