zkxw/backend/app/services/oa_overtime_hours.py

"""加班套件：工作日/休息日窗口重叠小时（与前端 oa-overtime-hours.ts 对齐）"""
from __future__ import annotations

import re
from datetime import datetime, timedelta
from typing import Any, Callable, Dict, List, Optional, Tuple

Interval = Tuple[float, float]

_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_OVERTIME_WINDOWS = {
    "workdayOvertimeStart": "17:30",
    "workdayOvertimeEnd": "08:30",
    "restBreakStart": "11:30",
    "restBreakEnd": "13: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 _format_ymd(d: datetime) -> str:
    return d.strftime("%Y-%m-%d")


def _day_start_ms(d: datetime) -> float:
    day0 = datetime(d.year, d.month, d.day)
    return day0.timestamp() * 1000


def _day_at_minutes(d: datetime, minutes: int) -> float:
    return _day_start_ms(d) + minutes * 60000


def _merge_intervals(intervals: List[Interval]) -> List[Interval]:
    if not intervals:
        return []
    sorted_iv = sorted(intervals, key=lambda x: x[0])
    merged: List[Interval] = [(sorted_iv[0][0], sorted_iv[0][1])]
    for s, e in sorted_iv[1:]:
        last = merged[-1]
        if s <= last[1]:
            merged[-1] = (last[0], max(last[1], e))
        else:
            merged.append((s, e))
    return merged


def _intersect_interval(a: Interval, b: Interval) -> Optional[Interval]:
    lo = max(a[0], b[0])
    hi = min(a[1], b[1])
    return (lo, hi) if hi > lo else None


def _subtract_interval(base: Interval, sub: Interval) -> List[Interval]:
    bs, be = base
    ss, se = sub
    if se <= bs or ss >= be:
        return [base]
    out: List[Interval] = []
    if ss > bs:
        out.append((bs, ss))
    if se < be:
        out.append((se, be))
    return out


def _subtract_intervals(base: Interval, subs: List[Interval]) -> List[Interval]:
    result: List[Interval] = [base]
    for sub in subs:
        nxt: List[Interval] = []
        for r in result:
            nxt.extend(_subtract_interval(r, sub))
        result = nxt
    return result


def _interval_minutes(intervals: List[Interval]) -> float:
    return sum((b - a) / 60000 for a, b in intervals)


def _build_workday_windows(anchor_day: datetime, windows: Dict[str, str]) -> Optional[Interval]:
    start_min = _parse_hm(windows.get("workdayOvertimeStart"))
    end_min = _parse_hm(windows.get("workdayOvertimeEnd"))
    if start_min is None or end_min is None:
        return None
    anchor_start = _day_at_minutes(anchor_day, start_min)
    next_day = anchor_day + timedelta(days=1)
    anchor_end = _day_at_minutes(next_day, end_min)
    if anchor_end <= anchor_start:
        return None
    return (anchor_start, anchor_end)


def _build_rest_day_segments(day: datetime, windows: Dict[str, str]) -> List[Interval]:
    break_start = _parse_hm(windows.get("restBreakStart"))
    break_end = _parse_hm(windows.get("restBreakEnd"))
    if break_start is None or break_end is None:
        return []

    day0 = _day_start_ms(day)
    next_day = day + timedelta(days=1)
    day_end = _day_start_ms(next_day)

    segments: List[Interval] = []
    if break_start > 0:
        segments.append((day0, _day_at_minutes(day, break_start)))
    if break_end < 24 * 60:
        segments.append((_day_at_minutes(day, break_end), day_end))
    return [(a, b) for a, b in segments if b > a]


def calc_overtime_hours(
    start_str: Any,
    end_str: Any,
    windows: Optional[Dict[str, str]] = None,
    is_rest_day_fn: Optional[Callable[[str], bool]] = None,
) -> Optional[float]:
    """返回 1 位小数小时；非法或 end<=start 返回 None。"""
    if is_rest_day_fn is None:
        raise TypeError("is_rest_day_fn is required")

    win = dict(DEFAULT_OVERTIME_WINDOWS)
    if isinstance(windows, dict):
        win.update({k: str(windows.get(k) or win[k]) for k in DEFAULT_OVERTIME_WINDOWS})

    start = _parse_dt(start_str)
    end = _parse_dt(end_str)
    if not start or not end or end <= start:
        return None

    query: Interval = (start.timestamp() * 1000, end.timestamp() * 1000)
    start_day = datetime(start.year, start.month, start.day)
    end_day = datetime(end.year, end.month, end.day)

    workday_intervals: List[Interval] = []
    anchor = start_day - timedelta(days=1)
    last_anchor = end_day
    while anchor <= last_anchor:
        ymd = _format_ymd(anchor)
        if not is_rest_day_fn(ymd):
            win_iv = _build_workday_windows(anchor, win)
            if win_iv:
                hit = _intersect_interval(win_iv, query)
                if hit:
                    workday_intervals.append(hit)
        anchor += timedelta(days=1)

    workday_merged = _merge_intervals(workday_intervals)

    rest_intervals: List[Interval] = []
    day = start_day
    while day <= end_day:
        ymd = _format_ymd(day)
        if is_rest_day_fn(ymd):
            for seg in _build_rest_day_segments(day, win):
                hit = _intersect_interval(seg, query)
                if hit:
                    rest_intervals.extend(_subtract_intervals(hit, workday_merged))
        day += timedelta(days=1)

    total_minutes = _interval_minutes(workday_merged) + _interval_minutes(rest_intervals)
    return round(total_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

    start_field: Optional[str] = None
    end_field: Optional[str] = None
    duration_field: Optional[str] = None
    date_fields: List[str] = []

    for c in children:
        if not isinstance(c, dict):
            continue
        field = c.get("field")
        if not field:
            continue
        c_props = c.get("props") if isinstance(c.get("props"), dict) else {}
        kit_field = c_props.get("_fc_kit_field")
        if kit_field == "overtimeStart":
            start_field = str(field)
            continue
        if kit_field == "overtimeEnd":
            end_field = str(field)
            continue
        if kit_field == "overtimeDuration":
            duration_field = str(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)

    if start_field is None and len(date_fields) >= 1:
        start_field = date_fields[0]
    if end_field is None and len(date_fields) >= 2:
        end_field = date_fields[1]
    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_overtime_kit_hours(
    form_schema: Any,
    form_data: Dict[str, Any],
    is_rest_day_fn: Callable[[str], bool],
) -> Dict[str, Any]:
    """
    遍历 overtimeKit：未手改则覆盖时长；已手改（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 "") != "overtimeKit":
            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 {}
        win = {
            "workdayOvertimeStart": str(
                props.get("workdayOvertimeStart") or DEFAULT_OVERTIME_WINDOWS["workdayOvertimeStart"]
            ),
            "workdayOvertimeEnd": str(
                props.get("workdayOvertimeEnd") or DEFAULT_OVERTIME_WINDOWS["workdayOvertimeEnd"]
            ),
            "restBreakStart": str(
                props.get("restBreakStart") or DEFAULT_OVERTIME_WINDOWS["restBreakStart"]
            ),
            "restBreakEnd": str(
                props.get("restBreakEnd") or DEFAULT_OVERTIME_WINDOWS["restBreakEnd"]
            ),
        }
        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_overtime_hours(data.get(start_f), data.get(end_f), win, is_rest_day_fn)
        if hours is not None:
            data[dur_f] = hours
    return data


def validate_overtime_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 "") != "overtimeKit":
            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