from sqlalchemy import Column, Integer, String, DateTime, Boolean, JSON from sqlalchemy.sql import func from pydantic import BaseModel, Field from typing import Optional, List from datetime import datetime from .database import Base from app.core.imageprocessor import process_image, ImageFormat class PointProductDB(Base): __tablename__ = "point_products" id = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(100), nullable=False) product_image = Column(String(200), nullable=False) tags = Column(String(100), nullable=True) # 商品标签,JSON数组 description = Column(String(500), nullable=False) point_amount = Column(Integer, nullable=False) # 所需积分 is_active = Column(Boolean, nullable=False, default=True) # 是否上架 create_time = Column(DateTime(timezone=True), server_default=func.now()) update_time = Column(DateTime(timezone=True), onupdate=func.now()) @property def optimized_product_image(self): """获取优化后的商品图片""" if self.product_image: return process_image(self.product_image).thumbnail(800, 800).format(ImageFormat.WEBP).build() return None # Pydantic 模型 class PointProductCreate(BaseModel): name: str = Field(..., max_length=100) product_image: str = Field(..., max_length=200) tags: Optional[str] = None description: str = Field(..., max_length=500) point_amount: int = Field(..., gt=0) is_active: bool = Field(default=True) class PointProductUpdate(BaseModel): name: Optional[str] = Field(None, max_length=100) product_image: Optional[str] = Field(None, max_length=200) tags: Optional[List[str]] = None description: Optional[str] = Field(None, max_length=500) point_amount: Optional[int] = Field(None, gt=0) is_active: Optional[bool] = None class PointProductInfo(BaseModel): id: int name: str product_image: str optimized_product_image: Optional[str] tags: Optional[str] description: str point_amount: int is_active: bool create_time: datetime class Config: from_attributes = True