Files
WonderQ-Admin/app/schemas.py
duanshuwen bac68842ef 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

179 lines
5.3 KiB
Python

from datetime import datetime, timedelta
from typing import Literal
from pydantic import BaseModel, EmailStr, Field, ValidationInfo, field_validator
ProductStatus = Literal["draft", "published", "archived"]
LeadStatus = Literal["new", "assigned", "contacted", "planning", "won", "invalid"]
class LoginIn(BaseModel):
email: EmailStr
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
sortOrder: int = 0
class TextBlock(BaseModel):
type: Literal["text"]
text: str = ""
class ImageBlock(BaseModel):
type: Literal["image"]
url: str = ""
alt: str | None = None
class ProductDetailSectionIn(BaseModel):
key: str = ""
label: str = ""
title: str | None = None
blocks: list[TextBlock | ImageBlock] = Field(default_factory=list)
class ProductCreateIn(BaseModel):
title: str = Field(min_length=2)
subtitle: str | None = None
destinationId: str | None = None
priceAmount: int | None = Field(default=None, ge=0)
priceUnit: str | None = None
tags: list[str] = Field(default_factory=list)
coverImage: str | None = None
summary: str | None = None
images: list[ProductImageIn] | None = None
detailSections: list[ProductDetailSectionIn] | None = None
status: ProductStatus = "draft"
sortWeight: int = 0
class ProductUpdateIn(BaseModel):
title: str | None = Field(default=None, min_length=2)
subtitle: str | None = None
destinationId: str | None = None
priceAmount: int | None = Field(default=None, ge=0)
priceUnit: str | None = None
tags: list[str] | None = None
coverImage: str | None = None
summary: str | None = None
images: list[ProductImageIn] | None = None
detailSections: list[ProductDetailSectionIn] | None = None
status: ProductStatus | None = None
sortWeight: int | None = None
class LeadCreateIn(BaseModel):
destination: str | None = None
phone: str = Field(min_length=2, max_length=64)
travelDate: datetime | None = None
peopleCount: int | None = Field(default=None, gt=0)
budgetMin: int | None = Field(default=None, ge=0)
budgetMax: int | None = Field(default=None, ge=0)
note: str | None = Field(default=None, max_length=1000)
sourcePage: str | None = None
sourceProductId: str | None = None
@field_validator("phone")
@classmethod
def normalize_phone(cls, value: str) -> str:
normalized = " ".join(value.strip().split())
if not normalized:
raise ValueError("联系方式不能为空")
return normalized
@field_validator("travelDate", mode="before")
@classmethod
def parse_date_only_travel_date(cls, value):
if isinstance(value, str) and len(value) == 10:
try:
return datetime.strptime(value, "%Y-%m-%d")
except ValueError:
return value
return value
class LeadStatusIn(BaseModel):
status: LeadStatus
class ProductQuery(BaseModel):
keyword: str | None = None
destinationId: str | None = None
status: ProductStatus = "published"
take: int = Field(default=48, ge=1, le=100)
class AdminProductQuery(BaseModel):
keyword: str | None = None
status: ProductStatus | None = None
take: int = Field(default=100, ge=1, le=200)
class LeadQuery(BaseModel):
status: LeadStatus | None = None
sourcePage: str | None = None
keyword: str | None = None
createdFrom: datetime | None = None
createdTo: datetime | None = None
take: int = Field(default=100, ge=1, le=200)
@field_validator("createdFrom", "createdTo", mode="before")
@classmethod
def parse_date_only_range(cls, value, info: ValidationInfo):
if isinstance(value, str) and len(value) == 10:
try:
parsed = datetime.strptime(value, "%Y-%m-%d")
return parsed + timedelta(days=1) if info.field_name == "createdTo" else parsed
except ValueError:
return value
return value
class SiteConfigPatchIn(BaseModel):
title: str | None = None
kicker: str | None = None
subtitle: str | None = None
name: str | None = None
slug: str | None = None
region: str | None = None
label: str | None = None
keyword: str | None = None
spots: str | None = None
alt: str | None = None
image: str | None = None
description: str | None = None
coverImage: str | None = None
priceAmount: int | None = Field(default=None, ge=0)
priceUnit: str | None = None
tags: list[str] | None = None
steps: list[str] | None = None
destinationLabel: str | None = None
destinationPlaceholder: str | None = None
phoneLabel: str | None = None
phonePlaceholder: str | None = None
noteLabel: str | None = None
notePlaceholder: str | None = None
submitLabel: str | None = None
chips: list[str] | None = None
productIds: list[str] | None = None
status: str | None = None
startsAt: datetime | None = None
endsAt: datetime | None = None
targetType: str | None = None
targetValue: str | None = None
isHot: bool | None = None
isActive: bool | None = None
sortOrder: int | None = None
class SiteConfigReorderIn(BaseModel):
itemIds: list[str] = Field(default_factory=list)