87 lines
3.7 KiB
Python
87 lines
3.7 KiB
Python
"""支付子域管理后台路由.
|
|
|
|
仅含管理后台用的支付流水列表/详情. URL prefix 与 tags 与拆分前完全一致
|
|
(``/admin/commerce/payments`` + ``commerce-admin-payments``), 保证对外契约不变.
|
|
共享 schema ``OrderResponse`` 与 repository 从父子域绝对 import.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Annotated, Literal
|
|
from urllib.parse import urlparse
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from pydantic import BaseModel, ConfigDict
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from luffy_agent.commerce.bootstrap import _secret_text, jsapi_ready
|
|
from luffy_agent.commerce.orders.console.schemas import (
|
|
AdminOrderRow,
|
|
to_admin_order_row,
|
|
to_admin_order_rows,
|
|
)
|
|
from luffy_agent.commerce.orders.repositories import order_repository
|
|
from luffy_agent.commerce.payments.wechat_jsapi import repositories as jsapi_repo
|
|
from luffy_agent.commerce.payments.wechat_jsapi import service as jsapi_service
|
|
from luffy_agent.core.db import get_db
|
|
from luffy_agent.core.exceptions import AppException
|
|
from luffy_agent.core.params import Int4Query
|
|
from luffy_agent.core.security.fernet import FernetSecretService, MasterKeyMissingError
|
|
from luffy_agent.core.setting import get_settings
|
|
from luffy_agent.core.setting.schemas import Settings
|
|
from luffy_agent.user.auth.deps import ConsolePrincipal, require_permission
|
|
from luffy_agent.user.console.rbac.service import (
|
|
bind_console_audit_request,
|
|
console_rbac_service,
|
|
)
|
|
|
|
router = APIRouter(prefix="/admin/commerce/payments", tags=["commerce-admin-payments"])
|
|
|
|
|
|
class CredentialPresence(BaseModel):
|
|
mp_app_secret: bool
|
|
wechat_pay_api_v3_key: bool
|
|
wechat_pay_private_key: bool
|
|
jsapi_intent_master_key: bool
|
|
@router.get("/capability", response_model=PaymentCapabilityDiagnostic)
|
|
async def get_payment_capability(
|
|
request: Request,
|
|
db: AsyncSession = Depends(get_db),
|
|
_: object = Depends(require_permission("commerce.payments:view")),
|
|
) -> PaymentCapabilityDiagnostic:
|
|
"""admin: JSAPI/Native/H5 只读诊断. 不含密钥、openid、签名或密文."""
|
|
settings = get_settings()
|
|
oauth_host = urlparse(settings.WECHAT_MP_OAUTH_REDIRECT_URI).hostname or ""
|
|
allowed = list(settings.PAYMENT_RETURN_ALLOWED_ORIGINS)
|
|
request_origin = request.headers.get("origin")
|
|
pay_origin = request_origin if request_origin in allowed else None
|
|
hint_origins = [pay_origin] if pay_origin else allowed
|
|
private_key_path = settings.WECHAT_PAY_PRIVATE_KEY_PATH
|
|
count, earliest = await jsapi_repo.recovery_required_summary(db)
|
|
return PaymentCapabilityDiagnostic(
|
|
native_enabled=bool(settings.WECHAT_PAY_MCH_ID),
|
|
h5_enabled=settings.WECHAT_H5_ENABLED,
|
|
jsapi_enabled=settings.WECHAT_JSAPI_ENABLED,
|
|
jsapi_ready=jsapi_ready(settings),
|
|
pay_app_id_masked=_mask_app_id(settings.WECHAT_PAY_APP_ID),
|
|
mp_app_id_masked=_mask_app_id(settings.WECHAT_MP_APP_ID),
|
|
credentials_present=CredentialPresence(
|
|
mp_app_secret=bool(_secret_text(settings.WECHAT_MP_APP_SECRET)),
|
|
wechat_pay_api_v3_key=bool(settings.WECHAT_PAY_API_V3_KEY),
|
|
wechat_pay_private_key=bool(private_key_path)
|
|
and Path(private_key_path).is_file(),
|
|
jsapi_intent_master_key=_intent_key_present(settings),
|
|
),
|
|
oauth_callback_host=oauth_host,
|
|
pay_origin=pay_origin,
|
|
allowed_pay_origins=allowed,
|
|
jsapi_auth_dir_hint=[f"{origin}/" for origin in hint_origins if origin],
|
|
domain_whitelist="unknown_manual_check",
|
|
merchant_binding="unknown_manual_check",
|
|
recovery_required_count=count,
|
|
recovery_required_earliest=earliest,
|
|
)
|