170 lines
5.6 KiB
Python
170 lines
5.6 KiB
Python
#!/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}')
|