payment-platform/backend/app/plugin/module_payment/account/service.py

from datetime import datetime, timedelta
from decimal import Decimal
from typing import Any, Optional

from redis.asyncio import Redis

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_str
from app.plugin.module_payment.enterprise.crud import EnterpriseCRUD

from .crud import AccountCRUD, TransferCRUD, DepositCRUD, WithdrawCRUD
from .enums import (
    DepositStatusEnum,
    WithdrawStatusEnum,
    TransferStatusEnum,
)
from .schema import (
    AccountAuthorizeApplySchema,
    AccountAuthorizeApplyOutSchema,
    AccountCreateSchema,
    AccountDepositSchema,
    AccountDepositOutSchema,
    AccountOperationOutSchema,
    AccountQuerySchema,
    AccountTransferSchema,
    AccountTransferOutSchema,
    AccountWithdrawSchema,
    ReceiptApplySchema,
    TransferListOutSchema,
    TransferOutSchema,
    TenantTransferCreate,
    TenantTransferResponse,
)
from ..openapi.crud import OpenTransferCRUD

# 支付宝资金专户转账错误码 → 友好提示
_TRANSFER_ERROR_HINTS = {
    "SYSTEM_ERROR": "系统繁忙，请稍后重试",
    "INVALID_PARAMETER": "请求参数有误，请检查后重试",
    "AMOUNT_LESS_THAN_ONE_CENT": "转账金额不能低于 0.01 元",
    "BALANCE_IS_NOT_ENOUGH": "企业余额不足，建议充值",
    "BANK_RESPONSE_ERROR": "银行处理失败：账户异常",
    "CARD_BIN_ERROR": "收款银行账号不正确，请确认",
    "DUPLICATE_DIFFERENT_REQUEST": "重复请求但参数不一致，请检查",
    "EXCEED_LIMIT_SM_MIN_AMOUNT": "转账金额不能低于 0.1 元",
    "EXCEED_LIMIT_DM_MAX_AMOUNT": "超出单日转账限额，请明天再试或联系管理员提升限额",
    "INVALID_ACCOUNT_BOOK": "资金专户不存在，请检查专户号",
    "INVALID_CARDNO": "无效的收款银行卡号",
    "INVALID_IDENTITY_TYPE": "收款方身份类型不匹配",
    "NO_AGREEMENT": "无转账权限，请联系管理员",
    "PAYEE_CARD_INFO_ERROR": "收款方账号或银行卡信息有误，请核实",
    "PAYEE_NOT_EXIST": "收款账号不存在或姓名有误",
    "PAYER_BALANCE_NOT_ENOUGH": "付款方余额不足，建议充值",
    "REQUEST_PROCESSING": "系统处理中，请稍后重试",
    "TRANS_AUTH_NO_EXIST": "转账授权协议不存在，请先签约",
}


def _parse_dt(val: str | None) -> datetime | None:
    """解析支付宝日期字符串"""
    if not val:
        return None
    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
        try:
            return datetime.strptime(val, fmt)
        except ValueError:
            continue
    return None


class AccountService:
    """资金专户服务层"""

    @classmethod
    async def stat_transfer_amount_service(
        cls,
        auth: AuthSchema,
        tenant_id: Optional[int] = None,
        enterprise_id: Optional[str] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        payee_type: Optional[str] = None,
    ) -> Decimal:
        """
        统计转账金额（✅）

        统计企业在指定时间范围内的转账总金额以及每天的转账金额。
        """
        crud = TransferCRUD(auth)
        
        return await crud.get_transfer_amount(
            tenant_id=tenant_id,
            enterprise_id=enterprise_id,
            start_date=start_date,
            end_date=end_date,
            payee_type=payee_type,
        )

    @classmethod
    async def stat_consume_amount_service(
        cls,
        auth: AuthSchema,
        tenant_id: Optional[int] = None,
        enterprise_id: Optional[str] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
    ) -> Decimal:
        """
        统计消费金额（✅）

        统计企业在指定时间范围内的消费总金额。
        数据来源: pay_bill 表，consume_type=CONSUME, status=PROCESSED
        """
        from app.plugin.module_payment.notification.crud import BillCRUD

        crud = BillCRUD(auth)
        return await crud.get_consume_amount(
            tenant_id=tenant_id,
            enterprise_id=enterprise_id,
            start_date=start_date,
            end_date=end_date,
        )

    @classmethod
    async def stat_summary_amount_service(
        cls,
        auth: AuthSchema,
        tenant_id: Optional[int] = None,
        enterprise_id: Optional[str] = None,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        payee_type: Optional[str] = None,
    ) -> Decimal:
        """
        汇总统计金额（✅）

        汇总 = 消费统计 + 转账统计，对应时间段结果相加
        payee_type 仅过滤转账部分
        """
        transfer_amount = await cls.stat_transfer_amount_service(
            auth=auth,
            tenant_id=tenant_id,
            enterprise_id=enterprise_id,
            start_date=start_date,
            end_date=end_date,
            payee_type=payee_type,
        )
        consume_amount = await cls.stat_consume_amount_service(
            auth=auth,
            tenant_id=tenant_id,
            enterprise_id=enterprise_id,
            start_date=start_date,
            end_date=end_date,
        )
        return transfer_amount + consume_amount

    @classmethod
    async def authorize_apply_service(
        cls,
        auth: AuthSchema,
        data: AccountAuthorizeApplySchema
    ) -> AccountAuthorizeApplyOutSchema:
        """
        申请转账授权签约（✅）

        调用: alipay.commerce.ec.trans.authorize.apply
        """
        from alipay.aop.api.request.AlipayCommerceEcTransAuthorizeApplyRequest import (
            AlipayCommerceEcTransAuthorizeApplyRequest,
        )
        from alipay.aop.api.domain.AlipayCommerceEcTransAuthorizeApplyModel import (
            AlipayCommerceEcTransAuthorizeApplyModel,
        )
        from alipay.aop.api.response.AlipayCommerceEcTransAuthorizeApplyResponse import (
            AlipayCommerceEcTransAuthorizeApplyResponse,
        )

        model = AlipayCommerceEcTransAuthorizeApplyModel()
        model.enterprise_id = data.enterprise_id

        request = AlipayCommerceEcTransAuthorizeApplyRequest()
        request.biz_model = model

        client = AlipayClient.get_client()
        response = client.execute(request)

        if not response:
            raise CustomException(msg="申请转账授权失败: 无响应")

        result = AlipayCommerceEcTransAuthorizeApplyResponse()
        result.parse_response_content(response)

        if not result.is_success():
            log.error(f"支付宝接口调用失败: {result.code} - {result.msg}")
            raise CustomException(msg=f"申请转账授权失败: {result.msg}")

        return AccountAuthorizeApplyOutSchema(
            sign_url=result.sign_url,
        )

    @classmethod
    async def create_account_service(
        cls,
        auth: AuthSchema,
        data: AccountCreateSchema
    ) -> AccountOperationOutSchema:
        """
        开通资金专户（✅）

        调用: alipay.commerce.ec.trans.account.create
        """
        from alipay.aop.api.request.AlipayCommerceEcTransAccountCreateRequest import (
            AlipayCommerceEcTransAccountCreateRequest,
        )
        from alipay.aop.api.domain.AlipayCommerceEcTransAccountCreateModel import (
            AlipayCommerceEcTransAccountCreateModel,
        )
        from alipay.aop.api.response.AlipayCommerceEcTransAccountCreateResponse import (
            AlipayCommerceEcTransAccountCreateResponse,
        )

        model = AlipayCommerceEcTransAccountCreateModel()
        model.enterprise_id = data.enterprise_id
        # model.account_type = data.account_type or "ALL"  # 收支全能户
        # model.scene = data.scene or "B2B_TRANS"  # ToB转账场景
        model.account_type = "ALL"
        model.scene = "B2B_TRANS"

        request = AlipayCommerceEcTransAccountCreateRequest()
        request.biz_model = model

        client = AlipayClient.get_client()
        response = client.execute(request)

        if not response:
            raise CustomException(msg="开通资金专户失败: 无响应")

        result = AlipayCommerceEcTransAccountCreateResponse()
        result.parse_response_content(response)

        if not result.is_success():
            log.error(f"支付宝接口调用失败: {result.code} - {result.msg}")
            raise CustomException(msg=f"开通资金专户失败: {result.msg}")

        account_data = AccountCreateSchema(
            enterprise_id=model.enterprise_id,
            account_book_id=result.account_book_id,
            account_type=model.account_type,
            scene=model.scene,
        )

        if result.account_book_id:
            account_data.account_book_id = result.account_book_id

        await AccountCRUD(auth).create(account_data)

        return AccountOperationOutSchema(
            enterprise_id=account_data.enterprise_id,
            account_book_id=account_data.account_book_id,
        )

    @classmethod
    async def deposit_service(
        cls,
        auth: AuthSchema,
        data: AccountDepositSchema
    ) -> AccountDepositOutSchema:
        """
        资金专户充值（✅）

        调用: alipay.commerce.ec.trans.account.deposit
        """
        from alipay.aop.api.request.AlipayCommerceEcTransAccountDepositRequest import (
            AlipayCommerceEcTransAccountDepositRequest,
        )
        from alipay.aop.api.domain.AlipayCommerceEcTransAccountDepositModel import (
            AlipayCommerceEcTransAccountDepositModel,
        )
        from alipay.aop.api.response.AlipayCommerceEcTransAccountDepositResponse import (
            AlipayCommerceEcTransAccountDepositResponse,
        )

        model = AlipayCommerceEcTransAccountDepositModel()
        model.enterprise_id = data.enterprise_id
        model.account_book_id = data.account_book_id
        model.amount = str(data.amount)
        model.out_biz_no = get_snowflake_id_str(auth.tenant_id)

        request = AlipayCommerceEcTransAccountDepositRequest()
        request.biz_model = model

        client = AlipayClient.get_client()
        response = client.execute(request)

        if not response:
            raise CustomException(msg="充值失败: 无响应")

        result = AlipayCommerceEcTransAccountDepositResponse()
        result.parse_response_content(response)

        if not result.is_success():
            log.error(f"支付宝接口调用失败: {result.code} - {result.msg}")
            raise CustomException(msg=f"充值失败: {result.msg}")

        deposit_crud = DepositCRUD(auth)
        deposit_data = {
            "enterprise_id": data.enterprise_id,
            "out_biz_no": model.out_biz_no,
            "account_book_id": data.account_book_id,
            "amount": data.amount,
            "url": result.url,
            "status": DepositStatusEnum.DEALING.value,
            "remark": data.remark,
        }
        await deposit_crud.create(deposit_data)

        return AccountDepositOutSchema(
            url=result.url,
        )

    @classmethod
    async def transfer_service(
        cls,
        auth: AuthSchema,
        data: AccountTransferSchema
    ) -> AccountTransferOutSchema:
        """
        资金专户转账（✅）

        调用: alipay.commerce.ec.trans.account.transfer
        """
        from alipay.aop.api.request.AlipayCommerceEcTransAccountTransferRequest import (
            AlipayCommerceEcTransAccountTransferRequest,
        )
        from alipay.aop.api.domain.AlipayCommerceEcTransAccountTransferModel import (
            AlipayCommerceEcTransAccountTransferModel,
        )
        from alipay.aop.api.response.AlipayCommerceEcTransAccountTransferResponse import (
            AlipayCommerceEcTransAccountTransferResponse,
        )
        from alipay.aop.api.domain.TransParticipant import (
            TransParticipant,
        )
        from alipay.aop.api.domain.BankCardExtInfoDTO import (
            BankCardExtInfoDTO,
        )
        
        # 检查资金专户是否存在
        account = await AccountCRUD(auth).get_by_account_book_id(data.account_book_id)
        if not account:
            raise CustomException(msg="资金账户不存在")
        if account.tenant_id != auth.tenant_id:
            raise CustomException(msg="无权限操作")
        if data.enterprise_id and account.enterprise_id != data.enterprise_id:
            raise CustomException(msg="参数错误")
        
        if not data.order_title and account.enterprise_id:
            enterprise = await EnterpriseCRUD(auth).get_by_enterprise_id(account.enterprise_id)
            if not enterprise:
                raise CustomException(msg="资金账户所属企业不存在")
            data.order_title = f"来自{enterprise.name}转账"

        model = AlipayCommerceEcTransAccountTransferModel()
        model.enterprise_id = account.enterprise_id
        model.account_book_id = account.account_book_id
        model.out_biz_no = get_snowflake_id_str(auth.tenant_id)
        # 转账总金额，单位为元，精确到小数点后两位
        model.amount = str(data.amount)
        model.order_title = data.order_title

        payee_info = TransParticipant()
        payee_info.identity_type = data.payee_info.identity_type
        payee_info.name = data.payee_info.name
        payee_info.identity = data.payee_info.identity
        if data.payee_info.bankcard_ext_info:
            payee_info.bankcard_ext_info = BankCardExtInfoDTO.from_alipay_dict(
                    data.payee_info.bankcard_ext_info.model_dump(exclude_none=True)
            )

        model.payee_info = payee_info

        request = AlipayCommerceEcTransAccountTransferRequest()
        request.biz_model = model

        client = AlipayClient.get_client()
        response = client.execute(request)

        if not response:
            raise CustomException(msg="转账失败: 无响应")

        result = AlipayCommerceEcTransAccountTransferResponse()
        result.parse_response_content(response)
        sub_code = getattr(result, 'sub_code', '') or ''
        sub_msg = getattr(result, 'sub_msg', '') or ''
        # 构建转账记录数据，但延迟写入：
        # - 成功时在当前会话写入
        # - 失败时使用独立事务写入并提交，避免被外层回滚吞掉
        transfer_crud = TransferCRUD(auth)
        transfer_data = {
            "enterprise_id": model.enterprise_id,
            "out_biz_no": model.out_biz_no,
            "account_book_id": model.account_book_id,
            "amount": model.amount,
            "order_title": model.order_title,
            "payee_info": data.payee_info.model_dump() if data.payee_info else None,
            "payee_type": data.payee_info.identity_type if data.payee_info else None,
            "status": result.status,
            "order_no": result.order_no,
            "fund_order_id": result.fund_order_id,
            "remark": "",
        }

        if result.status == TransferStatusEnum.DEALING.value:
            transfer_data["retry_count"] = 0
            transfer_data["next_retry_at"] = datetime.now() + timedelta(minutes=5)

        log.info(f"记录转账: {transfer_data}")

        if not result.is_success():
            # 优先用 sub_code 匹配
            hint = _TRANSFER_ERROR_HINTS.get(sub_code)
            # sub_code 无匹配时，尝试从 sub_msg 中提取错误码（支付宝部分接口sub_code返回unknown-sub-code）
            if not hint:
                for code_key, code_hint in _TRANSFER_ERROR_HINTS.items():
                    if code_key in sub_msg:
                        hint = code_hint
                        break
            hint = hint or sub_msg or result.msg or "转账失败"

            log.error(f"支付宝接口调用失败: {result.code} - {result.msg} (sub_code={sub_code}, sub_msg={sub_msg})")

            # 使用独立的 session/事务保证失败记录能被持久化
            from app.core.database import async_db_session

            async with async_db_session() as _session:
                async with _session.begin():
                    new_auth = AuthSchema(db=_session, check_data_scope=False)
                    # 保持 tenant_id
                    new_auth.tenant_id = getattr(auth, "tenant_id", None)
                    transfer_data["status"]="FAIL"
                    transfer_data["remark"]=f"{result.msg} ({sub_code} {sub_msg})"
                    await TransferCRUD(new_auth).create(transfer_data)

            raise CustomException(msg=f"转账失败: {hint}")

        # 成功时写入当前会话
        await transfer_crud.create(transfer_data)

        return AccountTransferOutSchema(
            status=result.status,
            order_no=result.order_no,
            fund_order_id=result.fund_order_id,
            out_biz_no=model.out_biz_no,
        )

    @classmethod
    async def tenant_transfer_service(
        cls,
        auth: AuthSchema,
        tenant_id: int,
        data: TenantTransferCreate,
        request_ip: str,
        api_key_id: int | None = None,
    ) -> TenantTransferResponse:
        """
        租户API转账（通过API Key认证）

        调用: alipay.commerce.ec.trans.account.transfer
        """
        from alipay.aop.api.request.AlipayCommerceEcTransAccountTransferRequest import (
            AlipayCommerceEcTransAccountTransferRequest,
        )
        from alipay.aop.api.domain.AlipayCommerceEcTransAccountTransferModel import (
            AlipayCommerceEcTransAccountTransferModel,
        )
        from alipay.aop.api.response.AlipayCommerceEcTransAccountTransferResponse import (
            AlipayCommerceEcTransAccountTransferResponse,
        )
        from alipay.aop.api.domain.TransParticipant import (
            TransParticipant,
        )
        from alipay.aop.api.domain.BankCardExtInfoDTO import (
            BankCardExtInfoDTO,
        )
        
        # 检查资金专户是否存在
        account = await AccountCRUD(auth).get_by_account_book_id(data.account_book_id)
        if not account:
            raise CustomException(msg="资金账户不存在")
        if account.tenant_id != tenant_id:
            raise CustomException(msg="无权限操作")
        if data.enterprise_id and account.enterprise_id != data.enterprise_id:
            raise CustomException(msg="参数错误")
        
        if not data.order_title and account.enterprise_id:
            enterprise = await EnterpriseCRUD(auth).get_by_enterprise_id(account.enterprise_id)
            if not enterprise:
                raise CustomException(msg="资金账户所属企业不存在")
            data.order_title = f"来自{enterprise.name}转账"

        model = AlipayCommerceEcTransAccountTransferModel()
        model.enterprise_id = account.enterprise_id
        model.account_book_id = account.account_book_id
        model.out_biz_no = get_snowflake_id_str(tenant_id)
        # 转账总金额，单位为元，精确到小数点后两位
        model.amount = str(data.amount)
        model.order_title = data.order_title

        payee_info = TransParticipant()
        payee_info.identity_type = data.payee_info.identity_type
        payee_info.name = data.payee_info.name
        payee_info.identity = data.payee_info.identity
        if data.payee_info.bankcard_ext_info:
            payee_info.bankcard_ext_info = BankCardExtInfoDTO.from_alipay_dict(
                    data.payee_info.bankcard_ext_info.model_dump(exclude_none=True)
            )

        model.payee_info = payee_info

        request = AlipayCommerceEcTransAccountTransferRequest()
        request.biz_model = model

        client = AlipayClient.get_client()
        response = client.execute(request)

        if not response:
            raise CustomException(msg="转账失败: 无响应")

        result = AlipayCommerceEcTransAccountTransferResponse()
        result.parse_response_content(response)

        if not result.is_success():
            log.error(f"支付宝接口调用失败: {result.code} - {result.msg}")
            raise CustomException(msg=f"转账失败: {result.sub_msg or result.msg or result.code}")

        transfer_crud = TransferCRUD(auth)
        transfer_data = {
            "enterprise_id": model.enterprise_id,
            "out_biz_no": model.out_biz_no,
            "account_book_id": model.account_book_id,
            "amount": model.amount,
            "order_title": model.order_title,
            "payee_info": data.payee_info.model_dump() if data.payee_info else None,
            "payee_type": data.payee_info.identity_type if data.payee_info else None,
            "status": result.status,
            "order_no": result.order_no,
            "fund_order_id": result.fund_order_id,
        }

        if result.status == TransferStatusEnum.DEALING.value:
            transfer_data["retry_count"] = 0
            transfer_data["next_retry_at"] = datetime.now() + timedelta(minutes=5)

        await transfer_crud.create(transfer_data)

        return TenantTransferResponse(
            status=result.status,
            order_no=result.order_no,
            fund_order_id=result.fund_order_id,
        )

    @classmethod
    async def withdraw_service(
        cls,
        auth: AuthSchema,
        data: AccountWithdrawSchema
    ) -> AccountOperationOutSchema:
        """
        资金专户提现

        调用: alipay.commerce.ec.trans.account.withdraw

        接口文档: https://opendocs.alipay.com/pre-open/d651859b_alipay.commerce.ec.trans.account.withdraw

        参数说明:
        - enterprise_id: 企业ID
        - account_book_id: 资金专户号
        - amount: 提现金额
        - out_biz_no: 商家侧订单号（唯一）
        """
        from alipay.aop.api.request.AlipayCommerceEcTransAccountWithdrawRequest import (
            AlipayCommerceEcTransAccountWithdrawRequest,
        )
        from alipay.aop.api.domain.AlipayCommerceEcTransAccountWithdrawModel import (
            AlipayCommerceEcTransAccountWithdrawModel,
        )
        from alipay.aop.api.response.AlipayCommerceEcTransAccountWithdrawResponse import (
            AlipayCommerceEcTransAccountWithdrawResponse,
        )

        crud = AccountCRUD(auth)
        enterprise = await crud.get_by_enterprise_id(data.enterprise_id)

        if not enterprise:
            raise CustomException(msg="企业不存在")

        model = AlipayCommerceEcTransAccountWithdrawModel()
        model.enterprise_id = enterprise.enterprise_id
        model.account_book_id = data.account_book_id
        model.amount = str(data.amount)
        model.out_biz_no = get_snowflake_id_str(auth.tenant_id)
        if data.remark:
            model.remark = data.remark

        request = AlipayCommerceEcTransAccountWithdrawRequest()
        request.biz_model = model

        client = AlipayClient.get_client()
        response = client.execute(request)

        if not response:
            raise CustomException(msg="提现失败: 无响应")

        result = AlipayCommerceEcTransAccountWithdrawResponse()
        result.parse_response_content(response)

        if not result.is_success():
            log.error(f"支付宝接口调用失败: {result.code} - {result.msg}")
            raise CustomException(msg=f"提现失败: {result.msg}")

        withdraw_crud = WithdrawCRUD(auth)
        withdraw_data = {
            "enterprise_id": data.enterprise_id,
            "out_biz_no": model.out_biz_no,
            "account_book_id": data.account_book_id,
            "amount": data.amount,
            # 专户提现到余额户是同步操作，要么执行成功，要么执行异常，
            # 出参status设计多余，遵循规范使用业务码区分成功与失败
            "status": WithdrawStatusEnum.SUCCESS.value,
            "order_no": result.order_no,
        }
        await withdraw_crud.create(withdraw_data)

        log.info(f"资金专户提现发起成功: 企业: {data.enterprise_id}, 金额: {data.amount}")

        return AccountOperationOutSchema(
            enterprise_id=data.enterprise_id,
            account_book_id=data.account_book_id,
        )

    @classmethod
    async def query_account_service(
        cls,
        auth: AuthSchema,
        data: AccountQuerySchema
    ) -> list[Any]:
        """
        查询资金专户（调用支付宝接口）

        调用: alipay.commerce.ec.trans.account.query
        """
        from alipay.aop.api.request.AlipayCommerceEcTransAccountQueryRequest import (
            AlipayCommerceEcTransAccountQueryRequest,
        )
        from alipay.aop.api.domain.AlipayCommerceEcTransAccountQueryModel import (
            AlipayCommerceEcTransAccountQueryModel,
        )
        from alipay.aop.api.response.AlipayCommerceEcTransAccountQueryResponse import (
            AlipayCommerceEcTransAccountQueryResponse,
        )
        from alipay.aop.api.domain.FundAccountApiDTO import (
            FundAccountApiDTO,
        )

        model = AlipayCommerceEcTransAccountQueryModel()
        model.enterprise_id = data.enterprise_id

        request = AlipayCommerceEcTransAccountQueryRequest()
        request.biz_model = model

        client = AlipayClient.get_client()
        response = client.execute(request)

        if not response:
            raise CustomException(msg="查询资金专户失败: 无响应")

        result = AlipayCommerceEcTransAccountQueryResponse()
        result.parse_response_content(response)

        if not result.is_success():
            log.error(f"支付宝接口调用失败: {result.code} - {result.msg}")
            raise CustomException(msg=f"查询资金专户失败: {result.msg}")

        collect = []
        for v in list(result.account_list or []):
            if not hasattr(v, "account_book_id"):
                continue
            if not hasattr(v, "scene") or v.scene != "B2B_TRANS":
                continue

            account = FundAccountApiDTO.to_alipay_dict(v)
            collect.append(account)
        
        return collect

    @classmethod
    async def transfer_detail_service(
        cls,
        auth: AuthSchema,
        out_biz_no: str
    ) -> TransferOutSchema:
        """
        查询转账记录详情
        """
        crud = TransferCRUD(auth)
        transfer = await crud.get_by_out_biz_no(out_biz_no)

        if not transfer:
            raise CustomException(msg="转账记录不存在")

        transfer_result = TransferOutSchema.model_validate(transfer)

        # 查询三方订单号
        open_transfer_crud = OpenTransferCRUD(auth)
        open_transfer_data = await open_transfer_crud.get(out_biz_no=transfer.out_biz_no)

        if open_transfer_data:
            transfer_result.third_biz_no = open_transfer_data.third_biz_no

        return transfer_result

    @classmethod
    async def transfer_list_service(
        cls,
        auth: AuthSchema,
        page_no: int = 1,
        page_size: int = 20,
        search: dict | None = None,
    ) -> dict:
        """
        查询转账记录列表
        """
        log.info(f"查询转账记录列表: {page_no}, {page_size}, {search}")
        crud = TransferCRUD(auth)
        offset = (page_no - 1) * page_size
        return await crud.page(
            offset=offset,
            limit=page_size,
            order_by=[{"id": "desc"}],
            search=search or {},
            out_schema=TransferListOutSchema,
        )

    @classmethod
    async def transfer_export_service(
        cls,
        auth: AuthSchema,
        start_time: str,
        end_time: str,
        enterprise_id: Optional[str] = None,
    ) -> bytes:
        """
        导出转账记录报表为Excel文件
        """
        log.info(f"导出转账记录报表: {start_time} -> {end_time}")
        
        crud = TransferCRUD(auth)
        
        search = {
            "created_time__gte": start_time,
            "created_time__lte": end_time,
        }
        
        if enterprise_id:
            search["enterprise_id"] = enterprise_id
        
        records = await crud.list(
            search=search,
            order_by=[{"id": "desc"}],
        )
        
        from app.utils.excel_util import ExcelUtil
        
        status_map = {
            "DEALING": "处理中",
            "SUCCESS": "成功",
            "FAIL": "失败",
            "REFUND": "退票",
        }
        
        payee_type_map = {
            "ALIPAY_ACCOUNT": "支付宝账户",
            "BANK_CARD": "银行卡",
        }
        
        list_data = []
        for i, record in enumerate(records, start=1):
            payee_info = record.payee_info or {}
            list_data.append({
                "序号": i,
                "订单号": record.out_biz_no or "",
                "商户订单号": record.order_no or "",
                "金额(元)": str(record.amount or 0),
                "收款方姓名": payee_info.get("name", ""),
                "收款方类型": payee_type_map.get(payee_info.get("identity_type", ""), ""),
                "状态": status_map.get(record.status, record.status),
                "转账标题": record.order_title or "",
                "创建时间": record.created_time.strftime("%Y-%m-%d %H:%M:%S") if record.created_time else "",
            })
        
        mapping_dict = {
            "序号": "序号",
            "订单号": "订单号",
            "商户订单号": "商户订单号",
            "金额(元)": "金额(元)",
            "收款方姓名": "收款方姓名",
            "收款方类型": "收款方类型",
            "状态": "状态",
            "转账标题": "转账标题",
            "创建时间": "创建时间",
        }
        
        return ExcelUtil.export_list2excel(list_data, mapping_dict)


    @classmethod
    async def apply_receipt_service(
        cls,
        auth: AuthSchema,
        redis: Redis,
        data: ReceiptApplySchema,
    ) -> str:
        """
        申请转账业务回单
        
        调用: alipay.commerce.ec.trans.receipt.apply
        
        参数:
        - enterprise_id: 企业ID
        - order_no: 支付宝转账单号
        
        返回: file_id
        """
        from app.core.redis_crud import RedisCURD
        
        redis_crud = RedisCURD(redis)
        
        cache_key = f"receipt:{data.enterprise_id}:{data.order_no}"
        cached_file_id = await redis_crud.get(cache_key)
        
        if cached_file_id:
            log.info(f"使用缓存的 file_id: {cached_file_id}")
            return cached_file_id
        
        crud = EnterpriseCRUD(auth)
        enterprise = await crud.get_by_enterprise_id(data.enterprise_id)
        if not enterprise:
            raise CustomException(msg="企业不存在")
        
        from alipay.aop.api.request.AlipayCommerceEcTransReceiptApplyRequest import (
            AlipayCommerceEcTransReceiptApplyRequest,
        )
        from alipay.aop.api.domain.AlipayCommerceEcTransReceiptApplyModel import (
            AlipayCommerceEcTransReceiptApplyModel,
        )
        from alipay.aop.api.response.AlipayCommerceEcTransReceiptApplyResponse import (
            AlipayCommerceEcTransReceiptApplyResponse,
        )
        
        model = AlipayCommerceEcTransReceiptApplyModel()
        model.enterprise_id = data.enterprise_id
        model.order_no = data.order_no
        
        request = AlipayCommerceEcTransReceiptApplyRequest()
        request.biz_model = model
        
        client = AlipayClient.get_client()
        response = client.execute(request)
        
        if not response:
            raise CustomException(msg="申请回单失败: 无响应")
        
        result = AlipayCommerceEcTransReceiptApplyResponse()
        result.parse_response_content(response)
        
        if not result.is_success():
            # 清除缓存
            await redis_crud.delete(cache_key)
            raise CustomException(msg=f"申请回单失败: {result.msg}")
        
        file_id = str(result.file_id)
        await redis_crud.set(cache_key, file_id, expire=172800)
        
        log.info(f"申请回单成功: order_no={data.order_no}, file_id={file_id}")
        return file_id


    @classmethod
    async def query_receipt_service(cls, enterprise_id: str, file_id: str) -> dict:
        """
        查询回单状态
        
        调用: alipay.commerce.ec.trans.receipt.query
        
        参数:
        - file_id: 文件申请号
        
        返回: {file_id, status, download_url, error_message}
        """   
        from alipay.aop.api.request.AlipayCommerceEcTransReceiptQueryRequest import (
            AlipayCommerceEcTransReceiptQueryRequest,
        )
        from alipay.aop.api.response.AlipayCommerceEcTransReceiptQueryResponse import (
            AlipayCommerceEcTransReceiptQueryResponse,
        )
        from alipay.aop.api.domain.AlipayCommerceEcTransReceiptQueryModel import (
            AlipayCommerceEcTransReceiptQueryModel,
        )
        
        model = AlipayCommerceEcTransReceiptQueryModel()
        model.enterprise_id = enterprise_id
        model.file_id = file_id
        
        request = AlipayCommerceEcTransReceiptQueryRequest()
        request.biz_model = model

        client = AlipayClient.get_client()
        response = client.execute(request)
        
        if not response:
            raise CustomException(msg="查询回单失败: 无响应")
        
        result = AlipayCommerceEcTransReceiptQueryResponse()
        result.parse_response_content(response)
        
        if not result.is_success():
            raise CustomException(msg=f"查询回单失败: {result.msg}")
        
        data = {
            "file_id": file_id,
            "status": result.status,
            "download_url": result.download_url,
            "error_message": result.error_message,
        }
        
        return data


    @classmethod
    async def transfer_sync_status_service(
        cls,
        auth: AuthSchema,
        data: "TransferSyncStatusSchema",
    ) -> dict:
        """
        手动同步转账状态（管理员补录）
        用于修复因通知丢失而卡在 DEALING 的转账记录
        """
        from app.plugin.module_payment.account.crud import TransferCRUD
        from app.plugin.module_payment.account.schema import TransferSyncStatusSchema

        crud = TransferCRUD(auth)
        transfer = await crud.get_by_out_biz_no(data.out_biz_no)
        if not transfer:
            raise CustomException(msg=f"转账记录不存在: {data.out_biz_no}")

        if transfer.status != "DEALING" and data.status == "SUCCESS":
            raise CustomException(msg=f"转账记录当前状态为 {transfer.status}，无需同步")

        update_data = {"status": data.status}
        if data.error_code:
            update_data["error_code"] = data.error_code
        if data.error_msg:
            update_data["error_msg"] = data.error_msg

        for key, value in update_data.items():
            if hasattr(transfer, key):
                setattr(transfer, key, value)

        await auth.db.flush()
        await auth.db.refresh(transfer)

        log.info(f"手动同步转账状态成功: out_biz_no={data.out_biz_no}, {transfer.status}")
        return {
            "out_biz_no": transfer.out_biz_no,
            "status": transfer.status,
            "error_code": transfer.error_code,
            "error_msg": transfer.error_msg,
        }


    @classmethod
    async def transfer_sync_all_service(
        cls,
        auth: AuthSchema,
    ) -> dict:
        """
        全量同步转账状态（使用 trans.order.query）
        """
        from sqlalchemy import select
        from app.plugin.module_payment.account.model import TransferModel
        from app.plugin.module_payment.account.enums import TransferStatusEnum

        stmt = select(TransferModel).where(
            TransferModel.status == TransferStatusEnum.DEALING.value,
        ).order_by(TransferModel.id.asc())
        result = await auth.db.execute(stmt)
        all_transfers = list(result.scalars().all())

        synced = 0
        errors = 0
        details = []

        for transfer in all_transfers:
            out_biz_no = transfer.out_biz_no
            eid = transfer.enterprise_id
            order_no = transfer.order_no or ""
            if not out_biz_no or not eid:
                continue
            try:
                new_status = await cls._sync_transfer_detail(auth, out_biz_no, eid, order_no)
                if new_status:
                    synced += 1
                    details.append({"out_biz_no": out_biz_no, "old_status": transfer.status, "new_status": new_status})
                else:
                    details.append({"out_biz_no": out_biz_no, "status": transfer.status, "action": "no_change"})
            except Exception as e:
                errors += 1
                details.append({"out_biz_no": out_biz_no, "status": transfer.status, "error": str(e)})
                log.warning(f"全量同步 - 查询失败: out_biz_no={out_biz_no}, err={e}")

        if synced > 0:
            await auth.db.flush()

        return {
            "total": len(all_transfers),
            "synced": synced,
            "errors": errors,
            "details": details,
        }

    @classmethod
    async def _sync_transfer_detail(
        cls,
        auth: AuthSchema,
        out_biz_no: str,
        enterprise_id: str,
        order_no: str = "",
    ) -> str | None:
        """调 alipay.commerce.ec.trans.order.query 查询单笔转账状态并更新
        有 out_biz_no 传 out_biz_no，有 order_no 也传 order_no（两个都传）
        返回: 新状态str / None(不可查/无变化)
        """
        from sqlalchemy import update as sa_update
        from app.plugin.module_payment.account.model import TransferModel
        from app.core.alipay import AlipayClient

        try:
            from alipay.aop.api.request.AlipayCommerceEcTransOrderQueryRequest import (
                AlipayCommerceEcTransOrderQueryRequest,
            )
            from alipay.aop.api.domain.AlipayCommerceEcTransOrderQueryModel import (
                AlipayCommerceEcTransOrderQueryModel,
            )
            from alipay.aop.api.response.AlipayCommerceEcTransOrderQueryResponse import (
                AlipayCommerceEcTransOrderQueryResponse,
            )
        except ImportError:
            log.warning("SDK 不支持 trans.order.query")
            return None

        model = AlipayCommerceEcTransOrderQueryModel()
        model.enterprise_id = enterprise_id
        if out_biz_no:
            model.out_biz_no = out_biz_no
        if order_no:
            model.order_no = order_no

        request = AlipayCommerceEcTransOrderQueryRequest()
        request.biz_model = model

        client = AlipayClient.get_client()
        response = client.execute(request)

        if not response:
            return None

        result = AlipayCommerceEcTransOrderQueryResponse()
        result.parse_response_content(response)

        if not result.is_success():
            sub_code = getattr(result, 'sub_code', '') or ''
            sub_msg = getattr(result, 'sub_msg', '') or ''
            if sub_code == "ORDER_NOT_EXIST":
                return None
            log.warning(f"trans.order.query 失败: out_biz_no={out_biz_no}, sub_code={sub_code}, sub_msg={sub_msg}")
            return None

        alipay_status = getattr(result, 'status', None)
        if not alipay_status or alipay_status == "DEALING":
            return None

        update_data = {"status": alipay_status}
        api_order_no = getattr(result, 'order_no', None)
        api_amount = getattr(result, 'amount', None)
        api_fund_order_id = getattr(result, 'pay_fund_order_id', None)

        if api_order_no:
            update_data["order_no"] = api_order_no
        if api_fund_order_id:
            update_data["fund_order_id"] = api_fund_order_id
        if api_amount:
            update_data["amount"] = Decimal(str(api_amount))
        if alipay_status == "FAIL":
            sub_msg = getattr(result, 'sub_msg', '') or getattr(result, 'msg', '') or ''
            if sub_msg:
                update_data["remark"] = sub_msg

        upd = sa_update(TransferModel).where(
            TransferModel.out_biz_no == out_biz_no
        ).values(**update_data)
        await auth.db.execute(upd)
        log.info(f"转账同步(trans.order.query) - out_biz_no={out_biz_no}, status={alipay_status}")
        return alipay_status


    # ─────────── 退避时间映射 ───────────
    _RETRY_INTERVALS = [5, 10, 45, 60]  # 累计: +5min, +15min, +1h, +2h

    @classmethod
    async def retry_dealing_transfers(cls) -> dict:
        """定时任务：反查处理中的到卡转账结果（DB 锁防多 worker 重复执行）
        启动时立即执行一次，之后每分钟执行一次。使用独立数据库会话。
        """
        from sqlalchemy import select, text
        from app.plugin.module_payment.account.model import TransferModel
        from app.plugin.module_payment.account.enums import TransferStatusEnum
        from app.core.database import async_db_session

        db = async_db_session()
        try:
            async with db.begin():
                # 抢 PostgreSQL advisory lock
                lock = await db.execute(text("SELECT pg_try_advisory_lock(99999)"))
                if not lock.scalar():
                    log.debug("[重试任务] 锁已被其他 worker 持有，跳过")
                    return {"scanned": 0, "processed": 0, "skipped_lock": True}

                try:
                    log.info("[重试任务] ====== 开始执行 ======")
                    now = datetime.now()

                    stmt = (
                        select(TransferModel)
                        .where(
                            TransferModel.status == TransferStatusEnum.DEALING.value,
                            TransferModel.retry_count < 4,
                            TransferModel.next_retry_at.isnot(None),
                            TransferModel.next_retry_at <= now,
                        )
                        .order_by(TransferModel.next_retry_at.asc())
                    )
                    result = await db.execute(stmt)
                    pending = list(result.scalars().all())

                    if not pending:
                        return {"scanned": 0, "processed": 0}

                    log.info(f"[重试任务] 扫描到 {len(pending)} 条")

                    processed = 0
                    for transfer in pending:
                        out_biz_no = transfer.out_biz_no
                        enterprise_id = transfer.enterprise_id
                        order_no = transfer.order_no or ""
                        retry_count = transfer.retry_count

                        try:
                            sync_result = await cls._sync_transfer_detail(
                                AuthSchema(db=db, check_data_scope=False, tenant_id=transfer.tenant_id),
                                out_biz_no, enterprise_id, order_no,
                            )
                        except Exception as e:
                            log.warning(f"[重试任务] trans.order.query 异常: out_biz_no={out_biz_no}, err={e}")
                            continue

                        if sync_result:  # SUCCESS 或 FAIL
                            transfer.retry_count = 4
                            transfer.next_retry_at = None
                            log.info(f"[重试任务] out_biz_no={out_biz_no}, retry={retry_count+1}/4, result={sync_result}, 已更新")
                            try:
                                auth = AuthSchema(db=db, check_data_scope=False, tenant_id=transfer.tenant_id)
                                await OpenTransferService.open_return_service(auth, order_no or out_biz_no)
                            except Exception:
                                pass
                        else:
                            # 区分退避策略：对卡转账走多次退避，支付宝转账仅重试1次
                            is_card_transfer = transfer.payee_type == "BANK_CARD"
                            max_retries = 3 if is_card_transfer else 0

                            if retry_count < max_retries:
                                accumulated = sum(cls._RETRY_INTERVALS[:retry_count + 2])
                                transfer.retry_count = retry_count + 1
                                transfer.next_retry_at = transfer.created_time + timedelta(minutes=accumulated)
                                log.info(f"[重试任务] out_biz_no={out_biz_no}, payee_type={transfer.payee_type}, retry={retry_count+1}/{max_retries+1}, result=DEALING, 下次={transfer.next_retry_at}")
                                processed += 1
                            else:
                                transfer.status = TransferStatusEnum.FAIL.value
                                transfer.retry_count = 4
                                transfer.next_retry_at = None
                                fail_reason = "反查超时判定为失败" if is_card_transfer else "支付宝转账仅重试1次仍DEALING"
                                transfer.error_msg = fail_reason
                                processed += 1
                                log.info(f"[重试任务] out_biz_no={out_biz_no}, payee_type={transfer.payee_type}, retry={retry_count+1}/{max_retries+1}次DEALING, 判FAIL: {fail_reason}")
                                try:
                                    auth = AuthSchema(db=db, check_data_scope=False, tenant_id=transfer.tenant_id)
                                    await OpenTransferService.open_return_service(auth, order_no or out_biz_no)
                                except Exception:
                                    pass

                    # db.begin() 会自动 commit

                    log.info(f"[重试任务] ====== 执行完成 ====== 处理 {processed} 条，扫描 {len(pending)} 条")
                    return {"scanned": len(pending), "processed": processed}

                finally:
                    await db.execute(text("SELECT pg_advisory_unlock(99999)"))
        except Exception:
            log.error("[重试任务] 执行异常", exc_info=True)
            return {"scanned": 0, "processed": 0}
        finally:
            await db.close()


    @classmethod
    async def update_transfer_status_service(
        cls,
        auth: AuthSchema,
        order_no: str,
        status: str,
        ext_info: dict = {}
    ) -> None:
        """
        更新转账状态（由通知处理器调用）
        """
        crud = TransferCRUD(auth)
        transfer = await crud.get_by_order_no(order_no)

        if not transfer:
            log.warning(f"转账记录不存在: {order_no}")
            return

        update_data = {}
        update_data["status"] = status
        if ext_info:
            update_data["ext_info"] = ext_info

        await crud.update_by_order_no(order_no, update_data)

    @classmethod
    async def update_deposit_status_service(
        cls,
        auth: AuthSchema,
        out_biz_no: str,
        status: str,
    ) -> None:
        """
        更新充值状态（由通知处理器调用）
        """
        crud = DepositCRUD(auth)
        deposit = await crud.get_by_out_biz_no(out_biz_no)

        if not deposit:
            log.warning(f"充值记录不存在: {out_biz_no}")
            return

        update_data = {"status": status}
        await crud.update_by_out_biz_no(out_biz_no, update_data)



    @classmethod
    async def update_withdraw_status_service(
        cls,
        auth: AuthSchema,
        out_biz_no: str,
        status: str,
        error_code: str | None = None,
        error_msg: str | None = None,
    ) -> None:
        """
        更新提现状态（由通知处理器调用）
        """
        crud = WithdrawCRUD(auth)
        withdraw = await crud.get_by_out_biz_no(out_biz_no)

        if not withdraw:
            log.warning(f"提现记录不存在: {out_biz_no}")
            return

        update_data = {"status": status}
        if error_code:
            update_data["error_code"] = error_code
        if error_msg:
            update_data["error_msg"] = error_msg

        await crud.update_by_out_biz_no(out_biz_no, update_data)

    @classmethod
    async def consume_detail_query_service(
        cls,
        auth: AuthSchema,
        pay_no: str,
        enterprise_id: str | None = None,
        ant_shop_id: str | None = None,
        query_options: list[str] | None = None,
    ) -> dict:
        """
        账单详情查询（✅）

        调用: alipay.commerce.ec.consume.detail.query

        用于查询企业码账单详情，支持查询关联退款、订单、票据等信息。
        """
        from alipay.aop.api.request.AlipayCommerceEcConsumeDetailQueryRequest import (
            AlipayCommerceEcConsumeDetailQueryRequest,
        )
        from alipay.aop.api.domain.AlipayCommerceEcConsumeDetailQueryModel import (
            AlipayCommerceEcConsumeDetailQueryModel,
        )
        from alipay.aop.api.response.AlipayCommerceEcConsumeDetailQueryResponse import (
            AlipayCommerceEcConsumeDetailQueryResponse,
        )

        model = AlipayCommerceEcConsumeDetailQueryModel()
        model.pay_no = pay_no

        if enterprise_id:
            model.enterprise_id = enterprise_id
        if ant_shop_id:
            model.ant_shop_id = ant_shop_id
        if query_options:
            model.query_options = query_options

        request = AlipayCommerceEcConsumeDetailQueryRequest()
        request.biz_model = model

        client = AlipayClient.get_client()
        response = client.execute(request)

        if not response:
            raise CustomException(msg="账单详情查询失败: 无响应")

        result = AlipayCommerceEcConsumeDetailQueryResponse()
        result.parse_response_content(response)

        if not result.is_success():
            log.error(f"支付宝接口调用失败: {result.code} - {result.msg}")
            raise CustomException(msg=f"账单详情查询失败: {result.msg}")

        consume_info = result.consume_info
        if not consume_info:
            raise CustomException(msg="账单详情查询失败: 无账单信息")

        return {
            "account_id": consume_info.account_id,
            "pay_no": consume_info.pay_no,
            "consume_type": consume_info.consume_type,
            "gmt_biz_create": consume_info.gmt_biz_create,
            "consume_biz_type": consume_info.consume_biz_type,
            "consume_amount": consume_info.consume_amount,
            "order_complete_label": consume_info.order_complete_label,
            "refund_status": consume_info.refund_status,
            "refund_amount": consume_info.refund_amount,
            "peer_payer_card_name": consume_info.peer_payer_card_name,
            "user_id": getattr(consume_info, 'user_id', None),
            "open_id": getattr(consume_info, 'open_id', None),
            "enterprise_id": consume_info.enterprise_id,
            "employee_id": consume_info.employee_id,
            "enterprise_name": getattr(consume_info, 'enterprise_name', None),
            "employee_name": getattr(consume_info, 'employee_name', None),
            "consume_scene_code": getattr(consume_info, 'consume_scene_code', None),
            "consume_type_sub_category": getattr(consume_info, 'consume_type_sub_category', None),
            "consume_title": getattr(consume_info, 'consume_title', None),
            "gmt_pay": getattr(consume_info, 'gmt_pay', None),
            "gmt_refund": getattr(consume_info, 'gmt_refund', None),
            "pay_amount": getattr(consume_info, 'pay_amount', None),
            "invoice_amount": getattr(consume_info, 'invoice_amount', None),
            "peer_pay_amount": getattr(consume_info, 'peer_pay_amount', None),
            "subsidy_amount": getattr(consume_info, 'subsidy_amount', None),
            "ext_infos": getattr(consume_info, 'ext_infos', None),
        }