276 lines
10 KiB
Python
276 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Process May intervention: replace mileage with GPS, add TBOX/test columns."""
|
|
import os, shutil, openpyxl, pymysql, warnings
|
|
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
|
|
from openpyxl.utils import get_column_letter
|
|
from datetime import datetime, date, timedelta
|
|
from collections import defaultdict
|
|
|
|
warnings.filterwarnings('ignore')
|
|
os.chdir('/Users/kkfluous/Downloads')
|
|
|
|
SRC = '租赁任务考核_2026年5月.xlsx'
|
|
BAK = '租赁任务考核_2026年5月_原版备份.xlsx'
|
|
OUT = SRC # overwrite
|
|
|
|
# ── Step 1: Backup ──
|
|
if not os.path.exists(BAK):
|
|
shutil.copy2(SRC, BAK)
|
|
print(f"✅ Backup: {BAK}")
|
|
else:
|
|
print(f"跳过备份（已存在）")
|
|
|
|
# ── Step 2: Load intervention plates ──
|
|
wb_i = openpyxl.load_workbook('标黄为5月干预车辆.xlsx')
|
|
ws_i = wb_i['Sheet1']
|
|
yellow_plates = set()
|
|
for row in ws_i.iter_rows(min_row=2, values_only=False):
|
|
cell = row[0]
|
|
fill = cell.fill
|
|
is_yellow = (fill and fill.patternType == 'solid' and fill.start_color
|
|
and fill.start_color.rgb and 'FFFF00' in str(fill.start_color.rgb))
|
|
if is_yellow:
|
|
p = str(cell.value).strip() if cell.value else ''
|
|
if p:
|
|
yellow_plates.add(p)
|
|
wb_i.close()
|
|
print(f"干预车辆: {len(yellow_plates)}")
|
|
|
|
# ── Step 3: Read source records ──
|
|
wb_s = openpyxl.load_workbook(SRC, data_only=True)
|
|
ws_s = wb_s['业务考核视图']
|
|
src_h = [c.value for c in next(ws_s.iter_rows(min_row=1, max_row=1))]
|
|
rows_raw = list(ws_s.iter_rows(min_row=2, values_only=True))
|
|
wb_s.close()
|
|
|
|
records = []
|
|
for row in rows_raw:
|
|
if not row[0]:
|
|
continue
|
|
records.append(dict(zip(src_h, row)))
|
|
print(f"源数据: {len(records)} 条")
|
|
|
|
# ── Step 4: GPS from DB ──
|
|
print("查询数据库...")
|
|
conn = pymysql.connect(host='101.133.130.65', user='root', password='ln123!@#',
|
|
database='hydrogen_energy', charset='utf8mb4')
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT plate_number, dates, total_mileage FROM ln_vehicle_g7_mileage WHERE dates >= '2026-05-01' AND dates <= '2026-06-08'")
|
|
db_rows = cur.fetchall()
|
|
conn.close()
|
|
|
|
# Group: plate -> {date: mileage}
|
|
pmap = defaultdict(lambda: defaultdict(int))
|
|
for p, d, m in db_rows:
|
|
dd = d if isinstance(d, date) else (d.date() if isinstance(d, datetime) else date.fromisoformat(str(d)[:10]))
|
|
pmap[p][dd] = m or 0
|
|
|
|
print(f"DB plates loaded: {len(pmap)}")
|
|
|
|
# Compute GPS km per plate per考核期
|
|
def get_gps_km(plate, start_v, end_v):
|
|
"""GPS里程(km) = sum of daily total_mileage / 100000"""
|
|
if plate not in pmap:
|
|
return 0.0
|
|
|
|
# Parse dates
|
|
if isinstance(start_v, datetime):
|
|
sd = start_v.date()
|
|
elif isinstance(start_v, date):
|
|
sd = start_v
|
|
else:
|
|
return 0.0
|
|
|
|
if isinstance(end_v, datetime):
|
|
ed = end_v.date()
|
|
elif isinstance(end_v, date):
|
|
ed = end_v
|
|
else:
|
|
return 0.0
|
|
|
|
total = 0
|
|
d = sd
|
|
daily = pmap[plate]
|
|
while d <= ed:
|
|
total += daily.get(d, 0)
|
|
d += timedelta(days=1)
|
|
return total / 100000.0
|
|
|
|
# Build plate -> gps for intervention vehicles
|
|
plate_gps = {}
|
|
for rec in records:
|
|
p = str(rec['车牌号']).strip()
|
|
if p not in yellow_plates:
|
|
continue
|
|
if p in plate_gps:
|
|
continue
|
|
km = get_gps_km(p, rec.get('考核开始日期'), rec.get('考核结束日期'))
|
|
plate_gps[p] = round(km, 2)
|
|
|
|
print(f"GPS数据获取: {len(plate_gps)}/{len(yellow_plates)}")
|
|
gps_zero = sum(1 for v in plate_gps.values() if v == 0)
|
|
print(f"GPS=0: {gps_zero} 辆")
|
|
|
|
# ── Step 5: Generate new workbook ──
|
|
YELLOW = PatternFill('solid', fgColor='FFFF00')
|
|
RED = PatternFill('solid', fgColor='FF0000')
|
|
REDF = Font(color='FFFFFF', bold=True)
|
|
HFILL = PatternFill('solid', fgColor='4472C4')
|
|
HFONT = Font(bold=True, color='FFFFFF', size=10)
|
|
BORD = Border(Side('thin','B4B4B4'), Side('thin','B4B4B4'), Side('thin','B4B4B4'), Side('thin','B4B4B4'))
|
|
CC = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
|
NFONT = Font(size=10)
|
|
|
|
wb = openpyxl.Workbook()
|
|
ws = wb.active
|
|
ws.title = '业务考核视图'
|
|
|
|
# New header layout:
|
|
# 车牌号(0) 考核年份(1) 考核月份(2) 部门名称(3) 销售经理(4) 客户名称(5) 合同编号(6) 项目名称(7)
|
|
# 考核目标ID(8) 考核目标(9) 考核开始日期(10) 考核结束日期(11) 考核天数(12) 应考核里程(13)
|
|
# 原TBOX里程(14) 实际行驶里程=GPS(15) 测试里程(16) 完成率(17) 是否达标(18) 考核状态(19) ...
|
|
# Column map: src_col -> dst_col (0-indexed)
|
|
# 0-13 stay same, 14(old actual) -> split into 14(TBOX) 15(new actual) 16(test), 15(old rate) -> 17, 16(old pass)->18, 17(status)->19, 18-20 -> 20-22
|
|
|
|
new_h = ['车牌号', '考核年份', '考核月份', '部门名称', '销售经理', '客户名称',
|
|
'合同编号', '项目名称', '考核目标ID', '考核目标', '考核开始日期',
|
|
'考核结束日期', '考核天数', '应考核里程(km)',
|
|
'原TBOX里程(km)', '实际行驶里程(km)', '测试里程(km)',
|
|
'完成率(%)', '是否达标', '考核状态',
|
|
'交车日期', '预计还车日期', '实际还车日期']
|
|
|
|
for ci, h in enumerate(new_h):
|
|
c = ci + 1
|
|
cell = ws.cell(row=1, column=c, value=h)
|
|
cell.fill = HFILL
|
|
cell.font = HFONT
|
|
cell.alignment = CC
|
|
cell.border = BORD
|
|
|
|
# Write data
|
|
for ri, rec in enumerate(records):
|
|
r = ri + 2
|
|
plate = str(rec['车牌号']).strip()
|
|
is_intv = plate in yellow_plates
|
|
|
|
# Columns 0-13: copy directly (0-indexed in src_h)
|
|
for si in range(14): # 车牌号 ~ 应考核里程(km)
|
|
val = rec.get(src_h[si])
|
|
if val is not None:
|
|
ws.cell(row=r, column=si + 1, value=val)
|
|
ws.cell(row=r, column=si + 1).border = BORD
|
|
ws.cell(row=r, column=si + 1).alignment = CC
|
|
ws.cell(row=r, column=si + 1).font = NFONT
|
|
|
|
# Column 14 (0-indexed): 原TBOX里程(km) - old actual mileage
|
|
tbox = float(rec.get('实际行驶里程(km)', 0) or 0)
|
|
ws.cell(row=r, column=15, value=tbox).border = BORD
|
|
ws.cell(row=r, column=15).alignment = CC
|
|
ws.cell(row=r, column=15).font = NFONT
|
|
|
|
# Column 15 (0-indexed:16): 实际行驶里程(km) - GPS for intv, TBOX for others
|
|
target_km = float(rec.get('应考核里程(km)', 0) or 0)
|
|
if is_intv:
|
|
new_actual = plate_gps.get(plate, 0)
|
|
else:
|
|
new_actual = tbox
|
|
|
|
cell_actual = ws.cell(row=r, column=16, value=new_actual)
|
|
cell_actual.border = BORD
|
|
cell_actual.alignment = CC
|
|
cell_actual.font = NFONT
|
|
|
|
# Column 16 (0-indexed:17): 测试里程(km) = TBOX - GPS
|
|
test_km = round(tbox - new_actual, 2) if is_intv else 0
|
|
ws.cell(row=r, column=17, value=test_km).border = BORD
|
|
ws.cell(row=r, column=17).alignment = CC
|
|
ws.cell(row=r, column=17).font = NFONT
|
|
|
|
# Column 17 (0-indexed:18): 完成率(%)
|
|
rate = round(new_actual / target_km * 100, 2) if target_km > 0 else 0
|
|
ws.cell(row=r, column=18, value=rate).border = BORD
|
|
ws.cell(row=r, column=18).alignment = CC
|
|
ws.cell(row=r, column=18).font = NFONT
|
|
|
|
# Column 18 (0-indexed:19): 是否达标
|
|
new_pass = '达标' if new_actual >= target_km and target_km > 0 else '未达标'
|
|
ws.cell(row=r, column=19, value=new_pass).border = BORD
|
|
ws.cell(row=r, column=19).alignment = CC
|
|
ws.cell(row=r, column=19).font = NFONT
|
|
|
|
# Column 19 (0-indexed:20): 考核状态
|
|
val = rec.get('考核状态')
|
|
ws.cell(row=r, column=20, value=val if val else '进行中').border = BORD
|
|
ws.cell(row=r, column=20).alignment = CC
|
|
ws.cell(row=r, column=20).font = NFONT
|
|
|
|
# Columns 20-22 (0-indexed): 交车日期, 预计还车日期, 实际还车日期
|
|
for si in range(18, 21):
|
|
val = rec.get(src_h[si])
|
|
if val is not None:
|
|
ws.cell(row=r, column=si + 1 + 5, value=val) # +5 offset for new cols
|
|
else:
|
|
ws.cell(row=r, column=si + 1 + 5, value='')
|
|
ws.cell(row=r, column=si + 1 + 5).border = BORD
|
|
ws.cell(row=r, column=si + 1 + 5).alignment = CC
|
|
ws.cell(row=r, column=si + 1 + 5).font = NFONT
|
|
|
|
# ── Formatting: yellow for intervention, red for GPS=0 ──
|
|
if is_intv:
|
|
for c in range(1, len(new_h) + 1):
|
|
cell = ws.cell(row=r, column=c)
|
|
# Don't override red cells
|
|
existing_fill = cell.fill
|
|
is_red = (existing_fill.patternType == 'solid' and
|
|
existing_fill.start_color and
|
|
existing_fill.start_color.rgb and
|
|
'FF0000' in str(existing_fill.start_color.rgb))
|
|
if not is_red:
|
|
cell.fill = YELLOW
|
|
else:
|
|
# Red stays, but also yellow background? No, red means GPS=0, keep it
|
|
pass
|
|
|
|
# Red highlight for GPS=0
|
|
if plate_gps.get(plate, 0) == 0:
|
|
ws.cell(row=r, column=16).fill = RED # 实际行驶里程 column
|
|
ws.cell(row=r, column=16).font = REDF
|
|
|
|
# Column widths
|
|
widths = [12, 8, 8, 12, 10, 26, 18, 22, 10, 22, 13, 13, 10, 14, 16, 16, 14, 12, 10, 8, 12, 12, 12]
|
|
for i, w in enumerate(widths):
|
|
ws.column_dimensions[get_column_letter(i+1)].width = w
|
|
|
|
# ── Save ──
|
|
wb.save(OUT)
|
|
print(f"\n✅ 已保存: {OUT} ({len(records)} rows, {len(new_h)} cols)")
|
|
|
|
# ── Stats ──
|
|
old_p = sum(1 for r in records if str(r.get('是否达标','')).strip() == '达标')
|
|
new_p = 0
|
|
flip_down = []
|
|
flip_up = []
|
|
for rec in records:
|
|
p = str(rec['车牌号']).strip()
|
|
tgt = float(rec.get('应考核里程(km)', 0) or 0)
|
|
old_status = str(rec.get('是否达标','')).strip()
|
|
if p in yellow_plates:
|
|
actual = plate_gps.get(p, 0)
|
|
else:
|
|
actual = float(rec.get('实际行驶里程(km)', 0) or 0)
|
|
n = '达标' if actual >= tgt and tgt > 0 else '未达标'
|
|
if n == '达标':
|
|
new_p += 1
|
|
if old_status == '达标' and n == '未达标':
|
|
flip_down.append((p, float(rec.get('实际行驶里程(km)',0) or 0), actual, tgt))
|
|
elif old_status == '未达标' and n == '达标':
|
|
flip_up.append((p, float(rec.get('实际行驶里程(km)',0) or 0), actual, tgt))
|
|
|
|
print(f"达标变化: {old_p} -> {new_p} ({new_p - old_p:+d})")
|
|
print(f"达→未达标: {len(flip_down)} 辆")
|
|
for p, tb, gp, tg in flip_down[:10]:
|
|
print(f" {p} TBOX={tb:.1f} GPS={gp:.2f} 目标={tg:.0f}")
|
|
print(f"未→达标: {len(flip_up)} 辆")
|
|
for p, tb, gp, tg in flip_up[:10]:
|
|
print(f" {p} TBOX={tb:.1f} GPS={gp:.2f} 目标={tg:.0f}")
|