shishengliang
65904c8270
Preserve August hydrogen revision and September audit evidence; calculate interrupted vehicle history across all prior months. Record remaining carry-policy and layout-baseline discrepancies.
122 lines
5.8 KiB
Python
122 lines
5.8 KiB
Python
"""文件适配器。原表作为数据读取；不执行 Query 页或单元格公式。"""
|
|
from io import BytesIO
|
|
from datetime import datetime
|
|
from zipfile import ZipFile
|
|
|
|
import openpyxl
|
|
from openpyxl.styles import Alignment, Font, PatternFill
|
|
|
|
from .core import HEADERS, day, number, plate
|
|
|
|
|
|
def open_book(blob):
|
|
if len(blob) > 20 * 1024 * 1024:
|
|
raise ValueError("文件超过 20 MB")
|
|
with ZipFile(BytesIO(blob)) as z:
|
|
if sum(x.file_size for x in z.infolist()) > 200 * 1024 * 1024:
|
|
raise ValueError("解压后文件超过 200 MB")
|
|
return openpyxl.load_workbook(BytesIO(blob), read_only=True, data_only=True)
|
|
|
|
|
|
def rows_with_headers(blob, required):
|
|
wb = open_book(blob)
|
|
try:
|
|
for ws in wb:
|
|
rows = iter(ws.values)
|
|
for n in range(1, 11):
|
|
raw = next(rows, None)
|
|
if raw is None:
|
|
break
|
|
header = [str(c).strip() if c is not None else "" for c in raw]
|
|
if required <= set(header):
|
|
if len([h for h in header if h]) != len(set(h for h in header if h)):
|
|
raise ValueError("表头存在重复字段")
|
|
result = [(i, dict(zip(header, r))) for i, r in enumerate(rows, n + 1)
|
|
if any(v is not None and v != "" for v in r)]
|
|
return ws.title, result
|
|
raise ValueError("未找到所需表头: " + ", ".join(sorted(required)))
|
|
finally:
|
|
wb.close()
|
|
|
|
|
|
def leases(blob):
|
|
fields = {"vehicle_id", "plate_number", "contract_id", "contract_code", "project_name", "customer_name",
|
|
"business_department_name", "business_manager_name", "delivery_vehicle_id", "delivery_time",
|
|
"return_vehicle_task_id", "return_time"}
|
|
sheet, raw = rows_with_headers(blob, fields)
|
|
result, issues = [], []
|
|
for n, r in raw:
|
|
p = plate(r["plate_number"])
|
|
try:
|
|
for key in ("plate_number", "delivery_vehicle_id", "delivery_time", "contract_id", "contract_code",
|
|
"customer_name", "business_department_name", "business_manager_name"):
|
|
if r[key] is None or str(r[key]).strip() == "":
|
|
raise ValueError(f"缺少 {key}")
|
|
start = datetime.fromisoformat(str(r["delivery_time"]))
|
|
finish = datetime.fromisoformat(str(r["return_time"])) if r["return_time"] else None
|
|
if finish and finish < start:
|
|
raise ValueError("还车时间早于交车时间")
|
|
if r["return_vehicle_task_id"] and finish is None:
|
|
raise ValueError("有还车任务但无还车时间，不能直接当作未还车")
|
|
result.append(dict(plate=p, vehicle_id=str(r["vehicle_id"]), lease_id=str(r["delivery_vehicle_id"]),
|
|
contract_id=str(r["contract_id"]), contract_code=str(r["contract_code"]),
|
|
project=r["project_name"] or "", customer=r["customer_name"],
|
|
department=r["business_department_name"], manager=r["business_manager_name"],
|
|
delivery_time=start.isoformat(), return_time=finish.isoformat() if finish else None,
|
|
source_row=n))
|
|
except (ValueError, TypeError) as exc:
|
|
issues.append(dict(code="invalid_lease", plate=p, source_row=n, message=str(exc)))
|
|
return dict(kind="leases", sheet=sheet, raw_rows=len(raw), rows=result, issues=issues)
|
|
|
|
|
|
def mileage(blob, source, provider):
|
|
if source not in ("hydrogen", "tbox", "gps"):
|
|
raise ValueError("source 只支持 hydrogen、tbox、gps")
|
|
if not provider or len(provider) > 100:
|
|
raise ValueError("必须提供 provider，最长 100 字符")
|
|
# 第一版采用明确的长表契约，不猜供应商列序或原始单位。
|
|
sheet, raw = rows_with_headers(blob, {"plate_number", "date", "mileage_km"})
|
|
result, issues = [], []
|
|
for n, r in raw:
|
|
p = plate(r["plate_number"])
|
|
try:
|
|
if not p:
|
|
raise ValueError("缺少车牌")
|
|
result.append(dict(plate=p, date=day(r["date"]).isoformat(),
|
|
km=str(number(r["mileage_km"])), source_row=n))
|
|
except (ValueError, TypeError) as exc:
|
|
issues.append(dict(code="invalid_mileage", plate=p, source_row=n, message=str(exc)))
|
|
return dict(kind="mileage", source=source, provider=provider, sheet=sheet,
|
|
raw_rows=len(raw), rows=result, issues=issues)
|
|
|
|
|
|
def export(result):
|
|
if result["status"] != "ready":
|
|
raise ValueError("尚有待核对问题，不能导出正式考核源表")
|
|
wb = openpyxl.Workbook()
|
|
ws = wb.active
|
|
ws.title = "业务考核视图"
|
|
ws.append(HEADERS)
|
|
date_fields = {"考核开始日期", "考核结束日期", "交车日期", "预计还车日期", "实际还车日期"}
|
|
for r in result["records"]:
|
|
ws.append([datetime.fromisoformat(r[h]) if h in date_fields and r[h] else r[h] for h in HEADERS])
|
|
# 文本始终按字符串保存，合同/客户等内容不能变成 Excel 公式。
|
|
for row in ws:
|
|
for cell in row:
|
|
if isinstance(cell.value, str):
|
|
cell.data_type = "s"
|
|
if isinstance(cell.value, datetime):
|
|
cell.number_format = "yyyy-mm-dd"
|
|
cell.alignment = Alignment(vertical="center")
|
|
for cell in ws[1]:
|
|
cell.fill = PatternFill("solid", fgColor="4472C4")
|
|
cell.font = Font(bold=True, color="FFFFFF")
|
|
for i, h in enumerate(HEADERS, 1):
|
|
ws.column_dimensions[openpyxl.utils.get_column_letter(i)].width = 24 if h in date_fields else 20
|
|
ws.freeze_panes = "A2"
|
|
ws.auto_filter.ref = ws.dimensions
|
|
stream = BytesIO()
|
|
wb.save(stream)
|
|
wb.close()
|
|
return stream.getvalue()
|