diff --git a/backend/app/services/oa_bank_receipt.py b/backend/app/services/oa_bank_receipt.py new file mode 100644 index 00000000..b3aa087f --- /dev/null +++ b/backend/app/services/oa_bank_receipt.py @@ -0,0 +1,171 @@ +"""OA 实例银行回执单（biz_attachment 关联）。""" +from __future__ import annotations + +import uuid +from datetime import datetime +from typing import Any, Dict, List, Optional + +from sqlalchemy.orm import Session + +from app.models.models import BizAttachment, OaInstance + +BUSINESS_TYPE = "oa_bank_receipt" + + +class OaBankReceiptError(Exception): + pass + + +def _require_instance(db: Session, instance_id: str, tenant_no: str) -> OaInstance: + iid = (instance_id or "").strip() + if not iid: + raise OaBankReceiptError("实例ID无效") + row = ( + db.query(OaInstance) + .filter( + OaInstance.id == iid, + OaInstance.tenant_no == tenant_no, + OaInstance.del_flag == 0, + ) + .first() + ) + if not row: + raise OaBankReceiptError("审批实例不存在") + return row + + +def _row_to_item(row: BizAttachment) -> Dict[str, Any]: + return { + "id": row.id, + "fileName": row.original_name or "", + "fileUrl": row.file_url or "", + "fileType": row.file_type or "", + "fileSize": row.file_size, + "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "", + } + + +def list_bank_receipts(db: Session, *, instance_id: str, tenant_no: str) -> List[Dict[str, Any]]: + _require_instance(db, instance_id, tenant_no) + rows = ( + db.query(BizAttachment) + .filter( + BizAttachment.tenant_no == tenant_no, + BizAttachment.business_type == BUSINESS_TYPE, + BizAttachment.business_id == instance_id, + BizAttachment.del_flag == 0, + ) + .order_by(BizAttachment.create_time.asc(), BizAttachment.id.asc()) + .all() + ) + return [_row_to_item(r) for r in rows] + + +def replace_bank_receipts( + db: Session, + *, + instance_id: str, + tenant_no: str, + items: List[Dict[str, Any]], + user_id: str, + dept_id: str = "", +) -> List[Dict[str, Any]]: + _require_instance(db, instance_id, tenant_no) + now = datetime.now() + desired: List[Dict[str, Any]] = [] + seen_urls = set() + for raw in items or []: + url = str((raw or {}).get("fileUrl") or (raw or {}).get("file_url") or "").strip() + if not url or url in seen_urls: + continue + seen_urls.add(url) + desired.append( + { + "fileUrl": url, + "fileName": str((raw or {}).get("fileName") or (raw or {}).get("file_name") or "").strip() + or url.rsplit("/", 1)[-1], + "fileType": str((raw or {}).get("fileType") or (raw or {}).get("file_type") or "").strip(), + "fileSize": (raw or {}).get("fileSize") + if (raw or {}).get("fileSize") is not None + else (raw or {}).get("file_size"), + } + ) + + existing = ( + db.query(BizAttachment) + .filter( + BizAttachment.tenant_no == tenant_no, + BizAttachment.business_type == BUSINESS_TYPE, + BizAttachment.business_id == instance_id, + BizAttachment.del_flag == 0, + ) + .all() + ) + existing_by_url = {r.file_url: r for r in existing if r.file_url} + desired_urls = {d["fileUrl"] for d in desired} + + for row in existing: + if row.file_url not in desired_urls: + row.del_flag = 1 + row.update_by = user_id or None + row.update_time = now + + for d in desired: + url = d["fileUrl"] + row: Optional[BizAttachment] = existing_by_url.get(url) + if row is None: + row = ( + db.query(BizAttachment) + .filter( + BizAttachment.tenant_no == tenant_no, + BizAttachment.file_url == url, + BizAttachment.del_flag == 0, + ) + .order_by(BizAttachment.create_time.desc()) + .first() + ) + if row is None: + ext = "" + if "." in d["fileName"]: + ext = d["fileName"].rsplit(".", 1)[-1][:20] + size = d["fileSize"] + try: + size_int = int(size) if size is not None else None + except (TypeError, ValueError): + size_int = None + row = BizAttachment( + id=uuid.uuid4().hex, + business_type=BUSINESS_TYPE, + business_id=instance_id, + original_name=d["fileName"][:200], + file_ext=ext or None, + file_url=url, + file_size=size_int, + file_type=(d["fileType"] or None), + tenant_no=tenant_no, + create_by=user_id or None, + create_dept_id=dept_id or "", + create_time=now, + update_by=user_id or None, + update_time=now, + del_flag=0, + ) + db.add(row) + else: + row.business_type = BUSINESS_TYPE + row.business_id = instance_id + row.del_flag = 0 + if d["fileName"]: + row.original_name = d["fileName"][:200] + if d["fileType"]: + row.file_type = d["fileType"] + if d["fileSize"] is not None: + try: + row.file_size = int(d["fileSize"]) + except (TypeError, ValueError): + pass + row.update_by = user_id or None + row.update_time = now + + db.flush() + return list_bank_receipts(db, instance_id=instance_id, tenant_no=tenant_no) diff --git a/backend/tests/test_oa_bank_receipt.py b/backend/tests/test_oa_bank_receipt.py new file mode 100644 index 00000000..b311faa6 --- /dev/null +++ b/backend/tests/test_oa_bank_receipt.py @@ -0,0 +1,235 @@ +"""OA 银行回执单：biz_attachment 挂载 instance。""" +from __future__ import annotations + +import unittest +from datetime import datetime + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from app.models.base import Base +from app.models.models import BizAttachment, OaApp, OaAppVersion, OaInstance +from app.services import oa_bank_receipt as svc + + +class TestOaBankReceipt(unittest.TestCase): + def setUp(self): + self.engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all( + self.engine, + tables=[ + OaApp.__table__, + OaAppVersion.__table__, + OaInstance.__table__, + BizAttachment.__table__, + ], + ) + Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False) + self.db = Session() + t0 = datetime(2026, 10, 16, 10, 0, 0) + self.db.add_all( + [ + OaApp( + id="app1", + app_name="付款申请", + initiator_scope="all", + admin_scope="all_oa_admin", + status="1", + tenant_no="zkxw", + create_dept_id="", + create_time=t0, + update_time=t0, + del_flag=0, + ), + OaAppVersion( + id="v1", + app_id="app1", + version_no=1, + status="published", + is_latest_published="1", + form_schema_json="{}", + flow_tree_json="{}", + advanced_settings_json="{}", + tenant_no="zkxw", + create_dept_id="", + create_time=t0, + update_time=t0, + del_flag=0, + ), + OaInstance( + id="inst-1", + app_id="app1", + app_version_id="v1", + title="付款-1", + initiator_user_id="u1", + status="completed", + current_round=1, + tenant_no="zkxw", + create_dept_id="", + create_time=t0, + update_time=t0, + del_flag=0, + ), + OaInstance( + id="inst-other-tenant", + app_id="app1", + app_version_id="v1", + title="付款-x", + initiator_user_id="u1", + status="completed", + current_round=1, + tenant_no="other", + create_dept_id="", + create_time=t0, + update_time=t0, + del_flag=0, + ), + ] + ) + self.db.commit() + + def tearDown(self): + self.db.close() + self.engine.dispose() + + def test_list_empty(self): + items = svc.list_bank_receipts(self.db, instance_id="inst-1", tenant_no="zkxw") + self.assertEqual(items, []) + + def test_replace_then_list(self): + out = svc.replace_bank_receipts( + self.db, + instance_id="inst-1", + tenant_no="zkxw", + items=[ + { + "fileUrl": "oa-bank-receipt/a.pdf", + "fileName": "回单A.pdf", + "fileType": "application/pdf", + "fileSize": 100, + }, + { + "fileUrl": "oa-bank-receipt/b.jpg", + "fileName": "回单B.jpg", + "fileType": "image/jpeg", + "fileSize": 200, + }, + ], + user_id="u1", + dept_id="d1", + ) + self.db.commit() + self.assertEqual(len(out), 2) + listed = svc.list_bank_receipts(self.db, instance_id="inst-1", tenant_no="zkxw") + self.assertEqual([x["fileUrl"] for x in listed], [ + "oa-bank-receipt/a.pdf", + "oa-bank-receipt/b.jpg", + ]) + self.assertEqual(listed[0]["fileName"], "回单A.pdf") + + def test_replace_removes_missing(self): + svc.replace_bank_receipts( + self.db, + instance_id="inst-1", + tenant_no="zkxw", + items=[ + {"fileUrl": "oa-bank-receipt/a.pdf", "fileName": "A.pdf"}, + {"fileUrl": "oa-bank-receipt/b.pdf", "fileName": "B.pdf"}, + ], + user_id="u1", + dept_id="", + ) + self.db.commit() + svc.replace_bank_receipts( + self.db, + instance_id="inst-1", + tenant_no="zkxw", + items=[{"fileUrl": "oa-bank-receipt/b.pdf", "fileName": "B.pdf"}], + user_id="u1", + dept_id="", + ) + self.db.commit() + listed = svc.list_bank_receipts(self.db, instance_id="inst-1", tenant_no="zkxw") + self.assertEqual([x["fileUrl"] for x in listed], ["oa-bank-receipt/b.pdf"]) + gone = ( + self.db.query(BizAttachment) + .filter(BizAttachment.file_url == "oa-bank-receipt/a.pdf") + .first() + ) + self.assertEqual(gone.del_flag, 1) + + def test_replace_empty_clears_all(self): + svc.replace_bank_receipts( + self.db, + instance_id="inst-1", + tenant_no="zkxw", + items=[{"fileUrl": "oa-bank-receipt/a.pdf", "fileName": "A.pdf"}], + user_id="u1", + dept_id="", + ) + self.db.commit() + svc.replace_bank_receipts( + self.db, + instance_id="inst-1", + tenant_no="zkxw", + items=[], + user_id="u1", + dept_id="", + ) + self.db.commit() + self.assertEqual( + svc.list_bank_receipts(self.db, instance_id="inst-1", tenant_no="zkxw"), + [], + ) + + def test_bind_orphan_upload_row(self): + """通用上传已写入无 business_type 的行时，PUT 应绑定而非再插一行。""" + t0 = datetime(2026, 10, 16, 11, 0, 0) + orphan = BizAttachment( + id="att-orphan", + business_type=None, + business_id=None, + original_name="orphan.pdf", + file_ext="pdf", + file_url="oa-bank-receipt/orphan.pdf", + file_size=10, + file_type="application/pdf", + tenant_no="zkxw", + create_dept_id="", + create_time=t0, + update_time=t0, + del_flag=0, + ) + self.db.add(orphan) + self.db.commit() + svc.replace_bank_receipts( + self.db, + instance_id="inst-1", + tenant_no="zkxw", + items=[{"fileUrl": "oa-bank-receipt/orphan.pdf", "fileName": "orphan.pdf"}], + user_id="u1", + dept_id="", + ) + self.db.commit() + rows = ( + self.db.query(BizAttachment) + .filter(BizAttachment.file_url == "oa-bank-receipt/orphan.pdf", BizAttachment.del_flag == 0) + .all() + ) + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0].business_type, svc.BUSINESS_TYPE) + self.assertEqual(rows[0].business_id, "inst-1") + + def test_missing_instance_raises(self): + with self.assertRaises(svc.OaBankReceiptError): + svc.list_bank_receipts(self.db, instance_id="nope", tenant_no="zkxw") + + def test_cross_tenant_raises(self): + with self.assertRaises(svc.OaBankReceiptError): + svc.list_bank_receipts( + self.db, instance_id="inst-other-tenant", tenant_no="zkxw" + ) + + +if __name__ == "__main__": + unittest.main()