""" 店铺报表服务 - 处理淘宝报表上传、解析、匹配和数据同步 核心功能: 1. 解密加密的Excel文件 2. 解析销售明细报表 3. 解析发货报表 4. 精确匹配子订单编号（一单多件） 5. 加密隐私信息并存储 6. 更新采购订单状态 """ import io import uuid import asyncio import logging from datetime import datetime from typing import Dict, List, Tuple, Optional import pandas as pd import msoffcrypto from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func, and_, or_ from app.models import ( ShopSalesReport, ShopShippingReport, ShopReportUpload, ProcurementOrder, ReportUploadStatus ) from app.utils.encryption import get_encryption_service logger = logging.getLogger(__name__) class ShopReportService: """店铺报表服务""" def __init__(self): """初始化服务""" self.encryption = get_encryption_service() async def process_upload_batch( self, db: AsyncSession, shop_configs: List[Dict], uploaded_by: str, upload_notes: str = None ) -> Tuple[str, Dict]: """ 处理一整批报表上传（多个店铺） Args: db: 数据库会话 shop_configs: 店铺配置列表，格式: [ { "shop_name": "小王国美妆", "sales_report_path": "/path/to/sales.xlsx", "shipping_report_path": "/path/to/shipping.xlsx", "shipping_report_password": "password123" }, ... ] uploaded_by: 上传操作员名称 upload_notes: 上传备注 Returns: (batch_id, result_dict) - 批次ID和处理结果 """ batch_id = str(uuid.uuid4()) try: all_sales_rows = [] all_shipping_rows = [] shop_names = [] sales_filenames = [] shipping_filenames = [] # 处理每个店铺的报表 for config in shop_configs: shop_name = config["shop_name"] shop_names.append(shop_name) logger.info(f"处理店铺: {shop_name}") # 解析销售明细报表 sales_df = pd.read_excel(config["sales_report_path"]) all_sales_rows.append(sales_df) sales_filenames.append(config.get("sales_report_filename", "sales_report.xlsx")) # 解密并解析发货报表 if config.get("shipping_report_password"): shipping_df = await self._decrypt_and_read_excel( config["shipping_report_path"], config["shipping_report_password"] ) all_shipping_rows.append(shipping_df) shipping_filenames.append(config.get("shipping_report_filename", "shipping_report.xlsx")) # 合并所有数据 sales_df_combined = pd.concat(all_sales_rows, ignore_index=True) if all_sales_rows else pd.DataFrame() shipping_df_combined = pd.concat(all_shipping_rows, ignore_index=True) if all_shipping_rows else pd.DataFrame() # 保存到数据库并进行匹配 result = await self._save_and_match_reports( db=db, batch_id=batch_id, sales_df=sales_df_combined, shipping_df=shipping_df_combined, shop_names=shop_names, sales_filenames=sales_filenames, shipping_filenames=shipping_filenames, uploaded_by=uploaded_by, upload_notes=upload_notes ) return batch_id, result except Exception as e: logger.error(f"处理报表批次失败 {batch_id}: {str(e)}", exc_info=True) raise async def _decrypt_and_read_excel( self, filepath: str, password: str ) -> pd.DataFrame: """ 解密并读取加密的Excel文件 Args: filepath: 文件路径 password: 解密密码 Returns: DataFrame: 解析后的数据 """ try: decrypted = io.BytesIO() with open(filepath, 'rb') as encrypted_file: office_file = msoffcrypto.OfficeFile(encrypted_file) office_file.load_key(password=password) office_file.decrypt(decrypted) decrypted.seek(0) df = pd.read_excel(decrypted) logger.info(f"成功解密并读取文件: {filepath}, 行数: {len(df)}") return df except Exception as e: logger.error(f"解密文件失败 {filepath}: {str(e)}") raise async def _save_and_match_reports( self, db: AsyncSession, batch_id: str, sales_df: pd.DataFrame, shipping_df: pd.DataFrame, shop_names: List[str], sales_filenames: List[str], shipping_filenames: List[str], uploaded_by: str, upload_notes: str ) -> Dict: """ 保存报表数据到数据库并进行子订单匹配 Args: db: 数据库会话 batch_id: 批次ID sales_df: 销售明细数据 shipping_df: 发货报表数据 shop_names: 店铺名称列表 sales_filenames: 销售明细文件名列表 shipping_filenames: 发货报表文件名列表 uploaded_by: 上传操作员 upload_notes: 上传备注 Returns: 处理结果字典 """ try: # 创建上传记录 upload_record = ShopReportUpload( batch_id=batch_id, shop_names=shop_names, sales_report_filenames=sales_filenames, shipping_report_filenames=shipping_filenames, sales_report_total_rows=len(sales_df), shipping_report_total_rows=len(shipping_df), uploaded_by=uploaded_by, upload_status="processing", notes=upload_notes, created_at=datetime.now() ) db.add(upload_record) await db.flush() # 获取ID upload_id = upload_record.id # 保存销售明细报表 matched_count = 0 unmatched_count = 0 for _, row in sales_df.iterrows(): try: sales_report = ShopSalesReport( upload_batch_id=batch_id, shop_name=row.get('店铺', ''), 子订单编号=str(row.get('子订单编号', '')), 主订单编号=str(row.get('主订单编号', '')), 商品标题=row.get('商品标题', ''), 商品属性=row.get('商品属性', ''), 商品价格=str(row.get('商品价格', '')), 购买数量=int(row.get('购买数量', 0)) if pd.notna(row.get('购买数量')) else 0, 订单状态=row.get('订单状态', ''), 退款状态=row.get('退款状态', ''), 交易状态=row.get('交易状态', ''), 商家备注=row.get('商家备注', ''), 主订单买家留言=row.get('主订单买家留言', ''), 联系方式备注=row.get('联系方式备注', ''), 订单创建时间=pd.to_datetime(row.get('订单创建时间'), errors='coerce'), 订单付款时间=pd.to_datetime(row.get('订单付款时间'), errors='coerce'), 发货时间=pd.to_datetime(row.get('发货时间'), errors='coerce'), 买家应付货款=str(row.get('买家应付货款', '')), 买家实付金额=str(row.get('买家实付金额', '')), 退款金额=str(row.get('退款金额', '')), 物流单号=row.get('物流单号', ''), 物流公司=row.get('物流公司', ''), 外部系统编号=row.get('外部系统编号', ''), 商家编码=row.get('商家编码', ''), 支付单号=row.get('支付单号', ''), 匹配状态="待匹配" ) db.add(sales_report) except Exception as e: logger.error(f"保存销售报表行失败: {str(e)}") continue await db.flush() # 保存发货报表并加密敏感信息 for _, row in shipping_df.iterrows(): try: # 加密收货人信息 phone = row.get('联系手机', '') address = row.get('收货地址', '') name = row.get('收货人姓名', '') shipping_report = ShopShippingReport( upload_batch_id=batch_id, shop_name=row.get('店铺', ''), 订单编号=str(row.get('订单编号', '')), 收货人姓名_encrypted=self.encryption.encrypt(str(name)) if name else None, 收货地址_encrypted=self.encryption.encrypt(str(address)) if address else None, 联系手机_encrypted=self.encryption.encrypt(str(phone)) if phone else None, 联系电话_encrypted=self.encryption.encrypt( str(row.get('联系电话', '')) ) if row.get('联系电话') else None, 商家备注=row.get('商家备注', ''), 买家留言=row.get('买家留言', ''), 运送方式=row.get('运送方式', ''), 物流单号=row.get('物流单号', ''), 物流公司=row.get('物流公司', ''), 发货时间=pd.to_datetime(row.get('发货时间'), errors='coerce'), 订单状态=row.get('订单状态', ''), 应发货时间=pd.to_datetime(row.get('应发货时间'), errors='coerce'), 买家应付货款=str(row.get('买家应付货款', '')), 买家实付金额=str(row.get('买家实付金额', '')), 商品标题=row.get('商品标题', ''), 商品属性SKU=row.get('商品属性SKU', ''), 支付单号=row.get('支付单号', ''), 宝贝总数量=int(row.get('宝贝总数量', 0)) if pd.notna(row.get('宝贝总数量')) else 0 ) db.add(shipping_report) except Exception as e: logger.error(f"保存发货报表行失败: {str(e)}") continue await db.flush() # 执行子订单匹配逻辑 matched_count, unmatched_count = await self._match_sub_orders( db=db, batch_id=batch_id, sales_df=sales_df, upload_id=upload_id ) # 更新上传记录状态 upload_record.upload_status = "completed" upload_record.matched_sub_orders = matched_count upload_record.unmatched_sub_orders = unmatched_count upload_record.completed_at = datetime.now() await db.commit() result = { "batch_id": batch_id, "status": "success", "sales_report_rows": len(sales_df), "shipping_report_rows": len(shipping_df), "matched_sub_orders": matched_count, "unmatched_sub_orders": unmatched_count, "shops_processed": len(shop_names), "message": f"成功处理 {len(shop_names)} 个店铺，匹配 {matched_count} 个子订单" } logger.info(f"批次 {batch_id} 处理完成: {result}") return result except Exception as e: logger.error(f"保存和匹配报表失败: {str(e)}", exc_info=True) upload_record.upload_status = "failed" upload_record.error_message = str(e) await db.commit() raise async def _match_sub_orders( self, db: AsyncSession, batch_id: str, sales_df: pd.DataFrame, upload_id: int ) -> Tuple[int, int]: """ 匹配子订单编号到采购订单 策略: 通过主订单编号 + 商品标题 + 商品属性精确匹配 Args: db: 数据库会话 batch_id: 报表批次ID sales_df: 销售明细DataFrame upload_id: 上传记录ID Returns: (matched_count, unmatched_count) - 匹配成功和失败的数量 """ matched_count = 0 unmatched_count = 0 # 按主订单编号分组 grouped = sales_df.groupby('主订单编号') for main_order_no, group in grouped: main_order_no_str = str(main_order_no) # 查询该主订单的所有采购订单 result = await db.execute( select(ProcurementOrder).where( ProcurementOrder.原始订单编号 == main_order_no_str ) ) po_orders = result.scalars().all() if not po_orders: unmatched_count += len(group) continue # 对于一单多件，需要精确匹配 if len(group) > 1: # 一单多件场景 for _, sales_row in group.iterrows(): title = sales_row['商品标题'] attr = sales_row['商品属性'] sub_order_no = str(sales_row['子订单编号']) # 查找匹配的采购订单 matched_po = None for po in po_orders: if (po.线上宝贝名称 == title and po.线上销售属性 == attr): matched_po = po break if matched_po: # 更新采购订单 matched_po.子订单编号 = sub_order_no matched_po.淘宝订单状态 = sales_row.get('订单状态', '') matched_po.淘宝退款状态 = sales_row.get('退款状态', '') matched_po.淘宝商家备注 = sales_row.get('商家备注', '') matched_po.关联报表批次ID = upload_id matched_po.最后同步时间 = datetime.now() matched_count += 1 else: unmatched_count += 1 else: # 一单一件场景，直接匹配 sales_row = group.iloc[0] po = po_orders[0] po.子订单编号 = str(sales_row['子订单编号']) po.淘宝订单状态 = sales_row.get('订单状态', '') po.淘宝退款状态 = sales_row.get('退款状态', '') po.淘宝商家备注 = sales_row.get('商家备注', '') po.关联报表批次ID = upload_id po.最后同步时间 = datetime.now() matched_count += 1 await db.commit() logger.info(f"批次 {batch_id} 子订单匹配完成: {matched_count} 成功, {unmatched_count} 失败") return matched_count, unmatched_count async def get_upload_history( self, db: AsyncSession, limit: int = 20, offset: int = 0, status: Optional[str] = None ) -> Tuple[List[ShopReportUpload], int]: """ 查询上传历史 Args: db: 数据库会话 limit: 限制数量 offset: 偏移量 status: 状态筛选(可选) Returns: (records, total) - 上传记录列表和总数 """ query = select(ShopReportUpload) if status: query = query.where(ShopReportUpload.upload_status == status) # 获取总数 count_query = select(func.count()).select_from(query.subquery()) total = await db.scalar(count_query) # 获取分页数据 query = query.order_by(ShopReportUpload.created_at.desc()).limit(limit).offset(offset) result = await db.execute(query) records = result.scalars().all() return records, total or 0 async def get_batch_details( self, db: AsyncSession, batch_id: str ) -> Dict: """ 获取上传批次的详细信息 Args: db: 数据库会话 batch_id: 批次ID Returns: 批次详细信息字典 """ # 查询上传记录 result = await db.execute( select(ShopReportUpload).where(ShopReportUpload.batch_id == batch_id) ) upload = result.scalar_one_or_none() if not upload: raise ValueError(f"找不到批次: {batch_id}") # 查询销售报表数据 sales_result = await db.execute( select(ShopSalesReport).where(ShopSalesReport.upload_batch_id == batch_id) ) sales_reports = sales_result.scalars().all() # 查询发货报表数据 shipping_result = await db.execute( select(ShopShippingReport).where(ShopShippingReport.upload_batch_id == batch_id) ) shipping_reports = shipping_result.scalars().all() return { "upload_record": upload, "sales_reports": sales_reports, "shipping_reports": shipping_reports, "sales_count": len(sales_reports), "shipping_count": len(shipping_reports) } # 全局服务实例 _shop_report_service = None def get_shop_report_service() -> ShopReportService: """获取全局报表服务实例""" global _shop_report_service if _shop_report_service is None: _shop_report_service = ShopReportService() return _shop_report_service