"""出差套件:半日制行程时长(与前端 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