""" 地址加密上传服务 处理密码保护的Excel文件，加密存储敏感信息到order_addresses表 This is part of the new independent shipping module that does not affect existing modules. """ import hashlib import pandas as pd import msoffcrypto import io import re from datetime import datetime from typing import Optional, Dict, List, Any from pathlib import Path import logging from sqlalchemy import select, and_, update from sqlalchemy.ext.asyncio import AsyncSession from fastapi import UploadFile from app.models.order_addresses import OrderAddress from app.models.shipping_upload_batches import ShippingUploadBatch from app.models.procurement_orders import ProcurementOrder from app.utils.encryption import EncryptionService logger = logging.getLogger(__name__) class AddressUploadService: """地址加密上传服务""" def __init__(self): self.encryption_service = EncryptionService() # Excel列名映射到数据库字段（42个字段） COLUMN_MAPPING = { '订单编号': '订单编号', '支付单号': '支付单号', '支付详情': '支付详情', '买家应付货款': '买家应付货款', '买家应付邮费': '买家应付邮费', '总金额': '总金额', '返点积分': '返点积分', '买家实付金额': '买家实付金额', '买家实付积分': '买家实付积分', '订单状态': '订单状态', '买家留言': '买家留言', '收货人姓名': '收货人姓名', # 需要加密 '收货地址': '收货地址', # 需要加密 '联系电话': '联系电话', # 需要加密 '联系手机': '联系手机', # 需要加密 '虚拟号过期时间': '虚拟号过期时间', '运送方式': '运送方式', '订单创建时间': '订单创建时间', '订单付款时间': '订单付款时间', '商品标题': '商品标题', '宝贝种类': '宝贝种类', '物流单号': '物流单号', '物流公司': '物流公司', '备注标签': '备注标签', '商家备注': '商家备注', '退款金额': '退款金额', '是否主动赔付': '是否主动赔付', '主动赔付金额': '主动赔付金额', '主动赔付出账时间': '主动赔付出账时间', '预售下沉订单': '预售下沉订单', '预售下沉状态': '预售下沉状态', '安心鉴订单': '安心鉴订单', '安心鉴订单二阶段物流': '安心鉴订单二阶段物流', '百亿补贴审核订单': '百亿补贴审核订单', '淘特订单': '淘特订单', 'C2B小额收款': 'C2B小额收款', '淘鲜达渠道': '淘鲜达渠道', '预售订单': '预售订单', '发货时间': '发货时间', '宝贝总数量': '宝贝总数量', '商品属性SKU': '商品属性SKU', '应发货时间': '应发货时间', } # 需要加密的字段 ENCRYPTED_FIELDS = ['收货人姓名', '收货地址', '联系电话', '联系手机'] async def process_address_file( self, db: AsyncSession, file: UploadFile, shop_name: Optional[str], operator: str, password: str = 'WIqo7ni4' ) -> Dict[str, Any]: """ 处理地址Excel文件上传 Args: db: 数据库会话 file: 上传的Excel文件（可能有密码保护） shop_name: 店铺名称（可选，如果未提供则自动识别） operator: 操作员 password: Excel密码（默认WIqo7ni4） Returns: 处理结果统计 """ try: # 读取文件内容（提前读取以便自动识别店铺） content = await file.read() file_hash = hashlib.sha256(content).hexdigest() # 尝试解密Excel文件 df = await self._decrypt_and_read_excel(content, password) total_rows = len(df) # 自动识别店铺 auto_identified_shop = await self.auto_identify_shop_from_excel(db, df, max_rows=5) # 店铺验证和确定逻辑 final_shop_name = None validation_message = None if shop_name and auto_identified_shop: # 两者都存在，验证是否一致 if shop_name == auto_identified_shop: final_shop_name = shop_name validation_message = f"✅ 店铺验证通过：用户选择与自动识别一致 ({shop_name})" logger.info(validation_message) else: # 不一致，返回警告 error_msg = f"❌ 店铺不一致：用户选择 '{shop_name}' 与自动识别 '{auto_identified_shop}' 不符" logger.warning(error_msg) return { 'success': False, 'error': 'shop_mismatch', 'message': error_msg, 'user_selected': shop_name, 'auto_identified': auto_identified_shop } elif auto_identified_shop: # 只有自动识别成功 final_shop_name = auto_identified_shop validation_message = f"✅ 自动识别店铺：{auto_identified_shop}" logger.info(validation_message) elif shop_name: # 只有用户提供 final_shop_name = shop_name validation_message = f"⚠️ 使用用户提供的店铺：{shop_name}（无法自动识别）" logger.warning(validation_message) else: # 两者都没有 error_msg = "❌ 无法确定店铺：请手动选择店铺或确保Excel中的订单号存在于系统中" logger.error(error_msg) return { 'success': False, 'error': 'no_shop', 'message': error_msg } # 创建上传批次 batch = ShippingUploadBatch( upload_type='address_list', 网店名称=final_shop_name, file_name=file.filename, 操作员=operator ) db.add(batch) await db.flush() batch.file_hash = file_hash batch.总行数 = total_rows logger.info(f"开始处理地址文件，店铺：{final_shop_name}，总行数：{total_rows}") if validation_message: logger.info(f"店铺识别结果：{validation_message}") # 处理统计 stats = { 'total': total_rows, 'new_addresses': 0, 'updated_addresses': 0, 'encrypted_fields': 0, 'errors': [] } # 处理每一行数据 for index, row in df.iterrows(): try: order_no = str(row.get('订单编号', '')) if not order_no: continue # 检查是否已存在 existing = await self._get_existing_address(db, order_no, final_shop_name) if existing: # 更新现有记录 await self._update_address(db, existing, row, batch.batch_id) stats['updated_addresses'] += 1 else: # 创建新记录 await self._create_address(db, final_shop_name, row, batch.batch_id) stats['new_addresses'] += 1 stats['encrypted_fields'] += len(self.ENCRYPTED_FIELDS) except Exception as e: logger.error(f"处理第{index+1}行时出错：{str(e)}") stats['errors'].append(f"Row {index+1}: {str(e)}") # 更新批次统计 batch.set_statistics(stats) batch.mark_completed() await db.commit() return { 'success': True, 'batch_id': batch.batch_id, 'statistics': stats, 'shop_name': final_shop_name, 'auto_identified': auto_identified_shop is not None, 'validation_message': validation_message, 'message': f"成功处理{total_rows}行地址数据" } except Exception as e: logger.error(f"处理地址文件失败：{str(e)}") if 'batch' in locals(): batch.mark_failed(str(e)) await db.rollback() return { 'success': False, 'error': str(e), 'message': f"处理失败：{str(e)}" } async def _decrypt_and_read_excel(self, content: bytes, password: str) -> pd.DataFrame: """解密并读取Excel文件""" try: # 尝试解密文件 file_stream = io.BytesIO(content) decrypted_stream = io.BytesIO() office_file = msoffcrypto.OfficeFile(file_stream) office_file.load_key(password=password) office_file.decrypt(decrypted_stream) # 读取解密后的Excel decrypted_stream.seek(0) return pd.read_excel(decrypted_stream, engine='openpyxl') except Exception as e: # 如果解密失败，尝试直接读取（可能文件没有密码） logger.warning(f"尝试解密失败，直接读取文件：{str(e)}") file_stream = io.BytesIO(content) return pd.read_excel(file_stream, engine='openpyxl') async def _get_existing_address( self, db: AsyncSession, order_no: str, shop_name: str ) -> Optional[OrderAddress]: """获取已存在的地址记录""" result = await db.execute( select(OrderAddress).where( and_( OrderAddress.订单编号 == order_no, OrderAddress.网店名称 == shop_name ) ) ) return result.scalar_one_or_none() async def _create_address( self, db: AsyncSession, shop_name: str, row: pd.Series, batch_id: str ): """创建新的地址记录""" # 提取敏感信息用于生成地址编码 name = str(row.get('收货人姓名', '')) address = str(row.get('收货地址', '')) phone = str(row.get('联系手机', '')) # 生成地址编码 address_encoding = self._generate_address_encoding(phone, name, address) # 创建新记录 new_address = OrderAddress( 网店名称=shop_name, upload_batch_id=batch_id, address_encoding=address_encoding ) # 处理所有字段 for excel_col, db_col in self.COLUMN_MAPPING.items(): if excel_col in row and pd.notna(row[excel_col]): value = row[excel_col] # 处理加密字段 if excel_col in self.ENCRYPTED_FIELDS: # 加密敏感信息 encrypted_value = self.encryption_service.encrypt(str(value)) setattr(new_address, f"{db_col}_encrypted", encrypted_value) else: # 处理日期字段 if '时间' in excel_col and isinstance(value, str): try: value = pd.to_datetime(value) except: pass # 处理数值字段 elif '金额' in excel_col or '积分' in excel_col or '数量' in excel_col: try: value = float(value) if pd.notna(value) else None except: pass setattr(new_address, db_col, value) db.add(new_address) async def _update_address( self, db: AsyncSession, address: OrderAddress, row: pd.Series, batch_id: str ): """更新现有的地址记录""" # 更新敏感信息和地址编码 name = str(row.get('收货人姓名', '')) addr = str(row.get('收货地址', '')) phone = str(row.get('联系手机', '')) # 重新生成地址编码 address.address_encoding = self._generate_address_encoding(phone, name, addr) address.upload_batch_id = batch_id address.updated_at = datetime.now() # 更新加密字段 for field in self.ENCRYPTED_FIELDS: if field in row and pd.notna(row[field]): encrypted_value = self.encryption_service.encrypt(str(row[field])) setattr(address, f"{field}_encrypted", encrypted_value) # 更新其他字段（不包括加密字段） for excel_col, db_col in self.COLUMN_MAPPING.items(): if excel_col not in self.ENCRYPTED_FIELDS and excel_col in row: value = row[excel_col] if pd.notna(value): # 处理日期字段 if '时间' in excel_col and isinstance(value, str): try: value = pd.to_datetime(value) except: pass # 处理数值字段 elif '金额' in excel_col or '积分' in excel_col or '数量' in excel_col: try: value = float(value) if pd.notna(value) else None except: pass setattr(address, db_col, value) def _generate_address_encoding(self, phone: str, name: str, address: str) -> str: """ 生成地址编码（SHA-256哈希） 用于匹配相同地址的订单 Args: phone: 手机号 name: 收货人姓名 address: 收货地址 Returns: SHA-256哈希值 """ # 标准化处理 phone_normalized = self._normalize_text(phone) name_normalized = self._normalize_text(name) address_normalized = self._normalize_text(address) # 生成编码字符串 encoding_str = f"{phone_normalized}|{name_normalized}|{address_normalized}" # 生成SHA-256哈希 return hashlib.sha256(encoding_str.encode('utf-8')).hexdigest() def _normalize_text(self, text: str) -> str: """ 标准化文本用于生成一致的哈希 Args: text: 原始文本 Returns: 标准化后的文本 """ if not text: return '' # 转换为字符串并去除空白 text = str(text).strip() # 移除所有空格和特殊字符 text = re.sub(r'\s+', '', text) # 转为小写 text = text.lower() # 处理虚拟号（移除-后面的扩展） if '-' in text and text.replace('-', '').isdigit(): text = text.split('-')[0] return text async def decrypt_address( self, db: AsyncSession, order_no: str, shop_name: Optional[str] = None ) -> Optional[Dict[str, str]]: """ 解密指定订单的地址信息 Args: db: 数据库会话 order_no: 订单号 shop_name: 店铺名称（可选） Returns: 解密后的地址信息 """ try: # 查询地址记录 query = select(OrderAddress).where(OrderAddress.订单编号 == order_no) if shop_name: query = query.where(OrderAddress.网店名称 == shop_name) result = await db.execute(query) address = result.scalar_one_or_none() if not address: return None # 解密敏感信息 decrypted_data = { '订单编号': address.订单编号, '收货人姓名': self.encryption_service.decrypt(address.收货人姓名_encrypted) if address.收货人姓名_encrypted else None, '收货地址': self.encryption_service.decrypt(address.收货地址_encrypted) if address.收货地址_encrypted else None, '联系电话': self.encryption_service.decrypt(address.联系电话_encrypted) if address.联系电话_encrypted else None, '联系手机': self.encryption_service.decrypt(address.联系手机_encrypted) if address.联系手机_encrypted else None, 'address_encoding': address.address_encoding } return decrypted_data except Exception as e: logger.error(f"解密地址信息失败：{str(e)}") return None async def identify_shop_from_order( self, db: AsyncSession, order_number: str ) -> Optional[str]: """ 根据订单号识别店铺 通过查询procurement_orders表匹配原始订单编号 Args: db: 数据库会话 order_number: 订单编号 Returns: 识别到的店铺名称，未找到返回None """ try: result = await db.execute( select(ProcurementOrder.网店名称) .where(ProcurementOrder.原始订单编号 == str(order_number)) .limit(1) ) shop_name = result.scalar_one_or_none() if shop_name: logger.info(f"自动识别店铺成功：订单号 {order_number} → 店铺 {shop_name}") else: logger.debug(f"未找到订单号 {order_number} 对应的店铺") return shop_name except Exception as e: logger.error(f"识别店铺时出错：{str(e)}") return None async def auto_identify_shop_from_excel( self, db: AsyncSession, df: pd.DataFrame, max_rows: int = 5 ) -> Optional[str]: """ 从Excel数据中自动识别店铺 逐行提取订单编号，直到找到匹配的店铺 Args: db: 数据库会话 df: Excel数据DataFrame max_rows: 最多检查的行数 Returns: 识别到的店铺名称，未找到返回None """ if '订单编号' not in df.columns: logger.warning("Excel中未找到'订单编号'列") return None # 提取前N行的订单编号 order_numbers = df['订单编号'].head(max_rows).astype(str).tolist() logger.info(f"开始自动识别店铺，检查前{len(order_numbers)}个订单号") # 逐个尝试识别 for order_no in order_numbers: if pd.isna(order_no) or str(order_no).strip() == '': continue shop_name = await self.identify_shop_from_order(db, order_no) if shop_name: logger.info(f"✅ 自动识别成功：通过订单号 {order_no} 识别为店铺 {shop_name}") return shop_name logger.warning(f"⚠️ 无法自动识别店铺：前{max_rows}行订单号均未匹配") return None