91 lines
4.4 KiB
Python
91 lines
4.4 KiB
Python
from ast import parse, walk, Call, Constant, keyword
|
|
from collections import defaultdict
|
|
from datetime import datetime, date
|
|
from pathlib import Path
|
|
from zipfile import ZipFile
|
|
from xml.etree import ElementTree as ET
|
|
import re
|
|
|
|
import openpyxl
|
|
import pymysql
|
|
|
|
BASE = Path('/Users/kkfluous/Downloads')
|
|
ASSESSMENT = BASE / '7月核算结果/租赁任务考核_2026年7月.xlsx'
|
|
TBOX = BASE / '车辆里程查询_20260701-20260731_1024辆.xlsx'
|
|
DETAILS = [BASE / '里程统计[天][2026-07-01至2026-07-31].xlsx', BASE / '里程统计[天][2026-07-01至2026-07-31] (1).xlsx']
|
|
NS = '{http://schemas.openxmlformats.org/spreadsheetml/2006/main}'
|
|
|
|
def norm(v): return re.sub(r'\s+', '', str(v or '')).upper()
|
|
|
|
def cfg():
|
|
tree=parse((BASE/'replace_gps_jun.py').read_text())
|
|
for node in walk(tree):
|
|
if isinstance(node,Call) and getattr(node.func,'attr',None)=='connect':
|
|
d={x.arg:x.value.value for x in node.keywords if isinstance(x.value,Constant)}
|
|
if {'host','user','password','database'}<=d.keys(): return d
|
|
raise RuntimeError('no db config')
|
|
|
|
def read_details():
|
|
out=defaultdict(dict)
|
|
for path in DETAILS:
|
|
with ZipFile(path) as z:
|
|
root=ET.fromstring(z.read('xl/worksheets/sheet1.xml'))
|
|
for row in root.findall(f'.//{NS}row'):
|
|
if int(row.get('r'))<4: continue
|
|
d={}
|
|
for c in row.findall(f'{NS}c'):
|
|
col=re.match(r'[A-Z]+',c.get('r')).group()
|
|
if c.get('t')=='inlineStr': d[col]=''.join(t.text or '' for t in c.findall(f'.//{NS}t'))
|
|
else:
|
|
v=c.find(f'{NS}v'); d[col]=v.text if v is not None else None
|
|
p=norm(d.get('H')); ds=str(d.get('O') or '').strip()
|
|
if p and p!='合计' and re.fullmatch(r'2026-07-\d{2}',ds):
|
|
out[p][datetime.strptime(ds,'%Y-%m-%d').date()]=float(d.get('P') or 0)
|
|
return out
|
|
|
|
awb=openpyxl.load_workbook(ASSESSMENT,data_only=True)
|
|
aws=awb['业务考核视图']
|
|
h=[c.value for c in aws[1]]; ix={v:i for i,v in enumerate(h)}
|
|
candidates=set(); records=[]
|
|
for ri,row in enumerate(aws.iter_rows(min_row=2,values_only=True),2):
|
|
fills={aws.cell(ri,c).fill.fgColor.rgb for c in range(1,aws.max_column+1) if aws.cell(ri,c).fill.fill_type=='solid'}
|
|
is_candidate=any(str(x).upper().endswith('FFFF00') for x in fills if x)
|
|
p=norm(row[ix['车牌号']]); start=row[ix['考核开始日期']].date(); end=row[ix['考核结束日期']].date()
|
|
records.append((p,start,end,is_candidate))
|
|
if is_candidate:candidates.add(p)
|
|
awb.close()
|
|
|
|
twb=openpyxl.load_workbook(TBOX,read_only=True,data_only=True); tws=twb['里程查询']
|
|
daycols={i:v.date() for i,v in enumerate(next(tws.iter_rows(min_row=6,max_row=6,values_only=True))) if isinstance(v,datetime)}
|
|
tbox=defaultdict(dict)
|
|
for row in tws.iter_rows(min_row=7,values_only=True):
|
|
p=norm(row[0])
|
|
if p:
|
|
for i,d in daycols.items():
|
|
if row[i] is not None:tbox[p][d]=float(row[i])
|
|
twb.close()
|
|
|
|
gps=read_details()
|
|
dc=cfg();dc.update(charset='utf8mb4',connect_timeout=8,read_timeout=30,write_timeout=30)
|
|
conn=pymysql.connect(**dc)
|
|
try:
|
|
with conn.cursor() as cur:
|
|
ph=','.join(['%s']*len(candidates))
|
|
cur.execute(f'SELECT plate_number,dates,total_mileage FROM ln_vehicle_g7_mileage WHERE dates >= %s AND dates <= %s AND plate_number IN ({ph})',[date(2026,7,1),date(2026,7,31),*sorted(candidates)])
|
|
rows=cur.fetchall()
|
|
finally:conn.close()
|
|
for p,d,m in rows:
|
|
if isinstance(d,datetime):d=d.date()
|
|
gps[norm(p)].setdefault(d,float(m or 0)/100000)
|
|
|
|
candidate_days=0; comparable=0; flagged=0; plates_no_tbox=set(); plates_no_gps=set(); flagged_plates=set()
|
|
for p,start,end,cand in records:
|
|
if not cand:continue
|
|
for n in range((end-start).days+1):
|
|
d=start.fromordinal(start.toordinal()+n);candidate_days+=1
|
|
if d not in tbox.get(p,{}):plates_no_tbox.add(p);continue
|
|
if d not in gps.get(p,{}):plates_no_gps.add(p);continue
|
|
comparable+=1
|
|
if tbox[p][d]-gps[p][d]>50:flagged+=1;flagged_plates.add(p)
|
|
print({'candidate_records':sum(x[3] for x in records),'candidate_plates':len(candidates),'candidate_days':candidate_days,'comparable_days':comparable,'flagged_days':flagged,'flagged_plates':len(flagged_plates),'plates_with_any_tbox_gap':len(plates_no_tbox),'plates_with_any_gps_gap':len(plates_no_gps),'no_tbox_sample':sorted(plates_no_tbox)[:20],'no_gps_sample':sorted(plates_no_gps)[:20]})
|