""" 待采购订单服务 V2 - 基于产品主表的优化版本 不再依赖标准化订单，直接从产品主表生成待采购订单 """ import json import logging from datetime import datetime from typing import Dict, List, Any, Optional from decimal import Decimal from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, update, delete, func, and_, or_, text from sqlalchemy.orm import selectinload from app.models.products_master import ProductMaster, ProductSourceMapping from app.models.procurement_orders import ProcurementOrder, ProcurementStatus, ProcurementMethod from app.models.raw_orders import RawOrder logger = logging.getLogger(__name__) class ProcurementOrderServiceV2: """基于产品主表的待采购订单服务""" async def generate_procurement_orders(self, db: AsyncSession, force_rebuild: bool = False, batch_size: int = 1000) -> Dict[str, Any]: """ 从原始订单生成待采购订单（修改为原始订单驱动） ⚠️ 重要变更：为保证历史订单保留，不再清空现有订单 一旦订单进入待采购列表，将永久保留（只更新字段） Args: db: 数据库会话 force_rebuild: 是否强制重建（现在仅影响是否重新处理所有订单） batch_size: 批处理大小 Returns: 处理结果统计 """ logger.info("开始从原始订单生成待采购订单...") # 获取需要处理的原始订单（保持原有筛选逻辑） raw_orders = await self._get_raw_orders_for_procurement(db, force_rebuild) result = { 'total_raw_orders': len(raw_orders), 'processed_orders': 0, 'created_orders': 0, 'updated_orders': 0, 'failed_orders': 0, 'errors': [] } if not raw_orders: logger.info("没有需要处理的原始订单") return result logger.info(f"找到 {len(raw_orders)} 个原始订单需要处理") # 分批处理原始订单 for i in range(0, len(raw_orders), batch_size): batch = raw_orders[i:i + batch_size] batch_result = await self._process_raw_orders_batch(db, batch) # 合并结果 result['processed_orders'] += batch_result['processed'] result['created_orders'] += batch_result['created'] result['updated_orders'] += batch_result['updated'] result['failed_orders'] += batch_result['failed'] result['errors'].extend(batch_result['errors']) logger.info(f"批次 {i//batch_size + 1} 完成: " f"处理订单 {batch_result['processed']}, " f"创建采购单 {batch_result['created']}, " f"更新采购单 {batch_result['updated']}") logger.info(f"待采购订单生成完成: " f"处理订单 {result['processed_orders']}, " f"创建采购单 {result['created_orders']}, " f"更新采购单 {result['updated_orders']}") return result async def _get_raw_orders_for_procurement(self, db: AsyncSession, force_rebuild: bool) -> List[RawOrder]: """获取需要生成采购订单的原始订单（保持原有筛选逻辑）""" # 基础过滤条件（排除不需要采购的订单） base_filters = [ RawOrder.原始订单编号.notlike('%合计%'), RawOrder.原始订单编号.notlike('%汇总%'), RawOrder.原始订单编号.notlike('%总计%'), RawOrder.原始订单编号.is_not(None), RawOrder.原始订单编号 != '', RawOrder.线上宝贝名称.is_not(None), # 必须有商品名称 RawOrder.订单单价.is_not(None), # 必须有单价 # 只保留需要采购的交易状态 RawOrder.交易状态.in_(['已付款', '部分发货']), # 保留正常、退款中、退款成功的订单（退款成功代表未满足的采购需求） RawOrder.退款状态.in_(['正常', '退款中', '退款成功']) ] if force_rebuild: # 强制重建：获取所有有效原始订单 query = (select(RawOrder) .where(and_(*base_filters)) .order_by(RawOrder.付款时间.desc(), RawOrder.created_at.desc())) else: # 增量处理：只处理还没有对应采购订单的原始订单 # 使用 ID 进行匹配，因为 original_order_id 是整数类型，关联到 raw_orders.id existing_orders_subquery = select(ProcurementOrder.original_order_id) query = (select(RawOrder) .where(and_(*base_filters)) .where(RawOrder.id.not_in(existing_orders_subquery)) # 修复：使用 ID 而不是订单编号 .order_by(RawOrder.付款时间.desc(), RawOrder.created_at.desc())) result = await db.execute(query) raw_orders = result.scalars().all() logger.info(f"获取到 {len(raw_orders)} 个原始订单需要生成采购订单") return raw_orders async def _process_raw_orders_batch(self, db: AsyncSession, raw_orders: List[RawOrder]) -> Dict[str, Any]: """ 处理原始订单批次 支持三种操作： - 'created': 创建新的采购订单 - 'updated': 更新现有采购订单的字段 - 'failed': 处理失败 """ result = { 'processed': 0, 'created': 0, 'updated': 0, 'failed': 0, 'errors': [] } for raw_order in raw_orders: try: # 为每个原始订单创建或更新采购订单 order_result = await self._create_procurement_order_from_raw_order(db, raw_order) if order_result['success']: action = order_result.get('action') if action == 'created': result['created'] += 1 elif action == 'updated': result['updated'] += 1 # 如果没有指定 action，默认按创建处理 else: result['created'] += 1 else: result['failed'] += 1 result['errors'].append(order_result['error']) result['processed'] += 1 except Exception as e: result['failed'] += 1 error_msg = f"处理原始订单 {raw_order.id} 失败: {str(e)}" result['errors'].append(error_msg) logger.error(error_msg, exc_info=True) # 提交批次处理结果 await db.commit() return result async def _get_product_info_for_raw_order(self, db: AsyncSession, raw_order: RawOrder) -> Optional[ProductMaster]: """为原始订单获取对应的产品主表信息（标准化SKU信息）""" # 通过产品来源映射查找对应的产品主表记录 query = (select(ProductMaster) .join(ProductSourceMapping, ProductSourceMapping.product_master_id == ProductMaster.id) .where(ProductSourceMapping.raw_order_id == raw_order.id)) result = await db.execute(query) product = result.scalar_one_or_none() if not product: logger.warning(f"原始订单 {raw_order.id} 未找到对应的产品主表记录") return product async def _create_procurement_order_from_raw_order(self, db: AsyncSession, raw_order: RawOrder) -> Dict[str, Any]: """ 基于原始订单创建或更新采购订单，从产品主表获取标准化SKU信息 ⚠️ 重要逻辑： - 如果订单不存在：创建新订单 - 如果订单已存在：更新字段，但保留采购相关的设置（status, 忽略采购） """ try: # 检查是否已存在采购订单 existing_order = await self._get_existing_procurement_order( db, raw_order.id) if existing_order: # 订单已存在，更新其字段（但保留采购相关设置） logger.debug(f"采购订单已存在，更新字段: 原始订单 {raw_order.id}") return await self._update_existing_procurement_order( db, existing_order, raw_order) # 获取对应的产品主表信息（标准化SKU信息） product = await self._get_product_info_for_raw_order(db, raw_order) # 创建采购订单 - 主要信息来自原始订单，标准化信息来自产品主表 procurement_order = ProcurementOrder( # 原始订单关联 original_order_id=raw_order.id, 原始订单编号=raw_order.原始订单编号, # 商品信息 - 优先从产品主表获取标准化信息，否则使用原始订单 线上宝贝名称=raw_order.线上宝贝名称, # 保持原始商品名称 线上销售属性=raw_order.线上销售属性, # 保持原始销售属性 品牌=product.品牌 if product else None, # 标准化品牌 货号=product.货号 if product else None, # 标准化货号 颜色=product.颜色 if product else None, # 标准化颜色 尺寸=product.尺寸 if product else None, # 标准化尺寸 # 订单详情 - 来自原始订单 数量=int(raw_order.数量) if raw_order.数量 else 1, 订单单价=Decimal(str(raw_order.订单单价)) if raw_order.订单单价 else Decimal('0'), 订单金额=Decimal(str(raw_order.订单金额)) if raw_order.订单金额 else Decimal('0'), # 时间信息 - 来自原始订单 付款时间=raw_order.付款时间, # 订单状态 - 来自原始订单 交易状态=raw_order.交易状态, 退款状态=raw_order.退款状态, 网店名称=raw_order.网店名称, # 采购相关 - 来自产品主表（包含ZT规则等） procurement_status=ProcurementStatus.PENDING, procurement_method=product.procurement_method.value if product and product.procurement_method else "NY", # SKU关联 product_key=product.sku_key if product else None, # 优先级 - 来自产品主表 is_urgent=product.is_urgent if product else False, priority_score=product.priority_score if product else 50, # 建议采购价 - 来自产品主表的平均价格 建议采购价=Decimal(str(product.avg_price)) if product and product.avg_price else None, # 图片信息 - 优先从产品主表获取，否则从原始订单获取 # 确保 image_urls 是字符串格式（JSON字符串），而不是列表 image_urls=( product.original_image_urls if product and product.original_image_urls and isinstance(product.original_image_urls, str) else ( json.dumps(product.original_image_urls) if product and product.original_image_urls and isinstance(product.original_image_urls, list) else f'["{raw_order.图片}"]' if raw_order.图片 else None ) ), main_image_path=product.main_image_path if product and product.main_image_path else None, # 新增字段 卖家备注=raw_order.卖家备注, # 从原始订单获取卖家备注 忽略采购=False # 默认不忽略 ) db.add(procurement_order) await db.flush() # 获取ID但不提交 logger.debug(f"创建采购订单: 原始订单 {raw_order.id}, 产品关联: {'是' if product else '否'}") return {'success': True, 'action': 'created'} except Exception as e: logger.error(f"创建采购订单失败: 原始订单 {raw_order.id}, 错误: {e}") return {'success': False, 'error': str(e)} async def _update_existing_procurement_order(self, db: AsyncSession, existing_order: ProcurementOrder, raw_order: RawOrder) -> Dict[str, Any]: """ 更新现有采购订单的字段，但保留采购相关设置 保护字段（不更新）： - procurement_status: 采购状态 - 忽略采购: 是否忽略 - 已推送: 是否已推送到采购系统 更新字段： - 交易状态、退款状态等原始订单信息 - 品牌、货号、颜色、尺寸等从产品主表获取的标准化信息 """ try: # 获取对应的产品主表信息（标准化SKU信息） product = await self._get_product_info_for_raw_order(db, raw_order) # 更新原始订单字段 existing_order.线上宝贝名称 = raw_order.线上宝贝名称 existing_order.线上销售属性 = raw_order.线上销售属性 existing_order.数量 = int(raw_order.数量) if raw_order.数量 else 1 existing_order.订单单价 = Decimal(str(raw_order.订单单价)) if raw_order.订单单价 else Decimal('0') existing_order.订单金额 = Decimal(str(raw_order.订单金额)) if raw_order.订单金额 else Decimal('0') existing_order.付款时间 = raw_order.付款时间 existing_order.交易状态 = raw_order.交易状态 # 退款状态不更新，保留用户手动修改的值 existing_order.卖家备注 = raw_order.卖家备注 # 更新产品主表获取的标准化字段 if product: existing_order.品牌 = product.品牌 existing_order.货号 = product.货号 existing_order.颜色 = product.颜色 existing_order.尺寸 = product.尺寸 existing_order.product_key = product.sku_key existing_order.建议采购价 = Decimal(str(product.avg_price)) if product.avg_price else None # 更新图片信息 if product.original_image_urls: if isinstance(product.original_image_urls, str): existing_order.image_urls = product.original_image_urls else: existing_order.image_urls = json.dumps(product.original_image_urls) if product.main_image_path: existing_order.main_image_path = product.main_image_path # ⚠️ 不更新这些字段： # - procurement_status（保留原有的采购状态） # - 忽略采购（保留原有的忽略标志） # - 已推送（保留原有的推送状态） await db.flush() # 获取ID但不提交 logger.debug(f"更新采购订单字段: 原始订单 {raw_order.id}, 产品关联: {'是' if product else '否'}") return {'success': True, 'action': 'updated'} except Exception as e: logger.error(f"更新采购订单失败: 原始订单 {raw_order.id}, 错误: {e}") return {'success': False, 'error': str(e)} async def _get_existing_procurement_order(self, db: AsyncSession, raw_order_id: int) -> Optional[ProcurementOrder]: """检查是否已存在采购订单""" result = await db.execute( select(ProcurementOrder) .where(ProcurementOrder.original_order_id == raw_order_id) ) return result.scalar_one_or_none() async def get_procurement_orders_stats(self, db: AsyncSession) -> Dict[str, Any]: """获取采购订单统计信息""" # 基础统计 basic_stats = await db.execute(text(""" SELECT COUNT(*) as total_orders, COUNT(DISTINCT product_key) as unique_products, COUNT(DISTINCT 品牌) as total_brands, SUM(数量) as total_quantity, SUM(CAST(订单金额 AS FLOAT)) as total_amount, COUNT(CASE WHEN is_urgent = 1 THEN 1 END) as urgent_orders, AVG(priority_score) as avg_priority FROM procurement_orders WHERE procurement_status = 'PENDING' """)) basic_row = basic_stats.fetchone() # 采购方式分布 method_distribution = await db.execute(text(""" SELECT procurement_method as method, COUNT(*) as order_count, COUNT(DISTINCT product_key) as sku_count, SUM(数量) as total_quantity, SUM(CAST(订单金额 AS FLOAT)) as total_amount FROM procurement_orders WHERE procurement_status = 'PENDING' GROUP BY procurement_method ORDER BY total_amount DESC """)) methods = [] for row in method_distribution.fetchall(): methods.append({ 'method': row.method, 'order_count': row.order_count, 'sku_count': row.sku_count, 'total_quantity': row.total_quantity, 'total_amount': float(row.total_amount or 0) }) # TOP品牌 top_brands = await db.execute(text(""" SELECT 品牌 as brand, COUNT(DISTINCT product_key) as sku_count, COUNT(*) as order_count, SUM(数量) as total_quantity, SUM(CAST(订单金额 AS FLOAT)) as total_amount FROM procurement_orders WHERE procurement_status = 'PENDING' AND 品牌 IS NOT NULL GROUP BY 品牌 ORDER BY total_amount DESC LIMIT 20 """)) brands = [] for row in top_brands.fetchall(): brands.append({ 'brand': row.brand, 'sku_count': row.sku_count, 'order_count': row.order_count, 'total_quantity': row.total_quantity, 'total_amount': float(row.total_amount or 0) }) return { 'basic_stats': { 'total_orders': basic_row.total_orders or 0, '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 } async def update_procurement_priorities(self, db: AsyncSession) -> Dict[str, Any]: """ 更新采购优先级 基于产品主表的统计信息智能调整优先级 """ # 获取需要更新优先级的采购订单 query = text(""" UPDATE procurement_orders SET priority_score = CASE WHEN pm.total_quantity >= 10 THEN 80 -- 高销量产品 WHEN pm.total_orders >= 5 THEN 70 -- 多订单产品 WHEN pm.avg_price >= 1000 THEN 60 -- 高价值产品 ELSE 50 END, is_urgent = CASE WHEN pm.total_quantity >= 20 OR pm.avg_price >= 2000 THEN 1 ELSE 0 END FROM products_master pm WHERE procurement_orders.product_key = pm.sku_key AND procurement_orders.procurement_status = 'PENDING' """) result = await db.execute(query) updated_count = result.rowcount await db.commit() logger.info(f"更新了 {updated_count} 个采购订单的优先级") return { 'updated_orders': updated_count, 'message': f'成功更新 {updated_count} 个采购订单的优先级' }