shishengliang
65904c8270
Preserve August hydrogen revision and September audit evidence; calculate interrupted vehicle history across all prior months. Record remaining carry-policy and layout-baseline discrepancies.
35 lines
2.5 KiB
Python
35 lines
2.5 KiB
Python
"""Assemble September assessment intervals and daily sources, before selecting mileage."""
|
|
from pathlib import Path
|
|
from datetime import date,timedelta
|
|
from collections import Counter
|
|
import json
|
|
P=Path(__file__).resolve().parent
|
|
d=json.loads((P/'inputs.json').read_text())
|
|
records=d['records'];byrow={r['源表行']:r for r in records};changes=[]
|
|
for x in d['overlaps']:
|
|
later=max((byrow[z[0]] for z in x['rows']),key=lambda r:r['考核开始日期'])
|
|
old=later['考核开始日期'][:10];new=str(date.fromisoformat(x['date'])+timedelta(days=1))
|
|
changes.append(dict(plate=later['车牌号'],sales=later['销售经理'],source_row=later['源表行'],old_start=old,new_start=new,reason='用户确认交接日计前段，后段次日起算'))
|
|
later['考核开始日期']=new
|
|
from sys import path
|
|
path.insert(0,str(P.parents[1]))
|
|
from calc_engine import RULES
|
|
prices={'交投40辆4.5T普货':150,'交投190辆4.5T冷链车':150,'羚牛136辆4.5T冷链车':300,'恒运50辆4.5T普货':300,'恒运50辆4.5T 普货':300,'羚牛100辆18T':150}
|
|
seen=set();daily=[]
|
|
for r in records:
|
|
p=r['车牌号'];a=date.fromisoformat(r['考核开始日期'][:10]);b=date.fromisoformat(r['考核结束日期'][:10]);days=(b-a).days+1
|
|
assert 0<days<=30
|
|
rule=RULES[r['考核目标']]
|
|
r['原考核天数']=r['考核天数'];r['原应考核里程(km)']=r['应考核里程(km)']
|
|
r['考核天数']=days;r['应考核里程(km)']=round(rule['km']*days/30,2)
|
|
r['月度目标里程']=rule['km'];r['月奖励']=prices[r['考核目标']] if p in d['policy_plates'] else rule['good']
|
|
r['新政策']=p in d['policy_plates'];r['测试候选']=p in d['tests']
|
|
r['实际行驶里程(km)']=None;r['完成率(%)']=None;r['是否达标']=None
|
|
for i in range(days):
|
|
day=str(a+timedelta(days=i));assert (p,day) not in seen;seen.add((p,day))
|
|
vals={k:x.get(p,{}).get(day) for k,x in d['matrices'].items()}
|
|
daily.append(dict(source_row=r['源表行'],plate=p,sales=r['销售经理'],date=day,test_candidate=r['测试候选'],**vals,provenance={k:x.get(p,{}).get(day) for k,x in d['provenance'].items()}))
|
|
out=dict(records=records,daily=daily,overlap_adjustments=changes,sources=d['sources'],tests=d['tests'],policy_plates=d['policy_plates'],metadata=d['vehicle_metadata'],headers=d['headers'])
|
|
(P/'prepared_records.json').write_text(json.dumps(out,ensure_ascii=False,indent=2))
|
|
print(json.dumps(dict(records=len(records),unique_vehicle_days=len(seen),overlap_adjustments=changes,policy_records=sum(r['新政策'] for r in records),targets=dict(Counter(r['考核目标'] for r in records))),ensure_ascii=False,indent=2))
|