"""请假套件:双段工时重叠小时(与前端 oa-leave-hours.ts 对齐)""" from __future__ import annotations import re from datetime import datetime, timedelta from typing import Any, Dict, List, Optional, Tuple _DT_RE = re.compile(r"^(\d{4})-(\d{2})-(\d{2})[ T](\d{2}):(\d{2})$") _HM_RE = re.compile(r"^(\d{1,2}):(\d{2})$") DEFAULT_LEAVE_WORK_HOURS = { "morningStart": "08:30", "morningEnd": "12:00", "afternoonStart": "13:00", "afternoonEnd": "17:30", } MANUAL_PREFIX = "__manual_" def _parse_hm(hm: Any) -> Optional[int]: m = _HM_RE.match(str(hm or "").strip()) if not m: return None h, mi = int(m.group(1)), int(m.group(2)) if h < 0 or h > 23 or mi < 0 or mi > 59: return None return h * 60 + mi def _parse_dt(s: Any) -> Optional[datetime]: m = _DT_RE.match(str(s or "").strip()) if not m: return None try: return datetime( int(m.group(1)), int(m.group(2)), int(m.group(3)), int(m.group(4)), int(m.group(5)), ) except ValueError: return None def _overlap_minutes_on_day( day: datetime, range_start: datetime, range_end: datetime, work_hours: Dict[str, str], ) -> float: day0 = datetime(day.year, day.month, day.day) am_a = _parse_hm(work_hours.get("morningStart")) am_b = _parse_hm(work_hours.get("morningEnd")) pm_a = _parse_hm(work_hours.get("afternoonStart")) pm_b = _parse_hm(work_hours.get("afternoonEnd")) if None in (am_a, am_b, pm_a, pm_b): return 0.0 if am_a >= am_b or pm_a >= pm_b: return 0.0 segments = [ (day0 + timedelta(minutes=am_a), day0 + timedelta(minutes=am_b)), (day0 + timedelta(minutes=pm_a), day0 + timedelta(minutes=pm_b)), ] total = 0.0 for a, b in segments: lo = max(range_start, a) hi = min(range_end, b) if hi > lo: total += (hi - lo).total_seconds() / 60.0 return total def calc_leave_hours( start_str: Any, end_str: Any, work_hours: Optional[Dict[str, str]] = None, ) -> Optional[float]: """返回 1 位小数小时;非法或 end<=start 返回 None。""" wh = dict(DEFAULT_LEAVE_WORK_HOURS) if isinstance(work_hours, dict): wh.update({k: str(work_hours.get(k) or wh[k]) for k in DEFAULT_LEAVE_WORK_HOURS}) start = _parse_dt(start_str) end = _parse_dt(end_str) if not start or not end or end <= start: return None minutes = 0.0 cursor = datetime(start.year, start.month, start.day) last = datetime(end.year, end.month, end.day) while cursor <= last: minutes += _overlap_minutes_on_day(cursor, start, end, wh) cursor += timedelta(days=1) return round(minutes / 60.0, 1) def _extract_rules(form_schema: Any) -> List[Dict[str, Any]]: if isinstance(form_schema, list): return [r for r in form_schema if isinstance(r, dict)] if isinstance(form_schema, dict): rules = form_schema.get("rule") or form_schema.get("rules") or form_schema.get("fields") if isinstance(rules, list): return [r for r in rules if isinstance(r, dict)] if form_schema.get("type") or form_schema.get("field"): return [form_schema] return [] def _walk(rules: List[Dict[str, Any]], out: List[Dict[str, Any]]) -> None: for r in rules: if not isinstance(r, dict): continue out.append(r) children = r.get("children") if isinstance(children, list): _walk([c for c in children if isinstance(c, dict)], out) props = r.get("props") if isinstance(r.get("props"), dict) else {} nested = props.get("rule") if isinstance(nested, list): _walk([c for c in nested if isinstance(c, dict)], out) def _kit_child_fields(kit: Dict[str, Any]) -> Tuple[Optional[str], Optional[str], Optional[str]]: """返回 (start_field, end_field, duration_field)。""" props = kit.get("props") if isinstance(kit.get("props"), dict) else {} start_f = props.get("startField") end_f = props.get("endField") dur_f = props.get("durationField") if start_f and end_f and dur_f: return str(start_f), str(end_f), str(dur_f) children = kit.get("children") if not isinstance(children, list): children = [] p_rule = props.get("rule") if isinstance(p_rule, list): children = p_rule date_fields: List[str] = [] duration_field: Optional[str] = None for c in children: if not isinstance(c, dict): continue field = c.get("field") if not field: continue t = str(c.get("type") or "") if t in ("datePicker", "date", "timePicker", "time"): date_fields.append(str(field)) elif t in ("inputNumber", "number") and duration_field is None: duration_field = str(field) elif t in ("inputNumber", "number") and "时长" in str(c.get("title") or ""): duration_field = str(field) start_field = date_fields[0] if len(date_fields) >= 1 else None end_field = date_fields[1] if len(date_fields) >= 2 else None return start_field, end_field, duration_field def _round1(v: Any) -> Optional[float]: try: n = float(v) except (TypeError, ValueError): return None if n < 0: return None return round(n, 1) def apply_leave_kit_hours(form_schema: Any, form_data: Dict[str, Any]) -> Dict[str, Any]: """ 遍历 leaveKit:未手改则覆盖时长;已手改(form_data[__manual_])则仅规范化一位小数。 """ data = dict(form_data or {}) flat: List[Dict[str, Any]] = [] _walk(_extract_rules(form_schema), flat) for rule in flat: if str(rule.get("type") or "") != "leaveKit": continue start_f, end_f, dur_f = _kit_child_fields(rule) if not start_f or not end_f or not dur_f: continue props = rule.get("props") if isinstance(rule.get("props"), dict) else {} wh = { "morningStart": str(props.get("morningStart") or DEFAULT_LEAVE_WORK_HOURS["morningStart"]), "morningEnd": str(props.get("morningEnd") or DEFAULT_LEAVE_WORK_HOURS["morningEnd"]), "afternoonStart": str( props.get("afternoonStart") or DEFAULT_LEAVE_WORK_HOURS["afternoonStart"] ), "afternoonEnd": str(props.get("afternoonEnd") or DEFAULT_LEAVE_WORK_HOURS["afternoonEnd"]), } manual_key = MANUAL_PREFIX + dur_f is_manual = bool(data.get(manual_key)) if is_manual: rounded = _round1(data.get(dur_f)) if rounded is not None: data[dur_f] = rounded continue hours = calc_leave_hours(data.get(start_f), data.get(end_f), wh) if hours is not None: data[dur_f] = hours return data def validate_leave_kit_form_values(form_schema: Any, form_data: Dict[str, Any]) -> List[str]: """提交校验:结束>=开始;手改时长合法。""" errors: List[str] = [] data = form_data or {} flat: List[Dict[str, Any]] = [] _walk(_extract_rules(form_schema), flat) for rule in flat: if str(rule.get("type") or "") != "leaveKit": continue start_f, end_f, dur_f = _kit_child_fields(rule) if not start_f or not end_f or not dur_f: continue start = _parse_dt(data.get(start_f)) end = _parse_dt(data.get(end_f)) if data.get(start_f) and data.get(end_f): if not start or not end: errors.append("请假起止时间格式须为 YYYY-MM-DD HH:mm") elif end <= start: errors.append("请假结束时间须晚于开始时间") if MANUAL_PREFIX + dur_f in data or data.get(MANUAL_PREFIX + dur_f): if data.get(dur_f) is not None and data.get(dur_f) != "": if _round1(data.get(dur_f)) is None: errors.append("请假时长须为非负数字(1 位小数)") return errors