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.
91 lines
4.5 KiB
Python
91 lines
4.5 KiB
Python
"""Read-only source coverage audit. Does not calculate or overwrite awards."""
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta
|
|
from collections import Counter
|
|
import json, re, warnings
|
|
import openpyxl
|
|
|
|
warnings.filterwarnings('ignore', category=UserWarning, module='openpyxl')
|
|
OUT = Path(__file__).resolve().parent
|
|
REPO = OUT.parents[1]
|
|
HROOT = Path('/tmp/mileage-hydrogen-review-0916/氢能源行驶里程')
|
|
METER = Path('/Users/shishengliang/Downloads/车辆里程查询_20260101-20260731_1024辆.xlsx')
|
|
JAN = Path('/Users/shishengliang/Code/lingniu/氢能车用氢行驶里程统计 1.17.xlsx')
|
|
|
|
def read(p, sheet=None):
|
|
w = openpyxl.load_workbook(p, read_only=True, data_only=True)
|
|
s = w[sheet] if sheet else w.worksheets[0]
|
|
s.reset_dimensions()
|
|
r = list(s.values)
|
|
w.close()
|
|
return r
|
|
|
|
def norm(x):
|
|
return re.sub(r'\s+', '', str(x or '')).upper()
|
|
|
|
h = {}
|
|
bad_files = []
|
|
for p in HROOT.rglob('*.xlsx'):
|
|
if int(p.parent.name[:-1]) > 6:
|
|
continue
|
|
rr = read(p)
|
|
headers = list(rr[0])
|
|
if '车牌号码' not in headers:
|
|
bad_files.append(str(p))
|
|
continue
|
|
for row in rr[1:]:
|
|
r = dict(zip(headers,row))
|
|
plate = norm(r.get('车牌号码'))
|
|
if not plate:
|
|
continue
|
|
d = str(r['日期范围'])[:10]
|
|
val = r.get('氢能源行驶里程(km)')
|
|
if val is not None and str(val).strip():
|
|
key = (plate,d)
|
|
assert key not in h, ('duplicate hydrogen',key)
|
|
h[key] = float(val)
|
|
jan_status = 'not read'
|
|
try:
|
|
rr=read(JAN)
|
|
assert '车牌号码' in rr[0]
|
|
for row in rr[1:]:
|
|
r=dict(zip(rr[0],row)); plate=norm(r.get('车牌号码'))
|
|
if plate and r.get('氢能源行驶里程(km)') is not None:
|
|
h[(plate,str(r['日期范围'])[:10])] = float(r['氢能源行驶里程(km)'])
|
|
jan_status='read'
|
|
except (OSError,AssertionError) as e:
|
|
jan_status=f'{type(e).__name__}: {e}'
|
|
|
|
(OUT/'hydrogen_daily.json').write_text(json.dumps([[p,d,v] for (p,d),v in sorted(h.items())],ensure_ascii=False))
|
|
# GPS and meter acquisition superseded by the user's monthly minimum rule.
|
|
t={};meter_months=Counter()
|
|
|
|
records=[]; summary=[]
|
|
for m in range(1,7):
|
|
rr=read(REPO/f'租赁任务考核_2026年{m}月.xlsx','业务考核视图')
|
|
month=[]
|
|
for rownum,row in enumerate(rr[1:],2):
|
|
r=dict(zip(rr[0],row)); plate=norm(r.get('车牌号'))
|
|
if not plate:
|
|
continue
|
|
def as_date(v):
|
|
return v.date() if isinstance(v,datetime) else datetime.fromisoformat(str(v)).date()
|
|
start,end=as_date(r['考核开始日期']),as_date(r['考核结束日期'])
|
|
dates=[(start+timedelta(days=i)).isoformat() for i in range((end-start).days+1)]
|
|
mh=[d for d in dates if (plate,d) not in h]
|
|
mt=[d for d in dates if (plate,d) not in t]
|
|
both=[d for d in mh if (plate,d) not in t]
|
|
x=dict(month=m,source_row=rownum,plate=plate,sales=r.get('销售经理'),start=str(start),end=str(end),days=len(dates),hydrogen_missing_dates=mh,meter_missing_dates=mt,both_missing_dates=both,original_assessment_km=r.get('实际行驶里程(km)'),test_decision='not applicable before April' if m<4 else 'pending GPS and complete same-period meter basis')
|
|
assert len(mh) in (0,len(dates)), 'Partial hydrogen needs a daily fallback decision'
|
|
x['hydrogen_km'] = round(sum(h[(plate,d)] for d in dates), 2) if not mh else None
|
|
x['candidate_non_test_km'] = x['hydrogen_km'] if not mh else x['original_assessment_km']
|
|
x['meter_comparison_km'] = x['original_assessment_km']
|
|
x['meter_basis'] = '用户确认历史考核源表实际里程即完整仪表区间合计'
|
|
x['test_decision'] = '不再采用测试车辆判定；按同组月合计取小值'
|
|
month.append(x);records.append(x)
|
|
summary.append(dict(month=m,records=len(month),records_with_hydrogen_gaps=sum(bool(x['hydrogen_missing_dates']) for x in month),records_with_meter_gaps=sum(bool(x['meter_missing_dates']) for x in month),records_with_both_gaps=sum(bool(x['both_missing_dates']) for x in month),both_missing_record_days=sum(len(x['both_missing_dates']) for x in month)))
|
|
|
|
result=dict(jan17_status=jan_status,meter_nonblank_vehicle_days_by_month=dict(sorted(meter_months.items())),summary=summary,records=records,invalid_archive_files=bad_files)
|
|
(OUT/'source_coverage.json').write_text(json.dumps(result,ensure_ascii=False,indent=2,default=str))
|
|
print(json.dumps({'jan17_status':jan_status,'summary':summary},ensure_ascii=False,indent=2))
|