feat: 账单模块

backend/app/plugin/module_payment/notification/crud.py

@@ -1,10 +1,12 @@
-from typing import Any
+from collections.abc import Sequence
+from typing import Any, List
 
 from app.api.v1.module_system.auth.schema import AuthSchema
 from app.core.base_crud import CRUDBase
 from app.core.exceptions import CustomException
 
-from .model import AlipayNotifyLogModel
+from .model import AlipayNotifyLogModel, PayBillModel, PayBillOrderModel, PayBillVoucherModel
+
 
 class AlipayNotifyLogCRUD(CRUDBase[AlipayNotifyLogModel, Any, Any]):
     """支付宝通知日志 CRUD 操作"""
@@ -35,4 +37,193 @@ class AlipayNotifyLogCRUD(CRUDBase[AlipayNotifyLogModel, Any, Any]):
         await self.auth.db.flush()
         await self.auth.db.refresh(obj)
 
-        return obj
+        return obj
+
+
+class BillCRUD(CRUDBase[PayBillModel, Any, Any]):
+    """账单 CRUD 操作"""
+
+    def __init__(self, auth: AuthSchema) -> None:
+        self.auth = auth
+        super().__init__(model=PayBillModel, auth=auth)
+
+    async def get_by_pay_no(
+        self, pay_no: str
+    ) -> PayBillModel | None:
+        """根据账单号查询账单"""
+        return await self.get(pay_no=pay_no)
+
+    async def get_by_enterprise_id(
+        self, enterprise_id: str
+    ) -> Sequence[PayBillModel]:
+        """根据企业ID查询账单列表"""
+        return await self.list({"enterprise_id": enterprise_id})
+
+    async def update_by_pay_no(
+        self, pay_no: str, data: dict
+    ) -> PayBillModel | None:
+        """根据账单号更新账单"""
+        obj = await self.get(pay_no=pay_no, preload=[])
+        if not obj:
+            raise CustomException(msg="账单不存在")
+
+        if self.auth.user and hasattr(obj, "updated_id"):
+            setattr(obj, "updated_id", self.auth.user.id)
+
+        for key, value in data.items():
+            if hasattr(obj, key):
+                setattr(obj, key, value)
+
+        await self.auth.db.flush()
+        await self.auth.db.refresh(obj)
+
+        return obj
+
+    async def create_or_update(
+        self, pay_no: str, data: dict
+    ) -> PayBillModel:
+        """创建或更新账单"""
+        obj = await self.get(pay_no=pay_no, preload=[])
+        
+        if obj:
+            if self.auth.user and hasattr(obj, "updated_id"):
+                setattr(obj, "updated_id", self.auth.user.id)
+            
+            for key, value in data.items():
+                if hasattr(obj, key):
+                    setattr(obj, key, value)
+            
+            await self.auth.db.flush()
+            await self.auth.db.refresh(obj)
+            return obj
+        else:
+            data["pay_no"] = pay_no
+            if "tenant_id" not in data:
+                data["tenant_id"] = self.auth.tenant_id
+            return await self.create(data=data, skip_tenant_id=True)
+
+
+class BillOrderCRUD(CRUDBase[PayBillOrderModel, Any, Any]):
+    """订单 CRUD 操作"""
+
+    def __init__(self, auth: AuthSchema) -> None:
+        self.auth = auth
+        super().__init__(model=PayBillOrderModel, auth=auth)
+
+    async def get_by_order_no(
+        self, order_no: str
+    ) -> PayBillOrderModel | None:
+        """根据订单号查询订单"""
+        return await self.get(order_no=order_no)
+
+    async def get_by_pay_no(
+        self, pay_no: str
+    ) -> Sequence[PayBillOrderModel]:
+        """根据账单号查询订单列表"""
+        return await self.list({"pay_no": pay_no})
+
+    async def update_by_order_no(
+        self, order_no: str, data: dict
+    ) -> PayBillOrderModel | None:
+        """根据订单号更新订单"""
+        obj = await self.get(order_no=order_no, preload=[])
+        if not obj:
+            raise CustomException(msg="订单不存在")
+
+        if self.auth.user and hasattr(obj, "updated_id"):
+            setattr(obj, "updated_id", self.auth.user.id)
+
+        for key, value in data.items():
+            if hasattr(obj, key):
+                setattr(obj, key, value)
+
+        await self.auth.db.flush()
+        await self.auth.db.refresh(obj)
+
+        return obj
+
+    async def create_or_update(
+        self, order_no: str, data: dict
+    ) -> PayBillOrderModel:
+        """创建或更新订单"""
+        obj = await self.get(order_no=order_no, preload=[])
+        
+        if obj:
+            if self.auth.user and hasattr(obj, "updated_id"):
+                setattr(obj, "updated_id", self.auth.user.id)
+            
+            for key, value in data.items():
+                if hasattr(obj, key):
+                    setattr(obj, key, value)
+            
+            await self.auth.db.flush()
+            await self.auth.db.refresh(obj)
+            return obj
+        else:
+            data["order_no"] = order_no
+            if "tenant_id" not in data:
+                data["tenant_id"] = self.auth.tenant_id
+            return await self.create(data=data, skip_tenant_id=True)
+
+
+class BillVoucherCRUD(CRUDBase[PayBillVoucherModel, Any, Any]):
+    """凭证 CRUD 操作"""
+
+    def __init__(self, auth: AuthSchema) -> None:
+        self.auth = auth
+        super().__init__(model=PayBillVoucherModel, auth=auth)
+
+    async def get_by_voucher_id(
+        self, voucher_id: str
+    ) -> PayBillVoucherModel | None:
+        """根据凭证ID查询凭证"""
+        return await self.get(voucher_id=voucher_id)
+
+    async def get_by_pay_no(
+        self, pay_no: str
+    ) -> Sequence[PayBillVoucherModel]:
+        """根据账单号查询凭证列表"""
+        return await self.list({"pay_no": pay_no})
+
+    async def update_by_voucher_id(
+        self, voucher_id: str, data: dict
+    ) -> PayBillVoucherModel | None:
+        """根据凭证ID更新凭证"""
+        obj = await self.get(voucher_id=voucher_id, preload=[])
+        if not obj:
+            raise CustomException(msg="凭证不存在")
+
+        if self.auth.user and hasattr(obj, "updated_id"):
+            setattr(obj, "updated_id", self.auth.user.id)
+
+        for key, value in data.items():
+            if hasattr(obj, key):
+                setattr(obj, key, value)
+
+        await self.auth.db.flush()
+        await self.auth.db.refresh(obj)
+
+        return obj
+
+    async def create_or_update(
+        self, voucher_id: str, data: dict
+    ) -> PayBillVoucherModel:
+        """创建或更新凭证"""
+        obj = await self.get(voucher_id=voucher_id, preload=[])
+        
+        if obj:
+            if self.auth.user and hasattr(obj, "updated_id"):
+                setattr(obj, "updated_id", self.auth.user.id)
+            
+            for key, value in data.items():
+                if hasattr(obj, key):
+                    setattr(obj, key, value)
+            
+            await self.auth.db.flush()
+            await self.auth.db.refresh(obj)
+            return obj
+        else:
+            data["voucher_id"] = voucher_id
+            if "tenant_id" not in data:
+                data["tenant_id"] = self.auth.tenant_id
+            return await self.create(data=data, skip_tenant_id=True)

backend/app/plugin/module_payment/notification/handlers/__init__.py

@@ -2,6 +2,7 @@ from .base_handler import BaseHandler
 from .enterprise_handler import EnterpriseHandler
 from .employee_handler import EmployeeHandler
 from .bill_handler import BillHandler, VoucherHandler
+from .order_handler import OrderHandler
 from .account_handler import AccountHandler
 
 __all__ = [
@@ -10,5 +11,6 @@ __all__ = [
     "EmployeeHandler",
     "BillHandler",
     "VoucherHandler",
+    "OrderHandler",
     "AccountHandler",
 ]

backend/app/plugin/module_payment/notification/handlers/bill_handler.py

@@ -1,10 +1,15 @@
+from datetime import datetime
+from decimal import Decimal
 from redis.asyncio import Redis
 
 from app.api.v1.module_system.auth.schema import AuthSchema
 from app.core.logger import log
 from app.plugin.module_payment.account.service import AccountService
+from app.core.alipay import AlipayClient
 
 from ..schemas import ConsumeChangeContent, VoucherChangeContent
+from ..model import PayBillModel, PayBillOrderModel, PayBillVoucherModel
+from ..crud import BillCRUD, BillOrderCRUD, BillVoucherCRUD
 from .base_handler import BaseHandler
 from ...openapi.service import OpenTransferService
 
@@ -15,6 +20,11 @@ class BillHandler(BaseHandler[dict]):
     async def handle(self, method: str, content: dict, auth: AuthSchema, redis: Redis) -> bool:
         """
         处理账单变动通知
+        
+        流程：
+        1. 保存账单基础数据
+        2. 调用 alipay.commerce.ec.consume.detail.query 查询详情
+        3. 保存账单和凭证详情数据
         """
         try:
             notify_data = ConsumeChangeContent(**content)
@@ -25,7 +35,7 @@ class BillHandler(BaseHandler[dict]):
         try:
             return await self._process_bill(notify_data, auth)
         except Exception as e:
-            log.error(f"处理账单变动通知异常: {e}")
+            log.error(f"处理账单变动通知异常: {e}", exc_info=True)
             return False
 
     async def _process_bill(self, data: ConsumeChangeContent, auth: AuthSchema) -> bool:
@@ -37,6 +47,21 @@ class BillHandler(BaseHandler[dict]):
             f"consume_type={data.consume_type}"
         )
 
+        # 1. 保存账单基础数据
+        # await self._save_bill_base(data, auth)
+
+        # 2. 调用支付宝查询详情
+        # try:
+        #     detail = await self._query_bill_detail(data.pay_no, auth)
+        # except Exception as e:
+        #     log.warning(f"查询账单详情失败: {e}")
+        #     return True
+
+        # 3. 保存账单和凭证详情数据
+        # if detail:
+        #     await self._save_bill_detail(detail, auth)
+
+        # 处理转账类型
         if data.consume_type == "TRANSFER":
             await AccountService.update_transfer_status_service(
                 auth, data.pay_no, "SUCCESS", data.model_dump(exclude_none=True)
@@ -44,14 +69,188 @@ class BillHandler(BaseHandler[dict]):
             await OpenTransferService.open_return_service(auth, data.pay_no)
         
         return True
-    
-    async def _handle_trans_change(self, content: dict, auth: AuthSchema) -> bool:
-        """
-        处理转账变动通知
-        ● 功能说明：转账结果通知，单笔交易账单信息变化时通知。
-            ○ 通知中"consume_type":"TRANSFER"，可以用于判断转账是否成功。
-        """
-        return False
+
+    async def _save_bill_base(self, data: ConsumeChangeContent, auth: AuthSchema) -> None:
+        """保存账单基础数据"""
+        bill_crud = BillCRUD(auth)
+        
+        bill_data = {
+            "enterprise_id": data.enterprise_id,
+            "account_id": data.account_id,
+            "employee_id": data.employee_id,
+            "consume_type": data.consume_type,
+            "consume_amount": Decimal(data.consume_amount) if data.consume_amount else None,
+            "gmt_biz_create": datetime.strptime(data.gmt_biz_create, "%Y-%m-%d %H:%M:%S") if data.gmt_biz_create else None,
+            "gmt_recieve_pay": datetime.strptime(data.gmt_recieve_pay, "%Y-%m-%d %H:%M:%S") if data.gmt_recieve_pay else None,
+            "peer_pay_amount": Decimal(data.peer_pay_amount) if data.peer_pay_amount else None,
+            "notify_reason": data.notify_reason,
+            "notify_msg": data.notify_msg,
+            "related_pay_no": data.related_pay_no,
+            "expense_rule_group_id": data.expense_rule_group_id,
+            "expense_scene_code": data.expense_scene_code,
+            "expense_type": data.expense_type,
+            "status": "NEW",
+        }
+        
+        await bill_crud.create_or_update(data.pay_no, bill_data)
+        log.info(f"保存账单基础数据成功: pay_no={data.pay_no}")
+
+    async def _query_bill_detail(self, pay_no: str, enterprise_id: str, auth: AuthSchema) -> dict | None:
+        """调用 alipay.commerce.ec.consume.detail.query 查询账单详情"""
+        from alipay.aop.api.request.AlipayCommerceEcConsumeDetailQueryRequest import (
+            AlipayCommerceEcConsumeDetailQueryRequest,
+        )
+        from alipay.aop.api.response.AlipayCommerceEcConsumeDetailQueryResponse import (
+            AlipayCommerceEcConsumeDetailQueryResponse,
+        )
+        from alipay.aop.api.domain.AlipayCommerceEcConsumeDetailQueryModel import AlipayCommerceEcConsumeDetailQueryModel
+
+        model = AlipayCommerceEcConsumeDetailQueryModel()
+        model.pay_no = pay_no
+        model.enterprise_id = enterprise_id
+
+        request = AlipayCommerceEcConsumeDetailQueryRequest()
+        request.biz_model = model
+
+        client = AlipayClient.get_client()
+        response = client.execute(request)
+
+        if not response:
+            log.error("查询账单详情失败: 无响应")
+            return None
+
+        result = AlipayCommerceEcConsumeDetailQueryResponse()
+        result.parse_response_content(response)
+
+        if not result.is_success():
+            log.error(f"查询账单详情失败: {result.msg}")
+            return None
+
+        # 解析响应为字典
+        return {
+            "pay_no": getattr(result.consume_info, "pay_no", None),
+            "enterprise_id": getattr(result.consume_info, "enterprise_id", None),
+            "employee_id": getattr(result.consume_info, "employee_id", None),
+            "employee_name": getattr(result.consume_info, "employee_name", None),
+            "consume_amount": getattr(result, "consume_amount", None),
+            "peer_pay_amount": getattr(result, "peer_pay_amount", None),
+            "employee_pay_amount": getattr(result, "employee_pay_amount", None),
+            "order_info": self._parse_order_info(result),
+            "voucher_list": self._parse_voucher_list(result),
+        }
+
+    def _parse_order_info(self, result) -> dict | None:
+        """解析订单信息"""
+        order_info = getattr(result, "order_info", None)
+        if not order_info:
+            return None
+        
+        return {
+            "order_no": getattr(order_info, "order_no", None),
+            "trade_no": getattr(order_info, "trade_no", None),
+            "product_code": getattr(order_info, "product_code", None),
+            "order_title": getattr(order_info, "order_title", None),
+            "order_amount": getattr(order_info, "order_amount", None),
+            "order_status": getattr(order_info, "order_status", None),
+            "merchant_name": getattr(order_info, "merchant_name", None),
+            "merchant_id": getattr(order_info, "merchant_id", None),
+            "shop_name": getattr(order_info, "shop_name", None),
+            "gmt_payment": getattr(order_info, "gmt_payment", None),
+            "fund_channel": getattr(order_info, "fund_channel", None),
+        }
+
+    def _parse_voucher_list(self, result) -> list | None:
+        """解析凭证列表"""
+        voucher_list = getattr(result, "voucher_list", None)
+        if not voucher_list:
+            return None
+        
+        vouchers = []
+        for voucher in voucher_list:
+            vouchers.append({
+                "voucher_id": getattr(voucher, "voucher_id", None),
+                "voucher_type": getattr(voucher, "voucher_type", None),
+                "voucher_status": getattr(voucher, "voucher_status", None),
+                "invoice_code": getattr(voucher, "invoice_code", None),
+                "invoice_no": getattr(voucher, "invoice_no", None),
+                "invoice_amount": getattr(voucher, "invoice_amount", None),
+                "tax_amount": getattr(voucher, "tax_amount", None),
+                "issue_date": getattr(voucher, "issue_date", None),
+                "check_code": getattr(voucher, "check_code", None),
+                "pdf_url": getattr(voucher, "pdf_url", None),
+            })
+        return vouchers
+
+    async def _save_bill_detail(self, detail: dict, auth: AuthSchema) -> None:
+        """保存账单详情和凭证数据"""
+        bill_crud = BillCRUD(auth)
+        
+        pay_no = detail.get("pay_no")
+        if not pay_no:
+            return
+
+        # 更新账单详情
+        await bill_crud.update_by_pay_no(pay_no, {"status": "PROCESSED", "ext_infos": detail})
+
+        # 保存订单详情
+        order_info = detail.get("order_info")
+        if order_info:
+            await self._save_order_detail(pay_no, order_info, auth)
+
+        # 保存凭证列表
+        voucher_list = detail.get("voucher_list")
+        if voucher_list:
+            await self._save_voucher_list(pay_no, voucher_list, auth)
+
+        log.info(f"保存账单详情成功: pay_no={pay_no}")
+
+    async def _save_order_detail(self, pay_no: str, order_info: dict, auth: AuthSchema) -> None:
+        """保存订单详情"""
+        order_crud = BillOrderCRUD(auth)
+        
+        order_no = order_info.get("order_no")
+        if not order_no:
+            return
+
+        order_data = {
+            "pay_no": pay_no,
+            "trade_no": order_info.get("trade_no"),
+            "product_code": order_info.get("product_code"),
+            "order_title": order_info.get("order_title"),
+            "order_amount": Decimal(order_info.get("order_amount")) if order_info.get("order_amount") else None,
+            "order_status": order_info.get("order_status"),
+            "merchant_name": order_info.get("merchant_name"),
+            "merchant_id": order_info.get("merchant_id"),
+            "shop_name": order_info.get("shop_name"),
+            "gmt_payment": datetime.strptime(order_info.get("gmt_payment"), "%Y-%m-%d %H:%M:%S") if order_info.get("gmt_payment") else None,
+            "fund_channel": order_info.get("fund_channel"),
+        }
+        
+        await order_crud.create_or_update(order_no, order_data)
+
+    async def _save_voucher_list(self, pay_no: str, voucher_list: list, auth: AuthSchema) -> None:
+        """保存凭证列表"""
+        voucher_crud = BillVoucherCRUD(auth)
+
+        for voucher in voucher_list:
+            voucher_id = voucher.get("voucher_id")
+            if not voucher_id:
+                continue
+
+            voucher_data = {
+                "pay_no": pay_no,
+                "voucher_type": voucher.get("voucher_type"),
+                "voucher_status": voucher.get("voucher_status"),
+                "invoice_code": voucher.get("invoice_code"),
+                "invoice_no": voucher.get("invoice_no"),
+                "invoice_amount": Decimal(voucher.get("invoice_amount")) if voucher.get("invoice_amount") else None,
+                "tax_amount": Decimal(voucher.get("tax_amount")) if voucher.get("tax_amount") else None,
+                "issue_date": datetime.strptime(voucher.get("issue_date"), "%Y-%m-%d") if voucher.get("issue_date") else None,
+                "check_code": voucher.get("check_code"),
+                "pdf_url": voucher.get("pdf_url"),
+            }
+            
+            await voucher_crud.create_or_update(voucher_id, voucher_data)
 
 
 class VoucherHandler(BaseHandler[dict]):
@@ -60,6 +259,10 @@ class VoucherHandler(BaseHandler[dict]):
     async def handle(self, method: str, content: dict, auth: AuthSchema, redis: Redis) -> bool:
         """
         处理凭证变动通知
+        
+        流程：
+        1. 调用 alipay.commerce.ec.consume.detail.query 查询详情
+        2. 更新凭证信息
         """
         try:
             notify_data = VoucherChangeContent(**content)
@@ -75,7 +278,7 @@ class VoucherHandler(BaseHandler[dict]):
         try:
             return await self._process_voucher(notify_data, auth)
         except Exception as e:
-            log.error(f"处理凭证变动通知异常: {e}")
+            log.error(f"处理凭证变动通知异常: {e}", exc_info=True)
             return False
 
     async def _process_voucher(self, data: VoucherChangeContent, auth: AuthSchema) -> bool:
@@ -85,4 +288,65 @@ class VoucherHandler(BaseHandler[dict]):
             f"enterprise_id={data.enterprise_id}, "
             f"action={data.action}"
         )
-        return True
+
+        # 4.1: 调用支付宝查询账单及凭证详情
+        try:
+            # 使用 pay_no 或 voucher_id 查询，这里需要确定查询参数
+            # 由于凭证变动通知中没有直接提供 pay_no，我们先尝试通过 voucher_id 查询
+            detail = await self._query_bill_detail_by_voucher(data.voucher_id, auth)
+        except Exception as e:
+            log.warning(f"查询凭证详情失败: {e}")
+            return True
+
+        # 4.3: 更新凭证信息
+        if detail:
+            await self._update_voucher_info(detail, auth)
+
+        return True
+
+    async def _query_bill_detail_by_voucher(self, voucher_id: str, auth: AuthSchema) -> dict | None:
+        """通过凭证ID查询账单详情"""
+        from alipay.aop.api.request.AlipayCommerceEcConsumeDetailQueryRequest import (
+            AlipayCommerceEcConsumeDetailQueryRequest,
+        )
+        from alipay.aop.api.response.AlipayCommerceEcConsumeDetailQueryResponse import (
+            AlipayCommerceEcConsumeDetailQueryResponse,
+        )
+
+        # 注意：consume.detail.query 需要 pay_no 参数
+        # 如果没有 pay_no，可以考虑通过其他方式获取
+        # 这里先返回空，实际实现时需要根据业务场景补充
+        log.warning(f"凭证查询需要 pay_no，当前仅收到 voucher_id={voucher_id}")
+        return None
+
+    async def _update_voucher_info(self, detail: dict, auth: AuthSchema) -> None:
+        """更新凭证信息"""
+        voucher_crud = BillVoucherCRUD(auth)
+        
+        voucher_list = detail.get("voucher_list")
+        if not voucher_list:
+            return
+
+        pay_no = detail.get("pay_no")
+        
+        for voucher in voucher_list:
+            voucher_id = voucher.get("voucher_id")
+            if not voucher_id:
+                continue
+
+            voucher_data = {
+                "pay_no": pay_no,
+                "voucher_type": voucher.get("voucher_type"),
+                "voucher_status": voucher.get("voucher_status"),
+                "invoice_code": voucher.get("invoice_code"),
+                "invoice_no": voucher.get("invoice_no"),
+                "invoice_amount": Decimal(voucher.get("invoice_amount")) if voucher.get("invoice_amount") else None,
+                "tax_amount": Decimal(voucher.get("tax_amount")) if voucher.get("tax_amount") else None,
+                "issue_date": datetime.strptime(voucher.get("issue_date"), "%Y-%m-%d") if voucher.get("issue_date") else None,
+                "check_code": voucher.get("check_code"),
+                "pdf_url": voucher.get("pdf_url"),
+            }
+            
+            await voucher_crud.create_or_update(voucher_id, voucher_data)
+        
+        log.info(f"更新凭证信息成功: voucher_count={len(voucher_list)}")

backend/app/plugin/module_payment/notification/handlers/order_handler.py

@@ -0,0 +1,88 @@
+from datetime import datetime
+from decimal import Decimal
+from redis.asyncio import Redis
+
+from app.api.v1.module_system.auth.schema import AuthSchema
+from app.core.logger import log
+
+from ..schemas import OrderChangeContent
+from ..crud import BillOrderCRUD
+from .base_handler import BaseHandler
+
+
+class OrderHandler(BaseHandler[dict]):
+    """订单变动通知处理器"""
+
+    async def handle(self, method: str, content: dict, auth: AuthSchema, redis: Redis) -> bool:
+        """
+        处理订单变动通知
+        
+        接口名：alipay.ebpp.invoice.ecorder.order.changed
+        功能说明：交易订单信息，部分场景会有比账单中更详细的业务订单数据
+        
+        流程：
+        1. 保存/更新订单数据
+        2. 如果有账单号，关联到账单
+        """
+        try:
+            notify_data = OrderChangeContent(**content)
+        except Exception as e:
+            log.error(f"解析订单通知内容失败: {e}")
+            return False
+
+        action = notify_data.action
+        order_no = notify_data.order_no
+
+        log.info(f"处理订单变动通知: action={action}, order_no={order_no}")
+
+        try:
+            return await self._process_order(notify_data, auth)
+        except Exception as e:
+            log.error(f"处理订单变动通知异常: {e}", exc_info=True)
+            return False
+
+    async def _process_order(self, data: OrderChangeContent, auth: AuthSchema) -> bool:
+        """处理订单"""
+        log.info(
+            f"订单变动: order_no={data.order_no}, "
+            f"enterprise_id={data.enterprise_id}, "
+            f"action={data.action}, "
+            f"order_status={data.order_status}"
+        )
+
+        # 保存/更新订单数据
+        await self._save_order(data, auth)
+
+        return True
+
+    async def _save_order(self, data: OrderChangeContent, auth: AuthSchema) -> None:
+        """保存订单数据"""
+        order_crud = BillOrderCRUD(auth)
+
+        order_data = {
+            "enterprise_id": data.enterprise_id,
+            "pay_no": data.pay_no,
+            "trade_no": data.trade_no or data.out_biz_no,
+            "product_code": data.order_type,
+            "order_title": data.order_title,
+            "order_amount": Decimal(data.order_amount) if data.order_amount else None,
+            "order_status": data.order_status,
+            "merchant_name": data.merchant_name,
+            "merchant_id": data.merchant_id,
+            "shop_name": data.shop_name,
+            "gmt_payment": datetime.strptime(data.gmt_payment, "%Y-%m-%d %H:%M:%S") if data.gmt_payment else None,
+            "fund_channel": data.fund_channel,
+            "gmt_create": datetime.strptime(data.gmt_create, "%Y-%m-%d %H:%M:%S") if data.gmt_create else (
+                datetime.strptime(data.change_time, "%Y-%m-%d %H:%M:%S") if data.change_time else None
+            ),
+            "ext_infos": {
+                "action": data.action,
+                "employee_id": data.employee_id,
+                "out_biz_no": data.out_biz_no,
+                "invoice_status": data.invoice_status,
+                "ext_info": data.ext_info,
+            },
+        }
+
+        await order_crud.create_or_update(data.order_no, order_data)
+        log.info(f"保存订单数据成功: order_no={data.order_no}")

backend/app/plugin/module_payment/notification/model.py

@@ -1,7 +1,12 @@
 from datetime import datetime
-from sqlalchemy import JSON, DateTime, String, Text, Boolean
+from decimal import Decimal
+
+from sqlalchemy import JSON, DateTime, String, Text, Boolean, Numeric
 from sqlalchemy.orm import Mapped, mapped_column
-from app.core.base_model import PaymentModelMixin, TenantMixin
+
+from app.core.base_model import PaymentModelMixin, TenantMixin, EnterpriseMixin
+from app.common.enums import PermissionFilterStrategy
+
 
 class AlipayNotifyLogModel(PaymentModelMixin):
     """支付宝通知日志模型"""
@@ -32,4 +37,136 @@ class AlipayNotifyLogModel(PaymentModelMixin):
     )
     received_at: Mapped[datetime] = mapped_column(
         DateTime, default=datetime.now, nullable=False, comment="接收时间"
-    )
+    )
+
+
+class PayBillModel(PaymentModelMixin, TenantMixin, EnterpriseMixin):
+    """账单主表"""
+
+    __tablename__ = "pay_bill"
+    __table_args__ = {"comment": "企业码账单表"}
+    __permission_strategy__ = PermissionFilterStrategy.ENTERPRISE_BASED
+
+    pay_no: Mapped[str] = mapped_column(
+        String(64), unique=True, index=True, comment="支付宝账单号"
+    )
+    account_id: Mapped[str] = mapped_column(String(64), comment="共同账户ID")
+    employee_id: Mapped[str] = mapped_column(String(64), comment="员工ID")
+    consume_type: Mapped[str] = mapped_column(
+        String(32), comment="账单类型: CONSUME/REFUND/TRANSFER"
+    )
+    consume_amount: Mapped[Decimal] = mapped_column(
+        Numeric(12, 2), comment="账单金额"
+    )
+    gmt_biz_create: Mapped[datetime] = mapped_column(
+        DateTime, comment="账单创建时间"
+    )
+    gmt_recieve_pay: Mapped[datetime | None] = mapped_column(
+        DateTime, comment="支付时间"
+    )
+    peer_pay_amount: Mapped[Decimal | None] = mapped_column(
+        Numeric(12, 2), comment="企业代付金额"
+    )
+    notify_reason: Mapped[str] = mapped_column(String(255), comment="通知原因")
+    notify_msg: Mapped[str | None] = mapped_column(Text, comment="通知描述")
+    related_pay_no: Mapped[str | None] = mapped_column(
+        String(64), comment="关联账单号"
+    )
+    expense_rule_group_id: Mapped[str | None] = mapped_column(
+        String(64), comment="费控规则ID"
+    )
+    expense_scene_code: Mapped[str | None] = mapped_column(
+        String(32), comment="费用场景"
+    )
+    expense_type: Mapped[str | None] = mapped_column(
+        String(64), comment="费用类型"
+    )
+    status: Mapped[str] = mapped_column(
+        String(32), default="NEW", comment="状态: NEW/PROCESSED/FAILED"
+    )
+    ext_infos: Mapped[dict | None] = mapped_column(JSON, comment="扩展信息")
+
+
+class PayBillOrderModel(PaymentModelMixin, TenantMixin, EnterpriseMixin):
+    """订单明细表"""
+
+    __tablename__ = "pay_bill_order"
+    __table_args__ = {"comment": "账单订单明细表"}
+    __permission_strategy__ = PermissionFilterStrategy.ENTERPRISE_BASED
+
+    pay_no: Mapped[str] = mapped_column(
+        String(64), index=True, comment="关联账单号"
+    )
+    order_no: Mapped[str] = mapped_column(
+        String(64), unique=True, index=True, comment="支付宝订单号"
+    )
+    trade_no: Mapped[str | None] = mapped_column(
+        String(64), comment="支付宝交易号"
+    )
+    product_code: Mapped[str | None] = mapped_column(
+        String(64), comment="产品码"
+    )
+    order_title: Mapped[str | None] = mapped_column(
+        String(255), comment="订单标题"
+    )
+    order_amount: Mapped[Decimal | None] = mapped_column(
+        Numeric(12, 2), comment="订单金额"
+    )
+    order_status: Mapped[str | None] = mapped_column(
+        String(32), comment="订单状态"
+    )
+    merchant_name: Mapped[str | None] = mapped_column(
+        String(255), comment="商户名称"
+    )
+    merchant_id: Mapped[str | None] = mapped_column(
+        String(64), comment="商户ID"
+    )
+    shop_name: Mapped[str | None] = mapped_column(
+        String(255), comment="门店名称"
+    )
+    gmt_payment: Mapped[datetime | None] = mapped_column(
+        DateTime, comment="支付时间"
+    )
+    fund_channel: Mapped[str | None] = mapped_column(
+        String(64), comment="资金渠道"
+    )
+
+
+class PayBillVoucherModel(PaymentModelMixin, TenantMixin, EnterpriseMixin):
+    """凭证信息表"""
+
+    __tablename__ = "pay_bill_voucher"
+    __table_args__ = {"comment": "账单凭证信息表"}
+    __permission_strategy__ = PermissionFilterStrategy.ENTERPRISE_BASED
+
+    pay_no: Mapped[str] = mapped_column(
+        String(64), index=True, comment="关联账单号"
+    )
+    voucher_id: Mapped[str] = mapped_column(
+        String(64), index=True, comment="凭证ID"
+    )
+    voucher_type: Mapped[str | None] = mapped_column(
+        String(32), comment="凭证类型"
+    )
+    voucher_status: Mapped[str | None] = mapped_column(
+        String(32), comment="凭证状态"
+    )
+    invoice_code: Mapped[str | None] = mapped_column(
+        String(64), comment="发票代码"
+    )
+    invoice_no: Mapped[str | None] = mapped_column(
+        String(64), comment="发票号码"
+    )
+    invoice_amount: Mapped[Decimal | None] = mapped_column(
+        Numeric(12, 2), comment="发票金额"
+    )
+    tax_amount: Mapped[Decimal | None] = mapped_column(
+        Numeric(12, 2), comment="税额"
+    )
+    issue_date: Mapped[datetime | None] = mapped_column(
+        DateTime, comment="开票日期"
+    )
+    check_code: Mapped[str | None] = mapped_column(
+        String(64), comment="校验码"
+    )
+    pdf_url: Mapped[str | None] = mapped_column(Text, comment="PDF下载地址")

backend/app/plugin/module_payment/notification/schemas.py

@@ -142,4 +142,52 @@ class ConsumeChangeContent(BaseModel):
     expense_scene_code: str | None = Field(None, description="费用场景")
     expense_type: str | None = Field(None, description="费用类型")
     expense_type_sub_category: str | None = Field(None, description="费用类型子类目")
-    ext_infos: str | None = Field(None, description="扩展参数，JSON格式")
+    ext_infos: str | None = Field(None, description="扩展参数，JSON格式")
+
+
+class OrderChangeContent(BaseModel):
+    """订单变动通知内容
+    
+    接口名: alipay.ebpp.invoice.ecorder.order.changed
+    参考文档: https://opendocs.alipay.com/apis/03iu5y
+    """
+
+    order_no: str = Field(..., description="支付宝订单号")
+    enterprise_id: str = Field(..., description="企业ID")
+    employee_id: str | None = Field(None, description="员工ID")
+    out_biz_no: str | None = Field(None, description="外部业务号")
+    action: str = Field(..., description="变动动作")
+    change_time: str = Field(..., description="变动时间")
+    order_status: str | None = Field(None, description="订单状态")
+    
+    # 补充字段
+    pay_no: str | None = Field(None, description="账单号")
+    order_title: str | None = Field(None, description="订单标题")
+    order_amount: str | None = Field(None, description="订单金额")
+    trade_no: str | None = Field(None, description="交易号")
+    merchant_id: str | None = Field(None, description="商户ID")
+    merchant_name: str | None = Field(None, description="商户名称")
+    shop_id: str | None = Field(None, description="店铺ID")
+    shop_name: str | None = Field(None, description="店铺名称")
+    gmt_create: str | None = Field(None, description="创建时间")
+    gmt_payment: str | None = Field(None, description="支付时间")
+    fund_channel: str | None = Field(None, description="资金渠道")
+    order_type: str | None = Field(None, description="订单类型")
+    invoice_status: str | None = Field(None, description="发票状态")
+    ext_info: str | None = Field(None, description="扩展信息")
+
+
+class BillDetailQueryOut(BaseModel):
+    """账单详情查询响应"""
+
+    pay_no: str = Field(..., description="支付宝账单号")
+    enterprise_id: str = Field(..., description="企业ID")
+    employee_id: str = Field(..., description="员工ID")
+    employee_name: str | None = Field(None, description="员工姓名")
+    
+    consume_amount: str = Field(..., description="账单金额")
+    peer_pay_amount: str | None = Field(None, description="企业代付金额")
+    employee_pay_amount: str | None = Field(None, description="员工自付金额")
+    
+    order_info: dict | None = Field(None, description="订单详情")
+    voucher_list: list[dict] | None = Field(None, description="凭证列表")

backend/app/plugin/module_payment/notification/service.py

@@ -15,6 +15,7 @@ from .handlers import (
     EmployeeHandler,
     BillHandler,
     VoucherHandler,
+    OrderHandler,
     AccountHandler,
 )
 
@@ -27,6 +28,7 @@ class NotificationService:
         AlipayNotifyMethodEnum.EMPLOYEE_CHANGE.value: EmployeeHandler,
         AlipayNotifyMethodEnum.CONSUME_CHANGE.value: BillHandler,
         AlipayNotifyMethodEnum.VOUCHER_CHANGE.value: VoucherHandler,
+        AlipayNotifyMethodEnum.INVOICE_ORDER_CHANGE.value: OrderHandler,
         AlipayNotifyMethodEnum.TRANS_AUTHORIZE_NOTIFY.value: AccountHandler,
         AlipayNotifyMethodEnum.FUND_CHANGE_NOTIFY.value: AccountHandler,
     }