From ea8420b6488a3ba74e7d1d8236f8e4abe37b2536 Mon Sep 17 00:00:00 2001 From: zlwu <704541015@qq.com> Date: Sun, 20 Sep 2026 21:20:58 +0800 Subject: [PATCH] feat(oa): enumerate and validate comment form attachment refs Co-authored-by: Cursor --- backend/app/services/oa_instance_comment.py | 110 ++++++++++++++++++++ backend/tests/test_oa_instance_comment.py | 63 +++++++++++ 2 files changed, 173 insertions(+) create mode 100644 backend/app/services/oa_instance_comment.py create mode 100644 backend/tests/test_oa_instance_comment.py diff --git a/backend/app/services/oa_instance_comment.py b/backend/app/services/oa_instance_comment.py new file mode 100644 index 00000000..facaecb1 --- /dev/null +++ b/backend/app/services/oa_instance_comment.py @@ -0,0 +1,110 @@ +"""OA 审批实例评论：附件枚举、校验、CRUD""" +from __future__ import annotations + +import json +from datetime import datetime +from typing import Any, Dict, List, Optional + +from sqlalchemy.orm import Session + +from app.core.utils import generate_uuid +from app.models.models import OaAppVersion, OaInstance, OaInstanceComment +from app.services.oa_engine import OaEngineError, _get_instance, _loads, _user_name_map +from app.services.oa_related_approval import ( + UPLOAD_TYPES, + _extract_upload_urls, + _schema_rules, + _upload_display_name, +) + +CONTENT_MAX = 1024 + + +def enumerate_form_attachments( + schema: Any, form_data: Optional[Dict[str, Any]] +) -> List[Dict[str, Any]]: + data = form_data or {} + items: List[Dict[str, Any]] = [] + for rule in _schema_rules(schema): + t = str(rule.get("type") or "") + if t not in UPLOAD_TYPES: + continue + field = str(rule.get("field") or rule.get("name") or "").strip() + if not field: + continue + label = str(rule.get("title") or rule.get("label") or field).strip() or field + raw = data.get(field) + files: List[Dict[str, str]] = [] + if isinstance(raw, list): + for it in raw: + if isinstance(it, dict): + u = (_extract_upload_urls(it) or [""])[0] + if not u: + continue + files.append( + { + "url": u, + "fileName": _upload_display_name(it, u), + } + ) + else: + for u in _extract_upload_urls(it): + files.append({"url": u, "fileName": _upload_display_name(it, u)}) + else: + for u in _extract_upload_urls(raw): + files.append({"url": u, "fileName": _upload_display_name(raw, u)}) + if not files: + continue + items.append({"fieldKey": field, "fieldLabel": label, "files": files}) + return items + + +def _allowed_url_index( + schema: Any, form_data: Optional[Dict[str, Any]] +) -> Dict[str, Dict[str, str]]: + """url -> {fieldKey, fieldLabel, fileName}""" + index: Dict[str, Dict[str, str]] = {} + for group in enumerate_form_attachments(schema, form_data): + for f in group["files"]: + u = str(f.get("url") or "").strip() + if not u: + continue + index[u] = { + "fieldKey": group["fieldKey"], + "fieldLabel": group["fieldLabel"], + "fileName": str(f.get("fileName") or "").strip() or u, + } + return index + + +def normalize_attachment_refs( + schema: Any, + form_data: Optional[Dict[str, Any]], + refs: Optional[List[Dict[str, Any]]], +) -> List[Dict[str, str]]: + if not refs: + return [] + if not isinstance(refs, list): + raise OaEngineError("attachmentRefs 格式无效") + index = _allowed_url_index(schema, form_data) + out: List[Dict[str, str]] = [] + seen = set() + for raw in refs: + if not isinstance(raw, dict): + raise OaEngineError("引用的附件不存在或不属于本表单") + url = str(raw.get("url") or "").strip() + if not url or url not in index: + raise OaEngineError("引用的附件不存在或不属于本表单") + if url in seen: + continue + seen.add(url) + meta = index[url] + out.append( + { + "fieldKey": meta["fieldKey"], + "fieldLabel": meta["fieldLabel"], + "url": url, + "fileName": meta["fileName"], + } + ) + return out diff --git a/backend/tests/test_oa_instance_comment.py b/backend/tests/test_oa_instance_comment.py new file mode 100644 index 00000000..b50cb2fd --- /dev/null +++ b/backend/tests/test_oa_instance_comment.py @@ -0,0 +1,63 @@ +"""OA 实例评论：附件枚举与引用校验""" +import unittest + +from app.services.oa_instance_comment import ( + enumerate_form_attachments, + normalize_attachment_refs, +) + + +class TestEnumerateFormAttachments(unittest.TestCase): + def test_groups_upload_fields(self): + schema = { + "rule": [ + { + "type": "upload", + "field": "files", + "title": "附件", + }, + {"type": "input", "field": "name", "title": "名称"}, + ] + } + form = { + "files": [ + {"url": "https://x/a.pdf", "name": "a.pdf"}, + {"url": "https://x/b.pdf", "name": "b.pdf"}, + ], + "name": "demo", + } + items = enumerate_form_attachments(schema, form) + self.assertEqual(len(items), 1) + self.assertEqual(items[0]["fieldKey"], "files") + self.assertEqual(items[0]["fieldLabel"], "附件") + self.assertEqual(len(items[0]["files"]), 2) + self.assertEqual(items[0]["files"][0]["fileName"], "a.pdf") + + def test_skips_empty_upload(self): + schema = {"rule": [{"type": "fcUpload", "field": "f", "title": "F"}]} + self.assertEqual(enumerate_form_attachments(schema, {"f": []}), []) + + +class TestNormalizeAttachmentRefs(unittest.TestCase): + def test_accepts_valid_ref(self): + schema = {"rule": [{"type": "upload", "field": "files", "title": "附件"}]} + form = {"files": [{"url": "https://x/a.pdf", "name": "a.pdf"}]} + out = normalize_attachment_refs( + schema, + form, + [{"fieldKey": "files", "url": "https://x/a.pdf", "fileName": "wrong"}], + ) + self.assertEqual(out[0]["fileName"], "a.pdf") + self.assertEqual(out[0]["fieldLabel"], "附件") + + def test_rejects_foreign_url(self): + from app.services.oa_engine import OaEngineError + + schema = {"rule": [{"type": "upload", "field": "files", "title": "附件"}]} + form = {"files": [{"url": "https://x/a.pdf", "name": "a.pdf"}]} + with self.assertRaises(OaEngineError): + normalize_attachment_refs( + schema, + form, + [{"fieldKey": "files", "url": "https://evil/x.pdf"}], + )