from datetime import datetime from typing import Optional, List from sqlalchemy import Column, String, Integer, Float, DateTime, ForeignKey, Enum, Boolean from sqlalchemy.sql import func from pydantic import BaseModel, Field from .database import Base import enum from app.models.user import Gender from app.core.imageprocessor import process_image, ImageFormat class OrderStatus(str, enum.Enum): CREATED = "CREATED" # 已创建 CANCELLED = "CANCELLED" # 已取消 RECEIVED = "RECEIVED" # 已接单 DELIVERING = "DELIVERING" # 配送中 UNPAID = "UNPAID" # 未支付 COMPLETED = "COMPLETED" # 已完成 class DeliveryMethod(str, enum.Enum): DELIVERY_AT_DOORSTEP = "DELIVERY_AT_DOORSTEP" # 放在门口 DELIVERY_TO_ROOM = "DELIVERY_TO_ROOM" # 投递到家 # 数据库模型 class ShippingOrderDB(Base): __tablename__ = "shipping_orders" orderid = Column(String(32), primary_key=True) userid = Column(Integer, ForeignKey("users.userid"), index=True) # 配送地址信息 address_customer_name = Column(String(50), nullable=False, default='') # 客户名称快照 address_customer_phone = Column(String(11), nullable=False, default='') # 客户电话快照 address_customer_gender = Column(Enum(Gender), nullable=False, default=Gender.UNKNOWN) # 客户性别快照 address_community_name = Column(String(50), nullable=False, default='') # 小区名称快照 address_community_building_name = Column(String(50), nullable=False, default='') # 楼栋名称快照 address_detail = Column(String(100), nullable=False, default='') # 详细地址快照 delivery_method = Column(Enum(DeliveryMethod), nullable=False) package_count = Column(Integer, nullable=False) original_amount = Column(Float, nullable=False) coupon_discount_amount = Column(Float, default=0) point_discount_amount = Column(Float, default=0) coupon_id = Column(Integer, ForeignKey("user_coupons.id"), nullable=True) final_amount = Column(Float, nullable=False) status = Column(Enum(OrderStatus), nullable=False, default=OrderStatus.CREATED) cancel_reason = Column(String(200), nullable=True) # 取消原因 received_time = Column(DateTime(timezone=True), nullable=True) # 接单时间 pickup_time = Column(DateTime(timezone=True), nullable=True) # 取件时间 completed_time = Column(DateTime(timezone=True), nullable=True) # 完成时间 update_time = Column(DateTime(timezone=True), onupdate=func.now()) complete_images = Column(String(1000), nullable=True) # 完成订单的图片URL,多个URL用逗号分隔 create_time = Column(DateTime(timezone=True), server_default=func.now()) deliveryman_user_id = Column(Integer, ForeignKey("users.userid"), nullable=True) cancel_user_id = Column(Integer, ForeignKey("users.userid"), nullable=True) pay_status = Column(Boolean, default=False) # 支付状态 prepay_id = Column(String(64)) # 微信支付预支付ID pay_time = Column(DateTime(timezone=True)) # 支付时间 transaction_id = Column(String(64)) # 微信支付交易号 @property def optimized_complete_images(self): if self.complete_images: return [process_image(image).quality(80).thumbnail(width=450, height=450).format(ImageFormat.WEBP).build() for image in self.complete_images.split(",")] return [] class ShippingOrderPackageDB(Base): __tablename__ = "shipping_order_packages" id = Column(Integer, primary_key=True, autoincrement=True) orderid = Column(String(32), ForeignKey("shipping_orders.orderid"), index=True) station_id = Column(Integer, ForeignKey("stations.id"), index=True) station_name = Column(String(50), nullable=False) pickup_codes = Column(String(100), nullable=False) create_time = Column(DateTime(timezone=True), server_default=func.now()) # Pydantic 模型 class OrderPackage(BaseModel): station_id: int pickup_codes: str = Field(..., max_length=100) # 先定义 OrderPriceCalculateRequest class OrderPriceCalculateRequest(BaseModel): packages: List[OrderPackage] # 然后再定义 OrderCreate class OrderCreate(BaseModel): addressid: int price_request: OrderPriceCalculateRequest delivery_method: DeliveryMethod = Field( default=DeliveryMethod.DELIVERY_AT_DOORSTEP, description="配送方式:放在门口或投递到家" ) class OrderInfo(BaseModel): orderid: str userid: int address_customer_name: str address_customer_phone: str address_community_name: str address_community_building_name: str address_detail: str address_customer_gender: Gender package_count: int original_amount: float coupon_discount_amount: float point_discount_amount: float = 0 coupon_id: Optional[int] final_amount: float status: OrderStatus complete_images: Optional[List[str]] = None optimized_complete_images: Optional[List[str]] = None create_time: datetime delivery_method: DeliveryMethod deliveryman_user_id: Optional[int] = None cancel_reason: Optional[str] = None cancel_user_id: Optional[int] = None received_time: Optional[datetime] = None pickup_time: Optional[datetime] = None completed_time: Optional[datetime] = None def __init__(self, **data): super().__init__(**data) # 将逗号分隔的图片URL字符串转换为列表 if self.complete_images and isinstance(self.complete_images, str): self.complete_images = self.complete_images.split(",") class Config: from_attributes = True class OrderPackageInfo(BaseModel): id: int orderid: str station_id: int station_name: str pickup_codes: str create_time: datetime class Config: from_attributes = True class OrderPriceInfo(BaseModel): package_count: int original_amount: float coupon_discount_amount: float = 0 points_discount_amount: float = 0 coupon_id: Optional[int] = None final_amount: float # 添加取消订单请求模型 class OrderCancel(BaseModel): reason: str = Field(..., max_length=200, description="取消原因") # 完成订单请求模型 class OrderComplete(BaseModel): images: Optional[List[str]] = Field(None, max_items=5) # 最多5张图片,可选 class OrderPriceResult(BaseModel): """订单价格计算结果""" price_info: OrderPriceInfo used_coupon_id: Optional[int] = None # 使用的优惠券ID used_points: Optional[int] = None # 使用的积分数 price_detail_text: Optional[str] = None # 价格详情文本