chore: curate monthly assessment evidence and scripts
This commit is contained in:
kkfluous
@@ -0,0 +1,169 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Step 1: GPS里程替换 - 针对 租赁任务考核_2026年6月 (1).xlsx"""
|
||||
import openpyxl, pymysql, os, shutil
|
||||
from datetime import datetime
|
||||
from collections import defaultdict
|
||||
|
||||
os.chdir('/Users/kkfluous/Downloads')
|
||||
|
||||
# 36辆测试车辆（从上一次GPS替换留底提取）
|
||||
TEST_PLATES = {
|
||||
'粤AGE5480','粤AGF4535','粤AGG3490','粤AGG4135','粤AGG4191',
|
||||
'粤AGP2033','粤AGP3080','粤AGP3663','粤AGP3676','粤AGP5139',
|
||||
'粤AGP5156','粤AGP5163','粤AGP5168','粤AGP5636','粤AGP5767',
|
||||
'粤AGP5768','粤AGP5796','粤AGP9733','粤AGP9786','粤AGR0298',
|
||||
'粤AGR1586','粤AGR3288','粤AGR5056','粤AGR5068','粤AGR5088',
|
||||
'粤AGR6869','粤AGR6877','粤AGR6879','粤AGR6881','粤AGR8506',
|
||||
'粤AGR8551','粤AGR8558','粤AGR8799','粤AGR9866','粤AGR9893',
|
||||
'粤AGR9899',
|
||||
}
|
||||
|
||||
SRC = '租赁任务考核_2026年6月 (1).xlsx'
|
||||
DST = '租赁任务考核_2026年6月.xlsx'
|
||||
BAK = '租赁任务考核_2026年6月_备份_0714.xlsx'
|
||||
|
||||
# 0. 备份现有的6月文件
|
||||
if os.path.exists(DST):
|
||||
shutil.copy2(DST, BAK)
|
||||
print(f'已备份原6月文件 → {BAK}')
|
||||
|
||||
# 1. 复制新源文件为目标文件
|
||||
shutil.copy2(SRC, DST)
|
||||
print(f'源文件: {SRC}')
|
||||
print(f'目标文件: {DST}')
|
||||
|
||||
# 2. 读取目标文件
|
||||
wb = openpyxl.load_workbook(DST)
|
||||
ws = wb['业务考核视图']
|
||||
h = [c.value for c in next(ws.iter_rows(min_row=1, max_row=1))]
|
||||
print(f'列数: {len(h)}, 表头: {h}')
|
||||
|
||||
recs = []
|
||||
for row in ws.iter_rows(min_row=2, values_only=True):
|
||||
r = dict(zip(h, row))
|
||||
r['row_num'] = len(recs) + 2
|
||||
recs.append(r)
|
||||
|
||||
test_recs = [r for r in recs if r['车牌号'] in TEST_PLATES]
|
||||
print(f'测试车辆记录: {len(test_recs)}条 ({len(set(r["车牌号"] for r in test_recs))}辆)')
|
||||
|
||||
# 3. 连接数据库取GPS日里程
|
||||
conn = pymysql.connect(host='101.133.130.65', port=3306, user='root', password='ln123!@#',
|
||||
database='hydrogen_energy', connect_timeout=30)
|
||||
cur = conn.cursor()
|
||||
|
||||
def get_gps_km(plate, start_str, end_str):
|
||||
"""汇总GPS日里程(km)"""
|
||||
if not start_str or not end_str:
|
||||
return None
|
||||
try:
|
||||
sd = datetime.strptime(str(start_str)[:10], '%Y-%m-%d').date()
|
||||
ed = datetime.strptime(str(end_str)[:10], '%Y-%m-%d').date()
|
||||
except:
|
||||
return None
|
||||
|
||||
cur.execute("""
|
||||
SELECT dates, total_mileage FROM ln_vehicle_g7_mileage
|
||||
WHERE plate_number = %s AND dates >= %s AND dates <= %s
|
||||
ORDER BY dates
|
||||
""", (plate, sd, ed))
|
||||
rows = cur.fetchall()
|
||||
if not rows:
|
||||
return None
|
||||
|
||||
vals = [r[1] for r in rows]
|
||||
is_cumulative = all(vals[i] <= vals[i+1] for i in range(len(vals)-1)) if len(vals) > 1 else False
|
||||
|
||||
if is_cumulative:
|
||||
deltas = []
|
||||
prev = None
|
||||
for v in vals:
|
||||
if prev is not None and v > prev:
|
||||
deltas.append(v - prev)
|
||||
elif prev is None:
|
||||
deltas.append(v)
|
||||
prev = v
|
||||
total_km = sum(deltas) / 100000
|
||||
else:
|
||||
total_km = sum(vals) / 100000
|
||||
|
||||
return round(total_km, 1)
|
||||
|
||||
# 批量查询
|
||||
gps_cache = {}
|
||||
replaced = 0
|
||||
no_data = 0
|
||||
no_data_plates = set()
|
||||
|
||||
for r in test_recs:
|
||||
plate = r['车牌号']
|
||||
km = get_gps_km(plate, r.get('考核开始日期'), r.get('考核结束日期'))
|
||||
if km is None:
|
||||
no_data += 1
|
||||
no_data_plates.add(plate)
|
||||
else:
|
||||
gps_cache[r['row_num']] = km
|
||||
|
||||
conn.close()
|
||||
print(f'GPS查询: 成功{len(gps_cache)}条, 无数据{no_data}条')
|
||||
|
||||
# 4. 添加 原TBOX里程(km) 和 测试里程(km) 列
|
||||
if '原TBOX里程(km)' not in h:
|
||||
tbox_col = ws.max_column + 1
|
||||
test_col = ws.max_column + 2
|
||||
ws.cell(row=1, column=tbox_col, value='原TBOX里程(km)')
|
||||
ws.cell(row=1, column=test_col, value='测试里程(km)')
|
||||
# 刷新表头
|
||||
h_new = [c.value for c in next(ws.iter_rows(min_row=1, max_row=1))]
|
||||
else:
|
||||
h_new = h
|
||||
|
||||
# 找列索引
|
||||
actual_idx = h_new.index('实际行驶里程(km)') + 1
|
||||
tbox_idx = h_new.index('原TBOX里程(km)') + 1
|
||||
test_idx = h_new.index('测试里程(km)') + 1
|
||||
|
||||
# 5. 写入GPS数据
|
||||
for row_num, gps_km in gps_cache.items():
|
||||
old_val = ws.cell(row=row_num, column=actual_idx).value
|
||||
|
||||
# 写入原TBOX里程
|
||||
ws.cell(row=row_num, column=tbox_idx, value=float(old_val) if old_val else 0)
|
||||
# 写入测试里程
|
||||
tbox_val = float(old_val or 0)
|
||||
test_km = round(tbox_val - gps_km, 1)
|
||||
ws.cell(row=row_num, column=test_idx, value=test_km)
|
||||
# 替换实际行驶里程为GPS
|
||||
ws.cell(row=row_num, column=actual_idx, value=gps_km)
|
||||
replaced += 1
|
||||
|
||||
# 6. 测试车辆标记黄色背景
|
||||
from openpyxl.styles import PatternFill
|
||||
yellow_fill = PatternFill(start_color='FFFF00', end_color='FFFF00', fill_type='solid')
|
||||
for row_num in gps_cache.keys():
|
||||
for c in range(1, ws.max_column + 1):
|
||||
ws.cell(row=row_num, column=c).fill = yellow_fill
|
||||
|
||||
# 7. 重算完成率和是否达标
|
||||
DAYSM = {1:31,2:28,3:31,4:30,5:31,6:30,7:31,8:31,9:30,10:31,11:30,12:31}
|
||||
for r in recs:
|
||||
rn = r['row_num']
|
||||
actual = ws.cell(row=rn, column=actual_idx).value or 0
|
||||
target = float(r.get('应考核里程(km)') or 0)
|
||||
pct = round(actual / target * 100, 2) if target > 0 else 0
|
||||
q = '达标' if actual >= target and target > 0 else '未达标'
|
||||
# 找完成率列和是否达标列
|
||||
rate_idx = h_new.index('完成率(%)') + 1
|
||||
qualified_idx = h_new.index('是否达标') + 1
|
||||
ws.cell(row=rn, column=rate_idx, value=pct)
|
||||
ws.cell(row=rn, column=qualified_idx, value=q)
|
||||
|
||||
wb.save(DST)
|
||||
wb.close()
|
||||
|
||||
print(f'\n===== GPS替换完成 =====')
|
||||
print(f'替换成功: {replaced}条')
|
||||
print(f'无GPS数据: {no_data}条 ({len(no_data_plates)}辆)')
|
||||
if no_data_plates:
|
||||
print(f'无数据车牌: {sorted(no_data_plates)}')
|
||||
print(f'已保存: {DST}')
|
||||
Reference in New Issue
Block a user