服务时效差异审查

{
  "cutoff": "2026-09-15T00:00:00Z",
  "late_target": 0.1,
  "min_sample": 15,
  "review_gap_pp": 5,
  "priority_gap_pp": 15,
  "standard_weight": 0.5,
  "express_weight": 0.5
}

数据字典与口径

业务工作单

你加入冷链履约运营项目组。运营负责人需要回答：哪些订单与站点需要先处理、导出接口哪里存在控制缺口，以及下一版工作台应交付什么。四个站点为海港、东城、西岭、南湾。standard 承诺 24 h，express 承诺 8 h；具体截止时间以 orders.csv 的 due_at 为准。

材料覆盖 2026 年 9 月的订单及事件，统一复盘截止点为 rules.json 的 cutoff。所有业务时间转 UTC；本地显示时可以换时区，但计算不能把无时区字符串当成本地时间。输入可包含截止点之后的事件，用于检查是否错误使用未来信息。

orders.csv

• order_id：唯一订单号，字符串，不转成数字。
• hub：站点名；service：standard／express；route：线路编号。
• created_at：带时区的下单时刻；due_at：承诺交付时刻，UTC ISO 8601。
• 一行代表一个订单，事件不等于订单；不要按事件行数统计订单。

两个事件源

• A：event_id → event_id；order_ref → order_id；hub → hub；service → service；event → kind；event_time → happened_at。
• B：id → event_id；shipment.number → order_id；shipment.station → hub；shipment.product → service；action 的 OPEN／COLLECT／DROP／VOID 映射 created／picked_up／delivered／cancelled；at_ms 为整数 Unix 毫秒，输出为 UTC ISO 时间。
• 标准事件必须且只需包含 event_id、order_id、hub、service、kind、happened_at。来源文件与行号写入异常／重复清单。
• 同 event_id 标准内容完全相同：只留一条，其他计 exact_replay。同 event_id 内容不同：整组隔离，不能先到先得。坏字段、未知动作或无时区时间：隔离并保留来源及原因。
• 禁止原地修改原始材料。对账：原始事件数 = 标准事件数 + 隔离记录数 + 重复剔除数。

截止日履约口径

按订单聚合，先只保留 happened_at ≤ cutoff 的事件。取消单优先排除；due_at > cutoff 的非取消单记 not_mature。其他订单：截止前无交付记 overdue_open；首次交付时刻 > due_at 记 delivered_late；其余 on_time。恰好到期算成熟，恰好交付算准时。

成熟分母 = on_time + delivered_late + overdue_open。迟交分子 = delivered_late + overdue_open。取消与未到期均不得填为准时。已交付迟交分钟按交付时间减 due_at，未交付逾期分钟按 cutoff 减 due_at，两者单列。此材料不包含退款、食物损坏或实际成本，不能推出这些金额。

服务审查口径

rules.json 固定目标迟交率为 10%，最小成熟样本 15。差距用百分点，不用百分比变化。成熟样本 <15 优先归“待补样本”；差距 >15 个百分点归“优先处理”；>5 且 ≤15 归“需要复核”；≤5 归“达到阈值”。这些是本次运营审查规则，不是统计显著性或公平性的证明。

统一服务结构比较：standard 和 express 各 50% 权重。两个服务层均满足最小样本才计算站点标准化迟交率；否则保留空值并说明缺口。不得用无依据的总体平均填补小样本。业务假设须与观察事实分开。

访问日志与授权策略

access-log.csv 一行一个请求。request_id 唯一；at 为 UTC；user、session、role 为请求上下文；method、path 为路由；status 为响应码；rows、bytes 是网关响应计数。它们不证明唯一客户数，也不证明接收者已使用数据。

access-policy.json 是本次调查的授权基线。operator 只能读取被授权站点，不能 export。supervisor 与 service 允许导出。登录成功后 session 有效 30 分钟，恰好 30 分钟有效；之后应拒绝。401 请求代表被拒绝的尝试，不是成功泄露。svc-archive 是批准的归档账号，频繁大批导出不能只因次数高就定性。日志没有文件正文、终端取证或 VPN 会话记录，不能据此归属自然人、地理位置或入口路径。

"""Reference implementation for the Ape Intern cold-chain operations lab."""
import csv,json
from collections import defaultdict,Counter
from datetime import datetime,timezone
from pathlib import Path
from urllib.parse import urlparse,parse_qs
UTC=timezone.utc
KINDS={'OPEN':'created','COLLECT':'picked_up','DROP':'delivered','VOID':'cancelled'}
HUBS={'海港','东城','西岭','南湾'}
def parse_time(value):
    t=datetime.fromisoformat(value.replace('Z','+00:00'))
    if t.tzinfo is None:raise ValueError('timezone_required')
    return t.astimezone(UTC)
def iso(t):return t.isoformat(timespec='seconds').replace('+00:00','Z')
def readcsv(p):
    with Path(p).open(encoding='utf-8-sig',newline='') as f:return list(csv.DictReader(f))
def writecsv(p,rows,fields=None):
    rows=list(rows)
    with Path(p).open('w',encoding='utf-8-sig',newline='') as f:
        w=csv.DictWriter(f,fieldnames=fields or list(rows[0]));w.writeheader();w.writerows(rows)
def dump(p,value):Path(p).write_text(json.dumps(value,ensure_ascii=False,indent=2)+'\n')
def _normalize(record,source):
    if source=='A':
        e=dict(event_id=record['event_id'],order_id=record['order_ref'],hub=record['hub'],service=record['service'],kind=record['event'],happened_at=iso(parse_time(record['event_time'])))
    else:
        ms=record['at_ms']
        if type(ms) is not int:raise ValueError('epoch_must_be_integer')
        e=dict(event_id=record['id'],order_id=record['shipment']['number'],hub=record['shipment']['station'],service=record['shipment']['product'],kind=KINDS[record['action']],happened_at=iso(datetime.fromtimestamp(ms/1000,UTC)))
    if e['kind'] not in KINDS.values():raise ValueError('unsupported_event')
    if e['hub'] not in HUBS or e['service'] not in {'standard','express'}:raise ValueError('unknown_dimension')
    if not e['event_id'] or not e['order_id']:raise ValueError('empty_key')
    return e

def normalize(record,source):
    try:return _normalize(record,source)
    except (KeyError,TypeError,OverflowError) as error:raise ValueError(str(error).strip("'")) from error

def ingest(data):
    grouped=defaultdict(list);rejected=[];duplicates=[]
    for source,name in [('A','dispatch-a.jsonl'),('B','courier-b.jsonl')]:
        for line,text in enumerate((data/name).read_text().splitlines(),1):
            loc={'source':name,'line':line}
            try:
                event=normalize(json.loads(text),source);grouped[event['event_id']].append((event,loc))
            except (ValueError,KeyError,TypeError,OverflowError) as ex:
                rejected.append(dict(loc,reason=str(ex).strip("'")))
    accepted=[]
    for eid,items in grouped.items():
        distinct={json.dumps(e,sort_keys=True,ensure_ascii=False) for e,loc in items}
        if len(distinct)>1:
            rejected.extend(dict(loc,reason='immutable_id_conflict') for e,loc in items);continue
        accepted.append(items[0][0])
        duplicates.extend(dict(loc,event_id=eid,reason='exact_replay') for e,loc in items[1:])
    accepted.sort(key=lambda e:(e['happened_at'],e['event_id']))
    return accepted,rejected,duplicates

def classify_order(order,events,cutoff):
    # Only facts available at cutoff can influence a retrospective result.
    visible=[e for e in events if parse_time(e['happened_at'])<=cutoff]
    visible.sort(key=lambda e:parse_time(e['happened_at']))
    delivered=[parse_time(e['happened_at']) for e in visible if e['kind']=='delivered']
    cancelled=any(e['kind']=='cancelled' for e in visible)
    due=parse_time(order['due_at'])
    if cancelled:state='cancelled'
    elif due>cutoff:state='not_mature'
    elif not delivered:state='overdue_open'
    elif min(delivered)>due:state='delivered_late'
    else:state='on_time'
    delay=round((min(delivered)-due).total_seconds()/60) if state=='delivered_late' else 0
    overdue=round((cutoff-due).total_seconds()/60) if state=='overdue_open' else 0
    return dict(order,cohort=state,late_minutes=delay,overdue_minutes=overdue)

def aggregate(rows,keys):
    groups=defaultdict(list)
    for row in rows:groups[tuple(row[k] for k in keys)].append(row)
    result=[]
    for dims,records in sorted(groups.items()):
        counts=Counter(r['cohort'] for r in records)
        eligible=sum(counts[x] for x in ['on_time','delivered_late','overdue_open'])
        late=counts['delivered_late']+counts['overdue_open']
        result.append(dict(zip(keys,dims),total=len(records),eligible=eligible,late=late,late_rate=round(late/eligible,6) if eligible else None,on_time=counts['on_time'],delivered_late=counts['delivered_late'],overdue_open=counts['overdue_open'],cancelled=counts['cancelled'],not_mature=counts['not_mature'],delivered_late_minutes=sum(r['late_minutes'] for r in records),open_overdue_minutes=sum(r['overdue_minutes'] for r in records)))
    return result

def service_band(rate,n,rules):
    if n<rules['min_sample']:return '待补样本'
    gap=(rate-rules['late_target'])*100
    if gap>rules['priority_gap_pp']:return '优先处理'
    if gap>rules['review_gap_pp']:return '需要复核'
    return '达到阈值'
def fairness(hub_tiers,rules):
    rows=[]
    for x in hub_tiers:
        rate=x['late_rate'];rows.append(dict(hub=x['hub'],service=x['service'],eligible=x['eligible'],late=x['late'],late_rate=rate,target=rules['late_target'],gap_pp=round((rate-rules['late_target'])*100,3) if rate is not None else None,review=service_band(rate,x['eligible'],rules)))
    standardized=[]
    for hub in sorted(HUBS):
        parts=[r for r in rows if r['hub']==hub]
        enough=len(parts)==2 and all(r['eligible']>=rules['min_sample'] for r in parts)
        value=sum(r['late_rate']*rules[r['service']+'_weight'] for r in parts) if enough else None
        standardized.append(dict(hub=hub,standardized_late_rate=round(value,6) if value is not None else None,status='可比较' if enough else '待补样本'))
    return rows,standardized

def investigate(logs,policy):
    logins={};findings=[]
    for row in sorted(logs,key=lambda r:(r['at'],r['request_id'])):
        t=parse_time(row['at']);status=int(row['status']);route=urlparse(row['path']);query=parse_qs(route.query)
        if route.path=='/login' and status==200:logins[(row['user'],row['session'])]=t;continue
        if status!=200:continue
        types=[]
        if route.path=='/export' and row['role'] not in policy['export_roles']:types.append('unauthorised_export')
        hub=query.get('hub',[''])[0]
        if row['role']=='operator' and hub and hub!=policy['operator_hub'].get(row['user']):types.append('hub_scope_violation')
        login=logins.get((row['user'],row['session']))
        if login is None or (t-login).total_seconds()>policy['session_ttl_minutes']*60:types.append('expired_or_missing_session')
        if types:findings.append(dict(request_id=row['request_id'],at=row['at'],user=row['user'],session=row['session'],finding=';'.join(types),status=status,rows=int(row['rows']),bytes=int(row['bytes'])))
    return findings

def solve(data,out):
    data=Path(data);out=Path(out);out.mkdir(parents=True,exist_ok=True)
    events,rejects,dups=ingest(data)
    (out/'canonical-events.jsonl').write_text(''.join(json.dumps(e,ensure_ascii=False)+'\n' for e in events))
    writecsv(out/'rejected.csv',rejects);writecsv(out/'duplicates.csv',dups)
    rules=json.loads((data/'rules.json').read_text());cutoff=parse_time(rules['cutoff'])
    by_order=defaultdict(list)
    for e in events:by_order[e['order_id']].append(e)
    orders=readcsv(data/'orders.csv')
    records=[classify_order(o,by_order[o['order_id']],cutoff) for o in orders]
    writecsv(out/'order-review.csv',records)
    hubs=aggregate(records,['hub']);tiers=aggregate(records,['hub','service']);routes=aggregate(records,['route'])
    writecsv(out/'hub-metrics.csv',hubs);writecsv(out/'route-metrics.csv',routes)
    bands,std=fairness(tiers,rules)
    writecsv(out/'service-review.csv',bands);writecsv(out/'standardized-comparison.csv',std)
    findings=investigate(readcsv(data/'access-log.csv'),json.loads((data/'access-policy.json').read_text()))
    writecsv(out/'security-findings.csv',findings)
    summary=dict(orders=len(orders),accepted_events=len(events),rejected=len(rejects),duplicates=len(dups),cohorts=dict(Counter(r['cohort'] for r in records)),hubs=hubs,service_review=bands,standardized=std,security=dict(flagged_requests=len(findings),potential_export_rows=sum(f['rows'] for f in findings if 'unauthorised_export' in f['finding']),potential_export_bytes=sum(f['bytes'] for f in findings if 'unauthorised_export' in f['finding']),findings=findings))
    dump(out/'results.json',summary)
    return summary
if __name__=='__main__':
    import sys
    print(json.dumps(solve(sys.argv[1],sys.argv[2]),ensure_ascii=False,indent=2))