feat(auth): add wechat miniapp customer phone login

Implement Wechat Mini Program phone number login flow for end customers: - add PhoneLoginIn Pydantic request schema - create wechat.py module for Wechat API interactions and phone code exchange - add customer JWT utilities and require_customer authentication dependency - add new public API endpoints: /api/public/auth/phone-login and /api/public/auth/me - add required environment config variables and update example .env - add comprehensive test cases for the new auth flow and endpoints
2026-07-08 16:11:34 +08:00
parent fe85197e68
commit bac68842ef
8 changed files with 242 additions and 5 deletions
--- a/.env.example
+++ b/.env.example
@@ -7,3 +7,6 @@ OSS_ACCESS_KEY_ID="your-oss-access-key-id"
 OSS_ACCESS_KEY_SECRET="your-oss-access-key-secret"
 OSS_ENDPOINT="your-oss-endpoint"
 OSS_BUCKET_NAME="your-oss-bucket-name"
+WECHAT_MINIAPP_APPID="your-miniapp-appid"
+WECHAT_MINIAPP_SECRET="your-miniapp-secret"
+CUSTOMER_JWT_EXPIRES_HOURS=720
--- a/app/auth.py
+++ b/app/auth.py
@@ -6,7 +6,7 @@ from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from sqlalchemy.orm import Session
 from .config import get_settings
 from .database import get_db
-from .models import AdminUser
+from .models import AdminUser, Customer


 bearer = HTTPBearer(auto_error=False)
@@ -24,6 +24,7 @@ def create_token(user: AdminUser) -> str:
    settings = get_settings()
    now = datetime.now(timezone.utc)
    payload = {
+        "typ": "admin",
        "sub": user.id,
        "email": user.email,
        "role": user.role,
@@ -33,6 +34,18 @@ def create_token(user: AdminUser) -> str:
    return jwt.encode(payload, settings.jwt_secret, algorithm="HS256")


+def create_customer_token(customer: Customer) -> str:
+    settings = get_settings()
+    now = datetime.now(timezone.utc)
+    payload = {
+        "typ": "customer",
+        "sub": customer.id,
+        "iat": now,
+        "exp": now + timedelta(hours=settings.customer_jwt_expires_hours),
+    }
+    return jwt.encode(payload, settings.jwt_secret, algorithm="HS256")
+
+
 def require_admin(
    request: Request,
    credentials: HTTPAuthorizationCredentials | None = Depends(bearer),
@@ -44,6 +57,8 @@ def require_admin(
        payload = jwt.decode(credentials.credentials, get_settings().jwt_secret, algorithms=["HS256"])
    except jwt.PyJWTError as exc:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="请先登录后台") from exc
+    if payload.get("typ", "admin") != "admin":
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="请先登录后台")
    user_id = payload.get("sub")
    user = db.get(AdminUser, user_id) if user_id else None
    if not user or not user.isActive:
@@ -52,5 +67,26 @@ def require_admin(
    return user


+def require_customer(
+    request: Request,
+    credentials: HTTPAuthorizationCredentials | None = Depends(bearer),
+    db: Session = Depends(get_db),
+) -> Customer:
+    if credentials is None:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="请先登录")
+    try:
+        payload = jwt.decode(credentials.credentials, get_settings().jwt_secret, algorithms=["HS256"])
+    except jwt.PyJWTError as exc:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="请先登录") from exc
+    if payload.get("typ") != "customer":
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="请先登录")
+    customer_id = payload.get("sub")
+    customer = db.get(Customer, customer_id) if customer_id else None
+    if not customer:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="请先登录")
+    request.state.actor_id = customer.id
+    return customer
+
+
 def get_actor_id(request: Request) -> str | None:
    return getattr(request.state, "actor_id", None)
--- a/app/config.py
+++ b/app/config.py
@@ -15,6 +15,9 @@ class Settings(BaseSettings):
    oss_access_key_secret: str | None = Field(default=None)
    oss_endpoint: str | None = Field(default=None)
    oss_bucket_name: str | None = Field(default=None)
+    wechat_miniapp_appid: str | None = Field(default=None)
+    wechat_miniapp_secret: str | None = Field(default=None)
+    customer_jwt_expires_hours: int = Field(default=720)

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")

--- a/app/routers/public.py
+++ b/app/routers/public.py
@@ -2,15 +2,28 @@ from fastapi import APIRouter, Depends, HTTPException, Query, status
 from sqlalchemy import or_, select
 from sqlalchemy.orm import Session, selectinload
 from .shared import site_config
+from ..auth import create_customer_token, require_customer
 from ..database import get_db
-from ..models import Destination, DestinationAlias, Lead, Product
-from ..schemas import LeadCreateIn, ProductQuery, ProductStatus
+from ..models import Customer, Destination, DestinationAlias, Lead, Product
+from ..schemas import LeadCreateIn, PhoneLoginIn, ProductQuery, ProductStatus
 from ..serializers import destination_dict, public_product_dict
+from ..wechat import WechatApiError, WechatConfigError, exchange_phone_code


 router = APIRouter()


+def mask_phone(phone: str) -> str:
+    normalized = phone.strip()
+    if len(normalized) <= 7:
+        return "***"
+    return f"{normalized[:3]}****{normalized[-4:]}"
+
+
+def public_customer_dict(customer: Customer) -> dict:
+    return {"id": customer.id, "phoneMasked": mask_phone(customer.phone)}
+
+
@router.get("/health")
 def health():
    return {"ok": True, "service": "miniapp-api"}
@@ -83,6 +96,30 @@ def list_destinations(db: Session = Depends(get_db)):
    return {"items": [destination_dict(destination) for destination in destinations]}


+@router.post("/api/public/auth/phone-login")
+def phone_login(body: PhoneLoginIn, db: Session = Depends(get_db)):
+    try:
+        phone = exchange_phone_code(body.code)
+    except WechatConfigError as exc:
+        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)) from exc
+    except WechatApiError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
+
+    customer = db.scalar(select(Customer).where(Customer.phone == phone))
+    if not customer:
+        customer = Customer(phone=phone)
+        db.add(customer)
+        db.flush()
+    db.commit()
+    db.refresh(customer)
+    return {"token": create_customer_token(customer), "customer": public_customer_dict(customer)}
+
+
+@router.get("/api/public/auth/me")
+def current_customer(customer: Customer = Depends(require_customer)):
+    return public_customer_dict(customer)
+
+
@router.post("/api/public/leads", status_code=status.HTTP_201_CREATED)
 def create_lead(body: LeadCreateIn, db: Session = Depends(get_db)):
    lead = Lead(**body.model_dump(exclude_none=True), status="new")
--- a/app/schemas.py
+++ b/app/schemas.py
@@ -12,6 +12,10 @@ class LoginIn(BaseModel):
    password: str = Field(min_length=6)


+class PhoneLoginIn(BaseModel):
+    code: str = Field(min_length=1, max_length=256)
+
+
 class ProductImageIn(BaseModel):
    url: str = ""
    alt: str | None = None
--- a/app/wechat.py
+++ b/app/wechat.py
@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+import time
+
+import httpx
+
+from .config import Settings, get_settings
+
+
+class WechatConfigError(RuntimeError):
+    pass
+
+
+class WechatApiError(RuntimeError):
+    pass
+
+
+_access_token_cache: dict[str, object] = {"appid": None, "token": None, "expires_at": 0.0}
+
+
+def _require_miniapp_settings(settings: Settings) -> tuple[str, str]:
+    if not settings.wechat_miniapp_appid or not settings.wechat_miniapp_secret:
+        raise WechatConfigError("微信小程序登录未配置")
+    return settings.wechat_miniapp_appid, settings.wechat_miniapp_secret
+
+
+def _wechat_json(response: httpx.Response) -> dict:
+    try:
+        return response.json()
+    except ValueError as exc:
+        raise WechatApiError("微信登录服务返回异常") from exc
+
+
+def get_access_token(settings: Settings | None = None) -> str:
+    settings = settings or get_settings()
+    appid, secret = _require_miniapp_settings(settings)
+    now = time.time()
+    cached_token = _access_token_cache.get("token")
+    if _access_token_cache.get("appid") == appid and isinstance(cached_token, str) and now < float(_access_token_cache.get("expires_at", 0)):
+        return cached_token
+
+    try:
+        response = httpx.get(
+            "https://api.weixin.qq.com/cgi-bin/token",
+            params={"grant_type": "client_credential", "appid": appid, "secret": secret},
+            timeout=8,
+        )
+    except httpx.HTTPError as exc:
+        raise WechatApiError("微信登录服务暂时不可用") from exc
+
+    data = _wechat_json(response)
+    token = data.get("access_token")
+    if not isinstance(token, str) or not token:
+        raise WechatApiError("微信登录服务暂时不可用")
+    expires_in = data.get("expires_in")
+    ttl = int(expires_in) if isinstance(expires_in, int | str) and str(expires_in).isdigit() else 7200
+    _access_token_cache.update({"appid": appid, "token": token, "expires_at": now + max(ttl - 300, 60)})
+    return token
+
+
+def exchange_phone_code(code: str, settings: Settings | None = None) -> str:
+    access_token = get_access_token(settings)
+    try:
+        response = httpx.post(
+            "https://api.weixin.qq.com/wxa/business/getuserphonenumber",
+            params={"access_token": access_token},
+            json={"code": code},
+            timeout=8,
+        )
+    except httpx.HTTPError as exc:
+        raise WechatApiError("手机号授权服务暂时不可用") from exc
+
+    data = _wechat_json(response)
+    if data.get("errcode", 0) != 0:
+        raise WechatApiError("手机号授权失败，请重试")
+    phone_info = data.get("phone_info")
+    if not isinstance(phone_info, dict):
+        raise WechatApiError("手机号授权服务返回异常")
+    phone_number = phone_info.get("phoneNumber") or phone_info.get("purePhoneNumber")
+    if not isinstance(phone_number, str) or not phone_number.strip():
+        raise WechatApiError("手机号授权服务返回异常")
+    return phone_number.strip()
--- a/tests/test_api_contracts.py
+++ b/tests/test_api_contracts.py
@@ -5,11 +5,12 @@ from fastapi.testclient import TestClient
 from pydantic import ValidationError

 from app import serializers
-from app.auth import hash_password, require_admin
+from app.auth import create_customer_token, hash_password, require_admin
 from app.database import get_db
 from app.main import create_app
-from app.models import AdminUser, Campaign, CtaBanner, DemandFeatureCard, DemandForm, DemandHero, DemandRecommendation, DemandRecommendationProduct, Destination, DestinationAlias, DestinationHero, DestinationRegion, HeroSlide, HotelGroup, Lead, MapImage, MediaAsset, Product, ProductImage, RouteSection, RouteSectionProduct, ThemeCard, VehicleOption
+from app.models import AdminUser, Campaign, CtaBanner, Customer, DemandFeatureCard, DemandForm, DemandHero, DemandRecommendation, DemandRecommendationProduct, Destination, DestinationAlias, DestinationHero, DestinationRegion, HeroSlide, HotelGroup, Lead, MapImage, MediaAsset, Product, ProductImage, RouteSection, RouteSectionProduct, ThemeCard, VehicleOption
 from app.routers import admin as admin_router
+from app.routers import public as public_router
 from app.routers.admin import normalize_detail_sections, normalize_images
 from app.routers.shared import site_config
 from app.schemas import (
@@ -475,6 +476,65 @@ def test_public_leads_endpoint_accepts_date_only_and_returns_minimal_response():
    assert fake_db.added[0].travelDate == datetime(2027, 1, 1)


+def test_public_phone_login_creates_customer_and_returns_masked_session(monkeypatch):
+    monkeypatch.setattr(public_router, "exchange_phone_code", lambda code: "10000000000", raising=False)
+    fake_db = FakeDb(scalar_values=[None])
+    app = create_app()
+    app.dependency_overrides[get_db] = lambda: fake_db
+
+    try:
+        response = TestClient(app).post("/api/public/auth/phone-login", json={"code": "phone-code"})
+    finally:
+        app.dependency_overrides.clear()
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["token"]
+    assert body["customer"] == {"id": "customer-test-id", "phoneMasked": "100****0000"}
+    assert fake_db.added[0].phone == "10000000000"
+    assert fake_db.committed
+
+
+def test_public_phone_login_requires_wechat_configuration(monkeypatch):
+    def missing_config(_code):
+        raise public_router.WechatConfigError("微信小程序登录未配置")
+
+    monkeypatch.setattr(public_router, "exchange_phone_code", missing_config, raising=False)
+    app = create_app()
+    app.dependency_overrides[get_db] = lambda: FakeDb()
+
+    try:
+        response = TestClient(app).post("/api/public/auth/phone-login", json={"code": "phone-code"})
+    finally:
+        app.dependency_overrides.clear()
+
+    assert response.status_code == 503
+    assert response.json() == {"message": "微信小程序登录未配置"}
+
+
+def test_public_me_returns_customer_for_customer_token():
+    customer = Customer(id="customer-test", phone="10000000000")
+    fake_db = FakeDb(get_result=customer)
+    app = create_app()
+    app.dependency_overrides[get_db] = lambda: fake_db
+    token = create_customer_token(customer)
+
+    try:
+        response = TestClient(app).get("/api/public/auth/me", headers={"Authorization": f"Bearer {token}"})
+    finally:
+        app.dependency_overrides.clear()
+
+    assert response.status_code == 200
+    assert response.json() == {"id": "customer-test", "phoneMasked": "100****0000"}
+
+
+def test_public_me_requires_customer_token():
+    response = TestClient(create_app()).get("/api/public/auth/me")
+
+    assert response.status_code == 401
+    assert response.json() == {"message": "请先登录"}
+
+
 def test_admin_destinations_endpoint_uses_precomputed_product_counts():
    destination = make_destination()
    fake_db = FakeDb(scalar_results=[[destination]], execute_results=[[(destination.id, 7)]])
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -13,3 +13,15 @@ def test_oss_settings_are_loaded_from_prefixed_environment(monkeypatch):
    assert settings.oss_access_key_secret == "example-access-key-secret"
    assert settings.oss_endpoint == "oss-cn-example.aliyuncs.com"
    assert settings.oss_bucket_name == "example-bucket"
+
+
+def test_wechat_miniapp_settings_are_loaded_from_environment(monkeypatch):
+    monkeypatch.setenv("WECHAT_MINIAPP_APPID", "wx-test-appid")
+    monkeypatch.setenv("WECHAT_MINIAPP_SECRET", "test-secret")
+    monkeypatch.setenv("CUSTOMER_JWT_EXPIRES_HOURS", "24")
+
+    settings = Settings(_env_file=None)
+
+    assert settings.wechat_miniapp_appid == "wx-test-appid"
+    assert settings.wechat_miniapp_secret == "test-secret"
+    assert settings.customer_jwt_expires_hours == 24