运营异常工作台交付提案

工作台需求访谈纪要

运营负责人：早上先看尚未交付的到期订单，再看站点与线路。已交付迟交订单用于复盘，不应盖过仍在积压的订单。希望按站点、服务等级和履约状态筛选，打开订单看时间线。需要显示复盘截止时间与数据更新时间。

站点负责人：不能拿不同服务结构的总迟交率直接排绩效。南湾 express 数据少，需要补足后再比较。线路备注只是线索，不能让系统自动把延迟归因给某个员工。

安全负责人：operator 只能读取所属站点，导出另需 supervisor 或 service 权限，服务端每次请求都检查权限和 session。身份来自企业现有身份服务，不新增密码库。屏幕上选择站点不等于获得权限。

交付负责人：第一期做只读工作台、订单证据时间线与受控导出设计；不要加入退款、调度写入或客户通知。上线依赖客户提供身份服务、网关和订单接口。需要估算工时、明确验收、失败处理与回退。没有给定预算、合同 SLA、生产并发数或上线日期，方案须提出待确认项，不自行写成客户已确认。

{
  "session_ttl_minutes": 30,
  "export_roles": [
    "supervisor",
    "service"
  ],
  "operator_hub": {
    "op-1": "海港",
    "op-2": "东城",
    "op-3": "西岭",
    "op-4": "南湾",
    "op-7": "海港"
  },
  "network_note": "经企业 VPN 访问网关；IP 为文档保留地址，不能证明真实地理位置或攻击者身份。",
  "service_accounts": [
    "svc-archive"
  ]
}

{
  "version": "1.0",
  "as_of": "2026-09-15T00:00:00Z",
  "dataset_version": "coldchain-v1",
  "routes": [
    {
      "method": "GET",
      "path": "/v1/orders",
      "query": [
        "hub",
        "service",
        "cohort",
        "as_of",
        "cursor",
        "limit"
      ],
      "limit": {
        "min": 1,
        "max": 100
      },
      "response": [
        "items",
        "total",
        "next_cursor",
        "as_of",
        "dataset_version",
        "updated_at"
      ],
      "errors": [
        400,
        401,
        403,
        409,
        429,
        503
      ]
    },
    {
      "method": "GET",
      "path": "/v1/orders/{order_id}/events",
      "response": [
        "events",
        "visible_at_cutoff"
      ],
      "errors": [
        401,
        403,
        404,
        503
      ]
    },
    {
      "method": "GET",
      "path": "/v1/metrics",
      "response": [
        "eligible",
        "late",
        "by_hub",
        "by_service",
        "sample_status"
      ],
      "errors": [
        400,
        401,
        403,
        409,
        503
      ]
    },
    {
      "method": "POST",
      "path": "/v1/exports",
      "roles": [
        "supervisor",
        "service"
      ],
      "body": [
        "scope",
        "as_of",
        "dataset_version",
        "request_key"
      ],
      "accepted_status": 202,
      "errors": [
        400,
        401,
        403,
        409,
        429,
        503
      ]
    }
  ]
}

"""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))