|
|
@@ -0,0 +1,228 @@
|
|
|
+"""请假套件:双段工时重叠小时(与前端 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_<durationField>])则仅规范化一位小数。
|
|
|
+ """
|
|
|
+ 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
|