zkxw/backend/app/services/oa_instance_comment.py

"""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