829 lines
26 KiB
Python
829 lines
26 KiB
Python
"""关联审批单：字段扁平化与 form_data 抽链"""
|
||
from __future__ import annotations
|
||
|
||
from typing import Any, Dict, List, Optional, Sequence, Tuple
|
||
|
||
from app.models.models import OaApp, OaAppVersion, OaInstance, OaTask
|
||
from app.services.oa_engine import OaEngineError, _dept_name, _loads, _user_name_map
|
||
|
||
LAYOUT_SKIP = frozenset(
|
||
{
|
||
"fcRow",
|
||
"fcTable",
|
||
"col",
|
||
"row",
|
||
"elCard",
|
||
"space",
|
||
"divider",
|
||
"html",
|
||
"title",
|
||
"text",
|
||
"tag",
|
||
"card",
|
||
"group",
|
||
"object",
|
||
"leaveKit",
|
||
"confirmationKit",
|
||
"overtimeKit",
|
||
"resignationKit",
|
||
"tripKit",
|
||
}
|
||
)
|
||
UPLOAD_TYPES = frozenset({"upload", "fcUpload", "image", "oaAttachment"})
|
||
LONG_TYPES = frozenset({"textarea", "input"})
|
||
_ACCOUNT_TYPE_LABEL = {
|
||
"alipay": "支付宝",
|
||
"personal": "个人银行卡",
|
||
"corporate": "对公银行账户",
|
||
}
|
||
|
||
def _walk_rules(rules: Any, out: List[dict]) -> None:
|
||
if not isinstance(rules, list):
|
||
return
|
||
for r in rules:
|
||
if not isinstance(r, dict):
|
||
continue
|
||
out.append(r)
|
||
children = r.get("children") or (r.get("props") or {}).get("rule") or []
|
||
if isinstance(children, list):
|
||
_walk_rules(children, out)
|
||
|
||
|
||
def _schema_rules(schema: Any) -> List[dict]:
|
||
if isinstance(schema, list):
|
||
out: List[dict] = []
|
||
_walk_rules(schema, out)
|
||
return out
|
||
if not isinstance(schema, dict):
|
||
return []
|
||
out = []
|
||
_walk_rules(schema.get("rule") or schema.get("rules") or [], out)
|
||
return out
|
||
|
||
|
||
def _extract_upload_urls(raw: Any) -> List[str]:
|
||
"""Normalize FormCreate upload / 业务附件值为 URL 列表。"""
|
||
if raw is None or raw == "" or raw == []:
|
||
return []
|
||
if isinstance(raw, list):
|
||
items: List[Any] = raw
|
||
elif isinstance(raw, str):
|
||
items = [x.strip() for x in raw.split(",") if x.strip()]
|
||
elif isinstance(raw, dict):
|
||
items = [raw]
|
||
else:
|
||
items = [raw]
|
||
urls: List[str] = []
|
||
for it in items:
|
||
if it is None or it == "":
|
||
continue
|
||
if isinstance(it, dict):
|
||
u = it.get("url") or it.get("fileUrl") or ""
|
||
resp = it.get("response")
|
||
if not u and isinstance(resp, dict):
|
||
data = resp.get("data") if isinstance(resp.get("data"), dict) else {}
|
||
u = resp.get("url") or resp.get("fileUrl") or data.get("url") or data.get("fileUrl") or ""
|
||
u = str(u or "").strip()
|
||
if u:
|
||
urls.append(u)
|
||
continue
|
||
s = str(it).strip()
|
||
if s:
|
||
urls.append(s)
|
||
return urls
|
||
|
||
|
||
def _upload_display_name(raw_item: Any, fallback_url: str = "") -> str:
|
||
"""打印/摘要：优先对象 name，否则取路径最后一段文件名。"""
|
||
if isinstance(raw_item, dict):
|
||
named = str(raw_item.get("name") or "").strip()
|
||
if named:
|
||
return named
|
||
u = str(fallback_url or raw_item or "").strip()
|
||
if not u:
|
||
return ""
|
||
path = u.split("?", 1)[0]
|
||
base = path.rsplit("/", 1)[-1]
|
||
if not base:
|
||
return u
|
||
try:
|
||
from urllib.parse import unquote
|
||
|
||
return unquote(base) or base
|
||
except Exception:
|
||
return base
|
||
|
||
|
||
def _format_upload_display(raw: Any) -> str:
|
||
if raw is None or raw == "" or raw == []:
|
||
return "0 个"
|
||
if isinstance(raw, list):
|
||
items: List[Any] = raw
|
||
elif isinstance(raw, str):
|
||
items = [x.strip() for x in raw.split(",") if x.strip()]
|
||
elif isinstance(raw, dict):
|
||
items = [raw]
|
||
else:
|
||
items = [raw]
|
||
names: List[str] = []
|
||
for it in items:
|
||
if it is None or it == "":
|
||
continue
|
||
if isinstance(it, dict):
|
||
named = str(it.get("name") or "").strip()
|
||
if named:
|
||
names.append(named)
|
||
continue
|
||
urls = _extract_upload_urls(it)
|
||
if urls:
|
||
names.append(_upload_display_name(it, urls[0]))
|
||
continue
|
||
s = str(it).strip()
|
||
if s:
|
||
names.append(_upload_display_name(s, s))
|
||
return "、".join(names) if names else "0 个"
|
||
|
||
|
||
def _upload_files_meta(raw: Any) -> List[Dict[str, str]]:
|
||
"""摘要/列表用：上传字段规范化为 {url, name} 列表，供前端缩略图展示。"""
|
||
if raw is None or raw == "" or raw == []:
|
||
return []
|
||
if isinstance(raw, list):
|
||
items: List[Any] = raw
|
||
elif isinstance(raw, str):
|
||
items = [x.strip() for x in raw.split(",") if x.strip()]
|
||
elif isinstance(raw, dict):
|
||
items = [raw]
|
||
else:
|
||
items = [raw]
|
||
files: List[Dict[str, str]] = []
|
||
for it in items:
|
||
if it is None or it == "":
|
||
continue
|
||
if isinstance(it, dict):
|
||
urls = _extract_upload_urls(it)
|
||
if not urls:
|
||
continue
|
||
for u in urls:
|
||
files.append({"url": u, "name": _upload_display_name(it, u)})
|
||
continue
|
||
s = str(it).strip()
|
||
if not s:
|
||
continue
|
||
files.append({"url": s, "name": _upload_display_name(s, s)})
|
||
return files
|
||
|
||
|
||
def _normalize_related_ids(raw: Any) -> List[str]:
|
||
"""Order-preserving dedupe of related approval instance ids."""
|
||
if isinstance(raw, list):
|
||
ids = raw
|
||
elif raw is None or raw == "":
|
||
ids = []
|
||
else:
|
||
ids = [raw]
|
||
seen = set()
|
||
out: List[str] = []
|
||
for x in ids:
|
||
tid = str(x or "").strip()
|
||
if not tid or tid in seen:
|
||
continue
|
||
seen.add(tid)
|
||
out.append(tid)
|
||
return out
|
||
|
||
|
||
def _format_trip_leg(leg: Dict[str, Any], index: int) -> str:
|
||
transport = str(leg.get("transport") or "").strip() or "-"
|
||
trip_type = str(leg.get("tripType") or "").strip()
|
||
from_city = str(leg.get("fromCity") or "").strip() or "-"
|
||
to_city = str(leg.get("toCity") or "").strip() or "-"
|
||
start_date = str(leg.get("startDate") or "").strip()
|
||
start_half = str(leg.get("startHalf") or "").strip()
|
||
end_date = str(leg.get("endDate") or "").strip()
|
||
end_half = str(leg.get("endHalf") or "").strip()
|
||
duration = leg.get("duration")
|
||
if duration is None or duration == "":
|
||
duration_text = "-"
|
||
else:
|
||
try:
|
||
duration_text = f"{float(duration):g}天"
|
||
except (TypeError, ValueError):
|
||
duration_text = f"{duration}天"
|
||
route = f"{from_city} → {to_city}"
|
||
if start_date and end_date:
|
||
period = (
|
||
f"{start_date}{(' ' + start_half) if start_half else ''}"
|
||
f" ～ {end_date}{(' ' + end_half) if end_half else ''}"
|
||
)
|
||
else:
|
||
period = "-"
|
||
type_part = f" · {trip_type}" if trip_type else ""
|
||
return f"行程{index + 1}：{transport}{type_part} · {route} · {period} · {duration_text}"
|
||
|
||
|
||
def _format_trip_itinerary(raw: Any) -> str:
|
||
if isinstance(raw, dict):
|
||
legs = [raw]
|
||
elif isinstance(raw, list):
|
||
legs = [x for x in raw if isinstance(x, dict)]
|
||
else:
|
||
return ""
|
||
if not legs:
|
||
return ""
|
||
return "；".join(_format_trip_leg(leg, i) for i, leg in enumerate(legs))
|
||
|
||
|
||
def _format_receive_account(raw: Any) -> str:
|
||
if isinstance(raw, dict):
|
||
items = [raw]
|
||
elif isinstance(raw, list):
|
||
items = [x for x in raw if isinstance(x, dict)]
|
||
else:
|
||
return ""
|
||
lines: List[str] = []
|
||
for it in items:
|
||
at = str(it.get("accountType") or "").strip()
|
||
type_label = _ACCOUNT_TYPE_LABEL.get(at, at)
|
||
parts = [
|
||
type_label,
|
||
str(it.get("accountName") or "").strip(),
|
||
str(it.get("accountNo") or "").strip(),
|
||
]
|
||
if at in ("personal", "corporate"):
|
||
bank = str(it.get("bankName") or "").strip()
|
||
if bank:
|
||
parts.append(bank)
|
||
prov = str(it.get("bankProvince") or "").strip()
|
||
city = str(it.get("bankCity") or "").strip()
|
||
if prov and city:
|
||
parts.append(f"{prov} / {city}")
|
||
elif prov or city:
|
||
parts.append(prov or city)
|
||
branch = str(it.get("bankBranch") or "").strip()
|
||
if branch:
|
||
parts.append(branch)
|
||
remark = str(it.get("remark") or "").strip()
|
||
if remark:
|
||
parts.append(remark)
|
||
line = " · ".join(p for p in parts if p)
|
||
if line:
|
||
lines.append(line)
|
||
return "；".join(lines)
|
||
|
||
|
||
def _format_region_like(raw: Any) -> str:
|
||
"""省市区 / 级联：string[] 或 {province,city,district} 等。"""
|
||
if raw is None or raw == "":
|
||
return ""
|
||
if isinstance(raw, str):
|
||
return raw.strip()
|
||
if isinstance(raw, list):
|
||
parts: List[str] = []
|
||
for x in raw:
|
||
if x is None or x == "":
|
||
continue
|
||
if isinstance(x, dict):
|
||
n = str(
|
||
x.get("label") or x.get("name") or x.get("text") or x.get("value") or ""
|
||
).strip()
|
||
if n:
|
||
parts.append(n)
|
||
continue
|
||
parts.append(str(x).strip())
|
||
return " / ".join(p for p in parts if p)
|
||
if isinstance(raw, dict):
|
||
path = [
|
||
str(
|
||
raw.get("province")
|
||
or raw.get("provinceName")
|
||
or raw.get("province_name")
|
||
or ""
|
||
).strip(),
|
||
str(raw.get("city") or raw.get("cityName") or raw.get("city_name") or "").strip(),
|
||
str(
|
||
raw.get("district")
|
||
or raw.get("districtName")
|
||
or raw.get("area")
|
||
or raw.get("areaName")
|
||
or raw.get("district_name")
|
||
or ""
|
||
).strip(),
|
||
]
|
||
path = [p for p in path if p]
|
||
if path:
|
||
return " / ".join(path)
|
||
return _format_unknown_object(raw)
|
||
return str(raw).strip()
|
||
|
||
|
||
def _format_unknown_object(raw: Dict[str, Any]) -> str:
|
||
"""Never dump full dict/JSON; prefer common display keys then scalar values."""
|
||
named = str(
|
||
raw.get("label") or raw.get("name") or raw.get("text") or raw.get("value") or ""
|
||
).strip()
|
||
if named:
|
||
return named
|
||
if raw.get("accountType") or raw.get("accountNo") or raw.get("accountName"):
|
||
return _format_receive_account(raw)
|
||
if raw.get("fromCity") or raw.get("toCity") or raw.get("transport"):
|
||
return _format_trip_leg(raw, 0)
|
||
skip_keys = {
|
||
"id",
|
||
"uuid",
|
||
"key",
|
||
"durationManual",
|
||
"manual",
|
||
"__manual",
|
||
}
|
||
parts: List[str] = []
|
||
for k, v in raw.items():
|
||
ks = str(k)
|
||
if ks in skip_keys or ks.startswith("__") or ks.lower().endswith("id"):
|
||
continue
|
||
if isinstance(v, bool) or v is None or v == "":
|
||
continue
|
||
if isinstance(v, (str, int, float)) and not isinstance(v, bool):
|
||
parts.append(str(v).strip())
|
||
return "、".join(p for p in parts if p)
|
||
|
||
|
||
def _format_value(typ: str, raw: Any) -> Tuple[str, bool]:
|
||
"""returns (text, fullWidth)"""
|
||
if typ in UPLOAD_TYPES:
|
||
text = _format_upload_display(raw)
|
||
# 展示附件文件名（多条用中文顿号）；较长时占满一行
|
||
return text, True
|
||
if typ == "relatedApproval":
|
||
if isinstance(raw, list):
|
||
n = len([x for x in raw if str(x).strip()])
|
||
elif raw:
|
||
n = 1
|
||
else:
|
||
n = 0
|
||
return f"已关联 {n} 条", False
|
||
if (typ == "dept" or typ == "person" or typ == "post" or typ == "ownerDept" or typ == "ownerPerson"):
|
||
if isinstance(raw, list):
|
||
names = []
|
||
for x in raw:
|
||
if isinstance(x, dict):
|
||
n = str(x.get("name") or "").strip()
|
||
if n:
|
||
names.append(n)
|
||
return "、".join(names) if names else "", False
|
||
if isinstance(raw, dict):
|
||
return str(raw.get("name") or "").strip(), False
|
||
return "", False
|
||
if typ == "money":
|
||
if raw is None or raw == "":
|
||
return "", False
|
||
try:
|
||
if isinstance(raw, bool):
|
||
return "", False
|
||
v = float(str(raw).replace(",", "").strip())
|
||
except (TypeError, ValueError):
|
||
return str(raw), False
|
||
text = f"{v:,.2f}"
|
||
return text, False
|
||
if typ == "tripItinerary":
|
||
text = _format_trip_itinerary(raw)
|
||
return text, True
|
||
if typ == "receiveAccount":
|
||
text = _format_receive_account(raw)
|
||
return text, True
|
||
if typ in ("region", "cascader"):
|
||
return _format_region_like(raw), False
|
||
if typ in ("dateRange", "timeRange") or (
|
||
typ in ("datePicker", "date", "timePicker", "time") and isinstance(raw, list)
|
||
):
|
||
if isinstance(raw, list) and len(raw) >= 2:
|
||
a = str(raw[0] or "").strip()
|
||
b = str(raw[1] or "").strip()
|
||
if a or b:
|
||
return f"{a or '-'} ～ {b or '-'}", False
|
||
if isinstance(raw, list):
|
||
return "、".join(str(x).strip() for x in raw if x is not None and str(x).strip()), False
|
||
if isinstance(raw, list):
|
||
parts: List[str] = []
|
||
for x in raw:
|
||
if x is None or str(x) == "":
|
||
continue
|
||
if isinstance(x, dict):
|
||
n = _format_unknown_object(x)
|
||
if n:
|
||
parts.append(n)
|
||
continue
|
||
parts.append(str(x))
|
||
text = "、".join(parts)
|
||
elif isinstance(raw, dict):
|
||
text = _format_unknown_object(raw)
|
||
else:
|
||
text = "" if raw is None else str(raw)
|
||
full = typ in LONG_TYPES and len(text) > 40
|
||
return text, full
|
||
|
||
|
||
def flatten_form_fields_for_summary(schema: Any, form_data: Optional[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
||
data = form_data or {}
|
||
out: List[Dict[str, Any]] = []
|
||
for rule in _schema_rules(schema):
|
||
typ = str(rule.get("type") or "")
|
||
if not typ or typ in LAYOUT_SKIP:
|
||
continue
|
||
field = str(rule.get("field") or "").strip()
|
||
if not field:
|
||
continue
|
||
label = str(rule.get("title") or rule.get("label") or field)
|
||
raw = data.get(field)
|
||
if typ == "relatedApproval":
|
||
ids = _normalize_related_ids(raw)
|
||
item: Dict[str, Any] = {
|
||
"fieldKey": field,
|
||
"label": label,
|
||
"value": f"已关联 {len(ids)} 条",
|
||
"relatedIds": ids,
|
||
}
|
||
out.append(item)
|
||
continue
|
||
if typ in UPLOAD_TYPES:
|
||
text, full = _format_value(typ, raw)
|
||
item = {
|
||
"fieldKey": field,
|
||
"label": label,
|
||
"value": text,
|
||
"fieldType": typ,
|
||
"files": _upload_files_meta(raw),
|
||
}
|
||
if full:
|
||
item["fullWidth"] = True
|
||
out.append(item)
|
||
continue
|
||
text, full = _format_value(typ, raw)
|
||
item = {"fieldKey": field, "label": label, "value": text}
|
||
if full:
|
||
item["fullWidth"] = True
|
||
out.append(item)
|
||
return out
|
||
|
||
|
||
def extract_related_approval_links(
|
||
schema: Any, form_data: Optional[Dict[str, Any]]
|
||
) -> List[Tuple[str, str]]:
|
||
"""Ordered unique (field_key, to_instance_id)."""
|
||
data = form_data or {}
|
||
seen = set()
|
||
links: List[Tuple[str, str]] = []
|
||
for rule in _schema_rules(schema):
|
||
if str(rule.get("type") or "") != "relatedApproval":
|
||
continue
|
||
field = str(rule.get("field") or "").strip()
|
||
if not field:
|
||
continue
|
||
raw = data.get(field)
|
||
ids: Sequence[Any]
|
||
if isinstance(raw, list):
|
||
ids = raw
|
||
elif raw is None or raw == "":
|
||
ids = []
|
||
else:
|
||
ids = [raw]
|
||
for x in ids:
|
||
tid = str(x or "").strip()
|
||
if not tid:
|
||
continue
|
||
key = (field, tid)
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
links.append((field, tid))
|
||
return links
|
||
|
||
|
||
def merge_relation_targets(
|
||
schema: Any,
|
||
form_data: Optional[Dict[str, Any]],
|
||
legacy_ids: Optional[List[str]] = None,
|
||
) -> List[Tuple[Optional[str], str]]:
|
||
"""Returns (field_key|None, to_id) — form fields first, then legacy without key."""
|
||
links = extract_related_approval_links(schema, form_data)
|
||
result: List[Tuple[Optional[str], str]] = [(f, i) for f, i in links]
|
||
seen_ids = {i for _, i in links}
|
||
for x in legacy_ids or []:
|
||
tid = str(x or "").strip()
|
||
if not tid or tid in seen_ids:
|
||
continue
|
||
result.append((None, tid))
|
||
seen_ids.add(tid)
|
||
return result
|
||
|
||
|
||
_DONE_TASK_STATUSES = ("approved", "rejected", "returned", "transferred")
|
||
|
||
|
||
def _user_can_link_instance(db, *, inst: OaInstance, user_id: str, tenant_no: str) -> bool:
|
||
if (inst.initiator_user_id or "") == user_id:
|
||
return True
|
||
if db is None:
|
||
return False
|
||
done = (
|
||
db.query(OaTask.id)
|
||
.filter(
|
||
OaTask.tenant_no == tenant_no,
|
||
OaTask.del_flag == 0,
|
||
OaTask.instance_id == inst.id,
|
||
OaTask.assignee_user_id == user_id,
|
||
OaTask.task_type == "approval",
|
||
OaTask.status.in_(_DONE_TASK_STATUSES),
|
||
)
|
||
.first()
|
||
)
|
||
if done:
|
||
return True
|
||
cc = (
|
||
db.query(OaTask.id)
|
||
.filter(
|
||
OaTask.tenant_no == tenant_no,
|
||
OaTask.del_flag == 0,
|
||
OaTask.instance_id == inst.id,
|
||
OaTask.assignee_user_id == user_id,
|
||
OaTask.task_type == "cc",
|
||
)
|
||
.first()
|
||
)
|
||
return bool(cc)
|
||
|
||
|
||
def assert_linkable_instance(
|
||
inst: OaInstance,
|
||
*,
|
||
user_id: str,
|
||
tenant_no: str,
|
||
db=None,
|
||
) -> None:
|
||
if not inst or inst.del_flag != 0 or inst.tenant_no != tenant_no:
|
||
raise OaEngineError("关联的审批单不存在")
|
||
if not _user_can_link_instance(db, inst=inst, user_id=user_id, tenant_no=tenant_no):
|
||
raise OaEngineError("无权关联该审批单")
|
||
|
||
|
||
_STATUS_TEXT = {
|
||
"running": "处理中",
|
||
"completed": "已通过",
|
||
"rejected": "已拒绝",
|
||
"returned": "已打回",
|
||
"cancelled": "已撤销",
|
||
}
|
||
|
||
|
||
def related_status_text(status: str) -> str:
|
||
s = (status or "").strip()
|
||
return _STATUS_TEXT.get(s, s or "-")
|
||
|
||
|
||
def related_display_title(initiator_name: str, app_name: str, title: str) -> str:
|
||
name = (initiator_name or "").strip()
|
||
app = (app_name or "").strip()
|
||
if name and app:
|
||
return f"{name}提交的{app}"
|
||
return app or (title or "").strip() or "审批单"
|
||
|
||
|
||
def _instance_ids_for_tab(db, *, tenant_no: str, user_id: str, tab: str) -> Optional[List[str]]:
|
||
"""Return instance id list for done/cc tabs; None means mine (filter on initiator)."""
|
||
if tab == "mine":
|
||
return None
|
||
if tab == "done":
|
||
rows = (
|
||
db.query(OaTask.instance_id)
|
||
.filter(
|
||
OaTask.tenant_no == tenant_no,
|
||
OaTask.del_flag == 0,
|
||
OaTask.assignee_user_id == user_id,
|
||
OaTask.task_type == "approval",
|
||
OaTask.status.in_(_DONE_TASK_STATUSES),
|
||
)
|
||
.distinct()
|
||
.all()
|
||
)
|
||
return [r[0] for r in rows if r[0]]
|
||
if tab == "cc":
|
||
rows = (
|
||
db.query(OaTask.instance_id)
|
||
.filter(
|
||
OaTask.tenant_no == tenant_no,
|
||
OaTask.del_flag == 0,
|
||
OaTask.assignee_user_id == user_id,
|
||
OaTask.task_type == "cc",
|
||
)
|
||
.distinct()
|
||
.all()
|
||
)
|
||
return [r[0] for r in rows if r[0]]
|
||
raise OaEngineError(f"不支持的选单类型: {tab}")
|
||
|
||
|
||
def list_related_candidates(
|
||
db,
|
||
*,
|
||
tenant_no: str,
|
||
user_id: str,
|
||
tab: str = "mine",
|
||
keyword: str = "",
|
||
page: int = 1,
|
||
page_size: int = 20,
|
||
) -> Dict[str, Any]:
|
||
from sqlalchemy import or_
|
||
|
||
tab_n = (tab or "mine").strip() or "mine"
|
||
if tab_n not in ("mine", "done", "cc"):
|
||
raise OaEngineError(f"不支持的选单类型: {tab_n}")
|
||
|
||
q = db.query(OaInstance).filter(
|
||
OaInstance.tenant_no == tenant_no,
|
||
OaInstance.del_flag == 0,
|
||
)
|
||
tab_ids = _instance_ids_for_tab(db, tenant_no=tenant_no, user_id=user_id, tab=tab_n)
|
||
if tab_ids is None:
|
||
q = q.filter(OaInstance.initiator_user_id == user_id)
|
||
else:
|
||
if not tab_ids:
|
||
return {"list": [], "total": 0, "page": max(1, int(page or 1)), "pageSize": min(100, max(1, int(page_size or 20)))}
|
||
q = q.filter(OaInstance.id.in_(tab_ids))
|
||
|
||
kw = (keyword or "").strip()
|
||
if kw:
|
||
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: Dict[str, str] = {}
|
||
app_ids_page = list({r.app_id for r in rows if r.app_id})
|
||
if app_ids_page:
|
||
for a in db.query(OaApp).filter(OaApp.id.in_(app_ids_page)).all():
|
||
app_map[a.id] = a.app_name or ""
|
||
|
||
initiator_ids = [r.initiator_user_id for r in rows if r.initiator_user_id]
|
||
inst_ids = [r.id for r in rows]
|
||
pending_tasks = (
|
||
db.query(OaTask)
|
||
.filter(
|
||
OaTask.tenant_no == tenant_no,
|
||
OaTask.del_flag == 0,
|
||
OaTask.instance_id.in_(inst_ids),
|
||
OaTask.task_type == "approval",
|
||
OaTask.status == "pending",
|
||
)
|
||
.all()
|
||
if inst_ids
|
||
else []
|
||
)
|
||
pending_by_inst: Dict[str, List[OaTask]] = {}
|
||
for t in pending_tasks:
|
||
pending_by_inst.setdefault(t.instance_id, []).append(t)
|
||
if t.assignee_user_id:
|
||
initiator_ids.append(t.assignee_user_id)
|
||
name_map = _user_name_map(db, initiator_ids)
|
||
|
||
ver_ids = list({r.app_version_id for r in rows if r.app_version_id})
|
||
ver_by_id: Dict[str, OaAppVersion] = {}
|
||
if ver_ids:
|
||
for v in (
|
||
db.query(OaAppVersion)
|
||
.filter(OaAppVersion.id.in_(ver_ids), OaAppVersion.del_flag == 0)
|
||
.all()
|
||
):
|
||
ver_by_id[v.id] = v
|
||
|
||
items: List[Dict[str, Any]] = []
|
||
for r in rows:
|
||
app_name = app_map.get(r.app_id, "")
|
||
initiator_name = name_map.get(r.initiator_user_id or "", "") or ""
|
||
status = r.status or ""
|
||
approver_names = ""
|
||
if status == "running":
|
||
names: List[str] = []
|
||
for t in pending_by_inst.get(r.id, []):
|
||
uid = (t.assignee_user_id or "").strip()
|
||
if uid:
|
||
names.append(name_map.get(uid, uid))
|
||
approver_names = "、".join(names)
|
||
ver = ver_by_id.get(r.app_version_id or "")
|
||
schema = _loads(ver.form_schema_json, {}) if ver else {}
|
||
form_data = _loads(r.form_data_json, {}) if r.form_data_json else {}
|
||
preview = [
|
||
{"label": f.get("label") or "", "value": f.get("value") or ""}
|
||
for f in flatten_form_fields_for_summary(schema, form_data)[:2]
|
||
]
|
||
items.append(
|
||
{
|
||
"id": r.id,
|
||
"appName": app_name,
|
||
"title": r.title or "",
|
||
"displayTitle": related_display_title(initiator_name, app_name, r.title or ""),
|
||
"initiatorUserName": initiator_name,
|
||
"initiatorDeptName": _dept_name(
|
||
db, getattr(r, "initiator_dept_id", None), tenant_no=tenant_no
|
||
),
|
||
"applyTime": r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else "",
|
||
"status": status,
|
||
"statusText": related_status_text(status),
|
||
"currentApproverNames": approver_names,
|
||
"previewFields": preview,
|
||
}
|
||
)
|
||
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, db=db)
|
||
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
|