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
179 lines
5.3 KiB
Python
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)
|