""" 退款订单检测服务 实现多种检测算法识别退款/退货订单 """ import logging import re from datetime import datetime, timedelta from typing import List, Dict, Optional, Tuple from decimal import Decimal from dataclasses import dataclass from sqlalchemy.orm import Session from sqlalchemy import and_, or_, func, desc, asc, text from app.models import OrderItemNorm, RefundDetection logger = logging.getLogger(__name__) @dataclass class RefundDetectionResult: """退款检测结果""" order_id: int product_key: str detection_method: str confidence_score: float refund_amount: Optional[Decimal] refund_quantity: Optional[int] refund_reason: Optional[str] matched_original_order_id: Optional[int] = None class RefundDetectionService: """退款检测服务""" # 退款关键词模式 REFUND_KEYWORDS = [ '退款', '退货', '退单', '取消', '撤销', '作废', 'refund', 'return', 'cancel', 'void', 'reverse', '补差', '差价', '调整', '修正' ] # 退款原因关键词 REFUND_REASON_PATTERNS = { '质量问题': ['质量', '破损', '缺陷', '损坏', '不良'], '尺寸不符': ['尺寸', '大小', '尺码', '不合适', '太大', '太小'], '颜色问题': ['颜色', '色差', '褪色', '颜色不对'], '客户取消': ['不要', '取消', '不想要', '改变主意'], '物流问题': ['延迟', '丢失', '损坏', '快递'], '重复下单': ['重复', '多下', '误下', '重复购买'] } def __init__(self, db: Session): self.db = db def detect_refunds_batch(self, limit: int = 1000) -> List[RefundDetectionResult]: """ 批量检测退款订单 Args: limit: 处理的订单数量限制 Returns: List[RefundDetectionResult]: 检测结果列表 """ try: # 获取待检测的订单（最近的订单，排除已检测的） orders = self._get_orders_for_detection(limit) all_results = [] for order in orders: # 多种检测方法 results = [] # 方法1: 负金额检测 result = self._detect_by_negative_amount(order) if result: results.append(result) # 方法2: 关键词匹配 result = self._detect_by_keywords(order) if result: results.append(result) # 方法3: 重复产品模式检测 result = self._detect_by_duplicate_pattern(order) if result: results.append(result) # 选择置信度最高的结果 if results: best_result = max(results, key=lambda x: x.confidence_score) all_results.append(best_result) # 批量保存检测结果 self._save_detection_results(all_results) logger.info(f"Processed {len(orders)} orders, detected {len(all_results)} potential refunds") return all_results except Exception as e: logger.error(f"Error in batch refund detection: {e}") raise def _get_orders_for_detection(self, limit: int) -> List[OrderItemNorm]: """获取待检测的订单""" # 已检测的订单ID detected_order_ids = ( self.db.query(RefundDetection.original_order_id) .union( self.db.query(RefundDetection.refund_order_id).filter( RefundDetection.refund_order_id.isnot(None) ) ) ) # 获取未检测的订单 orders = ( self.db.query(OrderItemNorm) .filter( and_( ~OrderItemNorm.id.in_(detected_order_ids), OrderItemNorm.订单日期 >= datetime.now() - timedelta(days=365) # 只检测一年内的订单 ) ) .order_by(desc(OrderItemNorm.订单日期)) .limit(limit) .all() ) return orders def _detect_by_negative_amount(self, order: OrderItemNorm) -> Optional[RefundDetectionResult]: """通过负金额检测退款""" try: # 检查数量或金额是否为负 is_negative_quantity = order.数量 and order.数量 < 0 is_negative_amount = order.订单金额 and order.订单金额 < 0 if not (is_negative_quantity or is_negative_amount): return None confidence = 0.9 # 负金额/数量通常是强退款信号 # 计算退款金额和数量 refund_amount = abs(order.订单金额) if order.订单金额 and order.订单金额 < 0 else None refund_quantity = abs(order.数量) if order.数量 and order.数量 < 0 else None return RefundDetectionResult( order_id=order.id, product_key=order.product_key, detection_method='amount_negative', confidence_score=confidence, refund_amount=refund_amount, refund_quantity=refund_quantity, refund_reason='系统检测: 负金额/负数量' ) except Exception as e: logger.error(f"Error in negative amount detection: {e}") return None def _detect_by_keywords(self, order: OrderItemNorm) -> Optional[RefundDetectionResult]: """通过关键词检测退款""" try: # 检查的文本字段 text_fields = [ order.线上宝贝名称 or '', order.备注 or '', order.原始订单编号 or '' ] combined_text = ' '.join(text_fields).lower() # 检查退款关键词 keyword_matches = [] for keyword in self.REFUND_KEYWORDS: if keyword.lower() in combined_text: keyword_matches.append(keyword) if not keyword_matches: return None # 根据匹配的关键词数量计算置信度 confidence = min(0.8, 0.3 + len(keyword_matches) * 0.1) # 尝试匹配退款原因 refund_reason = self._extract_refund_reason(combined_text) return RefundDetectionResult( order_id=order.id, product_key=order.product_key, detection_method='keyword_match', confidence_score=confidence, refund_amount=order.订单金额, refund_quantity=order.数量, refund_reason=refund_reason or f'关键词匹配: {", ".join(keyword_matches)}' ) except Exception as e: logger.error(f"Error in keyword detection: {e}") return None def _detect_by_duplicate_pattern(self, order: OrderItemNorm) -> Optional[RefundDetectionResult]: """通过重复产品模式检测退款""" try: # 查找相同产品、相同客户、时间相近的订单 time_window = timedelta(days=30) # 30天时间窗口 # 构建查询条件 similar_orders = ( self.db.query(OrderItemNorm) .filter( and_( OrderItemNorm.product_key == order.product_key, OrderItemNorm.收货人姓名 == order.收货人姓名, OrderItemNorm.收货人手机 == order.收货人手机, OrderItemNorm.订单日期.between( order.订单日期 - time_window, order.订单日期 + time_window ), OrderItemNorm.id != order.id ) ) .order_by(OrderItemNorm.订单日期) .all() ) if not similar_orders: return None # 分析模式 total_quantity = sum(o.数量 or 0 for o in similar_orders) + (order.数量 or 0) total_amount = sum(o.订单金额 or 0 for o in similar_orders) + (order.订单金额 or 0) # 检查是否存在退款模式 has_negative = any(o.数量 and o.数量 < 0 for o in similar_orders + [order]) has_zero_net = abs(total_quantity) <= 1 or abs(total_amount) <= 1 if not (has_negative or has_zero_net): return None # 计算置信度 confidence = 0.6 if has_negative else 0.4 if has_zero_net: confidence += 0.2 # 找到可能的原始订单 original_order = None if order.数量 and order.数量 < 0: # 当前订单是负数，找正数订单 original_order = next( (o for o in similar_orders if o.数量 and o.数量 > 0), None ) result = RefundDetectionResult( order_id=order.id, product_key=order.product_key, detection_method='pattern_match', confidence_score=confidence, refund_amount=order.订单金额, refund_quantity=order.数量, refund_reason='模式检测: 重复产品订单模式' ) if original_order: result.matched_original_order_id = original_order.id return result except Exception as e: logger.error(f"Error in pattern detection: {e}") return None def _extract_refund_reason(self, text: str) -> Optional[str]: """从文本中提取退款原因""" for reason, keywords in self.REFUND_REASON_PATTERNS.items(): if any(keyword in text for keyword in keywords): return reason return None def _save_detection_results(self, results: List[RefundDetectionResult]) -> None: """保存检测结果到数据库""" try: for result in results: detection = RefundDetection( original_order_id=result.matched_original_order_id or result.order_id, refund_order_id=result.order_id if result.matched_original_order_id else None, product_key=result.product_key, detection_method=result.detection_method, confidence_score=Decimal(str(result.confidence_score)), refund_amount=result.refund_amount, refund_quantity=result.refund_quantity, refund_reason=result.refund_reason, original_order_date=self._get_order_date(result.matched_original_order_id or result.order_id), refund_order_date=self._get_order_date(result.order_id) if result.matched_original_order_id else None, status='detected' ) self.db.add(detection) self.db.commit() except Exception as e: logger.error(f"Error saving detection results: {e}") self.db.rollback() raise def _get_order_date(self, order_id: int) -> Optional[datetime]: """获取订单日期""" order = self.db.query(OrderItemNorm).filter(OrderItemNorm.id == order_id).first() return order.订单日期 if order else None def verify_refund_detection(self, detection_id: int, is_verified: bool, notes: Optional[str] = None) -> bool: """ 人工验证退款检测结果 Args: detection_id: 检测记录ID is_verified: 是否确认为退款 notes: 验证备注 Returns: bool: 是否成功 """ try: detection = self.db.query(RefundDetection).filter( RefundDetection.id == detection_id ).first() if not detection: return False detection.is_verified = True detection.status = 'confirmed' if is_verified else 'ignored' if notes: detection.notes = notes self.db.commit() logger.info(f"Verified refund detection {detection_id}: {is_verified}") return True except Exception as e: logger.error(f"Error verifying refund detection: {e}") self.db.rollback() return False def get_refund_summary(self, days: int = 30) -> Dict: """获取退款检测摘要""" try: since_date = datetime.now() - timedelta(days=days) # 总检测数量 total_detected = self.db.query(RefundDetection).filter( RefundDetection.detected_at >= since_date ).count() # 按状态分组 status_counts = ( self.db.query( RefundDetection.status, func.count(RefundDetection.id).label('count') ) .filter(RefundDetection.detected_at >= since_date) .group_by(RefundDetection.status) .all() ) # 按检测方法分组 method_counts = ( self.db.query( RefundDetection.detection_method, func.count(RefundDetection.id).label('count') ) .filter(RefundDetection.detected_at >= since_date) .group_by(RefundDetection.detection_method) .all() ) # 平均置信度 avg_confidence = self.db.query( func.avg(RefundDetection.confidence_score) ).filter(RefundDetection.detected_at >= since_date).scalar() # 退款总金额 total_refund_amount = self.db.query( func.coalesce(func.sum(RefundDetection.refund_amount), 0) ).filter( and_( RefundDetection.detected_at >= since_date, RefundDetection.status == 'confirmed' ) ).scalar() return { 'period_days': days, 'total_detected': total_detected, 'status_breakdown': {row.status: row.count for row in status_counts}, 'method_breakdown': {row.detection_method: row.count for row in method_counts}, 'average_confidence': float(avg_confidence) if avg_confidence else 0.0, 'total_refund_amount': float(total_refund_amount) if total_refund_amount else 0.0 } except Exception as e: logger.error(f"Error getting refund summary: {e}") return {} def get_pending_verifications(self, limit: int = 100) -> List[Dict]: """获取待验证的退款检测记录""" detections = ( self.db.query(RefundDetection) .filter( and_( RefundDetection.status == 'detected', RefundDetection.is_verified == False ) ) .order_by(desc(RefundDetection.confidence_score), desc(RefundDetection.detected_at)) .limit(limit) .all() ) return [ { 'id': d.id, 'product_key': d.product_key, 'detection_method': d.detection_method, 'confidence_score': float(d.confidence_score), 'refund_amount': float(d.refund_amount) if d.refund_amount else None, 'refund_quantity': d.refund_quantity, 'refund_reason': d.refund_reason, 'detected_at': d.detected_at.isoformat() if d.detected_at else None, 'original_order_id': d.original_order_id, 'refund_order_id': d.refund_order_id } for d in detections ]