""" SKU级别订单详情服务 提供ExportOrderList文件导入和SKU详情查询功能 """ import logging import hashlib from datetime import datetime from pathlib import Path from typing import Dict, List, Any, Optional from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func, and_ from sqlalchemy.exc import IntegrityError import pandas as pd from app.models.order_sku_details import OrderSkuDetail from app.models.raw_orders import RawOrder from app.utils.file_hash import calculate_file_sha1 from app.utils.attribute_normalizer import ( normalize_product_attributes, OrderStatusMapper, create_match_key ) logger = logging.getLogger(__name__) class OrderSkuDetailsService: """SKU级别订单详情服务""" # ExportOrderList文件的列名映射 COLUMN_MAP = { '子订单编号': 'sub_order_id', '主订单编号': 'main_order_id', '商品标题': 'product_title', '商品价格': 'product_price', '购买数量': 'quantity', '外部系统编号': 'external_system_id', '商品属性': 'product_attributes', '套餐信息': 'package_info', '联系方式备注': 'contact_note', '订单状态': 'order_status', '商家编码': 'merchant_code', '支付单号': 'payment_id', '买家应付货款': 'buyer_should_pay', '买家实付金额': 'buyer_actual_pay', '退款状态': 'refund_status', '退款金额': 'refund_amount', '订单创建时间': 'order_create_time', '订单付款时间': 'order_pay_time', '淘鲜达渠道': 'taoxianda_channel', '商品ID': 'product_id', '分阶段信息': 'phase_info', '备注标签': 'note_tags', '商家备注': 'merchant_note', '主订单买家留言': 'buyer_message', '发货时间': 'ship_time', '物流单号': 'tracking_number', '物流公司': 'logistics_company', '是否主动赔付': 'is_active_compensation', '主动赔付金额': 'active_compensation_amount', '主动赔付出账时间': 'active_compensation_time', } @staticmethod def _safe_float(value, default: float = 0.0) -> float: """ 安全地将值转换为float，处理非数字文本 Args: value: 待转换的值 default: 默认值（当无法转换时返回） Returns: 转换后的float值或默认值 """ # 处理NaN和None if pd.isna(value) or value is None: return default # 如果已经是数字类型 if isinstance(value, (int, float)): return float(value) # 处理字符串 if isinstance(value, str): value = value.strip() # 常见的非数字文本 if value in ['无退款申请', '无', '无退款', 'N/A', 'NA', '']: return default try: return float(value) except (ValueError, TypeError): logger.warning(f"Cannot convert '{value}' to float, using default {default}") return default return default async def import_excel( self, db: AsyncSession, file_path: Path, batch_size: int = 1000 ) -> Dict[str, Any]: """ 导入ExportOrderList Excel文件 Args: db: 数据库会话 file_path: Excel文件路径 batch_size: 批量处理大小 Returns: 导入结果统计 """ start_time = datetime.now() result = { 'file_name': file_path.name, 'total_rows': 0, 'imported': 0, 'skipped': 0, 'errors': 0, 'error_details': [], 'processing_time_seconds': 0 } try: # 计算文件哈希 file_hash = calculate_file_sha1(file_path) logger.info(f"Starting import of {file_path.name}, hash: {file_hash}") # 读取Excel文件 df = pd.read_excel(file_path) result['total_rows'] = len(df) logger.info(f"Read {len(df)} rows from {file_path.name}") # 批量处理 for batch_start in range(0, len(df), batch_size): batch_end = min(batch_start + batch_size, len(df)) batch_df = df.iloc[batch_start:batch_end] logger.debug(f"Processing batch {batch_start}-{batch_end}") for idx, row in batch_df.iterrows(): try: # 转换数据 sku_detail = await self._row_to_sku_detail(row, file_path.name, file_hash) # 检查是否已存在 existing = await db.execute( select(OrderSkuDetail).where( OrderSkuDetail.sub_order_id == sku_detail.sub_order_id ) ) if existing.scalar_one_or_none(): logger.debug(f"Sub-order {sku_detail.sub_order_id} already exists, skipping") result['skipped'] += 1 continue # 添加到数据库 db.add(sku_detail) result['imported'] += 1 except Exception as e: logger.error(f"Error processing row {idx}: {e}") result['errors'] += 1 result['error_details'].append({ 'row': int(idx) + 2, # +2 因为Excel从1开始且有表头 'error': str(e) }) # 提交批次 try: await db.commit() logger.info(f"Committed batch {batch_start}-{batch_end}") except Exception as e: logger.error(f"Error committing batch: {e}") await db.rollback() raise except Exception as e: logger.error(f"Error importing file {file_path}: {e}") result['error_details'].append({ 'row': 'N/A', 'error': str(e) }) await db.rollback() raise finally: processing_time = (datetime.now() - start_time).total_seconds() result['processing_time_seconds'] = processing_time logger.info(f"Import completed: {result}") return result async def _row_to_sku_detail( self, row: pd.Series, file_name: str, file_hash: str ) -> OrderSkuDetail: """ 将Excel行数据转换为OrderSkuDetail模型 Args: row: pandas Series（一行数据） file_name: 文件名 file_hash: 文件哈希 Returns: OrderSkuDetail实例 """ # 提取商品属性并规范化 product_attrs = str(row.get('商品属性', '')) if pd.notna(row.get('商品属性')) else "" normalized_attrs = normalize_product_attributes(product_attrs) # 提取订单和退款状态 order_status = str(row.get('订单状态', '')) if pd.notna(row.get('订单状态')) else "" refund_status = str(row.get('退款状态', '')) if pd.notna(row.get('退款状态')) else "" # 映射状态 mapped_order_status = OrderStatusMapper.map_order_status(order_status) mapped_refund_status = OrderStatusMapper.map_refund_status(refund_status) # 解析日期时间字段 def parse_datetime(value): if pd.isna(value): return None if isinstance(value, datetime): return value try: return pd.to_datetime(value) except: return None # 解析布尔字段 def parse_bool(value): if pd.isna(value): return False if isinstance(value, bool): return value if isinstance(value, str): return value.lower() in ['是', 'true', 'yes', '1'] return bool(value) # 创建OrderSkuDetail实例 sku_detail = OrderSkuDetail( # 订单标识 sub_order_id=str(row['子订单编号']), main_order_id=str(row['主订单编号']), # 商品信息 product_title=str(row.get('商品标题', '')) if pd.notna(row.get('商品标题')) else None, product_attributes=product_attrs, normalized_attributes=normalized_attrs, merchant_code=str(row.get('商家编码', '')) if pd.notna(row.get('商家编码')) else None, product_id=str(row.get('商品ID', '')) if pd.notna(row.get('商品ID')) else None, # 数量与金额 quantity=int(row.get('购买数量', 1)) if pd.notna(row.get('购买数量')) else 1, product_price=OrderSkuDetailsService._safe_float(row.get('商品价格'), 0), buyer_should_pay=OrderSkuDetailsService._safe_float(row.get('买家应付货款'), 0), buyer_actual_pay=OrderSkuDetailsService._safe_float(row.get('买家实付金额'), 0), # 状态信息 order_status=order_status, refund_status=refund_status, refund_amount=OrderSkuDetailsService._safe_float(row.get('退款金额'), 0), mapped_order_status=mapped_order_status, mapped_refund_status=mapped_refund_status, # 时间字段 order_create_time=parse_datetime(row.get('订单创建时间')), order_pay_time=parse_datetime(row.get('订单付款时间')), ship_time=parse_datetime(row.get('发货时间')), # 物流信息 tracking_number=str(row.get('物流单号', '')) if pd.notna(row.get('物流单号')) else None, logistics_company=str(row.get('物流公司', '')) if pd.notna(row.get('物流公司')) else None, # 其他信息 external_system_id=str(row.get('外部系统编号', '')) if pd.notna(row.get('外部系统编号')) else None, package_info=str(row.get('套餐信息', '')) if pd.notna(row.get('套餐信息')) else None, contact_note=str(row.get('联系方式备注', '')) if pd.notna(row.get('联系方式备注')) else None, buyer_message=str(row.get('主订单买家留言', '')) if pd.notna(row.get('主订单买家留言')) else None, merchant_note=str(row.get('商家备注', '')) if pd.notna(row.get('商家备注')) else None, note_tags=str(row.get('备注标签', '')) if pd.notna(row.get('备注标签')) else None, payment_id=str(row.get('支付单号', '')) if pd.notna(row.get('支付单号')) else None, # 赔付信息 is_active_compensation=parse_bool(row.get('是否主动赔付', False)), active_compensation_amount=OrderSkuDetailsService._safe_float(row.get('主动赔付金额'), 0) if pd.notna(row.get('主动赔付金额')) else None, active_compensation_time=parse_datetime(row.get('主动赔付出账时间')), # 文件信息 file_name=file_name, file_hash=file_hash, ) return sku_detail async def get_by_order_id( self, db: AsyncSession, main_order_id: str ) -> List[OrderSkuDetail]: """ 获取指定订单的所有SKU详情 Args: db: 数据库会话 main_order_id: 主订单号 Returns: SKU详情列表 """ result = await db.execute( select(OrderSkuDetail) .where(OrderSkuDetail.main_order_id == main_order_id) .order_by(OrderSkuDetail.sub_order_id) ) return result.scalars().all() async def get_statistics(self, db: AsyncSession) -> Dict[str, Any]: """ 获取SKU详情统计信息 Args: db: 数据库会话 Returns: 统计信息字典 """ # 总记录数 total_result = await db.execute(select(func.count(OrderSkuDetail.id))) total_count = total_result.scalar() # 订单数 order_result = await db.execute( select(func.count(func.distinct(OrderSkuDetail.main_order_id))) ) order_count = order_result.scalar() # 按状态统计 status_result = await db.execute( select( OrderSkuDetail.mapped_order_status, func.count(OrderSkuDetail.id) ).group_by(OrderSkuDetail.mapped_order_status) ) status_stats = {row[0]: row[1] for row in status_result.fetchall()} # 按退款状态统计 refund_result = await db.execute( select( OrderSkuDetail.mapped_refund_status, func.count(OrderSkuDetail.id) ).group_by(OrderSkuDetail.mapped_refund_status) ) refund_stats = {row[0]: row[1] for row in refund_result.fetchall()} return { 'total_sku_count': total_count, 'order_count': order_count, 'order_status_distribution': status_stats, 'refund_status_distribution': refund_stats, 'avg_sku_per_order': round(total_count / order_count, 2) if order_count > 0 else 0 } async def match_with_raw_orders( self, db: AsyncSession, limit: Optional[int] = None ) -> Dict[str, Any]: """ 将SKU详情与raw_orders进行匹配，统计匹配情况 Args: db: 数据库会话 limit: 限制处理数量（用于测试） Returns: 匹配统计结果 """ stats = { 'total_sku_details': 0, 'matched': 0, 'unmatched': 0, 'status_differences': [] } # 获取所有SKU详情 query = select(OrderSkuDetail) if limit: query = query.limit(limit) result = await db.execute(query) sku_details = result.scalars().all() stats['total_sku_details'] = len(sku_details) for sku in sku_details: # 尝试匹配raw_orders raw_order_result = await db.execute( select(RawOrder).where( and_( RawOrder.原始订单编号 == sku.main_order_id, # 这里需要规范化raw_orders的线上销售属性后再匹配 # 暂时先通过订单号匹配 ) ) ) raw_orders = raw_order_result.scalars().all() if raw_orders: stats['matched'] += 1 # 检查状态差异 for raw_order in raw_orders: if raw_order.交易状态 != sku.mapped_order_status or \ raw_order.退款状态 != sku.mapped_refund_status: stats['status_differences'].append({ 'order_id': sku.main_order_id, 'sku_order_status': sku.mapped_order_status, 'raw_order_status': raw_order.交易状态, 'sku_refund_status': sku.mapped_refund_status, 'raw_refund_status': raw_order.退款状态 }) else: stats['unmatched'] += 1 return stats