|
|
@@ -0,0 +1,247 @@
|
|
|
+"""出差套件：半日制行程时长（与前端 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
|
