zkxw/backend/app/services/oa_trip_days.py

"""出差套件：半日制行程时长（与前端 oa-trip-days.ts 对齐）"""
from __future__ import annotations

import re
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Tuple

_DATE_RE = re.compile(r"^(\d{4})-(\d{2})-(\d{2})$")

MANUAL_PREFIX = "__manual_"

_LEG_REQUIRED_FIELDS = (
    "transport",
    "tripType",
    "fromCity",
    "toCity",
    "startDate",
    "startHalf",
    "endDate",
    "endHalf",
)


def _day_index(date_str: Any) -> Optional[float]:
    m = _DATE_RE.match(str(date_str or "").strip())
    if not m:
        return None
    try:
        y, mo, d = int(m.group(1)), int(m.group(2)), int(m.group(3))
        dt = datetime(y, mo, d, tzinfo=timezone.utc)
    except ValueError:
        return None
    return dt.timestamp() / 86400.0


def _start_offset(half: Any) -> Optional[float]:
    h = str(half or "").strip()
    if h == "上午":
        return 0.0
    if h == "下午":
        return 0.5
    return None


def _end_offset(half: Any) -> Optional[float]:
    h = str(half or "").strip()
    if h == "上午":
        return 0.5
    if h == "下午":
        return 1.0
    return None


def calc_trip_leg_days(
    start_date: Any,
    start_half: Any,
    end_date: Any,
    end_half: Any,
) -> Optional[float]:
    """返回 0.5 步进天数；非法或 end<=start 返回 None。"""
    ds = _day_index(start_date)
    de = _day_index(end_date)
    so = _start_offset(start_half)
    eo = _end_offset(end_half)
    if ds is None or de is None or so is None or eo is None:
        return None
    duration = de + eo - (ds + so)
    if not (duration > 0):
        return None
    return round(duration * 2) / 2


def sum_trip_leg_days(legs: Any) -> float:
    total = 0.0
    for leg in legs or []:
        if not isinstance(leg, dict):
            continue
        try:
            n = float(leg.get("duration"))
        except (TypeError, ValueError):
            continue
        if n >= 0:
            total += n
    return round(total * 2) / 2


def is_half_day_multiple(n: Any) -> bool:
    try:
        v = float(n)
    except (TypeError, ValueError):
        return False
    if v < 0:
        return False
    return abs(v * 2 - round(v * 2)) < 1e-9


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_trip_fields(kit: Dict[str, Any]) -> Tuple[Optional[str], Optional[str]]:
    """返回 (trip_list_field, trip_days_field)。"""
    props = kit.get("props") if isinstance(kit.get("props"), dict) else {}
    list_f = props.get("tripListField")
    days_f = props.get("tripDaysField")
    if list_f and days_f:
        return str(list_f), str(days_f)

    children = kit.get("children")
    if not isinstance(children, list):
        children = []
        p_rule = props.get("rule")
        if isinstance(p_rule, list):
            children = p_rule

    list_field: Optional[str] = None
    days_field: Optional[str] = None
    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_key = str(c_props.get("_fc_kit_field") or "")
        if kit_key == "tripList":
            list_field = str(field)
        elif kit_key == "tripDays":
            days_field = str(field)

    return list_field, days_field


def _walk_trip_kits(form_schema: Any) -> List[Tuple[Optional[str], Optional[str]]]:
    flat: List[Dict[str, Any]] = []
    _walk(_extract_rules(form_schema), flat)
    kits: List[Tuple[Optional[str], Optional[str]]] = []
    for rule in flat:
        if str(rule.get("type") or "") != "tripKit":
            continue
        kits.append(_kit_trip_fields(rule))
    return kits


def apply_trip_kit_days(form_schema: Any, form_data: Dict[str, Any]) -> Dict[str, Any]:
    """
    遍历 tripKit：行程未手改则覆盖 duration；出差天数未手改则覆盖为各行程时长之和。
    """
    data = dict(form_data or {})
    for list_f, days_f in _walk_trip_kits(form_schema):
        if not list_f or not days_f:
            continue
        legs_raw = data.get(list_f)
        if not isinstance(legs_raw, list):
            continue
        legs: List[Dict[str, Any]] = []
        for leg in legs_raw:
            if not isinstance(leg, dict):
                legs.append(leg)  # type: ignore[arg-type]
                continue
            row = dict(leg)
            if not row.get("durationManual"):
                days = calc_trip_leg_days(
                    row.get("startDate"),
                    row.get("startHalf"),
                    row.get("endDate"),
                    row.get("endHalf"),
                )
                if days is not None:
                    row["duration"] = days
            legs.append(row)
        data[list_f] = legs
        if not data.get(MANUAL_PREFIX + days_f):
            data[days_f] = sum_trip_leg_days(legs)
    return data


def validate_trip_kit_form_values(form_schema: Any, form_data: Dict[str, Any]) -> List[str]:
    """提交校验：行程必填、结束晚于开始；时长/天数合法。"""
    errors: List[str] = []
    data = form_data or {}
    for list_f, days_f in _walk_trip_kits(form_schema):
        if not list_f or not days_f:
            continue
        legs_raw = data.get(list_f)
        if not isinstance(legs_raw, list) or len(legs_raw) < 1:
            errors.append("行程至少须保留 1 条")
            continue
        for i, leg in enumerate(legs_raw):
            if not isinstance(leg, dict):
                errors.append(f"行程第 {i + 1} 条格式无效")
                continue
            for key in _LEG_REQUIRED_FIELDS:
                val = leg.get(key)
                if val is None or str(val).strip() == "":
                    errors.append(f"行程第 {i + 1} 条必填项未填写完整")
                    break
            else:
                calc = calc_trip_leg_days(
                    leg.get("startDate"),
                    leg.get("startHalf"),
                    leg.get("endDate"),
                    leg.get("endHalf"),
                )
                if calc is None:
                    errors.append("行程结束时间须晚于开始时间")
            dur = leg.get("duration")
            if dur is not None and dur != "":
                try:
                    dv = float(dur)
                except (TypeError, ValueError):
                    errors.append("行程时长须为非负数字（0.5 天步进）")
                else:
                    if dv < 0 or not is_half_day_multiple(dv):
                        errors.append("行程时长须为非负数字（0.5 天步进）")
        days_val = data.get(days_f)
        if days_val is not None and days_val != "":
            try:
                dv = float(days_val)
            except (TypeError, ValueError):
                errors.append("出差天数须为非负数字（0.5 天步进）")
            else:
                if dv < 0 or not is_half_day_multiple(dv):
                    errors.append("出差天数须为非负数字（0.5 天步进）")
    return errors