248 lines
8.0 KiB
Python
248 lines
8.0 KiB
Python
"""出差套件：半日制行程时长（与前端 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
|