""" 待采购订单服务 负责从标准化订单生成和管理待采购订单 !!!重要警告!!! 此服务已废弃，请使用 ProcurementOrderServiceV2 原因：依赖已废弃的 order_items_norm 表（标准化订单表已不再使用） 迁移指南： - 生成采购订单：使用 ProcurementOrderServiceV2.generate_procurement_orders() - 查询采购订单：使用 ProcurementOrderServiceV2.get_procurement_orders() - 统计信息：使用 ProcurementOrderServiceV2.get_procurement_orders_stats() @deprecated since 2024-12-01 请勿在新代码中使用此服务！ """ import logging import warnings from datetime import datetime from typing import Dict, List, Any, Optional from decimal import Decimal from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import text, select, update, delete from sqlalchemy.orm import selectinload from app.models.normalized_orders import OrderItemNorm from app.models.procurement_orders import ProcurementOrder, ProcurementStatus, ProcurementMethod from app.utils.file_hash import calculate_data_sha1 from app.utils.sql_security import safe_like_param logger = logging.getLogger(__name__) class ProcurementOrderService: """待采购订单服务 - 已废弃""" def __init__(self): """初始化时发出废弃警告""" warnings.warn( "ProcurementOrderService 已废弃！\n" "此服务依赖已废弃的 order_items_norm 表。\n" "请使用 ProcurementOrderServiceV2 替代。\n" "详见: app/services/procurement_order_service_v2.py", DeprecationWarning, stacklevel=2 ) logger.warning("正在使用已废弃的 ProcurementOrderService，请尽快迁移到 ProcurementOrderServiceV2") @staticmethod def _determine_procurement_method(product_name: str, sales_attributes: str, brand: str = "") -> str: """ 根据业务规则确定采购方式 """ product_name = product_name.strip() if product_name else "" sales_attributes = sales_attributes.strip() if sales_attributes else "" brand = brand.strip() if brand else "" # 转为大写便于匹配 product_name_upper = product_name.upper() sales_attributes_upper = sales_attributes.upper() brand_upper = brand.upper() # 规则1: MC库存 - 商品名称以MC/ZT结尾 或 销售属性包含MC/ZT if (product_name_upper.endswith(' MC') or product_name_upper.endswith('MC') or product_name_upper.endswith(' ZT') or product_name_upper.endswith('ZT') or 'MC' in sales_attributes_upper or 'ZT' in sales_attributes_upper): return ProcurementMethod.MC # 规则2: AT/ATM Atelier if ((product_name_upper.endswith(' AT') or product_name_upper.endswith('AT') or product_name_upper.endswith(' ATM') or product_name_upper.endswith('ATM')) and 'MC' not in sales_attributes_upper and 'ZT' not in sales_attributes_upper): return ProcurementMethod.AT # 规则3: AP/APM 昌昌 if ((product_name_upper.endswith(' AP') or product_name_upper.endswith('AP') or product_name_upper.endswith(' APM') or product_name_upper.endswith('APM')) and 'MC' not in sales_attributes_upper and 'ZT' not in sales_attributes_upper): return ProcurementMethod.AP # 规则4: LA/LAM - 商品名称以LA/LAM结尾的直接归类为LA if (product_name_upper.endswith(' LA') or product_name_upper.endswith('LA') or product_name_upper.endswith(' LAM') or product_name_upper.endswith('LAM')): return ProcurementMethod.LA # 规则5: SS/SSM Saks if ((product_name_upper.endswith(' SS') or product_name_upper.endswith('SS') or product_name_upper.endswith(' SSM') or product_name_upper.endswith('SSM')) and 'MC' not in sales_attributes_upper and 'ZT' not in sales_attributes_upper): return ProcurementMethod.SS # 规则6: 没有特定后缀，但销售属性有独立的LA suffixes_to_exclude = ['LA', 'MC', 'ZT', 'LAM', 'SS', 'SSM', 'AT', 'ATM'] has_excluded_suffix = any( product_name_upper.endswith(f' {suffix}') or product_name_upper.endswith(suffix) for suffix in suffixes_to_exclude ) if not has_excluded_suffix and ProcurementOrderService._has_independent_la(sales_attributes_upper): return ProcurementMethod.LA # 规则7: 国内现货 - 现货关键词或特定品牌 if (ProcurementOrderService._has_stock_keywords(product_name) or ProcurementOrderService._is_domestic_brand(brand_upper)): return ProcurementMethod.GN # 规则8: 默认NY return ProcurementMethod.NY @staticmethod def _has_independent_la(sales_attributes_upper: str) -> bool: """检查销售属性中是否有独立的LA（LA后面只能是结束、_、[等分隔符）""" import re # LA后面必须是结束、下划线、空格、数字、方括号等分隔符，不能是字母 pattern = r'LA(?=_|\s|$|\d|\[)' return bool(re.search(pattern, sales_attributes_upper)) @staticmethod def _has_stock_keywords(product_name: str) -> bool: """检查商品名称是否包含现货关键词""" return ('现货' in product_name or 'GN现货' in product_name or '差价' in product_name) @staticmethod def _is_domestic_brand(brand_upper: str) -> bool: """检查是否是国内现货品牌""" domestic_brands = ['GRUNDZERO', 'JANSPORT', 'EMU'] return any(domestic_brand in brand_upper for domestic_brand in domestic_brands) async def generate_procurement_orders_from_normalized(self, db: AsyncSession) -> Dict[str, Any]: """ 从标准化订单生成待采购订单 """ logger.info("开始从标准化订单生成待采购订单...") # 1. 清理旧的待采购订单 await db.execute(delete(ProcurementOrder)) await db.commit() logger.info("清理旧的待采购订单完成") # 2. 查询符合采购条件的标准化订单 query = text(""" SELECT id, 原始订单编号, 线上宝贝名称, 线上销售属性, 品牌, 货号, 颜色, 尺寸, 数量, 订单单价, 订单金额, 付款时间, 交易状态, 退款状态, 网店名称 FROM order_items_norm WHERE 付款时间 IS NOT NULL AND (交易状态 = '已付款' OR 交易状态 = '部分发货') AND 退款状态 IN ('正常', '退款中') AND procurement_status = 'WAITING' ORDER BY 付款时间 DESC """) result = await db.execute(query) normalized_orders = result.fetchall() logger.info(f"找到 {len(normalized_orders)} 个符合条件的标准化订单") # 3. 生成待采购订单 created_count = 0 for order in normalized_orders: # 计算产品SKU sku_components = [ str(order.线上宝贝名称) if order.线上宝贝名称 else "", str(order.线上销售属性) if order.线上销售属性 else "" ] sku_key = '|'.join(sku_components) product_key = calculate_data_sha1(sku_key) # 确定采购方式 procurement_method = self._determine_procurement_method( order.线上宝贝名称 or "", order.线上销售属性 or "", order.品牌 or "" ) # 计算建议采购价（暂时使用订单单价的0.7倍作为建议价） suggested_price = None if order.订单单价: suggested_price = Decimal(str(order.订单单价)) * Decimal('0.7') # 处理颜色字段，应用新的颜色处理逻辑 processed_color = order.颜色 if order.颜色: from app.utils.color_processor import process_color_attribute processed_color_result = process_color_attribute(order.颜色) if processed_color_result: processed_color = processed_color_result # 处理付款时间 payment_time = None if order.付款时间: if isinstance(order.付款时间, str): try: # 尝试解析ISO格式的时间字符串 payment_time = datetime.fromisoformat(order.付款时间.replace('Z', '+00:00')) if payment_time.tzinfo: payment_time = payment_time.replace(tzinfo=None) # 移除时区信息 except: payment_time = None elif isinstance(order.付款时间, datetime): payment_time = order.付款时间 # 创建待采购订单记录 procurement_order = ProcurementOrder( original_order_id=order.id, 原始订单编号=order.原始订单编号, 线上宝贝名称=order.线上宝贝名称, 线上销售属性=order.线上销售属性, 品牌=order.品牌, 货号=order.货号, 颜色=processed_color, 尺寸=order.尺寸, 数量=order.数量, 订单单价=order.订单单价, 订单金额=order.订单金额, 付款时间=payment_time, 交易状态=order.交易状态, 退款状态=order.退款状态, 网店名称=order.网店名称, procurement_status=ProcurementStatus.PENDING, procurement_method=procurement_method, product_key=product_key, 建议采购价=suggested_price, is_urgent=False, # 后续可以根据付款时间等因素判断 priority_score=0, # 后续可以根据算法计算 已推送="否" # 新创建的订单默认为未推送 ) db.add(procurement_order) created_count += 1 # 每100条提交一次 if created_count % 100 == 0: await db.commit() logger.info(f"已创建 {created_count} 个待采购订单...") # 最终提交 await db.commit() # 4. 统计结果 method_stats = await db.execute(text(""" SELECT procurement_method, COUNT(*) as count FROM procurement_orders GROUP BY procurement_method ORDER BY count DESC """)) method_distribution = {} for row in method_stats: method_distribution[row.procurement_method] = row.count logger.info(f"待采购订单生成完成，总共创建 {created_count} 个订单") logger.info(f"采购方式分布: {method_distribution}") return { "total_created": created_count, "method_distribution": method_distribution, "message": f"成功生成 {created_count} 个待采购订单" } async def get_procurement_orders( self, db: AsyncSession, procurement_method: Optional[str] = None, brand: Optional[str] = None, status: Optional[str] = None, urgent_only: bool = False, page: int = 1, page_size: int = 50, sort_by: Optional[str] = None, sort_order: Optional[str] = None ) -> Dict[str, Any]: """ 获取待采购订单列表 """ # 构建查询条件 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 status: conditions.append("procurement_status = :status") params["status"] = status if urgent_only: conditions.append("is_urgent = true") where_clause = " AND ".join(conditions) # 构建排序子句 order_clause = "ORDER BY" if sort_by and sort_order: # 验证排序字段安全性 allowed_sort_fields = ['付款时间', '创建时间', '订单单价', '订单金额', '数量', '原始订单编号'] if sort_by in allowed_sort_fields: sort_direction = "DESC" if sort_order.lower() == 'desc' else "ASC" order_clause += f" {sort_by} {sort_direction}," # 默认排序：付款时间倒序，创建时间倒序 order_clause += " 付款时间 DESC, 创建时间 DESC" # 查询总数 count_query = text(f""" SELECT COUNT(*) as total FROM procurement_orders WHERE {where_clause} """) count_result = await db.execute(count_query, params) total = count_result.scalar() # 查询数据 offset = (page - 1) * page_size data_query = text(f""" SELECT * FROM procurement_orders WHERE {where_clause} {order_clause} LIMIT :limit OFFSET :offset """) params.update({"limit": page_size, "offset": offset}) data_result = await db.execute(data_query, params) orders = data_result.fetchall() return { "orders": [dict(order._mapping) for order in orders], "pagination": { "page": page, "page_size": page_size, "total": total, "pages": (total + page_size - 1) // page_size } }