""" 新的采购清单管理服务 基于待采购订单表(procurement_orders)生成SKU级别的采购清单 """ import logging import json from datetime import datetime from decimal import Decimal from typing import Dict, List, Optional, Any from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func, and_, or_, desc, asc, text from sqlalchemy.orm import selectinload from app.models.procurement_orders import ProcurementOrder, ProcurementStatus, ProcurementMethod from app.utils.sql_security import safe_like_param logger = logging.getLogger(__name__) class ProcurementListService: """基于待采购订单表的采购清单管理服务""" def _parse_image_urls(self, image_urls_json: str) -> List[str]: """解析图片URL JSON字符串""" if not image_urls_json: return [] try: # 第一次解析：去掉外层的JSON字符串包装 first_parse = json.loads(image_urls_json) # 如果第一次解析的结果是字符串，说明有双重编码，需要再解析一次 if isinstance(first_parse, str): urls = json.loads(first_parse) else: urls = first_parse # 处理解析结果 if isinstance(urls, list): return urls elif isinstance(urls, str): return [urls] else: return [] except (json.JSONDecodeError, TypeError): # 如果解析失败，尝试直接返回字符串 if isinstance(image_urls_json, str) and image_urls_json.startswith('http'): return [image_urls_json] return [] async def get_procurement_list_with_status( self, db: AsyncSession, procurement_method: Optional[str] = None, brand: Optional[str] = None, urgent_only: bool = False, payment_time_from: Optional[str] = None, include_all_status: bool = True, page: int = 1, page_size: int = 50 ) -> Dict[str, Any]: """ 获取采购清单（按SKU聚合，支持多状态统计） 直接从procurement_orders表（待采购订单列表）聚合数据，按SKU分组统计各状态数量 确保与待采购订单列表实时同步 """ # 构建基础查询条件（不限制status，获取所有状态） conditions = ["1=1"] # 基础条件，包含所有状态 params = {} if procurement_method: conditions.append("procurement_method = :procurement_method") params["procurement_method"] = procurement_method if brand: conditions.append("品牌 LIKE :brand") params["brand"] = safe_like_param(brand) if urgent_only: conditions.append("is_urgent = true") if payment_time_from: conditions.append("付款时间 >= :payment_time_from") params["payment_time_from"] = payment_time_from where_clause = " AND ".join(conditions) # SKU聚合查询 - 按商品名称+销售属性聚合，包含各状态数量统计 aggregation_query = text(f""" SELECT 线上宝贝名称, 线上销售属性, 品牌, 货号, 颜色, 尺寸, procurement_method, COUNT(*) as total_order_count, SUM(数量) as total_quantity, -- 按状态分组统计数量 SUM(CASE WHEN procurement_status = 'PENDING' THEN 数量 ELSE 0 END) as pending_quantity, SUM(CASE WHEN procurement_status = 'ORDERED' THEN 数量 ELSE 0 END) as ordered_quantity, SUM(CASE WHEN procurement_status = 'RECEIVED' THEN 数量 ELSE 0 END) as received_quantity, SUM(CASE WHEN procurement_status = 'CANCELLED' THEN 数量 ELSE 0 END) as cancelled_quantity, -- 按状态统计订单数 COUNT(CASE WHEN procurement_status = 'PENDING' THEN 1 END) as pending_order_count, COUNT(CASE WHEN procurement_status = 'ORDERED' THEN 1 END) as ordered_order_count, COUNT(CASE WHEN procurement_status = 'RECEIVED' THEN 1 END) as received_order_count, COUNT(CASE WHEN procurement_status = 'CANCELLED' THEN 1 END) as cancelled_order_count, AVG(订单单价) as avg_price, MIN(订单单价) as min_price, MAX(订单单价) as max_price, SUM(订单金额) as total_amount, AVG(建议采购价) as avg_suggested_price, MIN(付款时间) as earliest_payment, MAX(付款时间) as latest_payment, AVG(priority_score) as avg_priority, -- 紧急程度统计 CASE WHEN COUNT(CASE WHEN is_urgent = 1 THEN 1 END) > 0 THEN 1 ELSE 0 END as has_urgent, COUNT(CASE WHEN is_urgent = 1 THEN 1 END) as urgent_order_count, -- 最新状态时间 MAX(更新时间) as last_updated_at FROM procurement_orders WHERE {where_clause} GROUP BY 线上宝贝名称, 线上销售属性, 品牌, 货号, 颜色, 尺寸, procurement_method HAVING ( CASE WHEN :include_all_status = 1 THEN 1 ELSE (pending_quantity > 0) END ) = 1 ORDER BY has_urgent DESC, pending_quantity DESC, avg_priority DESC, total_quantity DESC, earliest_payment ASC LIMIT :limit OFFSET :offset """) # 计算总数的查询（包含having条件） count_query = text(f""" SELECT COUNT(*) as total FROM ( SELECT 线上宝贝名称, 线上销售属性, SUM(CASE WHEN procurement_status = 'PENDING' THEN 数量 ELSE 0 END) as pending_quantity FROM procurement_orders WHERE {where_clause} GROUP BY 线上宝贝名称, 线上销售属性 HAVING ( CASE WHEN :include_all_status = 1 THEN 1 ELSE (pending_quantity > 0) END ) = 1 ) as grouped """) # 添加参数 params["include_all_status"] = 1 if include_all_status else 0 # 执行总数查询 count_result = await db.execute(count_query, params) total = count_result.scalar() # 执行分页聚合查询 offset = (page - 1) * page_size params.update({"limit": page_size, "offset": offset}) result = await db.execute(aggregation_query, params) procurement_items = result.fetchall() # 转换结果 items = [] for item in procurement_items: item_dict = dict(item._mapping) # 生成SKU标识 sku_key = f"{item_dict['线上宝贝名称']}|{item_dict['线上销售属性'] or ''}" item_dict['sku_key'] = sku_key # 计算建议采购量（基于待采购数量） pending_qty = item_dict['pending_quantity'] or 0 item_dict['suggested_quantity'] = pending_qty # 计算预估采购成本（基于待采购数量） if item_dict['avg_suggested_price'] and pending_qty > 0: estimated_cost = Decimal(str(item_dict['avg_suggested_price'])) * pending_qty item_dict['estimated_cost'] = float(estimated_cost) else: # 如果没有建议价，使用平均售价的70%作为估算 avg_price = item_dict['avg_price'] or 0 estimated_cost = Decimal(str(avg_price)) * Decimal('0.7') * pending_qty item_dict['estimated_cost'] = float(estimated_cost) # 计算采购进度百分比 total_qty = item_dict['total_quantity'] or 0 received_qty = item_dict['received_quantity'] or 0 if total_qty > 0: item_dict['completion_rate'] = round((received_qty / total_qty) * 100, 1) else: item_dict['completion_rate'] = 0.0 items.append(item_dict) return { "items": items, "pagination": { "page": page, "page_size": page_size, "total": total or 0, "pages": (total + page_size - 1) // page_size if total else 0 } } async def get_procurement_list( self, db: AsyncSession, procurement_method: Optional[str] = None, brand: Optional[str] = None, urgent_only: bool = False, payment_time_from: Optional[str] = None, page: int = 1, page_size: int = 50 ) -> Dict[str, Any]: """ 获取待采购清单（按SKU聚合）- 兼容性方法 基于procurement_orders表生成SKU级别的采购清单 """ return await self.get_procurement_list_with_status( db=db, procurement_method=procurement_method, brand=brand, urgent_only=urgent_only, payment_time_from=payment_time_from, include_all_status=False, # 只显示有待采购数量的项目 page=page, page_size=page_size ) async def batch_update_procurement_status( self, db: AsyncSession, order_ids: List[int], new_status: str, supplier: Optional[str] = None, purchase_order_no: Optional[str] = None, notes: Optional[str] = None ) -> Dict[str, Any]: """ 批量更新采购订单状态 """ try: if not order_ids: return {"success": False, "error": "No order IDs provided"} # 验证状态值 valid_statuses = ['PENDING', 'ORDERED', 'RECEIVED', 'CANCELLED'] if new_status not in valid_statuses: return {"success": False, "error": f"Invalid status: {new_status}"} # 构建更新语句 update_fields = [ "procurement_status = :new_status", "更新时间 = :updated_at" ] update_params = { "new_status": new_status, "updated_at": datetime.now() } # 根据新状态添加相关字段 if new_status == 'ORDERED': if supplier: update_fields.append("采购供应商 = :supplier") update_params["supplier"] = supplier if purchase_order_no: update_fields.append("procurement_notes = :purchase_order_no") update_params["purchase_order_no"] = f"采购单号: {purchase_order_no}" if notes: # 将notes添加到procurement_notes字段中 current_notes = f"备注: {notes}" if purchase_order_no and new_status == 'ORDERED': current_notes = f"采购单号: {purchase_order_no}; 备注: {notes}" update_fields.append("procurement_notes = :notes") update_params["notes"] = current_notes # 执行批量更新 - 修复IN子句语法 placeholders = ','.join([f':id_{i}' for i in range(len(order_ids))]) update_query = text(f""" UPDATE procurement_orders SET {', '.join(update_fields)} WHERE id IN ({placeholders}) """) # 添加每个订单ID作为独立参数 for i, order_id in enumerate(order_ids): update_params[f'id_{i}'] = order_id result = await db.execute(update_query, update_params) await db.commit() updated_count = result.rowcount logger.info(f"Batch updated {updated_count} orders to status {new_status}") return { "success": True, "updated_orders": updated_count, "message": f"Successfully updated {updated_count} orders to {new_status}", "new_status": new_status } except Exception as e: await db.rollback() logger.error(f"Failed to batch update procurement status: {e}") return { "success": False, "error": str(e) } async def get_orders_by_sku_key( self, db: AsyncSession, sku_key: str, procurement_method: Optional[str] = None, status_filter: Optional[str] = None ) -> Dict[str, Any]: """ 根据SKU键获取相关订单，支持状态筛选 """ try: # 解析SKU键 parts = sku_key.split('|', 1) product_name = parts[0] sales_attrs = parts[1] if len(parts) > 1 else '' # 构建查询条件 conditions = [ "线上宝贝名称 = :product_name", "(线上销售属性 = :sales_attrs OR (线上销售属性 IS NULL AND :sales_attrs = ''))" ] params = { "product_name": product_name, "sales_attrs": sales_attrs } if procurement_method: conditions.append("procurement_method = :procurement_method") params["procurement_method"] = procurement_method if status_filter: conditions.append("procurement_status = :status_filter") params["status_filter"] = status_filter where_clause = " AND ".join(conditions) # 查询订单 query = text(f""" SELECT id, 原始订单编号, 线上宝贝名称, 线上销售属性, 品牌, 货号, 颜色, 尺寸, procurement_method, 数量, 订单单价, 订单金额, 付款时间, 交易状态, procurement_status, is_urgent, priority_score, 采购供应商, procurement_notes, 更新时间 FROM procurement_orders WHERE {where_clause} ORDER BY procurement_status ASC, 付款时间 DESC, is_urgent DESC, priority_score DESC """) result = await db.execute(query, params) orders = result.fetchall() # 按状态分组统计 status_stats = {} for order in orders: status = order.procurement_status if status not in status_stats: status_stats[status] = {"count": 0, "quantity": 0, "amount": 0} status_stats[status]["count"] += 1 status_stats[status]["quantity"] += order.数量 status_stats[status]["amount"] += float(order.订单金额 or 0) return { "sku_key": sku_key, "orders": [dict(order._mapping) for order in orders], "total_orders": len(orders), "status_statistics": status_stats } except Exception as e: logger.error(f"Failed to get orders by SKU key {sku_key}: {e}") return {"error": str(e)} async def get_procurement_list_stats_with_status(self, db: AsyncSession) -> Dict[str, Any]: """获取采购清单统计信息（包含状态分布）""" # 基础统计查询 - 包含所有状态 basic_stats_query = text(""" SELECT COUNT(*) as total_orders, (SELECT COUNT(*) FROM ( SELECT DISTINCT 线上宝贝名称, 线上销售属性 FROM procurement_orders )) as unique_skus, SUM(数量) as total_quantity, SUM(订单金额) as total_amount, COUNT(DISTINCT 品牌) as total_brands, COUNT(CASE WHEN is_urgent = 1 THEN 1 END) as urgent_orders, -- 按状态分组统计 COUNT(CASE WHEN procurement_status = 'PENDING' THEN 1 END) as pending_orders, COUNT(CASE WHEN procurement_status = 'ORDERED' THEN 1 END) as ordered_orders, COUNT(CASE WHEN procurement_status = 'RECEIVED' THEN 1 END) as received_orders, COUNT(CASE WHEN procurement_status = 'CANCELLED' THEN 1 END) as cancelled_orders, SUM(CASE WHEN procurement_status = 'PENDING' THEN 数量 ELSE 0 END) as pending_quantity, SUM(CASE WHEN procurement_status = 'ORDERED' THEN 数量 ELSE 0 END) as ordered_quantity, SUM(CASE WHEN procurement_status = 'RECEIVED' THEN 数量 ELSE 0 END) as received_quantity, SUM(CASE WHEN procurement_status = 'CANCELLED' THEN 数量 ELSE 0 END) as cancelled_quantity, SUM(CASE WHEN procurement_status = 'PENDING' THEN 订单金额 ELSE 0 END) as pending_amount, SUM(CASE WHEN procurement_status = 'ORDERED' THEN 订单金额 ELSE 0 END) as ordered_amount, SUM(CASE WHEN procurement_status = 'RECEIVED' THEN 订单金额 ELSE 0 END) as received_amount, SUM(CASE WHEN procurement_status = 'CANCELLED' THEN 订单金额 ELSE 0 END) as cancelled_amount FROM procurement_orders """) basic_result = await db.execute(basic_stats_query) basic_stats = basic_result.fetchone() # 按采购方式和状态统计 method_status_stats_query = text(""" SELECT procurement_method, procurement_status, COUNT(*) as order_count, SUM(数量) as total_quantity, SUM(订单金额) as total_amount, (SELECT COUNT(*) FROM ( SELECT DISTINCT 线上宝贝名称, 线上销售属性 FROM procurement_orders p2 WHERE p2.procurement_method = p1.procurement_method AND p2.procurement_status = p1.procurement_status )) as sku_count FROM procurement_orders p1 GROUP BY procurement_method, procurement_status ORDER BY procurement_method ASC, procurement_status ASC """) method_result = await db.execute(method_status_stats_query) method_status_stats = method_result.fetchall() # 按品牌和状态统计（TOP 10 品牌） brand_status_stats_query = text(""" SELECT 品牌, procurement_status, COUNT(*) as order_count, SUM(数量) as total_quantity, SUM(订单金额) as total_amount FROM procurement_orders WHERE 品牌 IS NOT NULL AND 品牌 IN ( SELECT 品牌 FROM procurement_orders WHERE 品牌 IS NOT NULL GROUP BY 品牌 ORDER BY SUM(订单金额) DESC LIMIT 10 ) GROUP BY 品牌, procurement_status ORDER BY 品牌 ASC, procurement_status ASC """) brand_result = await db.execute(brand_status_stats_query) brand_status_stats = brand_result.fetchall() return { "basic_stats": { "total_orders": basic_stats.total_orders or 0, "unique_skus": basic_stats.unique_skus or 0, "total_quantity": basic_stats.total_quantity or 0, "total_amount": float(basic_stats.total_amount or 0), "total_brands": basic_stats.total_brands or 0, "urgent_orders": basic_stats.urgent_orders or 0 }, "status_distribution": { "orders": { "PENDING": basic_stats.pending_orders or 0, "ORDERED": basic_stats.ordered_orders or 0, "RECEIVED": basic_stats.received_orders or 0, "CANCELLED": basic_stats.cancelled_orders or 0 }, "quantities": { "PENDING": basic_stats.pending_quantity or 0, "ORDERED": basic_stats.ordered_quantity or 0, "RECEIVED": basic_stats.received_quantity or 0, "CANCELLED": basic_stats.cancelled_quantity or 0 }, "amounts": { "PENDING": float(basic_stats.pending_amount or 0), "ORDERED": float(basic_stats.ordered_amount or 0), "RECEIVED": float(basic_stats.received_amount or 0), "CANCELLED": float(basic_stats.cancelled_amount or 0) } }, "method_status_distribution": [ { "method": row.procurement_method, "status": row.procurement_status, "order_count": row.order_count, "sku_count": row.sku_count, "total_quantity": row.total_quantity, "total_amount": float(row.total_amount or 0) } for row in method_status_stats ], "brand_status_distribution": [ { "brand": row.品牌, "status": row.procurement_status, "order_count": row.order_count, "total_quantity": row.total_quantity, "total_amount": float(row.total_amount or 0) } for row in brand_status_stats ] } async def get_procurement_list_stats(self, db: AsyncSession) -> Dict[str, Any]: """获取采购清单统计信息（兼容现有接口）""" # 调用新的增强统计方法并转换为兼容格式 enhanced_stats = await self.get_procurement_list_stats_with_status(db) # 转换为原有格式，保持向后兼容 return { "basic_stats": enhanced_stats["basic_stats"], "method_distribution": [ { "method": item["method"], "order_count": sum(s["order_count"] for s in enhanced_stats["method_status_distribution"] if s["method"] == item["method"]), "sku_count": sum(s["sku_count"] for s in enhanced_stats["method_status_distribution"] if s["method"] == item["method"]), "total_quantity": sum(s["total_quantity"] for s in enhanced_stats["method_status_distribution"] if s["method"] == item["method"]), "total_amount": sum(s["total_amount"] for s in enhanced_stats["method_status_distribution"] if s["method"] == item["method"]) } for item in {{"method": s["method"]} for s in enhanced_stats["method_status_distribution"]} ], "top_brands": [ { "brand": item["brand"], "order_count": sum(s["order_count"] for s in enhanced_stats["brand_status_distribution"] if s["brand"] == item["brand"]), "total_quantity": sum(s["total_quantity"] for s in enhanced_stats["brand_status_distribution"] if s["brand"] == item["brand"]), "total_amount": sum(s["total_amount"] for s in enhanced_stats["brand_status_distribution"] if s["brand"] == item["brand"]) } for item in {{"brand": s["brand"]} for s in enhanced_stats["brand_status_distribution"]} ][:10], # 新增字段：状态分布信息 "status_distribution": enhanced_stats["status_distribution"], "method_status_distribution": enhanced_stats["method_status_distribution"], "brand_status_distribution": enhanced_stats["brand_status_distribution"] } async def get_sku_details( self, db: AsyncSession, sku_key: str ) -> Dict[str, Any]: """ 获取特定SKU的详细信息 sku_key格式: "商品名称|销售属性" """ try: # 解析SKU键 parts = sku_key.split('|', 1) product_name = parts[0] sales_attrs = parts[1] if len(parts) > 1 else '' # 查询该SKU的所有待采购订单 query = text(""" SELECT * FROM procurement_orders WHERE procurement_status = 'PENDING' AND 线上宝贝名称 = :product_name AND (线上销售属性 = :sales_attrs OR (线上销售属性 IS NULL AND :sales_attrs = '')) ORDER BY 付款时间 DESC, is_urgent DESC, priority_score DESC """) result = await db.execute(query, { "product_name": product_name, "sales_attrs": sales_attrs }) orders = result.fetchall() if not orders: return {"error": "SKU not found"} # 聚合信息 total_quantity = sum(order.数量 for order in orders) total_amount = sum(order.订单金额 for order in orders) avg_price = total_amount / total_quantity if total_quantity > 0 else 0 # 采购方式分布 method_distribution = {} for order in orders: method = order.procurement_method if method not in method_distribution: method_distribution[method] = {"count": 0, "quantity": 0} method_distribution[method]["count"] += 1 method_distribution[method]["quantity"] += order.数量 # 时间分析 payment_times = [order.付款时间 for order in orders if order.付款时间] earliest_payment = min(payment_times) if payment_times else None latest_payment = max(payment_times) if payment_times else None # 紧急订单统计 urgent_count = sum(1 for order in orders if order.is_urgent) return { "sku_key": sku_key, "product_info": { "线上宝贝名称": orders[0].线上宝贝名称, "线上销售属性": orders[0].线上销售属性, "品牌": orders[0].品牌, "货号": orders[0].货号, "颜色": orders[0].颜色, "尺寸": orders[0].尺寸 }, "aggregated_stats": { "total_orders": len(orders), "total_quantity": total_quantity, "total_amount": float(total_amount), "avg_price": float(avg_price), "urgent_orders": urgent_count, "earliest_payment": earliest_payment, "latest_payment": latest_payment }, "method_distribution": method_distribution, "orders": [dict(order._mapping) for order in orders] } except Exception as e: logger.error(f"Failed to get SKU details for {sku_key}: {e}") return {"error": str(e)} async def update_procurement_priority( self, db: AsyncSession, sku_key: str, priority_score: int, is_urgent: bool = False, notes: Optional[str] = None ) -> Dict[str, Any]: """ 更新SKU的采购优先级 """ try: # 解析SKU键 parts = sku_key.split('|', 1) product_name = parts[0] sales_attrs = parts[1] if len(parts) > 1 else '' # 更新该SKU的所有待采购订单 update_query = text(""" UPDATE procurement_orders SET priority_score = :priority_score, is_urgent = :is_urgent, manual_note = CASE WHEN :notes IS NOT NULL THEN :notes ELSE manual_note END, 更新时间 = :updated_at WHERE procurement_status = 'PENDING' AND 线上宝贝名称 = :product_name AND (线上销售属性 = :sales_attrs OR (线上销售属性 IS NULL AND :sales_attrs = '')) """) result = await db.execute(update_query, { "priority_score": priority_score, "is_urgent": is_urgent, "notes": notes, "updated_at": datetime.now(), "product_name": product_name, "sales_attrs": sales_attrs }) await db.commit() updated_count = result.rowcount logger.info(f"Updated priority for SKU {sku_key}: {updated_count} orders affected") return { "success": True, "updated_orders": updated_count, "message": f"Successfully updated priority for {updated_count} orders" } except Exception as e: await db.rollback() logger.error(f"Failed to update priority for SKU {sku_key}: {e}") return { "success": False, "error": str(e) } async def mark_as_ordered( self, db: AsyncSession, order_ids: List[int], supplier: Optional[str] = None, purchase_order_no: Optional[str] = None, notes: Optional[str] = None ) -> Dict[str, Any]: """ 标记订单为已采购状态 """ try: if not order_ids: return {"success": False, "error": "No order IDs provided"} # 更新订单状态 update_query = text(""" UPDATE procurement_orders SET procurement_status = 'ORDERED', supplier_name = :supplier, purchase_order_no = :purchase_order_no, manual_note = CASE WHEN :notes IS NOT NULL THEN :notes ELSE manual_note END, ordered_at = :ordered_at, 更新时间 = :updated_at WHERE id IN :order_ids AND procurement_status = 'PENDING' """) result = await db.execute(update_query, { "supplier": supplier, "purchase_order_no": purchase_order_no, "notes": notes, "ordered_at": datetime.now(), "updated_at": datetime.now(), "order_ids": tuple(order_ids) }) await db.commit() updated_count = result.rowcount logger.info(f"Marked {updated_count} orders as ORDERED") return { "success": True, "updated_orders": updated_count, "message": f"Successfully marked {updated_count} orders as ordered" } except Exception as e: await db.rollback() logger.error(f"Failed to mark orders as ordered: {e}") return { "success": False, "error": str(e) } async def get_product_aggregated_list( self, db: AsyncSession, procurement_method: str, brand: Optional[str] = None, product_name: Optional[str] = None, urgent_only: bool = False, payment_time_from: Optional[str] = None, page: int = 1, page_size: int = 50 ) -> Dict[str, Any]: """ 获取按商品名称汇总的采购清单 """ # 构建基础查询条件（带表别名） conditions = ["po.procurement_status = 'PENDING'", "po.procurement_method = :procurement_method"] params = {"procurement_method": procurement_method} if brand: conditions.append("po.品牌 LIKE :brand") params["brand"] = safe_like_param(brand) if product_name: conditions.append("po.线上宝贝名称 LIKE :product_name") params["product_name"] = safe_like_param(product_name) if urgent_only: conditions.append("po.is_urgent = true") if payment_time_from: conditions.append("po.付款时间 >= :payment_time_from") params["payment_time_from"] = payment_time_from where_clause = " AND ".join(conditions) # 主查询：按商品名称汇总 main_query = f""" SELECT po.线上宝贝名称 as product_name, po.品牌 as brand, COUNT(DISTINCT CONCAT(po.线上销售属性, '-', po.颜色, '-', po.尺寸)) as sku_count, COUNT(*) as total_orders, SUM(po.数量) as total_quantity, AVG(po.订单单价) as avg_price, SUM(po.订单单价 * po.数量) as total_cost, MAX(CASE WHEN po.is_urgent = true THEN 1 ELSE 0 END) as has_urgent FROM procurement_orders po WHERE {where_clause} GROUP BY po.线上宝贝名称, po.品牌 ORDER BY total_cost DESC LIMIT :limit OFFSET :offset """ # 总数查询 count_query = f""" SELECT COUNT(DISTINCT po.线上宝贝名称) as total FROM procurement_orders po WHERE {where_clause} """ params["limit"] = page_size params["offset"] = (page - 1) * page_size try: # 获取商品列表 result = await db.execute(text(main_query), params) products = result.fetchall() # 获取总数 count_result = await db.execute(text(count_query), params) total = count_result.scalar() # 获取每个商品的SKU详情 product_list = [] for product in products: # 获取该商品的所有SKU详情，包含图片信息 sku_query = f""" SELECT po.线上销售属性 as sales_attr, po.颜色 as color, po.尺寸 as size, po.货号 as product_code, COUNT(*) as order_count, SUM(po.数量) as total_quantity, AVG(po.订单单价) as avg_price, SUM(po.订单单价 * po.数量) as estimated_cost, MAX(CASE WHEN po.is_urgent = true THEN 1 ELSE 0 END) as has_urgent, norm.图片链接 as image_urls FROM procurement_orders po LEFT JOIN order_items_norm norm ON po.original_order_id = norm.id WHERE {where_clause} AND po.线上宝贝名称 = :product_name GROUP BY po.线上销售属性, po.颜色, po.尺寸, po.货号, norm.图片链接 ORDER BY estimated_cost DESC """ sku_params = dict(params) sku_params["product_name"] = product.product_name sku_result = await db.execute(text(sku_query), sku_params) sku_details = sku_result.fetchall() product_item = { "product_name": product.product_name, "brand": product.brand, "sku_count": product.sku_count, "total_orders": product.total_orders, "total_quantity": product.total_quantity, "avg_price": float(product.avg_price) if product.avg_price else 0, "total_cost": float(product.total_cost) if product.total_cost else 0, "has_urgent": bool(product.has_urgent), "sku_details": [ { "sales_attr": sku.sales_attr, "color": sku.color, "size": sku.size, "product_code": sku.product_code, "order_count": sku.order_count, "total_quantity": sku.total_quantity, "avg_price": float(sku.avg_price) if sku.avg_price else 0, "estimated_cost": float(sku.estimated_cost) if sku.estimated_cost else 0, "has_urgent": bool(sku.has_urgent), "image_urls": self._parse_image_urls(sku.image_urls) if sku.image_urls else [] } for sku in sku_details ] } product_list.append(product_item) return { "items": product_list, "pagination": { "page": page, "page_size": page_size, "total": total, "pages": (total + page_size - 1) // page_size } } except Exception as e: logger.error(f"获取商品汇总清单失败: {e}") raise e async def get_product_orders( self, db: AsyncSession, product_name: str, procurement_method: str ) -> Dict[str, Any]: """ 获取特定商品的相关订单 """ query = """ SELECT 原始订单编号, 线上销售属性, 颜色, 尺寸, 数量, 订单单价, 订单金额, 付款时间, 交易状态, is_urgent FROM procurement_orders WHERE 线上宝贝名称 = :product_name AND procurement_method = :procurement_method AND procurement_status = 'PENDING' ORDER BY 付款时间 DESC """ try: result = await db.execute(text(query), { "product_name": product_name, "procurement_method": procurement_method }) orders = result.fetchall() order_list = [ { "原始订单编号": order.原始订单编号, "线上销售属性": order.线上销售属性, "颜色": order.颜色, "尺寸": order.尺寸, "数量": order.数量, "订单单价": float(order.订单单价) if order.订单单价 else 0, "订单金额": float(order.订单金额) if order.订单金额 else 0, "付款时间": order.付款时间.isoformat() if order.付款时间 else None, "交易状态": order.交易状态, "is_urgent": bool(order.is_urgent) } for order in orders ] return { "orders": order_list, "total": len(order_list) } except Exception as e: logger.error(f"获取商品订单失败: {e}") raise e