60 lines
2.0 KiB
Python
60 lines
2.0 KiB
Python
from ast import parse, walk, Call, Constant, keyword
|
|
from datetime import date
|
|
from pathlib import Path
|
|
import re
|
|
|
|
import openpyxl
|
|
import pymysql
|
|
|
|
|
|
def connection_kwargs_from_existing_script():
|
|
tree = parse(Path('/Users/kkfluous/Downloads/replace_gps_jun.py').read_text())
|
|
for node in walk(tree):
|
|
if not isinstance(node, Call):
|
|
continue
|
|
func = node.func
|
|
if getattr(func, 'attr', None) != 'connect':
|
|
continue
|
|
values = {}
|
|
for item in node.keywords:
|
|
if isinstance(item, keyword) and isinstance(item.value, Constant):
|
|
values[item.arg] = item.value.value
|
|
if {'host', 'user', 'password', 'database'} <= values.keys():
|
|
return values
|
|
raise RuntimeError('未在既有脚本中找到数据库连接配置')
|
|
|
|
|
|
src = openpyxl.load_workbook(
|
|
'/Users/kkfluous/Downloads/7月核算结果/租赁任务考核_2026年7月.xlsx',
|
|
read_only=True,
|
|
data_only=True,
|
|
)['业务考核视图']
|
|
headers = [c.value for c in src[1]]
|
|
plate_col = headers.index('车牌号')
|
|
plates = sorted({re.sub(r'\s+', '', str(r[plate_col] or '')).upper() for r in src.iter_rows(min_row=2, values_only=True) if r[plate_col]})
|
|
|
|
cfg = connection_kwargs_from_existing_script()
|
|
cfg.update(charset='utf8mb4', connect_timeout=8, read_timeout=30, write_timeout=30)
|
|
conn = pymysql.connect(**cfg)
|
|
try:
|
|
with conn.cursor() as cur:
|
|
placeholders = ','.join(['%s'] * len(plates))
|
|
cur.execute(
|
|
f'''SELECT plate_number, dates, total_mileage
|
|
FROM ln_vehicle_g7_mileage
|
|
WHERE dates >= %s AND dates <= %s
|
|
AND plate_number IN ({placeholders})''',
|
|
[date(2026, 7, 1), date(2026, 7, 31), *plates],
|
|
)
|
|
rows = cur.fetchall()
|
|
finally:
|
|
conn.close()
|
|
|
|
db_plates = {re.sub(r'\s+', '', str(r[0] or '')).upper() for r in rows}
|
|
print({
|
|
'assessment_plates': len(plates),
|
|
'database_rows': len(rows),
|
|
'database_plates': len(db_plates),
|
|
'uncovered_plates': len(set(plates) - db_plates),
|
|
})
|