fix(expense): 费控额度改造 - 去除假额度+真实额度同步+问题修复

1. 去除创建/修改制度时往 pay_expense_quota 写假额度的逻辑
2. 修复 OutsourceNotifyService 金额单位错误（元→分）及本地幂等
3. BillHandler 消费通知时同步更新 pay_expense_quota 可用额度
4. 详情页额度查询加 limit(1000)
5. VoucherHandler 改用 pay_no 查账单，补全 VoucherChangeContent 字段

backend/app/plugin/module_payment/expense/institution/service.py

@@ -243,36 +243,9 @@ class InstitutionService:
                 except Exception as e:
                     log.warning(f"保存使用规则到本地失败: {e}")
 
-        # 第6步：保存发放规则到本地
-        if issuerule_data and issue_rule_id:
-            from app.plugin.module_payment.expense.quota.model import QuotaModel
-            # 根据适用范围确定 employee_id
-            scope = (raw_data or {}).get("applicable_scope", "")
-            emp_id = "ALL"
-            if scope == "employee":
-                owner_ids = (raw_data or {}).get("scope_owner_id_list") or []
-                emp_id = owner_ids[0] if owner_ids else "ALL"
-            elif scope == "department":
-                dept_id = (raw_data or {}).get("department_id", "")
-                emp_id = f"DEPT:{dept_id}" if dept_id else "ALL"
-            quota_save_data = {
-                "employee_id": emp_id,
-                "institution_id": institution_id,
-                "out_biz_no": issuerule_data.get("outer_source_id", f"issue_{institution_id}"),
-                "quota_id": issue_rule_id,
-                "total_amount": float(issuerule_data.get("issue_amount_value", 0)),
-                "available_amount": float(issuerule_data.get("issue_amount_value", 0)),
-                "status": "QUOTA_ACTIVE",
-                "enterprise_id": enterprise_id,
-                "tenant_id": auth.user.tenant_id if auth.user else 1,
-            }
-            try:
-                from sqlalchemy import insert
-                stmt = insert(QuotaModel).values(**quota_save_data)
-                await auth.db.execute(stmt)
-                await auth.db.flush()
-            except Exception as e:
-                log.warning(f"保存发放规则到本地失败: {e}")
+        # 第6步：保存发放规则到本地（不写入pay_expense_quota）
+        # 去除假额度写入：额度由外部消费同步时通过
+        # alipay.ebpp.invoice.expensecomsue.outsource.notify 写入真实数据
 
         return {
             "institution_id": institution_id,
@@ -430,7 +403,7 @@ class InstitutionService:
             result_dict["rule_list"] = rule_list
 
         # 查额度
-        quota_stmt = select(QuotaModel).where(QuotaModel.institution_id == institution_id)
+        quota_stmt = select(QuotaModel).where(QuotaModel.institution_id == institution_id).limit(1000)
         quota_result = await auth.db.execute(quota_stmt)
         quotas = quota_result.scalars().all()
         if quotas:
@@ -665,57 +638,9 @@ class InstitutionService:
                 await auth.db.flush()
                 log.info(f"已同步使用规则: institution_id={institution_id}")
 
-            # 同步发放规则（modify_issue_rule_detail_info）
-            issue_detail = (raw_data or {}).get("modify_issue_rule_detail_info") or {}
-            if issue_detail:
-                from app.plugin.module_payment.expense.quota.model import QuotaModel
-                from sqlalchemy import delete as sa_delete
-
-                # 删除发放规则
-                delete_ids = issue_detail.get("delete_issue_rule_id_list", [])
-                if delete_ids:
-                    d_stmt = sa_delete(QuotaModel).where(
-                        QuotaModel.quota_id.in_(delete_ids)
-                    )
-                    await auth.db.execute(d_stmt)
-
-                # 新增发放规则
-                add_list = issue_detail.get("add_issue_rule_list", [])
-                for rule in add_list:
-                    amount = float(rule.get("issue_amount_value", 0))
-                    ins_data = {
-                        "employee_id": "",
-                        "institution_id": institution_id,
-                        "out_biz_no": rule.get("outer_source_id", f"issue_{institution_id}"),
-                        "quota_id": rule.get("issue_rule_id"),
-                        "total_amount": amount,
-                        "available_amount": amount,
-                        "status": "QUOTA_ACTIVE",
-                        "enterprise_id": raw_data.get("enterprise_id", ""),
-                        "tenant_id": auth.user.tenant_id if auth.user else 1,
-                    }
-                    stmt = sa_insert(QuotaModel).values(**ins_data)
-                    await auth.db.execute(stmt)
-
-                # 修改发放规则
-                modify_rule = issue_detail.get("modify_issue_rule_list") or {}
-                if modify_rule.get("issue_rule_id"):
-                    q_id = modify_rule["issue_rule_id"].strip('"')
-                    update_q = {}
-                    if modify_rule.get("issue_amount_value"):
-                        update_q["total_amount"] = float(modify_rule["issue_amount_value"])
-                        update_q["available_amount"] = float(modify_rule["issue_amount_value"])
-                    if modify_rule.get("issue_rule_name"):
-                        update_q["standard_name"] = modify_rule["issue_rule_name"]
-                    if update_q and q_id:
-                        from sqlalchemy import update as sa_update
-                        u_stmt = sa_update(QuotaModel).where(
-                            QuotaModel.quota_id == q_id
-                        ).values(**update_q)
-                        await auth.db.execute(u_stmt)
-
-                await auth.db.flush()
-                log.info(f"已同步发放规则: institution_id={institution_id}")
+            # 同步发放规则（modify_issue_rule_detail_info）已去除
+            # 发放规则对应的额度数据由外部消费同步时通过
+            # alipay.ebpp.invoice.expensecomsue.outsource.notify 写入真实数据
 
         except Exception as e:
             log.warning(f"本地同步失败（不影响支付宝侧）: {e}")

backend/app/plugin/module_payment/expense/quota/__init__.py

@@ -0,0 +1,8 @@
+from .outsource_schema import OutsourceNotifySchema, OutsourceNotifyOutSchema
+from .outsource_service import OutsourceNotifyService
+
+__all__ = [
+    "OutsourceNotifySchema",
+    "OutsourceNotifyOutSchema",
+    "OutsourceNotifyService",
+]

backend/app/plugin/module_payment/expense/quota/controller.py

@@ -21,6 +21,8 @@ from .schema import (
     QuotaUpdateSchema,
 )
 from .service import QuotaService
+from .outsource_schema import OutsourceNotifySchema, OutsourceNotifyOutSchema
+from .outsource_service import OutsourceNotifyService
 
 QuotaRouter = APIRouter(
     route_class=OperationLogRoute,
@@ -129,3 +131,18 @@ async def list_quota_controller(
         auth=auth, page_no=page_no, page_size=page_size, search=search
     )
     return SuccessResponse(data=result, msg="查询额度列表成功")
+
+
+@QuotaRouter.post(
+    "/outsource/notify",
+    summary="外部消费额度同步",
+    description="将外部消费同步到支付宝额度系统 (alipay.ebpp.invoice.expensecomsue.outsource.notify)",
+    response_model=ResponseSchema[OutsourceNotifyOutSchema],
+)
+async def outsource_notify_controller(
+    data: OutsourceNotifySchema,
+    auth: Annotated[AuthSchema, Depends(AuthPermission(["module_payment:expense:quota:outsource:notify"]))],
+) -> JSONResponse:
+    result = await OutsourceNotifyService.notify_service(auth=auth, data=data)
+    log.info(f"外部消费额度同步: out_source_id={result.out_source_id}, success={result.success}")
+    return SuccessResponse(data=result, msg="外部消费额度同步成功" if result.success else "外部消费额度同步失败")

backend/app/plugin/module_payment/expense/quota/outsource_schema.py

@@ -0,0 +1,40 @@
+from datetime import datetime
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+
+class OutsourceNotifySchema(BaseModel):
+    """外部消费额度同步请求 (alipay.ebpp.invoice.expensecomsue.outsource.notify)
+
+    外部和企业码进行额度共管的场景下，外部以交易形式（支付、退款）将额度变更同步给支付宝
+    """
+
+    # 必填字段
+    enterprise_id: str = Field(description="企业ID")
+    employee_id: str = Field(description="员工ID")
+    standard_id: str = Field(description="使用规则ID（标准ID）")
+    amount: float = Field(description="消费金额（元），正数为消费/支付，负数为退款")
+    out_source_id: str = Field(description="外部业务号（幂等ID）")
+    deal_time: str = Field(description="交易时间，格式：yyyy-MM-dd HH:mm:ss")
+
+    # 可选字段
+    account_id: Optional[str] = Field(default=None, description="企业余额账户ID")
+    agreement_no: Optional[str] = Field(default=None, description="授权签约协议号")
+    employee_id_type: Optional[str] = Field(default=None, description="员工ID类型: EMPLOYEE/PHONE/ENTERPRISE_PAY_UID")
+    employee_open_id: Optional[str] = Field(default=None, description="员工open_id")
+    extend: Optional[str] = Field(default=None, description="扩展参数，JSON格式")
+    is_off_set: Optional[str] = Field(default=None, description="是否冲抵: Y/N")
+    platform: Optional[str] = Field(default=None, description="平台类型")
+    relate_no: Optional[str] = Field(default=None, description="关联单号")
+
+    # 本地补充字段
+    institution_id: Optional[str] = Field(default=None, description="制度ID（本地补充，用于写pay_expense_quota）")
+
+
+class OutsourceNotifyOutSchema(BaseModel):
+    """外部消费额度同步响应"""
+
+    success: bool = Field(description="是否成功")
+    out_source_id: str = Field(description="外部业务号")
+    message: str = Field(default="", description="响应消息")

backend/app/plugin/module_payment/expense/quota/outsource_service.py

@@ -0,0 +1,246 @@
+import asyncio
+from decimal import Decimal
+from sqlalchemy import select, insert, update as sa_update
+
+from app.api.v1.module_system.auth.schema import AuthSchema
+from app.core.alipay import AlipayClient
+from app.core.exceptions import CustomException
+from app.core.logger import log
+from app.utils.snowflake import get_snowflake_id
+
+from .model import QuotaModel
+from .enums import QuotaStatusEnum
+from .outsource_schema import OutsourceNotifySchema, OutsourceNotifyOutSchema
+
+
+class OutsourceNotifyService:
+    """外部消费额度同步服务
+
+    调用 alipay.ebpp.invoice.expensecomsue.outsource.notify 将外部消费同步给支付宝，
+    同时写入本地 pay_expense_quota 记录真实额度变动。
+    """
+
+    _alipay_request_cls = None
+    _alipay_model_cls = None
+    _alipay_response_cls = None
+
+    @classmethod
+    def _ensure_imports(cls):
+        """延迟导入支付宝SDK"""
+        if cls._alipay_request_cls is not None:
+            return
+        try:
+            from alipay.aop.api.request.AlipayEbppInvoiceExpensecomsueOutsourceNotifyRequest import (
+                AlipayEbppInvoiceExpensecomsueOutsourceNotifyRequest,
+            )
+            from alipay.aop.api.domain.AlipayEbppInvoiceExpensecomsueOutsourceNotifyModel import (
+                AlipayEbppInvoiceExpensecomsueOutsourceNotifyModel,
+            )
+            from alipay.aop.api.response.AlipayEbppInvoiceExpensecomsueOutsourceNotifyResponse import (
+                AlipayEbppInvoiceExpensecomsueOutsourceNotifyResponse,
+            )
+            cls._alipay_request_cls = AlipayEbppInvoiceExpensecomsueOutsourceNotifyRequest
+            cls._alipay_model_cls = AlipayEbppInvoiceExpensecomsueOutsourceNotifyModel
+            cls._alipay_response_cls = AlipayEbppInvoiceExpensecomsueOutsourceNotifyResponse
+        except ImportError:
+            raise CustomException(msg="支付宝SDK未正确安装(alipay-ebpp-invoice-expensecomsue-outsource-notify)")
+
+    @classmethod
+    def _execute_alipay(cls, request):
+        """同步执行支付宝调用"""
+        client = AlipayClient.get_client()
+        return client.execute(request)
+
+    @classmethod
+    def _build_alipay_model(cls, data: OutsourceNotifySchema) -> object:
+        """构建Alipay SDK Model
+
+        金额转换规则：
+        - schema.amount 单位为元（正数为消费/扣款，负数为退款）
+        - Alipay API 单位为分（正整数），is_off_set 控制方向
+        """
+        cls._ensure_imports()
+        model = cls._alipay_model_cls()
+        model.enterprise_id = data.enterprise_id
+        model.employee_id = data.employee_id
+        model.standard_id = data.standard_id
+
+        # 元→分，取绝对值（正值），is_off_set 控制方向
+        amount_cents = int(abs(data.amount) * 100)
+        model.amount = str(amount_cents)
+
+        # 方向：0=扣款 1=退款
+        model.is_off_set = 1 if data.amount < 0 else 0
+
+        model.out_source_id = data.out_source_id
+        model.deal_time = data.deal_time
+
+        if data.account_id:
+            model.account_id = data.account_id
+        if data.agreement_no:
+            model.agreement_no = data.agreement_no
+        if data.employee_id_type:
+            model.employee_id_type = data.employee_id_type
+        if data.employee_open_id:
+            model.employee_open_id = data.employee_open_id
+        if data.extend:
+            model.extend = data.extend
+        if data.platform:
+            model.platform = data.platform
+        if data.relate_no:
+            model.relate_no = data.relate_no
+
+        return model
+
+    @classmethod
+    async def notify_service(
+        cls, auth: AuthSchema, data: OutsourceNotifySchema
+    ) -> OutsourceNotifyOutSchema:
+        """调用支付宝外部消费额度同步，并写入本地额度记录
+
+        Args:
+            auth: 认证信息
+            data: 同步请求数据
+
+        Returns:
+            同步结果
+        """
+        out_source_id = data.out_source_id or str(get_snowflake_id())
+
+        # 第1步：调用支付宝
+        try:
+            cls._ensure_imports()
+            model = cls._build_alipay_model(data)
+
+            request = cls._alipay_request_cls()
+            request.biz_model = model
+
+            response = await asyncio.to_thread(cls._execute_alipay, request)
+
+            if not response:
+                raise CustomException(msg="外部消费额度同步失败: 无响应")
+
+            result = cls._alipay_response_cls()
+            result.parse_response_content(response)
+
+            if not result.is_success():
+                log.error(f"外部消费额度同步失败: {result.code} - {result.msg} ({result.sub_code}: {result.sub_msg})")
+                # 不抛异常，返回失败信息
+                return OutsourceNotifyOutSchema(
+                    success=False,
+                    out_source_id=out_source_id,
+                    message=f"{result.msg}: {result.sub_msg}" if result.sub_msg else result.msg,
+                )
+
+            log.info(
+                f"外部消费额度同步成功: out_source_id={out_source_id}, "
+                f"employee_id={data.employee_id}, amount={data.amount}"
+            )
+        except CustomException:
+            raise
+        except Exception as e:
+            log.error(f"外部消费额度同步异常: {e}")
+            return OutsourceNotifyOutSchema(
+                success=False,
+                out_source_id=out_source_id,
+                message=str(e),
+            )
+
+        # 第2步：同步写入本地额度记录
+        try:
+            await cls._sync_local_quota(auth, data, out_source_id)
+        except Exception as e:
+            log.warning(f"本地额度记录同步失败（不影响支付宝侧）: {e}")
+
+        return OutsourceNotifyOutSchema(
+            success=True,
+            out_source_id=out_source_id,
+            message="success",
+        )
+
+    @classmethod
+    async def _sync_local_quota(
+        cls, auth: AuthSchema, data: OutsourceNotifySchema, out_source_id: str
+    ):
+        """同步本地额度记录（pay_expense_quota）
+
+        根据支付宝外部消费同步的结果，更新或创建本地额度记录。
+        - 消费：减少可用额度
+        - 退款：增加可用额度
+
+        幂等：检查 out_biz_no 是否已存在，已存在则跳过。
+        """
+        # 幂等检查：out_source_id 已存在则直接跳过
+        existing_by_out = await auth.db.execute(
+            select(QuotaModel).where(QuotaModel.out_biz_no == out_source_id)
+        )
+        if existing_by_out.scalar_one_or_none():
+            log.info(f"外部消费额度同步已处理，跳过本地写入: out_source_id={out_source_id}")
+            return
+
+        institution_id = data.institution_id or ""
+
+        # 查询该员工在该制度下是否已有额度记录
+        stmt = select(QuotaModel).where(
+            QuotaModel.employee_id == data.employee_id,
+            QuotaModel.institution_id == institution_id,
+        )
+        result = await auth.db.execute(stmt)
+        existing = result.scalar_one_or_none()
+
+        amount_decimal = Decimal(str(abs(data.amount)))
+
+        if existing:
+            # 更新现有额度
+            if data.amount > 0:
+                # 消费：减少可用额度
+                new_available = (existing.available_amount or Decimal("0")) - amount_decimal
+                if new_available < 0:
+                    new_available = Decimal("0")
+            else:
+                # 退款：增加可用额度
+                new_available = (existing.available_amount or Decimal("0")) + amount_decimal
+
+            new_status = QuotaStatusEnum.QUOTA_ACTIVE.value
+            if new_available <= 0:
+                new_status = QuotaStatusEnum.QUOTA_EXHAUSTED.value
+
+            upd = (
+                sa_update(QuotaModel)
+                .where(QuotaModel.id == existing.id)
+                .values(
+                    available_amount=new_available,
+                    status=new_status,
+                )
+            )
+            await auth.db.execute(upd)
+            log.info(
+                f"更新本地额度: employee_id={data.employee_id}, "
+                f"institution_id={institution_id}, "
+                f"available_amount={new_available}, status={new_status}"
+            )
+        else:
+            # 新建额度记录（仅记录可用额度，无额度ID）
+            new_available = Decimal("0")
+            if data.amount < 0:
+                # 退款时创建正向记录
+                new_available = amount_decimal
+
+            ins = insert(QuotaModel).values(
+                employee_id=data.employee_id,
+                institution_id=institution_id,
+                out_biz_no=out_source_id,
+                total_amount=new_available,
+                available_amount=new_available,
+                status=QuotaStatusEnum.QUOTA_ACTIVE.value,
+                enterprise_id=data.enterprise_id,
+                tenant_id=auth.user.tenant_id if auth.user else 1,
+            )
+            await auth.db.execute(ins)
+            log.info(
+                f"新建本地额度: employee_id={data.employee_id}, "
+                f"institution_id={institution_id}, "
+                f"available_amount={new_available}"
+            )
+
+        await auth.db.flush()

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

@@ -68,6 +68,12 @@ class BillHandler(BaseHandler[dict]):
             if detail:
                 await self._save_bill_detail(detail, auth)
 
+            # 同步更新本地额度
+            try:
+                await self._sync_expense_quota(data, auth)
+            except Exception as e:
+                log.warning(f"同步额度失败（不影响主流程）: {e}")
+
         # ========== 转账通知（企业码扩展） ==========
         elif data.consume_type == "TRANSFER":
             from app.plugin.module_payment.account.enums import TransferStatusEnum
@@ -118,6 +124,78 @@ class BillHandler(BaseHandler[dict]):
         await bill_crud.create_or_update(data.pay_no, bill_data)
         log.info(f"保存账单基础数据成功: pay_no={data.pay_no}")
 
+    async def _sync_expense_quota(self, data: ConsumeChangeContent, auth: AuthSchema) -> None:
+        """消费/退款时同步更新本地额度（pay_expense_quota）
+
+        通过 expense_rule_group_id → pay_expense_rule.rule_id 找到 institution_id，
+        然后按 employee_id + institution_id 更新对应额度的可用金额。
+        """
+        if not data.expense_rule_group_id:
+            return
+
+        expense_amount = Decimal(data.consume_amount) if data.consume_amount else Decimal("0")
+        if expense_amount <= 0:
+            return
+
+        # 查使用规则 → 找到制度ID
+        from app.plugin.module_payment.expense.rule.model import ExpenseRuleModel
+        from app.plugin.module_payment.expense.quota.model import QuotaModel
+        from app.plugin.module_payment.expense.quota.enums import QuotaStatusEnum
+        from sqlalchemy import select, update as sa_update
+
+        rule_stmt = select(ExpenseRuleModel).where(
+            ExpenseRuleModel.rule_id == data.expense_rule_group_id
+        )
+        rule_result = await auth.db.execute(rule_stmt)
+        rule = rule_result.scalar_one_or_none()
+
+        if not rule:
+            log.debug(f"未找到对应使用规则: rule_id={data.expense_rule_group_id}")
+            return
+
+        institution_id = rule.institution_id
+        if not institution_id:
+            return
+
+        # 查该员工在该制度下的额度
+        quota_stmt = select(QuotaModel).where(
+            QuotaModel.employee_id == data.employee_id,
+            QuotaModel.institution_id == institution_id,
+        )
+        quota_result = await auth.db.execute(quota_stmt)
+        quota = quota_result.scalar_one_or_none()
+
+        if not quota:
+            log.debug(f"未找到额度记录: employee_id={data.employee_id}, institution_id={institution_id}")
+            return
+
+        # 更新可用额度
+        if data.consume_type == "REFUND":
+            # 退款：增加可用额度
+            new_available = (quota.available_amount or Decimal("0")) + expense_amount
+        else:
+            # 消费：减少可用额度
+            new_available = (quota.available_amount or Decimal("0")) - expense_amount
+            if new_available < 0:
+                new_available = Decimal("0")
+
+        new_status = QuotaStatusEnum.QUOTA_EXHAUSTED.value if new_available <= 0 else QuotaStatusEnum.QUOTA_ACTIVE.value
+
+        upd = (
+            sa_update(QuotaModel)
+            .where(QuotaModel.id == quota.id)
+            .values(available_amount=new_available, status=new_status)
+        )
+        await auth.db.execute(upd)
+        await auth.db.flush()
+        log.info(
+            f"消费通知同步额度: employee_id={data.employee_id}, "
+            f"institution_id={institution_id}, "
+            f"consume_type={data.consume_type}, "
+            f"amount={expense_amount}, "
+            f"available_amount={new_available}, status={new_status}"
+        )
+
     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 (
@@ -305,43 +383,30 @@ class VoucherHandler(BaseHandler[dict]):
             return False
 
     async def _process_voucher(self, data: VoucherChangeContent, auth: AuthSchema) -> bool:
-        """处理凭证"""
+        """处理凭证变动通知
+
+        通知中已携带 pay_no 字段，直接用 pay_no 查询账单详情并更新凭证信息。
+        """
         log.info(
-            f"凭证变动: voucher_id={data.voucher_id}, "
-            f"enterprise_id={data.enterprise_id}, "
-            f"action={data.action}"
+            f"凭证变动: pay_no={data.pay_no}, voucher_type={data.voucher_type}, "
+            f"enterprise_id={data.enterprise_id}, notify_reason={data.notify_reason}"
         )
 
-        # 4.1: 调用支付宝查询账单及凭证详情
+        if not data.pay_no:
+            log.warning(f"凭证变动通知缺少 pay_no: voucher_type={data.voucher_type}")
+            return True
+
         try:
-            # 使用 pay_no 或 voucher_id 查询，这里需要确定查询参数
-            # 由于凭证变动通知中没有直接提供 pay_no，我们先尝试通过 voucher_id 查询
-            detail = await self._query_bill_detail_by_voucher(data.voucher_id, auth)
+            detail = await self._query_bill_detail(data.pay_no, data.enterprise_id or "", auth)
         except Exception as e:
-            log.warning(f"查询凭证详情失败: {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)

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

@@ -89,14 +89,29 @@ class EmployeeChangeContent(BaseModel):
 
 
 class VoucherChangeContent(BaseModel):
-    """凭证变动通知内容"""
+    """凭证变动通知内容 (alipay.commerce.ec.voucher.change.notify)"""
 
-    account_id: str = Field(..., description="账户id")
-    voucher_id: str = Field(..., description="凭证id")
-    enterprise_id: str = Field(..., description="企业id")
-    out_biz_no: str = Field(..., description="外部业务号")
-    action: str = Field(..., description="变动动作")
-    change_time: str = Field(..., description="变动时间")
+    # 必选字段
+    account_id: str = Field(..., description="共同账户ID")
+    pay_no: str = Field(..., description="交易流水号（支付宝账单号）")
+    voucher_type: str = Field(..., description="凭证类型: Ticket/Multimedia/Compliance")
+    notify_reason: str = Field(..., description="通知原因: INVOICE_CREATE/INVOICE_INVALID/TICKET_BIND")
+    notify_desc: str = Field(..., description="通知描述")
+
+    # 二选一
+    user_id: str | None = Field(None, description="用户支付宝UID")
+    open_id: str | None = Field(None, description="用户open_id")
+
+    # 可选
+    ext_infos: str | None = Field(None, description="扩展参数，JSON格式")
+    enterprise_id: str | None = Field(None, description="企业ID")
+    employee_id: str | None = Field(None, description="员工ID")
+    voucher_id: str | None = Field(None, description="凭证ID")
+
+    # 本地补充
+    action: str | None = Field(None, description="变动动作")
+    out_biz_no: str | None = Field(None, description="外部业务号")
+    change_time: str | None = Field(None, description="变动时间")
 
 
 class FundChangeContent(BaseModel):