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,
    OaInstanceViewGrant,
    SysUser,
)
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
MENTION_MAX = 20


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 _attachment_url_candidates(url: str) -> List[str]:
    raw = (url or "").strip()
    path = raw.split("?", 1)[0].strip()
    found: List[str] = []
    for item in (raw, path, path.lstrip("/")):
        if item and item not in found:
            found.append(item)
    try:
        from app.core.config import settings

        base = (getattr(settings, "MINIO_BASE_URL", None) or "").rstrip("/")
    except Exception:
        base = ""
    if base and path.startswith(base + "/"):
        rel = path[len(base) + 1 :]
        if rel and rel not in found:
            found.append(rel)
    return found


def _find_uploaded_attachment(db: Session, tenant_no: str, url: str):
    from app.models.models import BizAttachment

    q = db.query(BizAttachment).filter(
        BizAttachment.del_flag == 0,
        BizAttachment.tenant_no == tenant_no,
    )
    for candidate in _attachment_url_candidates(url):
        row = q.filter(BizAttachment.file_url == candidate).first()
        if row is not None:
            return row
    return None


def normalize_attachment_refs(
    schema: Any,
    form_data: Optional[Dict[str, Any]],
    refs: Optional[List[Dict[str, Any]]],
    db: Optional[Session] = None,
    tenant_no: str = "",
) -> 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()
        source = str(raw.get("source") or "form").strip() or "form"
        if not url or url in seen:
            if not url:
                raise OaEngineError("引用的附件不存在或不属于本表单")
            continue
        if source == "upload":
            if db is None:
                raise OaEngineError("上传的附件不存在")
            row = _find_uploaded_attachment(db, tenant_no, url)
            if row is None:
                raise OaEngineError("上传的附件不存在")
            seen.add(url)
            out.append(
                {
                    "source": "upload",
                    "fieldKey": "",
                    "fieldLabel": "",
                    "url": str(row.file_url or url),
                    "fileName": str(row.original_name or "").strip() or url,
                }
            )
            continue
        if source != "form":
            raise OaEngineError("attachmentRefs 格式无效")
        if url not in index:
            raise OaEngineError("引用的附件不存在或不属于本表单")
        seen.add(url)
        meta = index[url]
        out.append(
            {
                "source": "form",
                "fieldKey": meta["fieldKey"],
                "fieldLabel": meta["fieldLabel"],
                "url": url,
                "fileName": meta["fileName"],
            }
        )
    return out


def _fmt_time(dt: Optional[datetime]) -> str:
    return dt.strftime("%Y-%m-%d %H:%M:%S") if dt else ""


def _mention_ids(row: OaInstanceComment) -> List[str]:
    raw = _loads(getattr(row, "mention_user_ids_json", None), [])
    if not isinstance(raw, list):
        return []
    out: List[str] = []
    for item in raw:
        uid = str(item or "").strip()
        if uid and uid not in out:
            out.append(uid)
    return out


def _mentions_payload(row: OaInstanceComment, name_map: Dict[str, str]) -> List[Dict[str, str]]:
    return [
        {"userId": uid, "userName": name_map.get(uid, uid)}
        for uid in _mention_ids(row)
    ]


def _comment_to_dict(
    row: OaInstanceComment,
    *,
    name_map: Dict[str, str],
    viewer_user_id: str,
    instance_status: str,
) -> Dict[str, Any]:
    refs = _loads(row.attachment_refs_json, [])
    if not isinstance(refs, list):
        refs = []
    status = (row.status or "normal").strip() or "normal"
    can_revoke = status == "normal" and (row.create_by or "") == (viewer_user_id or "")
    return {
        "id": row.id,
        "instanceId": row.instance_id,
        "content": row.content or "",
        "attachmentRefs": refs,
        "status": status,
        "revokedTime": _fmt_time(row.revoked_time),
        "createBy": row.create_by or "",
        "createByName": name_map.get(row.create_by or "", row.create_by or ""),
        "createTime": _fmt_time(row.create_time),
        "canRevoke": can_revoke,
        "mentions": _mentions_payload(row, name_map),
    }


def _load_version_schema(db: Session, inst: OaInstance) -> Any:
    ver = (
        db.query(OaAppVersion)
        .filter(OaAppVersion.id == inst.app_version_id, OaAppVersion.del_flag == 0)
        .first()
    )
    if not ver:
        return {}
    return _loads(ver.form_schema_json, {})


def list_comments(
    db: Session,
    *,
    instance_id: str,
    tenant_no: str,
    viewer_user_id: str,
) -> List[Dict[str, Any]]:
    inst = _get_instance(db, instance_id, tenant_no)
    rows = (
        db.query(OaInstanceComment)
        .filter(
            OaInstanceComment.instance_id == instance_id,
            OaInstanceComment.tenant_no == tenant_no,
            OaInstanceComment.del_flag == 0,
        )
        .order_by(OaInstanceComment.create_time.asc())
        .all()
    )
    name_ids = [r.create_by for r in rows if r.create_by]
    for r in rows:
        name_ids.extend(_mention_ids(r))
    name_map = _user_name_map(db, name_ids)
    return [
        _comment_to_dict(
            r,
            name_map=name_map,
            viewer_user_id=viewer_user_id,
            instance_status=inst.status or "",
        )
        for r in rows
    ]


def list_form_attachments(
    db: Session, *, instance_id: str, tenant_no: str
) -> List[Dict[str, Any]]:
    inst = _get_instance(db, instance_id, tenant_no)
    schema = _load_version_schema(db, inst)
    form = _loads(inst.form_data_json, {})
    if not isinstance(form, dict):
        form = {}
    return enumerate_form_attachments(schema, form)


def normalize_mention_user_ids(
    db: Session,
    *,
    tenant_no: str,
    actor_user_id: str,
    user_ids: Optional[List[str]],
) -> List[str]:
    if not user_ids:
        return []
    if not isinstance(user_ids, list):
        raise OaEngineError("mentionUserIds 格式无效")
    actor = (actor_user_id or "").strip()
    seen: List[str] = []
    for raw in user_ids:
        uid = str(raw or "").strip()
        if not uid or uid in seen or uid == actor:
            continue
        seen.append(uid)
    if len(seen) > MENTION_MAX:
        raise OaEngineError(f"一次最多@{MENTION_MAX}人")
    if not seen:
        return []
    rows = (
        db.query(SysUser.id)
        .filter(
            SysUser.id.in_(seen),
            SysUser.tenant_no == tenant_no,
            SysUser.del_flag == 0,
            SysUser.status == "1",
        )
        .all()
    )
    found = {str(r[0]) for r in rows}
    if any(uid not in found for uid in seen):
        raise OaEngineError("提及的用户不存在或已停用")
    return seen


def _ensure_view_grants(
    db: Session,
    *,
    instance_id: str,
    tenant_no: str,
    user_ids: List[str],
    comment_id: str,
    actor_user_id: str,
    dept_id: str,
) -> None:
    if not user_ids:
        return
    existing = (
        db.query(OaInstanceViewGrant.user_id)
        .filter(
            OaInstanceViewGrant.instance_id == instance_id,
            OaInstanceViewGrant.tenant_no == tenant_no,
            OaInstanceViewGrant.del_flag == 0,
            OaInstanceViewGrant.user_id.in_(user_ids),
        )
        .all()
    )
    have = {str(r[0]) for r in existing}
    now = datetime.now()
    for uid in user_ids:
        if uid in have:
            continue
        db.add(
            OaInstanceViewGrant(
                id=generate_uuid(),
                instance_id=instance_id,
                user_id=uid,
                source="comment_mention",
                comment_id=comment_id,
                granted_by=actor_user_id,
                tenant_no=tenant_no,
                create_by=actor_user_id,
                create_dept_id=dept_id or "",
                create_time=now,
                update_by=actor_user_id,
                update_time=now,
                del_flag=0,
            )
        )


def list_mentioned_instance_ids(
    db: Session, *, tenant_no: str, user_id: str
) -> List[str]:
    """评论@授权的实例。被@即出现在「提到我的」，不因发起人/审批人/抄送身份而隐藏。"""
    uid = (user_id or "").strip()
    if not uid:
        return []
    grants = (
        db.query(OaInstanceViewGrant)
        .filter(
            OaInstanceViewGrant.tenant_no == tenant_no,
            OaInstanceViewGrant.user_id == uid,
            OaInstanceViewGrant.del_flag == 0,
        )
        .order_by(OaInstanceViewGrant.create_time.desc(), OaInstanceViewGrant.id.desc())
        .all()
    )
    inst_ids: List[str] = []
    seen = set()
    for g in grants:
        iid = (g.instance_id or "").strip()
        if iid and iid not in seen:
            seen.add(iid)
            inst_ids.append(iid)
    if not inst_ids:
        return []
    allowed = {
        str(r[0])
        for r in db.query(OaInstance.id)
        .filter(
            OaInstance.tenant_no == tenant_no,
            OaInstance.del_flag == 0,
            OaInstance.id.in_(inst_ids),
        )
        .all()
    }
    return [iid for iid in inst_ids if iid in allowed]


def create_comment(
    db: Session,
    *,
    instance_id: str,
    tenant_no: str,
    user_id: str,
    dept_id: str,
    content: str,
    attachment_refs: Optional[List[Dict[str, Any]]],
    mention_user_ids: Optional[List[str]] = None,
) -> Dict[str, Any]:
    inst = _get_instance(db, instance_id, tenant_no)
    text = (content or "").strip()
    if len(text) > CONTENT_MAX:
        raise OaEngineError(f"评论内容不能超过{CONTENT_MAX}字")
    schema = _load_version_schema(db, inst)
    form = _loads(inst.form_data_json, {})
    if not isinstance(form, dict):
        form = {}
    refs = normalize_attachment_refs(schema, form, attachment_refs, db=db, tenant_no=tenant_no)
    mentions = normalize_mention_user_ids(
        db,
        tenant_no=tenant_no,
        actor_user_id=user_id,
        user_ids=mention_user_ids,
    )
    if not text and not mentions and not refs:
        raise OaEngineError("请填写评论、@人员或添加附件")
    now = datetime.now()
    row = OaInstanceComment(
        id=generate_uuid(),
        instance_id=instance_id,
        content=text,
        attachment_refs_json=json.dumps(refs, ensure_ascii=False) if refs else None,
        mention_user_ids_json=json.dumps(mentions, ensure_ascii=False) if mentions else None,
        status="normal",
        revoked_time=None,
        tenant_no=tenant_no,
        create_by=user_id,
        create_dept_id=dept_id or "",
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(row)
    _ensure_view_grants(
        db,
        instance_id=instance_id,
        tenant_no=tenant_no,
        user_ids=mentions,
        comment_id=row.id,
        actor_user_id=user_id,
        dept_id=dept_id or "",
    )
    db.commit()
    db.refresh(row)
    name_map = _user_name_map(db, [user_id, *mentions])
    return _comment_to_dict(
        row,
        name_map=name_map,
        viewer_user_id=user_id,
        instance_status=inst.status or "",
    )


def revoke_comment(
    db: Session,
    *,
    instance_id: str,
    comment_id: str,
    tenant_no: str,
    user_id: str,
) -> Dict[str, Any]:
    inst = _get_instance(db, instance_id, tenant_no)
    row = (
        db.query(OaInstanceComment)
        .filter(
            OaInstanceComment.id == comment_id,
            OaInstanceComment.instance_id == instance_id,
            OaInstanceComment.tenant_no == tenant_no,
            OaInstanceComment.del_flag == 0,
        )
        .first()
    )
    if not row:
        raise OaEngineError("评论不存在")
    if (row.create_by or "") != (user_id or ""):
        raise OaEngineError("只能撤回自己的评论")
    now = datetime.now()
    if (row.status or "") != "revoked":
        row.status = "revoked"
        row.revoked_time = now
        row.update_by = user_id
        row.update_time = now
        db.add(row)
        db.commit()
        db.refresh(row)
    name_map = _user_name_map(db, [row.create_by, *_mention_ids(row)] if row.create_by else _mention_ids(row))
    return _comment_to_dict(
        row,
        name_map=name_map,
        viewer_user_id=user_id,
        instance_status=inst.status or "",
    )