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.
67 lines
4.9 KiB
Python
67 lines
4.9 KiB
Python
from pathlib import Path
|
|
from datetime import date,datetime,timedelta
|
|
from collections import Counter,defaultdict
|
|
import json,re,hashlib,shutil,sys
|
|
import openpyxl
|
|
P=Path(__file__).resolve().parent
|
|
FILES={
|
|
'assessment':Path('/Users/shishengliang/Downloads/租赁任务考核_2026年9月.xlsx'),
|
|
'meter':Path('/Users/shishengliang/Downloads/车辆里程查询_20260901-20260930_1024辆_仪表里程.xlsx'),
|
|
'gps':Path('/Users/shishengliang/Downloads/车辆里程查询_20260901-20260930_1024辆_GPS里程.xlsx'),
|
|
'hydrogen':Path('/Users/shishengliang/Downloads/车辆纯氢气里程查询_20260901-20260930_1024辆.xlsx'),
|
|
'tests':Path('/Users/shishengliang/Library/Containers/com.tencent.xinWeChat/Data/Documents/xwechat_files/wxid_1704407055112_3af8/temp/RWTemp/2026-10/4340099467fbf08b4c6583c49adba4ee/测试车辆9月.xlsx'),
|
|
'policy':Path('/Users/shishengliang/Documents/羚牛/里程考核核算/8月/输入数据/新合同+新提车.xls'),
|
|
}
|
|
def norm(v):return re.sub(r'\s+','',str(v or '')).upper()
|
|
def dt(v):return v.date() if isinstance(v,datetime) else date.fromisoformat(str(v)[:10])
|
|
def read(path,sheet=None):
|
|
w=openpyxl.load_workbook(path,read_only=True,data_only=True);s=w[sheet] if sheet else w.worksheets[0];s.reset_dimensions();rs=list(s.values);w.close();return rs
|
|
def dump(name,value):(P/name).write_text(json.dumps(value,ensure_ascii=False,indent=2,default=str))
|
|
sources={};(P/'sources').mkdir(parents=True,exist_ok=True)
|
|
for name,path in FILES.items():
|
|
digest=hashlib.sha256(path.read_bytes()).hexdigest();target=P/'sources'/path.name
|
|
if not target.exists():shutil.copy2(path,target)
|
|
assert hashlib.sha256(target.read_bytes()).hexdigest()==digest
|
|
sources[name]=dict(path=str(path),snapshot=str(target),sha256=digest)
|
|
matrices={};provenance={};matrix_stats={};meta={}
|
|
for name in ['meter','gps','hydrogen']:
|
|
rs=read(FILES[name]);cols={i:dt(v) for i,v in enumerate(rs[5]) if isinstance(v,(date,datetime))}
|
|
assert set(cols.values())=={date(2026,9,i) for i in range(1,31)}
|
|
rows={};refs={};nonblank=0
|
|
for rn,r in enumerate(rs[6:],7):
|
|
if not r[0]:continue
|
|
p=norm(r[0]);assert p not in rows,(name,p)
|
|
rows[p]={};refs[p]={};meta.setdefault(p,dict(vin=r[1],brand=r[2],model=r[3]))
|
|
for i,d in cols.items():
|
|
v=r[i] if i<len(r) else None
|
|
assert v is None or isinstance(v,(int,float)) and v>=0,(name,p,d,v)
|
|
rows[p][str(d)]=v;refs[p][str(d)]=f'{FILES[name].name} / {openpyxl.utils.get_column_letter(i+1)}{rn}'
|
|
nonblank+=v is not None
|
|
subtotal=sum(v for v in rows[p].values() if v is not None)
|
|
assert r[-1] is None or abs(subtotal-r[-1])<.01,(name,p,subtotal,r[-1])
|
|
matrices[name]=rows;provenance[name]=refs;matrix_stats[name]=dict(vehicles=len(rows),nonblank_days=nonblank)
|
|
rs=read(FILES['tests']);tests=[norm(r[0]) for r in rs[2:] if r and r[0]]
|
|
assert len(tests)==len(set(tests))
|
|
rs=read(FILES['assessment']);headers=rs[0];records=[];ownership=defaultdict(list);date_errors=[];targets=[]
|
|
for rn,row in enumerate(rs[1:],2):
|
|
r=dict(zip(headers,row))
|
|
if not r.get('车牌号'):continue
|
|
p=norm(r['车牌号']);r['车牌号']=p;r['源表行']=rn
|
|
a,b=dt(r['考核开始日期']),dt(r['考核结束日期']);days=(b-a).days+1
|
|
if a<date(2026,9,1) or b>date(2026,9,30) or days<=0 or days!=r['考核天数']:date_errors.append([rn,p,str(a),str(b),r['考核天数']])
|
|
for i in range(max(0,days)):ownership[(p,str(a+timedelta(days=i)))].append([rn,r['销售经理'],r['合同编号']])
|
|
records.append(r)
|
|
overlaps=[dict(plate=p,date=d,rows=items) for (p,d),items in ownership.items() if len(items)>1]
|
|
coverage={name:sum(matrices[name].get(p,{}).get(d) is not None for p,d in ownership) for name in matrices}
|
|
sys.path.insert(0,'/tmp/mileage-september-read-deps')
|
|
import xlrd
|
|
w=xlrd.open_workbook(FILES['policy']);policy_rows={s.name:[s.row_values(i) for i in range(s.nrows)] for s in w.sheets()}
|
|
plates=set()
|
|
for rs in policy_rows.values():
|
|
for row in rs:
|
|
for value in row:
|
|
plates.update(re.findall(r'[粤沪浙苏鲁皖赣鄂湘闽川渝京津豫冀陕晋桂琼贵云甘宁青新蒙黑吉辽藏][A-Z][A-Z0-9]{5,6}',norm(value)))
|
|
summary=dict(records=len(records),vehicles=len({r['车牌号'] for r in records}),test_roster=len(tests),assessed_test_vehicles=len(set(tests)&{r['车牌号'] for r in records}),policy_roster=len(plates),assessed_policy_vehicles=len(plates&{r['车牌号'] for r in records}),assessed_vehicle_days=len(ownership),covered_days=coverage,matrix_stats=matrix_stats,overlap_days=len(overlaps),date_errors=date_errors,people=dict(Counter(r['销售经理'] for r in records)))
|
|
dump('inputs.json',dict(sources=sources,records=records,headers=headers,matrices=matrices,provenance=provenance,tests=tests,policy_plates=sorted(plates),policy_rows=policy_rows,vehicle_metadata=meta,overlaps=overlaps,summary=summary))
|
|
dump('source_audit.json',summary);print(json.dumps(summary,ensure_ascii=False,indent=2));print('Overlaps:',json.dumps(overlaps[:12],ensure_ascii=False));print('Policy headers:',json.dumps({k:v[:3] for k,v in policy_rows.items()},ensure_ascii=False))
|