""" 订单标准化服务 """ import logging from datetime import datetime from typing import Dict, List, Optional, Tuple, Any from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, update, and_, or_, func from sqlalchemy.exc import IntegrityError from app.core.config import settings from app.models.raw_orders import RawOrder from app.models.normalized_orders import OrderItemNorm, ProcurementStatus from app.services.alias_service import AliasService from app.utils.text_parser import BrandExtractor, ProductCodeExtractor, AttributeExtractor, generate_sku_id from app.utils.image_processor import ImageProcessor from app.utils.file_hash import calculate_data_sha1 logger = logging.getLogger(__name__) class NormalizationService: """订单标准化服务""" def __init__(self): self.alias_service = AliasService() self.image_processor = ImageProcessor(settings.IMAGE_SPLITTER) # 解析器将在处理时动态初始化 self.brand_extractor: Optional[BrandExtractor] = None self.product_code_extractor = ProductCodeExtractor() self.attribute_extractor: Optional[AttributeExtractor] = None # 缓存别名映射 self._brand_aliases_cache: Optional[Dict[str, str]] = None self._color_aliases_cache: Optional[Dict[str, str]] = None self._size_aliases_cache: Optional[Dict[str, str]] = None async def initialize_extractors(self, db: AsyncSession): """初始化解析器（加载别名数据）""" # 加载别名映射 self._brand_aliases_cache = await self.alias_service.load_brand_aliases(db) self._color_aliases_cache = await self.alias_service.load_color_aliases(db) self._size_aliases_cache = await self.alias_service.load_size_aliases(db) # 初始化解析器 self.brand_extractor = BrandExtractor(self._brand_aliases_cache) self.attribute_extractor = AttributeExtractor( self._color_aliases_cache, self._size_aliases_cache ) logger.info(f"Initialized extractors with {len(self._brand_aliases_cache)} brands, " f"{len(self._color_aliases_cache)} colors, {len(self._size_aliases_cache)} sizes") async def normalize_raw_orders(self, db: AsyncSession, batch_size: int = None, force_update: bool = False) -> Dict[str, Any]: """ 标准化原始订单数据 Args: db: 数据库会话 batch_size: 批处理大小 force_update: 是否强制更新已标准化的数据 Returns: 处理结果统计 """ batch_size = batch_size or settings.BATCH_SIZE # 初始化解析器 await self.initialize_extractors(db) # 获取需要处理的原始订单 raw_orders = await self._get_raw_orders_to_process(db, force_update) result = { 'total_raw_orders': len(raw_orders), 'processed': 0, 'created': 0, 'updated': 0, 'failed': 0, 'errors': [] } if not raw_orders: logger.info("No raw orders to normalize") return result # 分批处理 for i in range(0, len(raw_orders), batch_size): batch = raw_orders[i:i + batch_size] batch_result = await self._process_batch(db, batch, force_update) # 合并结果 result['processed'] += batch_result['processed'] result['created'] += batch_result['created'] result['updated'] += batch_result['updated'] result['failed'] += batch_result['failed'] result['errors'].extend(batch_result['errors']) logger.info(f"Processed batch {i//batch_size + 1}: " f"{batch_result['processed']} orders processed") # 处理品牌别名建议 await self._generate_brand_alias_suggestions(db) # 自动生成待采购订单（在标准化完成后） procurement_result = await self._auto_generate_procurement_orders(db) result['procurement_orders'] = procurement_result logger.info(f"Normalization completed: {result['processed']} processed, " f"{result['created']} created, {result['updated']} updated, " f"{result['failed']} failed") return result async def _get_raw_orders_to_process(self, db: AsyncSession, force_update: bool = False) -> List[RawOrder]: """获取需要处理的原始订单""" # 过滤条件：排除汇总行和无效数据 valid_filters = [ RawOrder.原始订单编号.notlike('%合计%'), RawOrder.原始订单编号.notlike('%汇总%'), RawOrder.原始订单编号.notlike('%总计%'), RawOrder.原始订单编号.is_not(None), RawOrder.原始订单编号 != '', RawOrder.订单单价.is_not(None) # 必须有单价 ] if force_update: # 强制更新模式：获取所有有效的原始订单 result = await db.execute( select(RawOrder) .where(and_(*valid_filters)) .order_by(RawOrder.created_at.desc()) ) else: # 增量模式：只获取未处理的有效原始订单 # 通过LEFT JOIN找出没有对应标准化记录的原始订单 result = await db.execute( select(RawOrder) .outerjoin( OrderItemNorm, OrderItemNorm.raw_order_id == RawOrder.id ) .where(and_( OrderItemNorm.id.is_(None), *valid_filters )) .order_by(RawOrder.created_at.desc()) ) return result.scalars().all() async def _process_batch(self, db: AsyncSession, raw_orders: List[RawOrder], force_update: bool = False) -> Dict[str, Any]: """处理一批原始订单""" result = { 'processed': 0, 'created': 0, 'updated': 0, 'failed': 0, 'errors': [] } for raw_order in raw_orders: try: # 标准化单个订单 norm_result = await self._normalize_single_order(db, raw_order, force_update) if norm_result['success']: result['processed'] += 1 if norm_result['action'] == 'created': result['created'] += 1 elif norm_result['action'] == 'updated': result['updated'] += 1 else: result['failed'] += 1 if norm_result.get('error'): result['errors'].append(f"Order {raw_order.id}: {norm_result['error']}") except Exception as e: result['failed'] += 1 error_msg = f"Order {raw_order.id}: {str(e)}" result['errors'].append(error_msg) logger.error(f"Failed to normalize order {raw_order.id}: {e}") return result async def _normalize_single_order(self, db: AsyncSession, raw_order: RawOrder, force_update: bool = False) -> Dict[str, Any]: """标准化单个原始订单""" try: # 生成产品键 product_key = self._generate_product_key(raw_order) # 检查是否已存在 existing = await self._get_existing_normalized_order(db, product_key) if existing and not force_update: # 存在且不强制更新，只更新可变字段 updated = await self._update_variable_fields(db, existing, raw_order) return {'success': True, 'action': 'updated' if updated else 'skipped'} # 解析商品信息 parsed_data = await self._parse_order_data(raw_order) if existing: # 更新现有记录 await self._update_normalized_order(db, existing, raw_order, parsed_data) return {'success': True, 'action': 'updated'} else: # 创建新记录 await self._create_normalized_order(db, raw_order, parsed_data, product_key) return {'success': True, 'action': 'created'} except Exception as e: import traceback error_details = traceback.format_exc() logger.error(f"Error normalizing order {raw_order.id}: {e}\n{error_details}") return {'success': False, 'error': str(e)} def _generate_product_key(self, raw_order: RawOrder) -> str: """生成产品去重键""" # 获取配置的去重字段 key_fields = [] for field_name in settings.dup_hash_fields_list: value = getattr(raw_order, field_name, None) key_fields.append(str(value) if value is not None else '') # 生成SHA1哈希 combined_key = '|'.join(key_fields) return calculate_data_sha1(combined_key) async def _get_existing_normalized_order(self, db: AsyncSession, product_key: str) -> Optional[OrderItemNorm]: """获取现有的标准化订单""" result = await db.execute( select(OrderItemNorm).where(OrderItemNorm.product_key == product_key) ) return result.scalar_one_or_none() async def _update_variable_fields(self, db: AsyncSession, existing: OrderItemNorm, raw_order: RawOrder) -> bool: """更新可变字段（交易状态、退款状态、付款时间、卖家备注）""" updated = False update_data = {} # 可变字段列表 variable_fields = [ '交易状态', '退款状态', '付款时间', '卖家备注' ] for field in variable_fields: new_value = getattr(raw_order, field, None) old_value = getattr(existing, field, None) if new_value != old_value: update_data[field] = new_value updated = True if updated: update_data['updated_at'] = datetime.now() await db.execute( update(OrderItemNorm) .where(OrderItemNorm.id == existing.id) .values(**update_data) ) await db.commit() logger.debug(f"Updated variable fields for order {existing.id}") return updated async def _parse_order_data(self, raw_order: RawOrder) -> Dict[str, Any]: """解析订单数据""" # 基础商品信息，确保数据类型安全 def safe_str_convert(value): if isinstance(value, list): if value: if len(value) == 1: return str(value[0]) if value[0] is not None else "" else: # 多个元素时，取第一个非空的或用空格连接 non_empty = [str(x) for x in value if x is not None and str(x).strip()] return non_empty[0] if non_empty else "" return "" elif value is None: return "" return str(value) product_name = safe_str_convert(raw_order.线上宝贝名称) sales_attrs = safe_str_convert(raw_order.线上销售属性) image_text = safe_str_convert(raw_order.图片) # 解析品牌 brand = None if self.brand_extractor: brand = self.brand_extractor.extract_brand(product_name) # 解析货号 product_code = self.product_code_extractor.extract_product_code(product_name, brand, sales_attrs) # 解析颜色和尺寸 attributes = {} if self.attribute_extractor: attributes = self.attribute_extractor.extract_attributes(product_name, sales_attrs) # 使用新的颜色处理器优化颜色处理 raw_color = attributes.get('color') color = raw_color if raw_color: # 导入新的颜色处理器 from app.utils.color_processor import process_color_attribute processed_color = process_color_attribute(raw_color) if processed_color: color = processed_color size = attributes.get('size') # 生成SKU ID sku_id = generate_sku_id( raw_order.线上商家编码, raw_order.商品编号, raw_order.SKU编号, color, size ) # 处理图片 image_links = self.image_processor.process_image_links(image_text) image_json = self.image_processor.serialize_images(image_links) image_count = len(image_links) return { '品牌': brand, '货号': product_code, '颜色': color, '尺寸': size, 'sku_id': sku_id, '图片链接': image_json, '图片数量': image_count } async def _create_normalized_order(self, db: AsyncSession, raw_order: RawOrder, parsed_data: Dict[str, Any], product_key: str): """创建标准化订单记录""" # 确定采购状态 procurement_status = ProcurementStatus.WAITING if raw_order.付款时间: procurement_status = ProcurementStatus.WAITING # 有付款时间的订单等待采购 norm_order = OrderItemNorm( product_key=product_key, raw_order_id=raw_order.id, # 基本信息 原始订单编号=raw_order.原始订单编号, 网店名称=raw_order.网店名称, 交易状态=raw_order.交易状态, 付款时间=raw_order.付款时间, 买家留言=raw_order.买家留言, 卖家备注=raw_order.卖家备注, 订单金额=raw_order.订单金额, 退款状态=raw_order.退款状态, # 商品信息 线上宝贝名称=raw_order.线上宝贝名称, 线上销售属性=raw_order.线上销售属性, 订单单价=raw_order.订单单价, 数量=raw_order.数量, # 编码信息 线上商家编码=raw_order.线上商家编码, 商品编号=raw_order.商品编号, SKU编号=raw_order.SKU编号, # 解析结果 品牌=parsed_data['品牌'], 货号=parsed_data['货号'], 颜色=parsed_data['颜色'], 尺寸=parsed_data['尺寸'], sku_id=parsed_data['sku_id'], 图片链接=parsed_data['图片链接'], 图片数量=parsed_data['图片数量'], # 采购状态 procurement_status=procurement_status ) try: db.add(norm_order) await db.commit() logger.debug(f"Created normalized order {norm_order.id} for raw order {raw_order.id}") except IntegrityError as e: await db.rollback() if "product_key" in str(e): # 产品键冲突，可能是并发处理导致 logger.warning(f"Product key conflict for raw order {raw_order.id}") else: raise async def _update_normalized_order(self, db: AsyncSession, existing: OrderItemNorm, raw_order: RawOrder, parsed_data: Dict[str, Any]): """更新现有标准化订单记录""" update_data = { # 更新所有字段 '交易状态': raw_order.交易状态, '付款时间': raw_order.付款时间, '买家留言': raw_order.买家留言, '卖家备注': raw_order.卖家备注, '订单金额': raw_order.订单金额, '退款状态': raw_order.退款状态, # 更新解析结果 '品牌': parsed_data['品牌'], '货号': parsed_data['货号'], '颜色': parsed_data['颜色'], '尺寸': parsed_data['尺寸'], 'sku_id': parsed_data['sku_id'], '图片链接': parsed_data['图片链接'], '图片数量': parsed_data['图片数量'], 'updated_at': datetime.now() } await db.execute( update(OrderItemNorm) .where(OrderItemNorm.id == existing.id) .values(**update_data) ) await db.commit() logger.debug(f"Updated normalized order {existing.id}") async def _generate_brand_alias_suggestions(self, db: AsyncSession): """生成品牌别名建议""" if not self.brand_extractor: return # 获取未匹配到品牌的标准化订单样本 result = await db.execute( select(OrderItemNorm.线上宝贝名称) .where(OrderItemNorm.品牌.is_(None)) .distinct() .limit(100) # 限制处理数量 ) product_names = result.scalars().all() suggestion_count = 0 for product_name in product_names: if not product_name: continue suggestions = self.brand_extractor.suggest_new_alias(product_name) for suggestion in suggestions: try: await self.alias_service.suggest_brand_alias( db, suggestion['alias'], suggestion['canonical'], suggestion['confidence'] ) suggestion_count += 1 except Exception as e: logger.debug(f"Failed to add brand suggestion {suggestion}: {e}") if suggestion_count > 0: logger.info(f"Generated {suggestion_count} brand alias suggestions") async def get_normalization_stats(self, db: AsyncSession) -> Dict[str, Any]: """获取标准化统计信息""" # 原始订单总数 raw_total = await db.scalar(select(func.count(RawOrder.id))) # 标准化订单总数 norm_total = await db.scalar(select(func.count(OrderItemNorm.id))) # 有品牌的订单数 branded_count = await db.scalar( select(func.count(OrderItemNorm.id)) .where(OrderItemNorm.品牌.is_not(None)) ) # 采购状态统计 procurement_stats = await db.execute( select( OrderItemNorm.procurement_status, func.count(OrderItemNorm.id).label('count') ) .group_by(OrderItemNorm.procurement_status) ) procurement_counts = {status.value: count for status, count in procurement_stats.fetchall()} # 待处理订单数（未标准化的原始订单） pending_count = (raw_total or 0) - (norm_total or 0) return { 'raw_orders_total': raw_total or 0, 'normalized_total': norm_total or 0, 'pending_normalization': max(0, pending_count), 'branded_orders': branded_count or 0, 'brand_coverage': (branded_count / norm_total * 100) if norm_total else 0, 'procurement_status': procurement_counts } async def _auto_generate_procurement_orders(self, db: AsyncSession) -> Dict[str, Any]: """ 标准化完成后自动生成待采购订单 """ try: # 导入采购订单服务（延迟导入避免循环依赖） from app.services.procurement_order_service_v2 import ProcurementOrderServiceV2 logger.info("开始自动生成待采购订单...") procurement_service = ProcurementOrderServiceV2() # 注意：V2版本使用不同的方法名，直接从raw_orders生成 result = await procurement_service.generate_procurement_orders(db, force_rebuild=False) logger.info(f"自动生成待采购订单完成: {result['total_created']} 个订单") return { 'auto_generated': True, 'total_created': result['total_created'], 'method_distribution': result['method_distribution'], 'message': f"自动生成 {result['total_created']} 个待采购订单" } except Exception as e: logger.error(f"自动生成待采购订单失败: {e}") return { 'auto_generated': False, 'error': str(e), 'message': f"自动生成待采购订单失败: {str(e)}" }