payment-platform/backend/app/plugin/module_payment/notification/handlers/bill_handler.py

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


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)
        except Exception as e:
            log.error(f"解析账单通知内容失败: {e}")
            return False

        try:
            return await self._process_bill(notify_data, auth)
        except Exception as e:
            log.error(f"处理账单变动通知异常: {e}", exc_info=True)
            return False

    async def _process_bill(self, data: ConsumeChangeContent, auth: AuthSchema) -> bool:
        """处理账单变动通知

        支付宝文档: alipay.commerce.ec.consume.change.notify
        consume_type 标准枚举: CONSUME(消费), REFUND(退款)
        notify_reason 标准枚举: COLLECT(归集), SUMMARY_SUCCESS(汇总成功), ORDER_COMPLETE(完结)

        注: TRANSFER(转账) 为支付宝企业码扩展值，不在标准文档枚举中
        """
        log.info(
            f"账单变动: pay_no={data.pay_no}, "
            f"enterprise_id={data.enterprise_id}, "
            f"consume_amount={data.consume_amount}, "
            f"consume_type={data.consume_type}, "
            f"notify_reason={data.notify_reason}"
        )

        # 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 == "CONSUME":
            log.info(f"消费账单通知: pay_no={data.pay_no}, amount={data.consume_amount}, reason={data.notify_reason}")

        elif data.consume_type == "REFUND":
            log.info(
                f"退款账单通知: pay_no={data.pay_no}, "
                f"related_pay_no={data.related_pay_no}, "
                f"amount={data.consume_amount}"
            )

        # ========== 转账通知（企业码扩展） ==========
        elif data.consume_type == "TRANSFER":
            from app.plugin.module_payment.account.enums import TransferStatusEnum

            reason = (data.notify_reason or "").upper()
            if "SUCCESS" in reason:
                status = TransferStatusEnum.SUCCESS.value
            elif "FAIL" in reason:
                status = TransferStatusEnum.FAIL.value
            else:
                status = TransferStatusEnum.DEALING.value
                log.warning(f"转账状态无法判断: pay_no={data.pay_no}, notify_reason={data.notify_reason}")

            await AccountService.update_transfer_status_service(
                auth, data.pay_no, status, data.model_dump(exclude_none=True)
            )

            # 无论成功/失败都回调商户，告知最终结果
            await OpenTransferService.open_return_service(auth, data.pay_no)

        else:
            log.info(f"未知账单类型: consume_type={data.consume_type}, pay_no={data.pay_no}")

        return True

    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]):
    """凭证变动通知处理器"""

    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)
        except Exception as e:
            log.error(f"解析凭证通知内容失败: {e}")
            return False

        action = notify_data.action
        voucher_id = notify_data.voucher_id

        log.info(f"处理凭证变动通知: action={action}, voucher_id={voucher_id}")

        try:
            return await self._process_voucher(notify_data, auth)
        except Exception as e:
            log.error(f"处理凭证变动通知异常: {e}", exc_info=True)
            return False

    async def _process_voucher(self, data: VoucherChangeContent, auth: AuthSchema) -> bool:
        """处理凭证"""
        log.info(
            f"凭证变动: voucher_id={data.voucher_id}, "
            f"enterprise_id={data.enterprise_id}, "
            f"action={data.action}"
        )

        # 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)}")