feat(oa): add related-approval candidate and summary APIs

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/api/oa_approval.py

@@ -629,6 +629,53 @@ async def get_oa_instance(
         return error_response(400, str(e))
 
 
+@router.get("/related-candidates")
+async def list_related_candidates(
+    keyword: str = Query(""),
+    page: int = Query(1),
+    pageSize: int = Query(20, alias="pageSize"),
+    user: dict = Depends(require_admin_or_super_admin),
+    db: Session = Depends(get_db),
+):
+    from app.services import oa_related_approval as rel
+
+    tenant_no = _get_tenant_no(db, user)
+    data = rel.list_related_candidates(
+        db,
+        tenant_no=tenant_no,
+        user_id=user.get("user_id") or "",
+        keyword=keyword,
+        page=page,
+        page_size=pageSize,
+    )
+    return success_response(data)
+
+
+@router.get("/related-summaries")
+async def get_related_summaries(
+    ids: str = Query("", description="comma-separated instance ids"),
+    mode: str = Query("picker"),
+    user: dict = Depends(require_admin_or_super_admin),
+    db: Session = Depends(get_db),
+):
+    from app.services import oa_related_approval as rel
+    from app.services.oa_engine import OaEngineError
+
+    tenant_no = _get_tenant_no(db, user)
+    id_list = [x.strip() for x in (ids or "").split(",") if x.strip()]
+    try:
+        items = rel.build_related_summaries(
+            db,
+            tenant_no=tenant_no,
+            user_id=user.get("user_id") or "",
+            ids=id_list,
+            mode=mode,
+        )
+    except OaEngineError as e:
+        return error_response(400, str(e))
+    return success_response({"list": items})
+
+
 @router.get("/tasks/{kind}")
 async def list_oa_tasks(
     kind: str,

backend/app/services/oa_related_approval.py

@@ -3,6 +3,9 @@ from __future__ import annotations
 
 from typing import Any, Dict, List, Optional, Sequence, Tuple
 
+from app.models.models import OaApp, OaAppVersion, OaInstance
+from app.services.oa_engine import OaEngineError, _loads
+
 LAYOUT_SKIP = frozenset(
     {"fcRow", "fcTable", "col", "row", "elCard", "space", "divider", "html", "title", "text", "tag"}
 )
@@ -112,3 +115,146 @@ def extract_related_approval_links(
             seen.add(key)
             links.append((field, tid))
     return links
+
+
+def assert_linkable_instance(inst: OaInstance, *, user_id: str, tenant_no: str) -> None:
+    if not inst or inst.del_flag != 0 or inst.tenant_no != tenant_no:
+        raise OaEngineError("关联的审批单不存在")
+    if (inst.initiator_user_id or "") != user_id:
+        raise OaEngineError("只能关联本人发起的审批单")
+    if (inst.status or "") != "completed":
+        raise OaEngineError("只能关联已通过的审批单")
+
+
+def list_related_candidates(
+    db,
+    *,
+    tenant_no: str,
+    user_id: str,
+    keyword: str = "",
+    page: int = 1,
+    page_size: int = 20,
+) -> Dict[str, Any]:
+    q = db.query(OaInstance).filter(
+        OaInstance.tenant_no == tenant_no,
+        OaInstance.del_flag == 0,
+        OaInstance.initiator_user_id == user_id,
+        OaInstance.status == "completed",
+    )
+    kw = (keyword or "").strip()
+    if kw:
+        from sqlalchemy import or_
+
+        app_ids = [
+            row[0]
+            for row in db.query(OaApp.id)
+            .filter(
+                OaApp.tenant_no == tenant_no,
+                OaApp.del_flag == 0,
+                OaApp.app_name.like(f"%{kw}%"),
+            )
+            .all()
+        ]
+        clauses = [OaInstance.title.like(f"%{kw}%")]
+        if app_ids:
+            clauses.append(OaInstance.app_id.in_(app_ids))
+        q = q.filter(or_(*clauses))
+    total = q.count()
+    page = max(1, int(page or 1))
+    page_size = min(100, max(1, int(page_size or 20)))
+    rows = (
+        q.order_by(OaInstance.create_time.desc())
+        .offset((page - 1) * page_size)
+        .limit(page_size)
+        .all()
+    )
+    app_map = {}
+    ids = list({r.app_id for r in rows if r.app_id})
+    if ids:
+        for a in db.query(OaApp).filter(OaApp.id.in_(ids)).all():
+            app_map[a.id] = a.app_name or ""
+    items = []
+    for r in rows:
+        items.append(
+            {
+                "id": r.id,
+                "appName": app_map.get(r.app_id, ""),
+                "title": r.title or "",
+                "applyTime": r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else "",
+                "status": r.status or "",
+            }
+        )
+    return {"list": items, "total": total, "page": page, "pageSize": page_size}
+
+
+def build_related_summaries(
+    db,
+    *,
+    tenant_no: str,
+    user_id: str,
+    ids: List[str],
+    mode: str = "picker",
+) -> List[Dict[str, Any]]:
+    mode = (mode or "picker").strip()
+    out: List[Dict[str, Any]] = []
+    clean = [str(i).strip() for i in ids if str(i).strip()]
+    if not clean:
+        return out
+    insts = (
+        db.query(OaInstance)
+        .filter(
+            OaInstance.tenant_no == tenant_no,
+            OaInstance.del_flag == 0,
+            OaInstance.id.in_(clean),
+        )
+        .all()
+    )
+    by_id = {i.id: i for i in insts}
+    for iid in clean:
+        inst = by_id.get(iid)
+        if not inst:
+            out.append(
+                {
+                    "id": iid,
+                    "unavailable": True,
+                    "appName": "",
+                    "applyTime": "",
+                    "status": "",
+                    "fields": [],
+                }
+            )
+            continue
+        if mode == "picker":
+            try:
+                assert_linkable_instance(inst, user_id=user_id, tenant_no=tenant_no)
+            except OaEngineError:
+                out.append(
+                    {
+                        "id": iid,
+                        "unavailable": True,
+                        "appName": "",
+                        "applyTime": "",
+                        "status": "",
+                        "fields": [],
+                    }
+                )
+                continue
+        app = db.query(OaApp).filter(OaApp.id == inst.app_id, OaApp.del_flag == 0).first()
+        ver = (
+            db.query(OaAppVersion)
+            .filter(OaAppVersion.id == inst.app_version_id, OaAppVersion.del_flag == 0)
+            .first()
+        )
+        schema = _loads(ver.form_schema_json, {}) if ver else {}
+        form_data = _loads(inst.form_data_json, {}) if inst.form_data_json else {}
+        out.append(
+            {
+                "id": inst.id,
+                "unavailable": False,
+                "appName": (app.app_name if app else "") or (inst.title or ""),
+                "applyTime": inst.create_time.strftime("%Y-%m-%d %H:%M:%S") if inst.create_time else "",
+                "status": inst.status or "",
+                "fields": flatten_form_fields_for_summary(schema, form_data),
+            }
+        )
+    return out

backend/tests/test_oa_related_approval.py

@@ -46,5 +46,16 @@ class TestFlatten(unittest.TestCase):
         self.assertEqual(links, [("relA", "x"), ("relB", "solo")])
 
 
+class TestValidateLinkable(unittest.TestCase):
+    def test_rejects_non_completed(self):
+        from app.services.oa_related_approval import assert_linkable_instance
+        from app.services.oa_engine import OaEngineError
+        from unittest.mock import MagicMock
+
+        inst = MagicMock(status="running", initiator_user_id="u1", tenant_no="t", del_flag=0)
+        with self.assertRaises(OaEngineError):
+            assert_linkable_instance(inst, user_id="u1", tenant_no="t")
+
+
 if __name__ == "__main__":
     unittest.main()