"""关联审批单：字段扁平化与 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