361 lines
11 KiB
Python
361 lines
11 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
|
|
from app.services.oa_engine import OaEngineError, _loads
|
|
|
|
LAYOUT_SKIP = frozenset(
|
|
{"fcRow", "fcTable", "col", "row", "elCard", "space", "divider", "html", "title", "text", "tag"}
|
|
)
|
|
UPLOAD_TYPES = frozenset({"upload", "fcUpload", "image", "oaAttachment"})
|
|
LONG_TYPES = frozenset({"textarea", "input"})
|
|
|
|
|
|
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 _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_value(typ: str, raw: Any) -> Tuple[str, bool]:
|
|
"""returns (text, fullWidth)"""
|
|
if typ in UPLOAD_TYPES:
|
|
urls = _extract_upload_urls(raw)
|
|
if not urls:
|
|
return "0 个", False
|
|
# 展示附件地址（多条用中文顿号）；地址较长默认占满一行
|
|
return "、".join(urls), 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 isinstance(raw, list):
|
|
text = "、".join(str(x) for x in raw if x is not None and str(x) != "")
|
|
elif isinstance(raw, dict):
|
|
text = str(raw.get("label") or raw.get("value") or 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] = {
|
|
"label": label,
|
|
"value": f"已关联 {len(ids)} 条",
|
|
"relatedIds": ids,
|
|
}
|
|
out.append(item)
|
|
continue
|
|
text, full = _format_value(typ, raw)
|
|
item = {"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
|
|
|
|
|
|
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
|