feat(oa): add related-approval flatten and extract helpers

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed 2026-07-22 16:46:21 +08:00
1 parent b8b0446542
commit 5147d79747
2 files changed
+164

No files matched your search

+114
View File
@@ -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
+50
View File
@@ -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()