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.
204 lines
10 KiB
Python
204 lines
10 KiB
Python
"""无 I/O 的考核计算。规则、候选车辆和来源优先级由每次请求明确给定。"""
|
|
import calendar
|
|
import re
|
|
from collections import defaultdict
|
|
from datetime import date, datetime, timedelta
|
|
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
|
|
|
|
|
|
HEADERS = ["车牌号", "考核年份", "考核月份", "部门名称", "销售经理", "客户名称",
|
|
"合同编号", "项目名称", "考核目标ID", "考核目标", "考核开始日期", "考核结束日期",
|
|
"考核天数", "应考核里程(km)", "实际行驶里程(km)", "完成率(%)", "是否达标",
|
|
"考核状态", "交车日期", "预计还车日期", "实际还车日期"]
|
|
|
|
|
|
def plate(value):
|
|
return re.sub(r"\s+", "", str(value or "")).upper()
|
|
|
|
|
|
def day(value):
|
|
if isinstance(value, datetime):
|
|
return value.date()
|
|
if isinstance(value, date):
|
|
return value
|
|
return datetime.fromisoformat(str(value).strip()).date()
|
|
|
|
|
|
def number(value):
|
|
if value is None or isinstance(value, bool) or str(value).strip() == "":
|
|
raise ValueError("缺少数值；空值不能替代为 0")
|
|
try:
|
|
result = Decimal(str(value))
|
|
except InvalidOperation as exc:
|
|
raise ValueError(f"非数值: {value}") from exc
|
|
if not result.is_finite() or result < 0:
|
|
raise ValueError("里程及目标必须为有限的非负数")
|
|
return result
|
|
|
|
|
|
def rounded(value):
|
|
return float(value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP))
|
|
|
|
|
|
def dates(start, end):
|
|
return [start + timedelta(days=i) for i in range((end - start).days + 1)]
|
|
|
|
|
|
def build(lease_import, mileage_imports, config):
|
|
"""返回草稿及逐日依据。仅 ready 状态可正式导出。
|
|
|
|
规则清单定义考核范围。未列入规则的车辆只记录为 excluded，不自动套用档位。
|
|
相同来源内冲突不能自动择大；不同提供方按请求给定的顺序按日选取。
|
|
"""
|
|
if not re.fullmatch(r"\d{4}-\d{2}", str(config.get("month", ""))):
|
|
raise ValueError("month 必须是 YYYY-MM")
|
|
first = day(config["month"] + "-01")
|
|
month_days = calendar.monthrange(first.year, first.month)[1]
|
|
last = first.replace(day=month_days)
|
|
if not isinstance(config.get("test_plates"), list):
|
|
raise ValueError("必须明确提供 test_plates；无测试车辆时传 []")
|
|
tests = {plate(p) for p in config["test_plates"]}
|
|
threshold = number(config.get("test_threshold_km", 50))
|
|
priorities = config.get("source_priority", {})
|
|
for source in ("hydrogen", "tbox", "gps"):
|
|
if not isinstance(priorities.get(source), list) or not priorities[source]:
|
|
raise ValueError(f"必须明确提供 source_priority.{source}")
|
|
if len(set(priorities[source])) != len(priorities[source]):
|
|
raise ValueError(f"来源优先级有重复: {source}")
|
|
rules = defaultdict(list)
|
|
if not isinstance(config.get("rules"), list) or not config["rules"]:
|
|
raise ValueError("rules 必须明确列出参与考核车辆和有效期，不可为空")
|
|
for raw in config["rules"]:
|
|
r = dict(raw)
|
|
for field in ("plate", "target_id", "target_name", "monthly_target_km", "valid_from", "valid_to"):
|
|
if r.get(field) is None or str(r[field]).strip() == "":
|
|
raise ValueError(f"规则缺字段: {field}")
|
|
r["plate"] = plate(r["plate"])
|
|
r["start"], r["end"] = day(r["valid_from"]), day(r["valid_to"])
|
|
r["target"] = number(r["monthly_target_km"])
|
|
if r["start"] > r["end"] or r["target"] <= 0:
|
|
raise ValueError("规则有效期或月目标不合法")
|
|
rules[r["plate"]].append(r)
|
|
for p, rs in rules.items():
|
|
rs.sort(key=lambda r: r["start"])
|
|
if any(a["end"] >= b["start"] for a, b in zip(rs, rs[1:])):
|
|
raise ValueError(f"规则有效期重叠: {p}")
|
|
|
|
issues = []
|
|
def issue(code, p, message, row=None, **extra):
|
|
issues.append(dict(code=code, plate=p, message=message, source_row=row, **extra))
|
|
|
|
# 以有效规则限定本月考核范围，不把全量租赁车辆直接当作考核车辆。
|
|
relevant = {p for p, rs in rules.items() if any(r["start"] <= last and r["end"] >= first for r in rs)}
|
|
for item in lease_import.get("issues", []):
|
|
if not item.get("plate") or item["plate"] in relevant:
|
|
issues.append(dict(item))
|
|
daily = defaultdict(dict)
|
|
for imp in mileage_imports:
|
|
source, provider = imp["source"], imp["provider"]
|
|
if provider not in priorities[source]:
|
|
raise ValueError(f"提供方未列入 {source} 优先级: {provider}")
|
|
for item in imp.get("issues", []):
|
|
if not item.get("plate") or item["plate"] in relevant:
|
|
issues.append(dict(item))
|
|
for r in imp["rows"]:
|
|
d = day(r["date"])
|
|
if r["plate"] not in relevant or not first <= d <= last:
|
|
continue
|
|
key = (r["plate"], d, source)
|
|
value = number(r["km"])
|
|
evidence = dict(km=rounded(value), provider=provider, import_id=imp.get("id"),
|
|
source_row=r["source_row"])
|
|
existing = daily[key].get(provider)
|
|
if existing is not None and existing[0] != value:
|
|
issue("mileage_conflict", r["plate"], f"{d} 的 {source}/{provider} 存在不同里程", r["source_row"])
|
|
else:
|
|
daily[key][provider] = (value, evidence)
|
|
|
|
def pick(p, d, source):
|
|
available = daily.get((p, d, source), {})
|
|
return next((available[v] for v in priorities[source] if v in available), None)
|
|
|
|
segments, excluded, seen, active_plates = [], [], {}, set()
|
|
for lease in lease_import["rows"]:
|
|
p = lease["plate"]
|
|
if p not in relevant:
|
|
excluded.append(dict(plate=p, lease_id=lease["lease_id"], reason="未列入本月考核规则"))
|
|
continue
|
|
ident = lease["lease_id"]
|
|
canonical = {k: v for k, v in lease.items() if k != "source_row"}
|
|
if ident in seen:
|
|
if seen[ident] != canonical:
|
|
issue("lease_conflict", p, f"交车记录 {ident} 对应多条不同记录", lease["source_row"])
|
|
continue
|
|
seen[ident] = canonical
|
|
start = max(first, day(lease["delivery_time"]))
|
|
finish = min(last, day(lease["return_time"])) if lease["return_time"] else last
|
|
if start > finish:
|
|
continue
|
|
active_plates.add(p)
|
|
for r in rules[p]:
|
|
begin, end = max(start, r["start"]), min(finish, r["end"])
|
|
if begin <= end:
|
|
segments.append((lease, r, begin, end))
|
|
|
|
occupancy = {}
|
|
for lease, rule, start, end in segments:
|
|
for d in dates(start, end):
|
|
key = lease["plate"], d
|
|
if key in occupancy:
|
|
issue("overlapping_lease_day", lease["plate"],
|
|
f"{d} 被交车记录 {occupancy[key]} 和 {lease['lease_id']} 同时占用", lease["source_row"])
|
|
occupancy[key] = lease["lease_id"]
|
|
|
|
records, trace = [], []
|
|
for lease, rule, start, end in segments:
|
|
p = lease["plate"]
|
|
days = dates(start, end)
|
|
has_hydrogen = any(pick(p, d, "hydrogen") is not None for d in days)
|
|
source = "hydrogen" if has_hydrogen else ("gps" if p in tests else "tbox")
|
|
total, complete = Decimal(0), True
|
|
for d in days:
|
|
values = {s: pick(p, d, s) for s in ("hydrogen", "tbox", "gps")}
|
|
is_test = False
|
|
if p in tests and has_hydrogen:
|
|
if values["tbox"] is None or values["gps"] is None:
|
|
complete = False
|
|
issue("test_detection_missing", p, f"{d} 缺 TBOX/GPS，无法判断测试日", lease["source_row"])
|
|
else:
|
|
is_test = values["tbox"][0] - values["gps"][0] > threshold
|
|
chosen = values[source]
|
|
included = None
|
|
if is_test:
|
|
included = Decimal(0)
|
|
elif chosen is None:
|
|
complete = False
|
|
issue("missing_mileage_day", p, f"{d} 缺少 {source} 里程，不能按 0 处理", lease["source_row"])
|
|
else:
|
|
included = chosen[0]
|
|
total += included
|
|
trace.append(dict(plate=p, lease_id=lease["lease_id"], date=d.isoformat(),
|
|
source=source, test_day=is_test,
|
|
included_km=rounded(included) if included is not None else None,
|
|
evidence={s: v[1] if v is not None else None for s, v in values.items()}))
|
|
target = Decimal(str(rounded(rule["target"] * len(days) / month_days)))
|
|
actual = Decimal(str(rounded(total))) if complete else None
|
|
vals = [p, first.year, first.month, lease["department"], lease["manager"], lease["customer"],
|
|
lease["contract_code"], lease["project"], rule["target_id"], rule["target_name"],
|
|
start.isoformat(), end.isoformat(), len(days), float(target),
|
|
float(actual) if actual is not None else None,
|
|
rounded(actual / target * 100) if actual is not None else None,
|
|
("达标" if actual >= target else "未达标") if actual is not None else "待核对",
|
|
"已完成" if lease["return_time"] and day(lease["return_time"]) <= last else "进行中",
|
|
day(lease["delivery_time"]).isoformat(), None,
|
|
day(lease["return_time"]).isoformat() if lease["return_time"] else None]
|
|
records.append(dict(zip(HEADERS, vals)))
|
|
if not records:
|
|
issue("no_assessment_records", "", "本月没有可生成的考核记录")
|
|
records.sort(key=lambda r: (r["车牌号"], r["考核开始日期"], r["合同编号"]))
|
|
return dict(status="needs_review" if issues else "ready", month=config["month"], records=records,
|
|
issues=issues, daily_trace=trace, excluded=excluded,
|
|
configured_plates_without_active_lease=sorted(relevant - active_plates),
|
|
summary=dict(records=len(records), vehicles=len({r["车牌号"] for r in records}),
|
|
issues=len(issues), excluded_lease_records=len(excluded)))
|