diff --git a/backend/app/services/oa_related_approval.py b/backend/app/services/oa_related_approval.py new file mode 100644 index 00000000..6b73f38c --- /dev/null +++ b/backend/app/services/oa_related_approval.py @@ -0,0 +1,114 @@ +"""关联审批单:字段扁平化与 form_data 抽链""" +from __future__ import annotations + +from typing import Any, Dict, List, Optional, Sequence, Tuple + +LAYOUT_SKIP = frozenset( + {"fcRow", "fcTable", "col", "row", "elCard", "space", "divider", "html", "title", "text", "tag"} +) +UPLOAD_TYPES = frozenset({"upload", "fcUpload", "image", "oaAttachment"}) +LONG_TYPES = frozenset({"textarea", "input"}) + + +def _walk_rules(rules: Any, out: List[dict]) -> None: + if not isinstance(rules, list): + return + for r in rules: + if not isinstance(r, dict): + continue + out.append(r) + children = r.get("children") or (r.get("props") or {}).get("rule") or [] + if isinstance(children, list): + _walk_rules(children, out) + + +def _schema_rules(schema: Any) -> List[dict]: + if isinstance(schema, list): + out: List[dict] = [] + _walk_rules(schema, out) + return out + if not isinstance(schema, dict): + return [] + out = [] + _walk_rules(schema.get("rule") or schema.get("rules") or [], out) + return out + + +def _format_value(typ: str, raw: Any) -> Tuple[str, bool]: + """returns (text, fullWidth)""" + if typ in UPLOAD_TYPES: + if raw is None or raw == "" or raw == []: + n = 0 + elif isinstance(raw, list): + n = len(raw) + else: + n = len([x for x in str(raw).split(",") if x.strip()]) + return f"{n} 个", False + if typ == "relatedApproval": + if isinstance(raw, list): + n = len([x for x in raw if str(x).strip()]) + elif raw: + n = 1 + else: + n = 0 + return f"已关联 {n} 条", False + if isinstance(raw, list): + text = "、".join(str(x) for x in raw if x is not None and str(x) != "") + elif isinstance(raw, dict): + text = str(raw.get("label") or raw.get("value") or raw) + else: + text = "" if raw is None else str(raw) + full = typ in LONG_TYPES and len(text) > 40 + return text, full + + +def flatten_form_fields_for_summary(schema: Any, form_data: Optional[Dict[str, Any]]) -> List[Dict[str, Any]]: + data = form_data or {} + out: List[Dict[str, Any]] = [] + for rule in _schema_rules(schema): + typ = str(rule.get("type") or "") + if not typ or typ in LAYOUT_SKIP: + continue + field = str(rule.get("field") or "").strip() + if not field: + continue + label = str(rule.get("title") or rule.get("label") or field) + text, full = _format_value(typ, data.get(field)) + item: Dict[str, Any] = {"label": label, "value": text} + if full: + item["fullWidth"] = True + out.append(item) + return out + + +def extract_related_approval_links( + schema: Any, form_data: Optional[Dict[str, Any]] +) -> List[Tuple[str, str]]: + """Ordered unique (field_key, to_instance_id).""" + data = form_data or {} + seen = set() + links: List[Tuple[str, str]] = [] + for rule in _schema_rules(schema): + if str(rule.get("type") or "") != "relatedApproval": + continue + field = str(rule.get("field") or "").strip() + if not field: + continue + raw = data.get(field) + ids: Sequence[Any] + if isinstance(raw, list): + ids = raw + elif raw is None or raw == "": + ids = [] + else: + ids = [raw] + for x in ids: + tid = str(x or "").strip() + if not tid: + continue + key = (field, tid) + if key in seen: + continue + seen.add(key) + links.append((field, tid)) + return links diff --git a/backend/tests/test_oa_related_approval.py b/backend/tests/test_oa_related_approval.py new file mode 100644 index 00000000..36ff5ff9 --- /dev/null +++ b/backend/tests/test_oa_related_approval.py @@ -0,0 +1,50 @@ +import unittest + +from app.services.oa_related_approval import ( + extract_related_approval_links, + flatten_form_fields_for_summary, +) + + +class TestFlatten(unittest.TestCase): + def test_skips_layout_and_summarizes_upload(self): + schema = { + "rule": [ + {"type": "html", "title": "说明"}, + {"type": "input", "field": "reason", "title": "事由"}, + {"type": "upload", "field": "files", "title": "附件"}, + {"type": "relatedApproval", "field": "prev", "title": "关联"}, + {"type": "textarea", "field": "long", "title": "详细说明"}, + ] + } + data = { + "reason": "出差", + "files": ["a.png", "b.png"], + "prev": ["id1", "id2"], + "long": "很长很长的说明文字" * 5, + } + fields = flatten_form_fields_for_summary(schema, data) + labels = [f["label"] for f in fields] + self.assertNotIn("说明", labels) + self.assertEqual(fields[0], {"label": "事由", "value": "出差"}) + self.assertEqual(fields[1]["value"], "2 个") + self.assertEqual(fields[2]["value"], "已关联 2 条") + self.assertTrue(fields[3].get("fullWidth")) + + def test_extract_links_with_field_key_and_dedupe(self): + schema = { + "rule": [ + {"type": "relatedApproval", "field": "relA"}, + { + "type": "fcRow", + "children": [{"type": "relatedApproval", "field": "relB"}], + }, + ] + } + data = {"relA": ["x", "x", ""], "relB": "solo"} + links = extract_related_approval_links(schema, data) + self.assertEqual(links, [("relA", "x"), ("relB", "solo")]) + + +if __name__ == "__main__": + unittest.main()