zkxw/backend/app/services/oa_leave_hours.py

"""请假套件：双段工时重叠小时（与前端 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