111 lines
3.6 KiB
Python
111 lines
3.6 KiB
Python
"""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
|