"""无 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)))