""" 产品主表模型 - 系统架构优化的核心表 直接从原始订单生成，替代标准化订单步骤 """ from datetime import datetime from typing import Optional from sqlalchemy import ( Column, Integer, String, Text, DateTime, Float, Boolean, JSON, Index, UniqueConstraint, Enum as SQLEnum ) from sqlalchemy.sql import func import enum from app.core.database import Base class ProcurementMethod(str, enum.Enum): """采购方式枚举""" GN = "GN" # 国内现货 MC = "MC" # 库存 AT = "AT" # Atelier AP = "AP" # 昌昌 LA = "LA" # 洛杉矶 SS = "SS" # Saks NY = "NY" # 纽约默认 SF = "SF" # 旧金山 FL = "FL" # 佛罗里达 CA = "CA" # 加拿大 class ProductStatus(str, enum.Enum): """产品状态枚举""" ACTIVE = "ACTIVE" # 活跃 INACTIVE = "INACTIVE" # 非活跃 ARCHIVED = "ARCHIVED" # 已归档 class ProductMaster(Base): """ 产品主表 - 系统的核心产品数据表 基于原始订单直接生成，包含所有必要的SKU属性 SKU唯一性: 线上宝贝名称 + 线上销售属性 """ __tablename__ = "products_master" # 主键和唯一标识 id = Column(Integer, primary_key=True, index=True) sku_key = Column(String(64), unique=True, nullable=False, comment="SKU唯一键(SHA256)") # 原始商品信息 (从 raw_orders 直接解析) 线上宝贝名称 = Column(Text, nullable=False, comment="商品名称") 线上销售属性 = Column(Text, nullable=True, comment="销售属性") # 解析后的属性 品牌 = Column(String(100), nullable=True, index=True, comment="解析的品牌") 货号 = Column(String(100), nullable=True, comment="解析的货号") 颜色 = Column(String(100), nullable=True, comment="解析的颜色") 尺寸 = Column(String(50), nullable=True, comment="解析的尺寸") # 网店信息 网店名称 = Column(JSON, nullable=True, comment="关联的网店名称列表") # 商品编码信息 线上商家编码 = Column(String(100), nullable=True, comment="商家编码") 商品编号 = Column(String(100), nullable=True, comment="商品编号") SKU编号 = Column(String(100), nullable=True, comment="SKU编号") # 采购相关 procurement_method = Column( SQLEnum(ProcurementMethod), default=ProcurementMethod.NY, nullable=False, comment="采购方式" ) # 图片信息 original_image_urls = Column(JSON, nullable=True, comment="原始图片链接JSON数组") local_image_paths = Column(JSON, nullable=True, comment="本地图片路径JSON数组") main_image_path = Column(Text, nullable=True, comment="主图本地路径") image_download_status = Column(String(20), default="pending", comment="图片下载状态: pending/completed/failed") # 价格统计 avg_price = Column(Float, nullable=True, comment="平均售价") min_price = Column(Float, nullable=True, comment="最低售价") max_price = Column(Float, nullable=True, comment="最高售价") latest_price = Column(Float, nullable=True, comment="最新售价") # 销量统计 total_quantity = Column(Integer, default=0, comment="总销量") total_orders = Column(Integer, default=0, comment="订单数量") total_revenue = Column(Float, default=0.0, comment="总收入") # 时间信息 first_order_time = Column(DateTime, nullable=True, comment="首次订单时间") last_order_time = Column(DateTime, nullable=True, comment="最后订单时间") # 采购需求 pending_procurement_qty = Column(Integer, default=0, comment="待采购数量") total_procurement_qty = Column(Integer, default=0, comment="总采购数量") # 优先级信息 is_urgent = Column(Boolean, default=False, comment="是否紧急") priority_score = Column(Integer, default=50, comment="优先级分数 0-100") # 状态管理 status = Column(SQLEnum(ProductStatus), default=ProductStatus.ACTIVE, comment="产品状态") is_active = Column(Boolean, default=True, comment="是否激活") # 系统字段 created_at = Column(DateTime, server_default=func.now(), comment="创建时间") updated_at = Column(DateTime, onupdate=func.now(), comment="更新时间") last_aggregated_at = Column(DateTime, nullable=True, comment="最后聚合时间") # 数据来源追踪 source_order_count = Column(Integer, default=0, comment="来源订单数量") last_source_order_id = Column(Integer, nullable=True, comment="最后关联的原始订单ID") # 备注信息 manual_notes = Column(Text, nullable=True, comment="人工备注") auto_classification = Column(String(100), nullable=True, comment="自动分类结果") # 索引优化 __table_args__ = ( # 核心查询索引 Index('idx_products_master_brand', '品牌'), Index('idx_products_master_procurement', 'procurement_method'), Index('idx_products_master_active_status', 'is_active', 'status'), Index('idx_products_master_urgent_priority', 'is_urgent', 'priority_score'), # 时间相关索引 Index('idx_products_master_order_time', 'last_order_time'), Index('idx_products_master_created', 'created_at'), # 采购相关索引 Index('idx_products_master_pending', 'pending_procurement_qty', 'procurement_method'), Index('idx_products_master_brand_method', '品牌', 'procurement_method'), # 销售统计索引 Index('idx_products_master_quantity', 'total_quantity', 'last_order_time'), Index('idx_products_master_revenue', 'total_revenue', '品牌'), # 图片状态索引 Index('idx_products_master_image_status', 'image_download_status'), # 动态品牌查询优化复合索引 Index('idx_products_master_dynamic_brand', 'procurement_method', '品牌', 'is_active'), Index('idx_products_master_brand_active', '品牌', 'is_active', 'main_image_path'), {"comment": "产品主表 - 系统核心产品数据"} ) def __repr__(self): return f"" class ProductSourceMapping(Base): """ 产品来源映射表 - 追踪产品主表与原始订单的关联关系 用于数据溯源和调试 """ __tablename__ = "product_source_mapping" id = Column(Integer, primary_key=True, index=True) product_master_id = Column(Integer, nullable=False, comment="产品主表ID") raw_order_id = Column(Integer, nullable=False, comment="原始订单ID") contribution_quantity = Column(Integer, default=1, comment="该订单对产品的数量贡献") contribution_revenue = Column(Float, default=0.0, comment="该订单对产品的收入贡献") # 时间信息 mapped_at = Column(DateTime, server_default=func.now(), comment="映射时间") # 索引 __table_args__ = ( Index('idx_product_source_product', 'product_master_id'), Index('idx_product_source_order', 'raw_order_id'), UniqueConstraint('product_master_id', 'raw_order_id', name='uk_product_source_unique'), {"comment": "产品来源映射表"} ) class ImageDownloadLog(Base): """ 图片下载日志表 - 记录图片下载过程和状态 """ __tablename__ = "image_download_log" id = Column(Integer, primary_key=True, index=True) product_master_id = Column(Integer, nullable=False, comment="产品主表ID") sku_key = Column(String(64), nullable=False, comment="产品SKU键") # 下载信息 original_url = Column(String(1000), nullable=False, comment="原始图片URL") local_file_path = Column(String(500), nullable=True, comment="本地文件路径") file_size = Column(Integer, nullable=True, comment="文件大小(字节)") file_name = Column(String(255), nullable=True, comment="文件名") # 状态信息 download_status = Column(String(20), default="pending", comment="下载状态: pending/downloading/completed/failed") error_message = Column(Text, nullable=True, comment="错误信息") retry_count = Column(Integer, default=0, comment="重试次数") # 时间信息 download_started_at = Column(DateTime, nullable=True, comment="开始下载时间") download_completed_at = Column(DateTime, nullable=True, comment="下载完成时间") created_at = Column(DateTime, server_default=func.now(), comment="记录创建时间") # 索引 __table_args__ = ( Index('idx_image_download_product', 'product_master_id'), Index('idx_image_download_status', 'download_status', 'created_at'), Index('idx_image_download_retry', 'retry_count', 'download_status'), {"comment": "图片下载日志表"} )