diff --git a/alembic/versions/094_points_orders_wechat_fields.py b/alembic/versions/094_points_orders_wechat_fields.py new file mode 100644 index 000000000..fbba7fb2a --- /dev/null +++ b/alembic/versions/094_points_orders_wechat_fields.py @@ -0,0 +1,50 @@ +"""points_orders 新增微信支付链路字段 + +Revision ID: 094_points_orders_wechat_fields +Revises: 093 +Create Date: 2026-10-03 + +新增列： +- prepay_id: 微信预支付ID +- product_name: 下单商品名称（冗余，便于对账） +- payer_openid: 支付者 openid +- expire_at: 订单过期时间（未支付超时关闭用） +""" + +import sqlalchemy as sa + +from alembic import op + +revision = "094_points_orders_wechat_fields" +down_revision = "093" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + conn = op.get_bind() + cols = {c["name"] for c in sa.inspect(conn).get_columns("points_orders")} + + if "out_trade_no" not in cols: + op.add_column("points_orders", sa.Column("out_trade_no", sa.String(64), nullable=True)) + op.create_index("ix_points_orders_out_trade_no", "points_orders", ["out_trade_no"]) + if "prepay_id" not in cols: + op.add_column("points_orders", sa.Column("prepay_id", sa.String(128), nullable=True)) + if "product_name" not in cols: + op.add_column("points_orders", sa.Column("product_name", sa.String(100), nullable=True)) + if "payer_openid" not in cols: + op.add_column("points_orders", sa.Column("payer_openid", sa.String(128), nullable=True)) + if "expire_at" not in cols: + op.add_column("points_orders", sa.Column("expire_at", sa.DateTime(), nullable=True)) + + +def downgrade() -> None: + conn = op.get_bind() + cols = {c["name"] for c in sa.inspect(conn).get_columns("points_orders")} + try: + op.drop_index("ix_points_orders_out_trade_no", table_name="points_orders") + except Exception: + pass + for col in ("out_trade_no", "prepay_id", "product_name", "payer_openid", "expire_at"): + if col in cols: + op.drop_column("points_orders", col) diff --git a/apps/api/app/api/router.py b/apps/api/app/api/router.py index 7764a83e6..43cd02b2a 100755 --- a/apps/api/app/api/router.py +++ b/apps/api/app/api/router.py @@ -21,6 +21,7 @@ from app.api.routes.health import router as health_check_router from app.api.routes.ingest_jobs import router as ingest_jobs_router from app.api.routes.internal_render import router as internal_render_router from app.api.routes.lipsync import router as lipsync_router +from app.api.routes.payment import router as payment_router from app.api.routes.points import router as points_router from app.api.routes.points import usage_router from app.api.routes.projects import router as projects_router @@ -226,6 +227,11 @@ api_router.include_router( prefix="/ai-avatar/render", tags=["AI Avatar Render"], ) +api_router.include_router( + payment_router, + prefix="/payment", + tags=["Payment"], +) api_router.include_router( points_router, prefix="/points", diff --git a/apps/api/app/api/routes/payment.py b/apps/api/app/api/routes/payment.py new file mode 100644 index 000000000..32b2d6db3 --- /dev/null +++ b/apps/api/app/api/routes/payment.py @@ -0,0 +1,69 @@ +"""微信支付回调路由。 + +POST /api/v1/payment/wechat/notify +- 不做用户鉴权（微信服务器调用），靠平台证书签名保证来源可信 +- 返回微信要求的 JSON：{"code": "SUCCESS", "message": "成功"} +""" + +from __future__ import annotations + +import logging + +from fastapi import APIRouter, Depends, Request, Response +from sqlalchemy.orm import Session + +from app.dependencies import get_db_session +from packages.adapters.wechat_cert_store import get_platform_public_key +from packages.adapters.wechat_pay import WeChatPayError +from packages.application.payment_service import PaymentConfigError, PaymentService + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +def _wx_response(code: str, message: str, http_status: int = 200) -> Response: + """构造微信要求的回调应答。""" + import json + + return Response( + content=json.dumps({"code": code, "message": message}, ensure_ascii=False), + media_type="application/json", + status_code=http_status, + ) + + +@router.post("/wechat/notify") +async def wechat_pay_notify( + request: Request, + db: Session = Depends(get_db_session), +): + """微信支付结果通知。 + + - 验签失败 / 配置缺失 / 解密失败 → 返回 FAIL（微信会重试） + - 履约成功（含重复通知幂等）→ 返回 SUCCESS + """ + body = await request.body() + headers = {k.lower(): v for k, v in request.headers.items()} + + svc = PaymentService() + try: + result = svc.handle_wechat_notification( + headers=headers, + body=body, + platform_public_key_loader=get_platform_public_key, + db=db, + ) + except PaymentConfigError as e: + logger.error("微信回调时支付配置异常: %s", e) + return _wx_response("ERROR", "支付配置异常", http_status=503) + except WeChatPayError as e: + logger.warning("微信回调处理失败: %s", e) + return _wx_response("FAIL", str(e)[:200]) + except Exception: # noqa: BLE001 + logger.exception("微信回调未知异常") + return _wx_response("FAIL", "系统繁忙") + + if result.get("ignored"): + return _wx_response("SUCCESS", "成功") + return _wx_response("SUCCESS", "成功") diff --git a/apps/api/app/api/routes/points.py b/apps/api/app/api/routes/points.py index 531a68cd6..82aad73e1 100644 --- a/apps/api/app/api/routes/points.py +++ b/apps/api/app/api/routes/points.py @@ -61,8 +61,14 @@ def _get_service() -> PointsService: def _is_member(user: AuthenticatedUser) -> bool: - """判断用户是否为付费会员。""" - return getattr(user.user, "is_member", False) + """判断用户是否为有效付费会员（实时判断到期时间）。""" + if not getattr(user.user, "is_member", False): + return False + expires_at = getattr(user.user, "member_expires_at", None) + if expires_at is not None and expires_at <= datetime.now(timezone.utc): + return False + # 已标记取消但未到期：会员权益仍有效 + return True def _member_type(user: AuthenticatedUser) -> str | None: @@ -281,32 +287,55 @@ def refund_points( ) -@points_router.post("/recharge", response_model=PointsOrderResponse) +@points_router.post("/create-order", response_model=PointsOrderResponse) +def create_points_purchase_order( + body: PointsRechargeRequest, + current_user: AuthenticatedUser = Depends(get_current_user), + db: Session = Depends(get_db_session), +): + """创建积分包购买订单（微信 JSAPI 支付）。 + + 返回小程序调起支付参数；支付成功后积分自动到账（微信回调履约）。 + """ + from packages.adapters.wechat_pay import WeChatPayError + from packages.application.payment_service import PaymentConfigError, PaymentService + + openid = getattr(current_user.user, "wechat_openid", None) + svc = PaymentService() + try: + result = svc.create_points_order( + user_id=current_user.user.id, + package_code=body.package_id, + openid=openid, + db=db, + ) + except PaymentConfigError as e: + raise HTTPException(status_code=503, detail=f"支付通道不可用：{e}") from None + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) from None + except WeChatPayError as e: + raise HTTPException(status_code=502, detail=f"微信支付下单失败：{e}") from None + + return PointsOrderResponse( + id=result["order_id"], + order_type="points", + product_code=body.package_id, + amount_cents=result["amount_cents"], + points_amount=result["points_amount"], + status="pending", + pay_params=result["pay_params"], + expire_at=result["expire_at"], + ) + + +@points_router.post("/recharge", response_model=PointsOrderResponse, deprecated=True) def create_recharge_order( body: PointsRechargeRequest, current_user: AuthenticatedUser = Depends(get_current_user), db: Session = Depends(get_db_session), ): - """创建积分充值订单。pay_params 在支付通道接入后填入 prepay_id/payment_url；当前为空 dict。""" - svc = _get_service() - try: - order = svc.create_order( - user_id=current_user.user.id, - order_type="points", - product_code=body.package_id, - db=db, - ) - except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) from None - - package = POINTS_PACKAGES.get(body.package_id, {}) - now = datetime.now(timezone.utc) - expire_at = now + timedelta(hours=48) - # TODO: 接入微信/支付宝后填充真实 prepay_id / payment_url - order["points_amount"] = package.get("points", 0) - order["pay_params"] = {} - order["expire_at"] = expire_at.isoformat() - return PointsOrderResponse(**order) + """[已废弃] 旧积分充值入口，保留向后兼容，内部转发到 /points/create-order。""" + return create_points_purchase_order(body, current_user, db) @points_router.get("/subscription/membership", response_model=MembershipStatusResponse) diff --git a/apps/api/app/api/routes/subscription.py b/apps/api/app/api/routes/subscription.py index 3067ac2fc..7ac533fb2 100755 --- a/apps/api/app/api/routes/subscription.py +++ b/apps/api/app/api/routes/subscription.py @@ -14,12 +14,20 @@ from app.schemas.subscription import ( BillingRecord, ChangePlanRequest, ChangePlanResponse, + CreateOrderRequest, + CreateOrderResponse, MembershipType, + OrderItem, + OrderListResponse, SimpleResponse, SubscriptionInfo, ToggleAutoRenewRequest, ) -from fastapi import APIRouter, Depends, HTTPException, status +from fastapi import APIRouter, Depends, HTTPException, Query, status + +from sqlalchemy.orm import Session + +from app.dependencies import get_db_session from packages.ports.user_repository import UserRepository @@ -43,30 +51,53 @@ def _get_plan_name(plan_id: str) -> str: def _build_subscription_info(user: AuthenticatedUser) -> SubscriptionInfo: - """构建订阅信息响应""" + """构建订阅信息响应（P1-8：实时判断是否过期）。""" now = datetime.now(UTC) - if user.user.subscription_expires_at: - period_end = user.user.subscription_expires_at.isoformat() - period_start = now.isoformat() - else: - period_start = now.isoformat() - period_end = now.isoformat() + + expires_at = user.user.subscription_expires_at + # 优先使用会员体系字段 + member_expires = getattr(user.user, "member_expires_at", None) + if member_expires is not None: + expires_at = member_expires plan_id = user.user.subscription_plan or MembershipType.FREE + # 会员类型字段（member_type 与积分体系一致） + member_type = getattr(user.user, "member_type", None) + if member_type: + plan_id = member_type # 旧档位（standard/pro/enterprise）统一降级为 monthly，避免前端炸掉 if plan_id in {"standard", "pro", "enterprise"}: plan_id = MembershipType.MONTHLY + # 实时过期判断：到期即降级 free / expired（不改库，查询时计算） + is_expired = expires_at is not None and expires_at <= now + cancelled = (user.user.subscription_status or "") == "cancelled" + + if is_expired: + effective_plan = MembershipType.FREE + effective_status = "expired" + elif cancelled: + effective_plan = plan_id + effective_status = "cancelled" + else: + effective_plan = plan_id + effective_status = user.user.subscription_status or ( + "active" if plan_id != MembershipType.FREE else "active" + ) + + period_start = now.isoformat() + period_end = expires_at.isoformat() if expires_at else now.isoformat() + return SubscriptionInfo( id=f"sub-{user.user.id[:8]}", - plan_id=plan_id, - plan_name=_get_plan_name(plan_id), - status=user.user.subscription_status or "active", + plan_id=effective_plan, + plan_name=_get_plan_name(effective_plan), + status=effective_status, billing_cycle=plan_id if plan_id != MembershipType.FREE else BillingCycle.MONTHLY, current_period_start=period_start, current_period_end=period_end, - amount=0 if plan_id == MembershipType.FREE else 0, # 金额由前端 /plans 接口展示 - auto_renew=True, + amount=0, # 金额由前端 /plans 接口展示 + auto_renew=False, # 一期不做自动续费 created_at=user.user.created_at.isoformat() if user.user.created_at else now.isoformat(), ) @@ -216,12 +247,24 @@ async def cancel_subscription( detail="免费用户无需取消订阅", ) + # 已过期/未生效：直接报错，不允许取消 + expires_at = user.subscription_expires_at + member_expires = getattr(user, "member_expires_at", None) + if member_expires is not None: + expires_at = member_expires + if expires_at is not None and expires_at <= datetime.now(UTC): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="订阅已到期，无需取消", + ) + updated_user = replace(user, subscription_status="cancelled") user_repository.save(updated_user) + end_text = expires_at.strftime("%Y-%m-%d") if expires_at else "当前周期结束" return SimpleResponse( success=True, - message="订阅已取消，当前周期结束后将降级为免费用户", + message=f"已取消续费，{end_text} 前仍可正常使用会员权益，到期后自动降级为免费用户", ) @@ -294,3 +337,80 @@ async def toggle_auto_renew( """切换自动续费""" status_text = "已开启自动续费" if request.enabled else "已关闭自动续费" return SimpleResponse(success=True, message=status_text) + + +# ════════════════════════════════════════════════════════════════ +# 微信支付：下单 / 订单查询 +# ════════════════════════════════════════════════════════════════ + + +@router.post("/create-order") +def create_membership_order( + body: CreateOrderRequest, + current_user: AuthenticatedUser = Depends(get_current_user), + db: Session = Depends(get_db_session), +): + """创建会员年卡订单（微信 JSAPI 支付）。 + + - 一期只做年卡，billing_cycle 默认 yearly + - 返回小程序调起支付所需的 pay_params + - 支付结果通过 POST /api/v1/payment/wechat/notify 异步通知履约 + """ + from packages.adapters.wechat_pay import WeChatPayError + from packages.application.payment_service import PaymentConfigError, PaymentService + + openid = getattr(current_user.user, "wechat_openid", None) + svc = PaymentService() + try: + result = svc.create_membership_order( + user_id=current_user.user.id, + plan_id=body.plan_id, + billing_cycle=body.billing_cycle or "yearly", + openid=openid, + db=db, + ) + except PaymentConfigError as e: + raise HTTPException(status_code=503, detail=f"支付通道不可用：{e}") from None + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) from None + except WeChatPayError as e: + raise HTTPException(status_code=502, detail=f"微信支付下单失败：{e}") from None + + return CreateOrderResponse(**result) + + +@router.get("/orders", response_model=OrderListResponse) +def list_my_orders( + order_type: str | None = Query(None, description="筛选类型: membership/points"), + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + current_user: AuthenticatedUser = Depends(get_current_user), + db: Session = Depends(get_db_session), +): + """查询当前用户的订单列表（会员+积分包，分页）。""" + from packages.application.payment_service import PaymentService + + return PaymentService().list_orders( + user_id=current_user.user.id, + order_type=order_type, + page=page, + page_size=page_size, + db=db, + ) + + +@router.get("/orders/{order_id}", response_model=OrderItem) +def get_my_order( + order_id: str, + current_user: AuthenticatedUser = Depends(get_current_user), + db: Session = Depends(get_db_session), +): + """查询单个订单详情（仅能查自己的订单）。""" + from packages.application.payment_service import PaymentService + + order = PaymentService().get_order( + user_id=current_user.user.id, order_id=order_id, db=db + ) + if order is None: + raise HTTPException(status_code=404, detail="订单不存在") + return order diff --git a/apps/api/app/schemas/subscription.py b/apps/api/app/schemas/subscription.py index 377e21e0d..68e881e7b 100644 --- a/apps/api/app/schemas/subscription.py +++ b/apps/api/app/schemas/subscription.py @@ -110,3 +110,61 @@ class ToggleAutoRenewRequest(BaseModel): """切换自动续费请求""" enabled: bool = Field(..., description="是否开启自动续费") + + +# ============ 微信支付下单/订单 Schemas ============ + + +class CreateOrderRequest(BaseModel): + """会员购买下单请求（一期只做年卡）。""" + + plan_id: str = Field(..., description="套餐ID（admin plans 表UUID）；兼容传 yearly") + billing_cycle: str = Field("yearly", description="计费周期，一期默认 yearly") + + +class WeChatPayParams(BaseModel): + """小程序调起微信支付参数。""" + + appId: str + timeStamp: str + nonceStr: str + package: str + signType: str + paySign: str + + +class CreateOrderResponse(BaseModel): + """会员下单响应。""" + + order_id: str + out_trade_no: str + prepay_id: str + amount_cents: int + pay_params: WeChatPayParams + expire_at: str + + +class OrderItem(BaseModel): + """订单条目。""" + + id: str + order_type: str + product_code: str + product_name: Optional[str] = None + amount_cents: int + points_amount: int = 0 + status: str + payment_method: Optional[str] = None + prepay_id: Optional[str] = None + paid_at: Optional[str] = None + expire_at: Optional[str] = None + created_at: Optional[str] = None + + +class OrderListResponse(BaseModel): + """订单列表（分页）。""" + + total: int + page: int + page_size: int + items: list[OrderItem] diff --git a/packages/adapters/sqlalchemy_impl/commerce_models.py b/packages/adapters/sqlalchemy_impl/commerce_models.py new file mode 100644 index 000000000..226ace4b6 --- /dev/null +++ b/packages/adapters/sqlalchemy_impl/commerce_models.py @@ -0,0 +1,55 @@ +"""商业交易模型 —— 与 xiaoxia-admin 共享表的 ORM 映射。 + +xiaoxia-admin（管理后台）与 xiaoxia-saas（用户端）共享同一个数据库： +- plans / subscriptions 表由 admin 的 alembic 迁移创建（admin_alembic_version） +- 用户端在支付链路中需要读取套餐、写入订阅，因此在此做最小映射 + +注意：不要在此给这些表建迁移；表结构变更走 xiaoxia-admin 仓库。 +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +from sqlalchemy import Boolean, Column, DateTime, Integer, Numeric, String, Text +from sqlalchemy.orm import declarative_base + +CommerceBase: Any = declarative_base() + + +def _now() -> datetime: + return datetime.now(UTC) + + +class PlanModel(CommerceBase): + """admin 端 plans 套餐表（只读映射）。""" + + __tablename__ = "plans" + + id = Column(String(36), primary_key=True) + level_id = Column(String(36), nullable=True) + plan_key = Column(String(50), nullable=False) + name = Column(String(100), nullable=False) + description = Column(Text, nullable=True) + monthly_price = Column(Numeric(10, 2), nullable=False, default=0) + yearly_price = Column(Numeric(10, 2), nullable=False, default=0) + is_enabled = Column(Boolean, nullable=False, default=True) + display_order = Column(Integer, default=0) + created_at = Column(DateTime, default=_now) + + +class SubscriptionRecordModel(CommerceBase): + """admin 端 subscriptions 订阅记录表（支付成功后写入）。""" + + __tablename__ = "subscriptions" + + id = Column(String(36), primary_key=True) + user_id = Column(String(36), nullable=False, index=True) + plan_id = Column(String(36), nullable=False, index=True) + status = Column(String(20), nullable=False, default="active") + billing_cycle = Column(String(10), nullable=False) + start_date = Column(DateTime, nullable=False) + end_date = Column(DateTime, nullable=False) + cancelled_at = Column(DateTime, nullable=True) + created_at = Column(DateTime, default=_now) + updated_at = Column(DateTime, default=_now) diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py index 993c21c49..53fd7918d 100755 --- a/packages/adapters/sqlalchemy_impl/models.py +++ b/packages/adapters/sqlalchemy_impl/models.py @@ -816,6 +816,12 @@ class PointsOrderModel(Base): payment_method = Column(String(50), nullable=True) payment_id = Column(String(100), nullable=True) paid_at = Column(DateTime, nullable=True) + # 微信支付链路补充字段 + out_trade_no = Column(String(64), nullable=True, index=True) + prepay_id = Column(String(128), nullable=True) + product_name = Column(String(100), nullable=True) + payer_openid = Column(String(128), nullable=True) + expire_at = Column(DateTime, nullable=True) created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC)) diff --git a/packages/adapters/wechat_cert_store.py b/packages/adapters/wechat_cert_store.py new file mode 100644 index 000000000..45e90a903 --- /dev/null +++ b/packages/adapters/wechat_cert_store.py @@ -0,0 +1,102 @@ +"""微信支付平台证书缓存。 + +回调验签需要「微信支付平台公钥」。通过 V3 接口 +GET /v3/certificates 获取（响应用本商户 APIv3 密钥加密）， +带内存缓存（证书有效期通常约12个月）。 + +部署为单进程时内存缓存足够；多副本部署可改为 Redis 缓存。 +""" + +from __future__ import annotations + +import base64 +import json +import logging +import time +from typing import Any + +import httpx +from cryptography.hazmat.primitives import serialization + +from packages.adapters.wechat_pay import ( + WECHAT_BASE_URL, + build_authorization, + load_private_key, +) + +logger = logging.getLogger(__name__) + +CERT_PATH = "/v3/certificates" + +# serial -> {"public_key": obj, "expires_at": ts} +_cache: dict[str, dict[str, Any]] = {} + + +def _decrypt_cert_blob(*, api_v3_key: str, nonce: str, associated_data: str, ciphertext_b64: str) -> bytes: + from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + + blob = base64.b64decode(ciphertext_b64) + tag, data = blob[-16:], blob[:-16] + decryptor = Cipher( + algorithms.AES(api_v3_key.encode()), modes.GCM(nonce.encode(), tag) + ).decryptor() + return ( + decryptor.update(data) + decryptor.finalize_with_associated_data(associated_data.encode()) + ) + + +def refresh_platform_certificates() -> dict[str, Any]: + """拉取并刷新平台证书缓存。""" + from app.config import settings + + s = settings + if not getattr(s, "wechat_pay_configured", False): + raise RuntimeError("微信支付未配置，无法拉取平台证书") + + private_key = load_private_key(s.wechat_private_key) + auth = build_authorization( + "GET", CERT_PATH, s.wechat_mch_id, s.wechat_appid, s.wechat_cert_serial, private_key + ) + + with httpx.Client(timeout=15.0) as client: + resp = client.get( + f"{WECHAT_BASE_URL}{CERT_PATH}", + headers={"Authorization": auth, "Accept": "application/json"}, + ) + resp.raise_for_status() + data = resp.json() + + new_cache: dict[str, dict[str, Any]] = {} + for item in data.get("data", []): + serial = item["serial_no"] + enc = item["encrypt_certificate"] + pem = _decrypt_cert_blob( + api_v3_key=s.wechat_api_v3_key, + nonce=enc["nonce"], + associated_data=enc["associated_data"], + ciphertext_b64=enc["ciphertext"], + ) + cert = serialization.load_pem_x509_certificate(pem) + public_key = cert.public_key() + expires_at = time.mktime(cert.not_valid_after_utc.timetuple()) if hasattr( + cert, "not_valid_after_utc" + ) else time.time() + 365 * 24 * 3600 + new_cache[serial] = {"public_key": public_key, "expires_at": expires_at} + + _cache.clear() + _cache.update(new_cache) + logger.info("微信平台证书已刷新，共 %s 张", len(new_cache)) + return {"count": len(new_cache), "serials": list(new_cache.keys())} + + +def get_platform_public_key(serial: str): + """按证书序列号取平台公钥；缓存缺失/过期时自动拉取。""" + now = time.time() + entry = _cache.get(serial) + if entry is None or entry["expires_at"] <= now: + try: + refresh_platform_certificates() + except Exception: # noqa: BLE001 + logger.exception("刷新微信平台证书失败") + entry = _cache.get(serial) + return entry["public_key"] if entry else None diff --git a/packages/adapters/wechat_pay.py b/packages/adapters/wechat_pay.py new file mode 100644 index 000000000..6183d5756 --- /dev/null +++ b/packages/adapters/wechat_pay.py @@ -0,0 +1,237 @@ +"""微信支付 V3 适配层 —— JSAPI 下单、回调验签与解密、前端支付参数签名。 + +纯适配，不含业务逻辑： +- create_jsapi_order: JSAPI 统一下单，返回 prepay_id +- build_jsapi_pay_params: 生成调起微信支付所需参数（商户私钥二次签名） +- verify_notification_signature: 校验回调平台证书签名（含时间戳防重放） +- decrypt_resource: 用 APIv3 密钥解密 resource（AEAD_AES_256_GCM） + +依赖：httpx + cryptography（requirements-base.txt 中已含 httpx）。 +""" + +from __future__ import annotations + +import base64 +import json +import logging +import time +import uuid +from typing import Any + +import httpx +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import padding + +logger = logging.getLogger(__name__) + +WECHAT_BASE_URL = "https://api.mch.weixin.qq.com" +JSAPI_ORDER_PATH = "/v3/pay/transactions/jsapi" + +# 回调重放窗口：通知时间戳距当前超过 5 分钟视为过期 +NOTIFY_MAX_AGE_SECONDS = 300 + + +class WeChatPayError(Exception): + """微信支付调用错误。""" + + +# ──────────────────────── 密钥/证书加载 ──────────────────────── + + +def load_private_key(private_key_src: str): + """加载商户 API 私钥。 + + 支持： + - PEM 文本（含 BEGIN PRIVATE KEY / BEGIN RSA PRIVATE KEY） + - PEM 文件绝对路径 + - 单行 base64 编码的 PEM（容器 env 不支持换行时使用） + """ + src = (private_key_src or "").strip() + if not src: + raise WeChatPayError("微信商户私钥为空") + + if "BEGIN" not in src and "\n" not in src: + # 先尝试文件路径 + try: + with open(src, "rb") as f: + data = f.read() + return serialization.load_pem_private_key(data, password=None) + except (OSError, ValueError): + pass + # 再尝试单行 base64 + try: + src = base64.b64decode(src).decode() + except Exception as e: + raise WeChatPayError("无法解析微信商户私钥（不是PEM文本/文件/base64）") from e + + if "\\n" in src: + src = src.replace("\\n", "\n") + + try: + return serialization.load_pem_private_key(src.encode(), password=None) + except ValueError as e: + raise WeChatPayError(f"微信商户私钥格式无效: {e}") from e + + +def _rsa_sign(private_key, message: str) -> str: + """RSA-SHA256 签名，返回 base64。""" + signature = private_key.sign(message.encode(), padding.PKCS1v15(), hashes.SHA256()) + return base64.b64encode(signature).decode() + + +# ──────────────────────── Authorization 头 ──────────────────────── + + +def build_authorization( + method: str, + url_path: str, + mch_id: str, + appid: str, + cert_serial: str, + private_key, + body: str = "", +) -> str: + """构造微信 V3 API 的 Authorization 头。""" + timestamp = str(int(time.time())) + nonce = uuid.uuid4().hex + message = f"{method}\n{url_path}\n{timestamp}\n{nonce}\n{body}\n" + signature = _rsa_sign(private_key, message) + return ( + f'WECHATPAY2-SHA256-RSA2048 mchid="{mch_id}",nonce_str="{nonce}",' + f'timestamp="{timestamp}",serial_no="{cert_serial}",signature="{signature}"' + ) + + +# ──────────────────────── JSAPI 统一下单 ──────────────────────── + + +def create_jsapi_order( + *, + appid: str, + mch_id: str, + cert_serial: str, + private_key, + description: str, + out_trade_no: str, + amount_cents: int, + openid: str, + notify_url: str, + attach: str = "", +) -> dict[str, Any]: + """调用 V3 JSAPI 统一下单，返回 {"prepay_id": ...}。""" + payload: dict[str, Any] = { + "appid": appid, + "mchid": mch_id, + "description": description, + "out_trade_no": out_trade_no, + "notify_url": notify_url, + "amount": {"total": int(amount_cents), "currency": "CNY"}, + "payer": {"openid": openid}, + } + if attach: + payload["attach"] = attach + + body = json.dumps(payload, separators=(",", ":"), ensure_ascii=False) + authorization = build_authorization( + "POST", JSAPI_ORDER_PATH, mch_id, appid, cert_serial, private_key, body + ) + + try: + with httpx.Client(timeout=15.0) as client: + resp = client.post( + f"{WECHAT_BASE_URL}{JSAPI_ORDER_PATH}", + content=body.encode("utf-8"), + headers={ + "Authorization": authorization, + "Accept": "application/json", + "Content-Type": "application/json", + }, + ) + except httpx.HTTPError as e: + raise WeChatPayError(f"微信下单网络异常: {e}") from e + + if resp.status_code >= 400: + logger.error("微信下单失败: status=%s body=%s", resp.status_code, resp.text) + raise WeChatPayError(f"微信下单失败({resp.status_code}): {resp.text}") + + data = resp.json() + if not data.get("prepay_id"): + raise WeChatPayError(f"微信下单未返回 prepay_id: {data}") + return data + + +def build_jsapi_pay_params(appid: str, prepay_id: str, private_key) -> dict[str, str]: + """生成小程序调起支付参数（对 prepay_id 二次签名）。""" + timestamp = str(int(time.time())) + nonce = uuid.uuid4().hex + package = f"prepay_id={prepay_id}" + message = f"{appid}\n{timestamp}\n{nonce}\n{package}\n" + return { + "appId": appid, + "timeStamp": timestamp, + "nonceStr": nonce, + "package": package, + "signType": "RSA", + "paySign": _rsa_sign(private_key, message), + } + + +# ──────────────────────── 回调验签与解密 ──────────────────────── + + +def verify_notification_signature( + *, + timestamp: str, + nonce: str, + body: bytes, + signature_b64: str, + platform_public_key, +) -> bool: + """用微信平台公钥校验回调签名，并检查时间戳防重放。""" + try: + ts = int(timestamp) + except (TypeError, ValueError): + return False + if abs(time.time() - ts) > NOTIFY_MAX_AGE_SECONDS: + logger.warning("微信回调时间戳超出允许窗口: %s", timestamp) + return False + + message = f"{timestamp}\n{nonce}\n".encode() + body + b"\n" + try: + platform_public_key.verify( + base64.b64decode(signature_b64), + message, + padding.PKCS1v15(), + hashes.SHA256(), + ) + return True + except Exception: + logger.warning("微信回调签名校验失败") + return False + + +def decrypt_resource( + *, + api_v3_key: str, + nonce: str, + associated_data: str, + ciphertext_b64: str, +) -> dict[str, Any]: + """解密回调 resource（AEAD_AES_256_GCM）。""" + try: + from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + + ciphertext = base64.b64decode(ciphertext_b64) + tag = ciphertext[-16:] + data = ciphertext[:-16] + decryptor = Cipher( + algorithms.AES(api_v3_key.encode("utf-8")), + modes.GCM(nonce.encode("utf-8"), tag), + ).decryptor() + plaintext = ( + decryptor.update(data) + + decryptor.finalize_with_associated_data(associated_data.encode("utf-8")) + ) + return json.loads(plaintext.decode("utf-8")) + except Exception as e: + raise WeChatPayError(f"回调解密失败: {e}") from e diff --git a/packages/application/payment_service.py b/packages/application/payment_service.py new file mode 100644 index 000000000..12cc021a0 --- /dev/null +++ b/packages/application/payment_service.py @@ -0,0 +1,538 @@ +"""支付应用服务 —— 会员年卡/积分包购买的下单、回调履约、订单查询。 + +编排 xiaoxia-saas 自身能力（积分账户/用户表）与微信支付适配层， +同时写入 admin 端共享的 subscriptions 表，保持后台数据一致。 + +一期范围： +- 会员年卡：plan_id 支持 admin plans 表 UUID（主）与 legacy "yearly"（兼容） +- 积分包：product_code 对应 POINTS_PACKAGES +- 自动续费不做，auto_renew 固定 false +""" + +from __future__ import annotations + +import logging +import uuid +from datetime import UTC, datetime, timedelta +from typing import Any + +from sqlalchemy.orm import Session + +from packages.adapters.wechat_pay import ( + WeChatPayError, + build_jsapi_pay_params, + create_jsapi_order, + decrypt_resource, + load_private_key, + verify_notification_signature, +) +from packages.domain.points_rules import MEMBERSHIP_PRICES, POINTS_PACKAGES +from packages.domain.points_service import PointsService + +logger = logging.getLogger(__name__) + +# legacy plan_id -> 会员天数 +_LEGACY_DURATION = { + "monthly": 30, + "quarterly": 90, + "yearly": 365, +} +_ORDER_TTL_HOURS = 48 + + +class PaymentConfigError(Exception): + """支付配置缺失。""" + + +class PaymentService: + """会员/积分支付编排。""" + + def __init__(self) -> None: + self.points_service = PointsService() + + # ──────────────────────── 内部工具 ──────────────────────── + + @staticmethod + def _settings(): + from app.config import settings + + return settings + + def _require_wechat(self): + """校验微信配置并返回 (settings, private_key)。""" + settings = self._settings() + if not getattr(settings, "wechat_pay_configured", False): + raise PaymentConfigError( + "微信支付未配置或配置不完整（WECHAT_PAY_ENABLED/AppID/MchId/APIv3Key/CertSerial/PrivateKey/NotifyUrl）" + ) + return settings, load_private_key(settings.wechat_private_key) + + @staticmethod + def _resolve_plan(db: Session, plan_id: str, billing_cycle: str) -> dict[str, Any]: + """解析套餐：优先 admin plans 表 UUID；否则按 legacy 会员类型。 + + Returns: + {"plan_id", "name", "amount_cents", "duration_days", "db_plan"} + """ + # 1) admin plans 表 + from packages.adapters.sqlalchemy_impl.commerce_models import PlanModel + + plan = db.query(PlanModel).filter(PlanModel.id == plan_id).first() + if plan is not None: + if not plan.is_enabled: + raise ValueError("套餐已下架") + if billing_cycle == "yearly": + price = plan.yearly_price + days = 365 + elif billing_cycle == "monthly": + price = plan.monthly_price + days = 30 + else: + raise ValueError("一期仅支持 monthly/yearly 计费周期") + amount_cents = int(round(float(price) * 100)) + if amount_cents <= 0: + raise ValueError("该计费周期价格未配置（价格为0），无法下单") + return { + "plan_id": plan.id, + "name": plan.name, + "amount_cents": amount_cents, + "duration_days": days, + "db_plan": plan, + } + + # 2) legacy：plan_id 本身是会员类型 + if plan_id in _LEGACY_DURATION and plan_id == billing_cycle: + info = MEMBERSHIP_PRICES.get(plan_id) + if not info: + raise ValueError(f"未知会员类型: {plan_id}") + return { + "plan_id": plan_id, + "name": info["name"], + "amount_cents": info["price_cents"], + "duration_days": info["duration_days"], + "db_plan": None, + } + + raise ValueError(f"套餐不存在: {plan_id}") + + # ──────────────────────── 会员下单 ──────────────────────── + + def create_membership_order( + self, + *, + user_id: str, + plan_id: str, + billing_cycle: str, + openid: str, + db: Session, + ) -> dict[str, Any]: + """创建会员订单并向微信下单，返回订单 + 前端支付参数。""" + from packages.adapters.sqlalchemy_impl.models import PointsOrderModel + + settings, private_key = self._require_wechat() + if not openid: + raise ValueError("用户未绑定微信（缺少 openid），请先使用微信登录") + + resolved = self._resolve_plan(db, plan_id, billing_cycle) + + now = datetime.now(UTC) + out_trade_no = f"mb{now.strftime('%Y%m%d%H%M%S')}{uuid.uuid4().hex[:10]}" + expire_at = now + timedelta(hours=_ORDER_TTL_HOURS) + + order = PointsOrderModel( + id=uuid.uuid4().hex, + user_id=user_id, + order_type="membership", + product_code=resolved["plan_id"], + amount_cents=resolved["amount_cents"], + original_amount_cents=resolved["amount_cents"], + product_name=resolved["name"], + payer_openid=openid, + status="pending", + expire_at=expire_at, + ) + db.add(order) + db.flush() # 先不 commit，微信失败则回滚 + + try: + wx_resp = create_jsapi_order( + appid=settings.wechat_appid, + mch_id=settings.wechat_mch_id, + cert_serial=settings.wechat_cert_serial, + private_key=private_key, + description=resolved["name"], + out_trade_no=out_trade_no, + amount_cents=resolved["amount_cents"], + openid=openid, + notify_url=settings.wechat_notify_url, + attach=order.id, + ) + except WeChatPayError as e: + db.rollback() + raise WeChatPayError(str(e)) from e + + prepay_id = wx_resp["prepay_id"] + order.prepay_id = prepay_id + order.out_trade_no = out_trade_no + db.commit() + + pay_params = build_jsapi_pay_params(settings.wechat_appid, prepay_id, private_key) + return { + "order_id": order.id, + "out_trade_no": out_trade_no, + "prepay_id": prepay_id, + "pay_params": pay_params, + "amount_cents": resolved["amount_cents"], + "expire_at": expire_at.isoformat(), + } + + # ──────────────────────── 积分包下单 ──────────────────────── + + def create_points_order( + self, + *, + user_id: str, + package_code: str, + openid: str, + db: Session, + ) -> dict[str, Any]: + """创建积分包订单并向微信下单。""" + from packages.adapters.sqlalchemy_impl.models import PointsOrderModel + + settings, private_key = self._require_wechat() + package = POINTS_PACKAGES.get(package_code) + if not package: + raise ValueError(f"积分包不存在: {package_code}") + if not openid: + raise ValueError("用户未绑定微信（缺少 openid），请先使用微信登录") + + now = datetime.now(UTC) + out_trade_no = f"pt{now.strftime('%Y%m%d%H%M%S')}{uuid.uuid4().hex[:10]}" + expire_at = now + timedelta(hours=_ORDER_TTL_HOURS) + + order = PointsOrderModel( + id=uuid.uuid4().hex, + user_id=user_id, + order_type="points", + product_code=package_code, + amount_cents=package["price_cents"], + original_amount_cents=package["price_cents"], + points_amount=package["points"], + product_name=package["name"], + payer_openid=openid, + status="pending", + expire_at=expire_at, + ) + db.add(order) + db.flush() + + try: + wx_resp = create_jsapi_order( + appid=settings.wechat_appid, + mch_id=settings.wechat_mch_id, + cert_serial=settings.wechat_cert_serial, + private_key=private_key, + description=package["name"], + out_trade_no=out_trade_no, + amount_cents=package["price_cents"], + openid=openid, + notify_url=settings.wechat_notify_url, + attach=order.id, + ) + except WeChatPayError as e: + db.rollback() + raise WeChatPayError(str(e)) from e + + prepay_id = wx_resp["prepay_id"] + order.prepay_id = prepay_id + order.out_trade_no = out_trade_no + db.commit() + + pay_params = build_jsapi_pay_params(settings.wechat_appid, prepay_id, private_key) + return { + "order_id": order.id, + "out_trade_no": out_trade_no, + "prepay_id": prepay_id, + "pay_params": pay_params, + "amount_cents": package["price_cents"], + "points_amount": package["points"], + "expire_at": expire_at.isoformat(), + } + + # ──────────────────────── 回调履约 ──────────────────────── + + def handle_wechat_notification( + self, + *, + headers: dict[str, str], + body: bytes, + platform_public_key_loader, + db: Session, + ) -> dict[str, Any]: + """处理微信支付回调：验签 → 解密 → 验金额 → 履约。 + + Args: + platform_public_key_loader: callable(serial: str) -> 公钥对象 | None + """ + from packages.adapters.sqlalchemy_impl.models import PointsOrderModel, UserModel + + settings = self._settings() + + timestamp = headers.get("wechatpay-timestamp", "") + nonce = headers.get("wechatpay-nonce", "") + serial = headers.get("wechatpay-serial", "") + signature = headers.get("wechatpay-signature", "") + + if not all([timestamp, nonce, serial, signature]): + raise WeChatPayError("回调头不完整") + + platform_public_key = platform_public_key_loader(serial) + if platform_public_key is None: + raise WeChatPayError("未找到对应微信平台证书") + + if not verify_notification_signature( + timestamp=timestamp, + nonce=nonce, + body=body, + signature_b64=signature, + platform_public_key=platform_public_key, + ): + raise WeChatPayError("回调签名校验失败") + + envelope = _json_loads(body) + resource = envelope.get("resource", {}) + event_type = envelope.get("event_type", "") + if event_type != "TRANSACTION.SUCCESS": + logger.info("忽略非支付成功事件: %s", event_type) + return {"ignored": True, "event_type": event_type} + + payment = decrypt_resource( + api_v3_key=settings.wechat_api_v3_key, + nonce=resource.get("nonce", ""), + associated_data=resource.get("associated_data", ""), + ciphertext_b64=resource.get("ciphertext", ""), + ) + + out_trade_no = payment.get("out_trade_no", "") + transaction_id = payment.get("transaction_id", "") + payer_total = int(payment.get("amount", {}).get("payer_total", -1)) + trade_state = payment.get("trade_state", "") + + # 行锁取订单： + # 1) out_trade_no 独立列（主） + # 2) attach（下单时写入的 order.id） + # 3) payment_id（兼容旧数据） + order = ( + db.query(PointsOrderModel) + .filter(PointsOrderModel.out_trade_no == out_trade_no) + .with_for_update() + .first() + ) + if order is None: + attach_id = payment.get("attach") or envelope.get("id") + if attach_id: + order = ( + db.query(PointsOrderModel) + .filter(PointsOrderModel.id == attach_id) + .with_for_update() + .first() + ) + if order is None: + order = ( + db.query(PointsOrderModel) + .filter(PointsOrderModel.payment_id == out_trade_no) + .with_for_update() + .first() + ) + if order is None: + raise WeChatPayError(f"订单不存在: {out_trade_no}") + + if order.status == "paid": + # 幂等：微信可能重复通知 + return {"success": True, "order_id": order.id, "idempotent": True} + + if trade_state and trade_state != "SUCCESS": + raise WeChatPayError(f"交易状态非成功: {trade_state}") + + if payer_total != order.amount_cents: + raise WeChatPayError( + f"金额不一致: 订单 {order.amount_cents} / 实付 {payer_total}" + ) + + # ── 履约 ── + now = datetime.now(UTC) + order.status = "paid" + order.payment_method = "wechat" + order.payment_id = transaction_id + order.paid_at = now + + if order.order_type == "points": + self.points_service.add_points( + user_id=order.user_id, + amount=order.points_amount, + source=f"recharge:{order.product_code}", + db=db, + description=f"积分充值: {order.product_name or order.product_code}", + ref_id=order.id, + ) + elif order.order_type == "membership": + duration = self._duration_for(order) + self._activate_membership( + user_id=order.user_id, + plan_id=order.product_code, + duration_days=duration, + order_id=order.id, + now=now, + db=db, + ) + else: + raise WeChatPayError(f"未知订单类型: {order.order_type}") + + db.commit() + return {"success": True, "order_id": order.id} + + @staticmethod + def _duration_for(order: Any) -> int: + """根据订单商品推断会员天数。""" + code = order.product_code + if code in _LEGACY_DURATION: + return _LEGACY_DURATION[code] + # admin plan：查 plan_key / 名称或固定365（年卡订单） + return 365 + + def _activate_membership( + self, + *, + user_id: str, + plan_id: str, + duration_days: int, + order_id: str, + now: datetime, + db: Session, + ) -> None: + """激活会员：更新 users 表 + 写 admin subscriptions + 发积分。""" + from packages.adapters.sqlalchemy_impl.commerce_models import ( + PlanModel, + SubscriptionRecordModel, + ) + from packages.adapters.sqlalchemy_impl.models import UserModel + + user = db.query(UserModel).filter(UserModel.id == user_id).with_for_update().first() + if user is None: + raise WeChatPayError(f"用户不存在: {user_id}") + + base = user.member_expires_at or now + if base < now: + base = now + expires_at = base + timedelta(days=duration_days) + + user.is_member = True + user.member_type = plan_id + user.member_expires_at = expires_at + user.subscription_plan = plan_id if plan_id in _LEGACY_DURATION else "yearly" + user.subscription_status = "active" + user.subscription_expires_at = expires_at + user.max_projects = -1 + user.max_storage_gb = 100 + + # 写 admin subscriptions 表（后台可见；UUID 套餐才存在真实 plan_id） + admin_plan = db.query(PlanModel).filter(PlanModel.id == plan_id).first() + if admin_plan is not None: + record = SubscriptionRecordModel( + id=uuid.uuid4().hex, + user_id=user_id, + plan_id=admin_plan.id, + status="active", + billing_cycle="yearly" if duration_days >= 365 else "monthly", + start_date=now, + end_date=expires_at, + ) + db.add(record) + + # 发积分：优先 quotas.monthly_credits，年卡一期按一年一次性发放 + monthly_credits = 0 + level_id = admin_plan.level_id + if level_id: + from sqlalchemy import text as sa_text + + row = db.execute( + sa_text("SELECT monthly_credits FROM membership_levels WHERE id = :lid"), + {"lid": level_id}, + ).first() + if row: + monthly_credits = int(row[0] or 0) + + # 年卡：一次性发放 12 个月积分（任务9简化方案） + grant = monthly_credits * (12 if duration_days >= 365 else 1) + if grant > 0: + self.points_service.add_points( + user_id=user_id, + amount=float(grant), + source=f"membership:{plan_id}", + db=db, + description=f"会员开通赠积分（{admin_plan.name}）", + ref_id=order_id, + ) + + # ──────────────────────── 订单查询 ──────────────────────── + + def list_orders( + self, + *, + user_id: str, + order_type: str | None, + page: int, + page_size: int, + db: Session, + ) -> dict[str, Any]: + from packages.adapters.sqlalchemy_impl.models import PointsOrderModel + + query = db.query(PointsOrderModel).filter(PointsOrderModel.user_id == user_id) + if order_type: + query = query.filter(PointsOrderModel.order_type == order_type) + total = query.count() + rows = ( + query.order_by(PointsOrderModel.created_at.desc()) + .offset((page - 1) * page_size) + .limit(page_size) + .all() + ) + return { + "total": total, + "page": page, + "page_size": page_size, + "items": [self._order_dict(o) for o in rows], + } + + def get_order(self, *, user_id: str, order_id: str, db: Session) -> dict[str, Any] | None: + from packages.adapters.sqlalchemy_impl.models import PointsOrderModel + + order = ( + db.query(PointsOrderModel) + .filter(PointsOrderModel.id == order_id, PointsOrderModel.user_id == user_id) + .first() + ) + return self._order_dict(order) if order else None + + @staticmethod + def _order_dict(order: Any) -> dict[str, Any]: + return { + "id": order.id, + "order_type": order.order_type, + "product_code": order.product_code, + "product_name": order.product_name, + "amount_cents": order.amount_cents, + "points_amount": order.points_amount, + "status": order.status, + "payment_method": order.payment_method, + "prepay_id": order.prepay_id, + "paid_at": order.paid_at.isoformat() if order.paid_at else None, + "expire_at": order.expire_at.isoformat() if order.expire_at else None, + "created_at": order.created_at.isoformat() if order.created_at else None, + } + + +def _json_loads(body: bytes) -> dict[str, Any]: + import json + + return json.loads(body.decode("utf-8")) diff --git a/packages/config/base.py b/packages/config/base.py index eb375c8f4..3a6bdaee7 100755 --- a/packages/config/base.py +++ b/packages/config/base.py @@ -138,6 +138,57 @@ class SharedSettings(BaseSettings): self.credits_enabled = bool(value) self.points_enabled_compat = False + # ── 微信支付（V3 API）────────────────────────────────────────────── + # 未配置时支付下单接口返回明确错误，不会发出真实请求。 + # staging 可先配沙箱参数；通过环境变量注入。 + wechat_pay_enabled: bool = Field( + default=False, + validation_alias=AliasChoices("WECHAT_PAY_ENABLED", "wechat_pay_enabled"), + ) + # 公众号/小程序 AppID + wechat_appid: str = Field( + default="", + validation_alias=AliasChoices("WECHAT_APPID", "wechat_appid"), + ) + # 微信支付商户号 + wechat_mch_id: str = Field( + default="", + validation_alias=AliasChoices("WECHAT_MCH_ID", "wechat_mch_id"), + ) + # APIv3 密钥（32字符，用于回调解密） + wechat_api_v3_key: str = Field( + default="", + validation_alias=AliasChoices("WECHAT_API_V3_KEY", "wechat_api_v3_key"), + ) + # 商户API证书序列号 + wechat_cert_serial: str = Field( + default="", + validation_alias=AliasChoices("WECHAT_CERT_SERIAL", "wechat_cert_serial"), + ) + # 商户API私钥（PEM 内容；也支持填写文件绝对路径） + wechat_private_key: str = Field( + default="", + validation_alias=AliasChoices("WECHAT_PRIVATE_KEY", "wechat_private_key"), + ) + # 支付回调通知地址（完整 https URL） + wechat_notify_url: str = Field( + default="", + validation_alias=AliasChoices("WECHAT_NOTIFY_URL", "wechat_notify_url"), + ) + + @property + def wechat_pay_configured(self) -> bool: + """微信支付必要配置是否齐全（启用 + 六要素非空）。""" + return bool( + self.wechat_pay_enabled + and self.wechat_appid + and self.wechat_mch_id + and self.wechat_api_v3_key + and self.wechat_cert_serial + and self.wechat_private_key + and self.wechat_notify_url + ) + # ── GPU MuseTalk 反向轮询 Worker ──────────────────────────────────── # Worker 用这个长期 Token 鉴权（不是用户 JWT）。多 Worker 共用同一个 Token； # worker_id 用于区分具体机器。生产必须配置；development 留空会跳过校验。 diff --git a/requirements-base.txt b/requirements-base.txt index 1eb77c820..dd0a0be56 100644 --- a/requirements-base.txt +++ b/requirements-base.txt @@ -26,6 +26,7 @@ oss2==2.18.4 # HTTP 客户端（pin 间接依赖防止版本漂移） httpx==0.27.2 +cryptography==50.0.2 httpcore==1.0.7 h2==4.1.0 diff --git a/tests/unit/test_points_routes.py b/tests/unit/test_points_routes.py index 2b8a7f08b..dc42abebe 100644 --- a/tests/unit/test_points_routes.py +++ b/tests/unit/test_points_routes.py @@ -31,49 +31,93 @@ def _make_cu(user_id="user-1", is_member=False, member_type=None): class TestRechargeOrderResponse: - def test_recharge_returns_pay_params_points_amount_expire_at(self): - """recharge 响应必须包含 pay_params / points_amount / expire_at。""" - from app.api.routes.points import create_recharge_order + """积分包下单接口（create-order + 兼容 recharge 别名）。""" + + def _patch_payment(self, result=None, side_effect=None): + from packages.application.payment_service import PaymentService + + inst = MagicMock() + if side_effect is not None: + inst.create_points_order.side_effect = side_effect + else: + inst.create_points_order.return_value = result + return patch.object(PaymentService, "__new__", return_value=inst), inst + + def test_create_order_returns_pay_params(self): + """create-order 响应必须包含 pay_params / points_amount / expire_at。""" + from app.api.routes.points import create_points_purchase_order from app.schemas.points import PointsRechargeRequest - svc = MagicMock() - svc.create_order.return_value = { - "id": "order-1", - "order_type": "points", - "product_code": "starter_pack", + result = { + "order_id": "order-1", + "out_trade_no": "pt123", + "prepay_id": "prepay-1", "amount_cents": 990, - "status": "pending", - "created_at": datetime.now(UTC).isoformat(), + "points_amount": 100, + "expire_at": (datetime.now(UTC) + timedelta(hours=48)).isoformat(), + "pay_params": {"appId": "wx", "paySign": "s"}, } - db = MagicMock() + p, _inst = self._patch_payment(result=result) cu = _make_cu() + cu.user.wechat_openid = "o-1" body = PointsRechargeRequest(package_id="starter_pack") - before = datetime.now(UTC) - with patch("app.api.routes.points._get_service", return_value=svc): - resp = create_recharge_order(body=body, current_user=cu, db=db) - after = datetime.now(UTC) + timedelta(hours=48) + with p: + resp = create_points_purchase_order(body=body, current_user=cu, db=MagicMock()) - assert resp.points_amount == 100 # starter_pack 100 分 - assert isinstance(resp.pay_params, dict) - assert resp.expire_at is not None - expire_dt = datetime.fromisoformat(resp.expire_at) - assert expire_dt >= before + timedelta(hours=47, minutes=55) - assert expire_dt <= after + assert resp.points_amount == 100 + assert resp.pay_params["appId"] == "wx" + assert resp.expire_at == result["expire_at"] + assert resp.id == "order-1" - def test_recharge_invalid_package_returns_400(self): + def test_create_order_invalid_package_returns_400(self): + from app.api.routes.points import create_points_purchase_order + from app.schemas.points import PointsRechargeRequest + + p, _inst = self._patch_payment(side_effect=ValueError("积分包不存在: x")) + cu = _make_cu() + cu.user.wechat_openid = "o-1" + body = PointsRechargeRequest(package_id="x") + + with p, pytest.raises(HTTPException) as exc: + create_points_purchase_order(body=body, current_user=cu, db=MagicMock()) + assert exc.value.status_code == 400 + + def test_recharge_alias_forwards_to_create_order(self): + """旧 /recharge 端点保留且内部转发。""" from app.api.routes.points import create_recharge_order from app.schemas.points import PointsRechargeRequest - svc = MagicMock() - svc.create_order.side_effect = ValueError("invalid package") - db = MagicMock() + result = { + "order_id": "order-2", + "out_trade_no": "pt2", + "prepay_id": "p", + "amount_cents": 990, + "points_amount": 100, + "expire_at": "2026-10-05T00:00:00+00:00", + "pay_params": {"appId": "wx"}, + } + p, _inst = self._patch_payment(result=result) cu = _make_cu() - body = PointsRechargeRequest(package_id="nonexistent") + cu.user.wechat_openid = "o-1" + body = PointsRechargeRequest(package_id="starter_pack") + with p: + resp = create_recharge_order(body=body, current_user=cu, db=MagicMock()) + assert resp.id == "order-2" - with pytest.raises(HTTPException) as exc, patch("app.api.routes.points._get_service", return_value=svc): - create_recharge_order(body=body, current_user=cu, db=db) - assert exc.value.status_code == 400 + def test_create_order_payment_disabled_returns_503(self): + """微信未配置时返回 503。""" + from app.api.routes.points import create_points_purchase_order + from app.schemas.points import PointsRechargeRequest + from packages.application.payment_service import PaymentConfigError + + p, _inst = self._patch_payment(side_effect=PaymentConfigError("未配置")) + cu = _make_cu() + cu.user.wechat_openid = "o-1" + body = PointsRechargeRequest(package_id="starter_pack") + with p, pytest.raises(HTTPException) as exc: + create_points_purchase_order(body=body, current_user=cu, db=MagicMock()) + assert exc.value.status_code == 503 # ── P0-2: check 任意 scene_key（已下线场景返回 cost=0） ──────────────────