""" 采购清单管理服务 V2 - 基于产品主表的优化版本 直接从产品主表获取属性信息，提升查询性能和数据一致性 """ 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.products_master import ProductMaster from app.models.procurement_orders import ProcurementOrder, ProcurementStatus, ProcurementMethod from app.services.image_download_service import ImageDownloadService from app.utils.sql_security import safe_like_param logger = logging.getLogger(__name__) class ProcurementListServiceV2: """基于产品主表的采购清单管理服务""" def __init__(self): self.image_service = ImageDownloadService() def _parse_image_urls(self, image_urls_json: str) -> List[str]: """解析图片URL JSON字符串""" if not image_urls_json: return [] try: # 解析JSON字符串 urls = json.loads(image_urls_json) if isinstance(urls, list): return [url for url in urls if url and url.strip()] elif isinstance(urls, str) and urls.strip(): return [urls.strip()] else: return [] except (json.JSONDecodeError, TypeError): # 如果解析失败，尝试直接返回字符串 if isinstance(image_urls_json, str) and image_urls_json.startswith('http'): return [image_urls_json] return [] def _get_local_image_urls(self, local_image_paths: str) -> List[str]: """获取本地图片URL""" if not local_image_paths: return [] try: paths = json.loads(local_image_paths) if isinstance(paths, list): return [self.image_service.get_local_image_url(path) for path in paths if path] elif isinstance(paths, str): return [self.image_service.get_local_image_url(paths)] return [] except (json.JSONDecodeError, TypeError): return [] 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聚合）- V2优化版本 基于产品主表获取属性信息，确保数据一致性 """ # 构建基础查询条件 conditions = [] params = {} # 基础条件：只查询有采购需求的产品 base_conditions = [ ProductMaster.is_active == True, ProductMaster.pending_procurement_qty > 0 ] # 采购方式过滤 if procurement_method: base_conditions.append(ProductMaster.procurement_method == procurement_method) # 品牌过滤 if brand: base_conditions.append(ProductMaster.品牌.like(f"%{brand}%")) # 紧急订单过滤 if urgent_only: base_conditions.append(ProductMaster.is_urgent == True) # 时间过滤 - 基于最后订单时间 if payment_time_from: try: time_filter = datetime.fromisoformat(payment_time_from.replace('Z', '+00:00')) base_conditions.append(ProductMaster.last_order_time >= time_filter) except ValueError: logger.warning(f"Invalid time format: {payment_time_from}") # 构建主查询 query = (select(ProductMaster) .where(and_(*base_conditions)) .order_by(ProductMaster.pending_procurement_qty.desc(), ProductMaster.priority_score.desc())) # 分页 offset = (page - 1) * page_size paginated_query = query.offset(offset).limit(page_size) # 执行查询 result = await db.execute(paginated_query) products = result.scalars().all() # 获取总数 count_query = select(func.count(ProductMaster.id)).where(and_(*base_conditions)) count_result = await db.execute(count_query) total_count = count_result.scalar() # 转换为采购清单格式 procurement_items = [] for product in products: # 获取相关的采购订单来计算订单统计 order_stats = await self._get_product_order_stats(db, product.sku_key) # 处理图片URL image_urls = [] if product.image_download_status == 'completed' and product.local_image_paths: # 优先使用本地图片 image_urls = self._get_local_image_urls(product.local_image_paths) elif product.original_image_urls: # 备选：使用原始图片URL image_urls = self._parse_image_urls(product.original_image_urls) procurement_item = { 'sku_key': product.sku_key, 'product_id': product.id, '线上宝贝名称': product.线上宝贝名称, '线上销售属性': product.线上销售属性, '品牌': product.品牌, '货号': product.货号, '颜色': product.颜色, '尺寸': product.尺寸, 'procurement_method': product.procurement_method.value if product.procurement_method else 'NY', # 数量和价格信息 'pending_procurement_qty': product.pending_procurement_qty, 'total_quantity': product.total_quantity, 'avg_price': float(product.avg_price) if product.avg_price else 0.0, 'latest_price': float(product.latest_price) if product.latest_price else 0.0, 'estimated_cost': float(product.pending_procurement_qty * (product.avg_price or 0)), # 订单统计 - 从采购订单表获取 'order_count': order_stats['order_count'], 'total_orders': product.total_orders, # 优先级和紧急状态 'is_urgent': product.is_urgent, 'priority_score': product.priority_score, # 时间信息 'first_order_time': product.first_order_time.isoformat() if product.first_order_time else None, 'last_order_time': product.last_order_time.isoformat() if product.last_order_time else None, # 图片信息 'image_urls': image_urls, 'image_download_status': product.image_download_status, 'has_local_images': product.image_download_status == 'completed', # 状态信息 'status': product.status.value if product.status else 'ACTIVE', 'created_at': product.created_at.isoformat() if product.created_at else None, 'updated_at': product.updated_at.isoformat() if product.updated_at else None } procurement_items.append(procurement_item) # 计算分页信息 total_pages = (total_count + page_size - 1) // page_size return { 'items': procurement_items, 'pagination': { 'page': page, 'page_size': page_size, 'total': total_count, 'total_pages': total_pages, 'has_next': page < total_pages, 'has_prev': page > 1 }, 'filters': { 'procurement_method': procurement_method, 'brand': brand, 'urgent_only': urgent_only, 'payment_time_from': payment_time_from } } async def _get_product_order_stats(self, db: AsyncSession, sku_key: str) -> Dict[str, Any]: """获取产品相关的采购订单统计""" result = await db.execute( select(func.count(ProcurementOrder.id)) .where(ProcurementOrder.product_key == sku_key) .where(ProcurementOrder.procurement_status == ProcurementStatus.PENDING) ) order_count = result.scalar() or 0 return { 'order_count': order_count } async def get_product_aggregated_list( self, db: AsyncSession, procurement_method: str, brand: Optional[str] = None, product_name: Optional[str] = None, shops: Optional[List[str]] = None, urgent_only: bool = False, payment_time_from: Optional[str] = None, page: int = 1, page_size: int = 50 ) -> Dict[str, Any]: """ 获取按商品名称聚合的采购清单 - V2优化版本 基于产品主表进行聚合，提升性能 """ # 构建查询条件 conditions = [ ProductMaster.is_active == True, ProductMaster.pending_procurement_qty > 0, ProductMaster.procurement_method == procurement_method ] if brand: conditions.append(ProductMaster.品牌.like(f"%{brand}%")) if product_name: conditions.append(ProductMaster.线上宝贝名称.like(f"%{product_name}%")) if shops and len(shops) > 0: # 由于网店名称是JSON数组，需要使用特殊的过滤方式 shop_conditions = [] for shop in shops: shop_conditions.append(func.json_contains(ProductMaster.网店名称, f'"{shop}"')) conditions.append(or_(*shop_conditions)) if urgent_only: conditions.append(ProductMaster.is_urgent == True) if payment_time_from: try: time_filter = datetime.fromisoformat(payment_time_from.replace('Z', '+00:00')) conditions.append(ProductMaster.last_order_time >= time_filter) except ValueError: logger.warning(f"Invalid time format: {payment_time_from}") # 使用SQL聚合查询按商品名称分组 aggregation_query = text(f""" SELECT 线上宝贝名称 as product_name, 品牌 as brand, procurement_method, COUNT(*) as sku_count, SUM(pending_procurement_qty) as total_pending_qty, SUM(total_quantity) as total_sold_qty, SUM(total_orders) as total_order_count, AVG(avg_price) as avg_price, SUM(pending_procurement_qty * COALESCE(avg_price, 0)) as estimated_cost, MAX(CASE WHEN is_urgent = 1 THEN 1 ELSE 0 END) as has_urgent, AVG(priority_score) as avg_priority, MAX(last_order_time) as latest_order_time, -- 获取第一个产品的图片信息作为代表 (优先使用已下载的本地图片) MAX(CASE WHEN main_image_path IS NOT NULL AND image_download_status = 'completed' THEN main_image_path WHEN local_image_paths IS NOT NULL THEN local_image_paths ELSE original_image_urls END) as sample_images, MAX(CASE WHEN main_image_path IS NOT NULL AND image_download_status = 'completed' THEN 1 ELSE 0 END) as has_local_images FROM products_master WHERE is_active = 1 AND pending_procurement_qty > 0 AND procurement_method = :procurement_method {f"AND 品牌 LIKE :brand" if brand else ""} {f"AND 线上宝贝名称 LIKE :product_name" if product_name else ""} {f"AND is_urgent = 1" if urgent_only else ""} {f"AND last_order_time >= :payment_time_from" if payment_time_from else ""} GROUP BY 线上宝贝名称, 品牌, procurement_method ORDER BY total_pending_qty DESC, estimated_cost DESC LIMIT :limit OFFSET :offset """) # 准备查询参数 query_params = { 'procurement_method': procurement_method, 'limit': page_size, 'offset': (page - 1) * page_size } if brand: query_params['brand'] = safe_like_param(brand) if product_name: query_params['product_name'] = safe_like_param(product_name) if payment_time_from: query_params['payment_time_from'] = payment_time_from # 执行聚合查询 result = await db.execute(aggregation_query, query_params) rows = result.fetchall() # 构建返回数据 aggregated_products = [] for row in rows: # 处理图片 - 优先使用本地已下载的图片 image_urls = [] if row.sample_images: if row.has_local_images and row.sample_images.startswith('images/'): # 本地图片路径，转换为静态文件URL filename = row.sample_images.replace('images/', '') image_urls = [f"/api/v1/static/images/{filename}"] elif row.sample_images.startswith('/'): # 其他本地图片路径 image_urls = self._get_local_image_urls(row.sample_images) else: # 原始图片URL image_urls = self._parse_image_urls(row.sample_images) product_item = { 'product_name': row.product_name, 'brand': row.brand, 'procurement_method': row.procurement_method, 'sku_count': row.sku_count, 'total_pending_qty': row.total_pending_qty, 'total_sold_qty': row.total_sold_qty, 'total_order_count': row.total_order_count, 'avg_price': float(row.avg_price) if row.avg_price else 0.0, 'estimated_cost': float(row.estimated_cost) if row.estimated_cost else 0.0, 'has_urgent': bool(row.has_urgent), 'avg_priority': float(row.avg_priority) if row.avg_priority else 50.0, 'latest_order_time': row.latest_order_time, 'image_urls': image_urls, 'has_local_images': bool(row.has_local_images) } aggregated_products.append(product_item) # 获取总数 count_query = text(f""" SELECT COUNT(DISTINCT 线上宝贝名称) FROM products_master WHERE is_active = 1 AND pending_procurement_qty > 0 AND procurement_method = :procurement_method {f"AND 品牌 LIKE :brand" if brand else ""} {f"AND 线上宝贝名称 LIKE :product_name" if product_name else ""} {f"AND is_urgent = 1" if urgent_only else ""} {f"AND last_order_time >= :payment_time_from" if payment_time_from else ""} """) count_params = {k: v for k, v in query_params.items() if k != 'limit' and k != 'offset'} count_result = await db.execute(count_query, count_params) total_count = count_result.scalar() total_pages = (total_count + page_size - 1) // page_size return { 'items': aggregated_products, 'pagination': { 'page': page, 'page_size': page_size, 'total': total_count, 'total_pages': total_pages, 'has_next': page < total_pages, 'has_prev': page > 1 }, 'filters': { 'procurement_method': procurement_method, 'brand': brand, 'product_name': product_name, 'urgent_only': urgent_only, 'payment_time_from': payment_time_from } } async def get_product_orders( self, db: AsyncSession, product_name: str, procurement_method: str ) -> Dict[str, Any]: """ 获取特定商品的相关订单（SKU级别详情）- V2优化版本 """ # 查询该商品名称下的所有SKU query = (select(ProductMaster) .where(ProductMaster.线上宝贝名称 == product_name) .where(ProductMaster.procurement_method == procurement_method) .where(ProductMaster.is_active == True) .where(ProductMaster.pending_procurement_qty > 0) .order_by(ProductMaster.pending_procurement_qty.desc())) result = await db.execute(query) products = result.scalars().all() # 构建SKU详情列表 sku_details = [] for product in products: # 处理图片 - 强制优先使用本地图片 image_urls = [] # 调试日志 print(f"处理SKU {product.sku_key} 图片:") print(f" - image_download_status: {product.image_download_status}") print(f" - main_image_path: {product.main_image_path}") print(f" - local_image_paths: {product.local_image_paths}") # 优先检查主图片路径 if product.main_image_path and product.image_download_status == 'completed': # 主图片已下载完成，使用本地主图片 if product.main_image_path.startswith('images/'): filename = product.main_image_path.replace('images/', '') image_urls = [f"/api/v1/static/images/{filename}"] print(f" - 使用main_image_path本地图片: {image_urls}") else: image_urls = [self.image_service.get_local_image_url(product.main_image_path)] print(f" - 使用main_image_path转换: {image_urls}") # 如果没有主图片，检查local_image_paths elif product.image_download_status == 'completed' and product.local_image_paths: image_urls = self._get_local_image_urls(product.local_image_paths) print(f" - 使用local_image_paths: {image_urls}") # 最后才使用原始图片URL elif product.original_image_urls: image_urls = self._parse_image_urls(product.original_image_urls) print(f" - 使用original_image_urls: {image_urls}") print(f" - 最终image_urls: {image_urls}") sku_detail = { 'sku_key': product.sku_key, 'product_id': product.id, 'sales_attr': product.线上销售属性, 'color': product.颜色, 'size': product.尺寸, 'product_code': product.货号, 'brand': product.品牌, 'pending_procurement_qty': product.pending_procurement_qty, 'total_quantity': product.total_quantity, 'order_count': product.total_orders, 'avg_price': float(product.avg_price) if product.avg_price else 0.0, 'estimated_cost': float(product.pending_procurement_qty * (product.avg_price or 0)), 'is_urgent': product.is_urgent, 'priority_score': product.priority_score, 'image_urls': image_urls, 'has_local_images': product.image_download_status == 'completed', 'last_order_time': product.last_order_time.isoformat() if product.last_order_time else None } sku_details.append(sku_detail) return { 'product_name': product_name, 'procurement_method': procurement_method, 'orders': sku_details, 'total_skus': len(sku_details), 'total_pending_qty': sum(item['pending_procurement_qty'] for item in sku_details), 'total_estimated_cost': sum(item['estimated_cost'] for item in sku_details) } async def get_batch_product_orders( self, db: AsyncSession, product_names: List[str], procurement_method: str ) -> Dict[str, Any]: """ 批量获取多个商品的相关订单（SKU级别详情） 优化版本，减少数据库查询次数，提升前端加载性能 """ if not product_names: return {'products': [], 'total_products': 0} # 一次性查询所有商品的SKU信息 query = (select(ProductMaster) .where(ProductMaster.线上宝贝名称.in_(product_names)) .where(ProductMaster.procurement_method == procurement_method) .where(ProductMaster.is_active == True) .where(ProductMaster.pending_procurement_qty > 0) .order_by(ProductMaster.线上宝贝名称, ProductMaster.pending_procurement_qty.desc())) result = await db.execute(query) products = result.scalars().all() # 按商品名称分组 products_dict = {} for product in products: product_name = product.线上宝贝名称 if product_name not in products_dict: products_dict[product_name] = [] products_dict[product_name].append(product) # 构建结果 batch_results = [] for product_name in product_names: product_skus = products_dict.get(product_name, []) # 构建SKU详情列表 sku_details = [] for product in product_skus: # 处理图片 - 优先使用本地图片 image_urls = [] # 优先检查主图片路径 if product.main_image_path and product.image_download_status == 'completed': if product.main_image_path.startswith('images/'): filename = product.main_image_path.replace('images/', '') image_urls = [f"/api/v1/static/images/{filename}"] else: image_urls = [self.image_service.get_local_image_url(product.main_image_path)] # 如果没有主图片，检查local_image_paths elif product.image_download_status == 'completed' and product.local_image_paths: image_urls = self._get_local_image_urls(product.local_image_paths) # 最后才使用原始图片URL elif product.original_image_urls: image_urls = self._parse_image_urls(product.original_image_urls) sku_detail = { 'sku_key': product.sku_key, 'product_id': product.id, 'sales_attr': product.线上销售属性, 'color': product.颜色, 'size': product.尺寸, 'product_code': product.货号, 'brand': product.品牌, 'pending_procurement_qty': product.pending_procurement_qty, 'total_quantity': product.total_quantity, 'order_count': product.total_orders, 'avg_price': float(product.avg_price) if product.avg_price else 0.0, 'estimated_cost': float(product.pending_procurement_qty * (product.avg_price or 0)), 'is_urgent': product.is_urgent, 'priority_score': product.priority_score, 'image_urls': image_urls, 'has_local_images': product.image_download_status == 'completed', 'last_order_time': product.last_order_time.isoformat() if product.last_order_time else None } sku_details.append(sku_detail) # 添加到批量结果中 product_result = { 'product_name': product_name, 'procurement_method': procurement_method, 'orders': sku_details, 'total_skus': len(sku_details), 'total_pending_qty': sum(item['pending_procurement_qty'] for item in sku_details), 'total_estimated_cost': sum(item['estimated_cost'] for item in sku_details) } batch_results.append(product_result) logger.info(f"批量获取商品订单完成: {len(product_names)}个商品, 共{sum(len(p['orders']) for p in batch_results)}个SKU") return { 'products': batch_results, 'total_products': len(batch_results), 'requested_products': len(product_names), 'total_skus': sum(p['total_skus'] for p in batch_results), 'total_pending_qty': sum(p['total_pending_qty'] for p in batch_results), 'total_estimated_cost': sum(p['total_estimated_cost'] for p in batch_results) } async def get_procurement_list_stats(self, db: AsyncSession) -> Dict[str, Any]: """ 获取采购清单统计信息 - V2优化版本 基于产品主表统计，性能更佳 """ # 基础统计 basic_stats_query = text(""" SELECT COUNT(*) as unique_products, COUNT(DISTINCT 品牌) as total_brands, SUM(pending_procurement_qty) as total_quantity, SUM(pending_procurement_qty * COALESCE(avg_price, 0)) as total_amount, COUNT(CASE WHEN is_urgent = 1 THEN 1 END) as urgent_orders, AVG(priority_score) as avg_priority FROM products_master WHERE is_active = 1 AND pending_procurement_qty > 0 """) basic_result = await db.execute(basic_stats_query) basic_row = basic_result.fetchone() # 采购方式分布 method_stats_query = text(""" SELECT procurement_method as method, COUNT(*) as sku_count, SUM(pending_procurement_qty) as total_quantity, SUM(pending_procurement_qty * COALESCE(avg_price, 0)) as total_amount, COUNT(CASE WHEN is_urgent = 1 THEN 1 END) as urgent_count FROM products_master WHERE is_active = 1 AND pending_procurement_qty > 0 GROUP BY procurement_method ORDER BY total_amount DESC """) method_result = await db.execute(method_stats_query) methods = [] for row in method_result.fetchall(): methods.append({ 'method': row.method, 'sku_count': row.sku_count, 'total_quantity': row.total_quantity, 'total_amount': float(row.total_amount or 0), 'urgent_count': row.urgent_count }) # TOP品牌 brand_stats_query = text(""" SELECT 品牌 as brand, COUNT(*) as sku_count, SUM(pending_procurement_qty) as total_quantity, SUM(pending_procurement_qty * COALESCE(avg_price, 0)) as total_amount FROM products_master WHERE is_active = 1 AND pending_procurement_qty > 0 AND 品牌 IS NOT NULL GROUP BY 品牌 ORDER BY total_amount DESC LIMIT 20 """) brand_result = await db.execute(brand_stats_query) brands = [] for row in brand_result.fetchall(): brands.append({ 'brand': row.brand, 'sku_count': row.sku_count, 'total_quantity': row.total_quantity, 'total_amount': float(row.total_amount or 0) }) return { 'basic_stats': { 'unique_products': basic_row.unique_products or 0, 'total_brands': basic_row.total_brands or 0, 'total_quantity': basic_row.total_quantity or 0, 'total_amount': float(basic_row.total_amount or 0), 'urgent_orders': basic_row.urgent_orders or 0, 'avg_priority': float(basic_row.avg_priority or 50) }, 'method_distribution': methods, 'top_brands': brands }