""" FIFO采购确认服务 实现先进先出的采购匹配算法 """ import logging 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 from app.models import ( ProcurementConfirmation, FIFOMatch, OrderItemNorm, Product, PurchaseList, RefundDetection ) logger = logging.getLogger(__name__) @dataclass class FIFOMatchResult: """FIFO匹配结果""" matched_quantity: int remaining_procurement: int remaining_orders: int cost_per_unit: Decimal match_score: float @dataclass class MatchCandidate: """匹配候选项""" order_id: int order_date: datetime product_key: str quantity: int unit_price: Decimal already_matched: int class FIFOService: """FIFO采购确认服务""" def __init__(self, db: Session): self.db = db def confirm_procurement( self, purchase_list_id: int, product_key: str, confirmed_quantity: int, unit_cost: Optional[Decimal] = None, supplier: Optional[str] = None, expected_arrival: Optional[datetime] = None, notes: Optional[str] = None ) -> ProcurementConfirmation: """ 确认采购并创建采购确认记录 Args: purchase_list_id: 采购清单ID product_key: 产品键 confirmed_quantity: 确认采购数量 unit_cost: 采购单价 supplier: 供应商 expected_arrival: 预计到货时间 notes: 备注 Returns: ProcurementConfirmation: 采购确认记录 """ try: # 创建采购确认记录 confirmation = ProcurementConfirmation( purchase_list_id=purchase_list_id, product_key=product_key, confirmed_quantity=confirmed_quantity, unit_cost=unit_cost, total_cost=unit_cost * confirmed_quantity if unit_cost else None, supplier=supplier, expected_arrival=expected_arrival, remaining_quantity=confirmed_quantity, # 初始时全部为剩余 notes=notes ) self.db.add(confirmation) self.db.flush() # 获取ID logger.info(f"Created procurement confirmation {confirmation.id} for product {product_key}") # 自动执行FIFO匹配 self.execute_fifo_matching(confirmation.id) return confirmation except Exception as e: logger.error(f"Error confirming procurement: {e}") self.db.rollback() raise def execute_fifo_matching(self, confirmation_id: int) -> List[FIFOMatch]: """ 执行FIFO匹配算法 Args: confirmation_id: 采购确认ID Returns: List[FIFOMatch]: 创建的匹配记录列表 """ try: # 获取采购确认记录 confirmation = self.db.query(ProcurementConfirmation).filter( ProcurementConfirmation.id == confirmation_id ).first() if not confirmation: raise ValueError(f"Procurement confirmation {confirmation_id} not found") # 获取待匹配的订单（FIFO排序） candidates = self._get_fifo_candidates(confirmation.product_key) matches = [] remaining_procurement = confirmation.remaining_quantity for candidate in candidates: if remaining_procurement <= 0: break # 计算可匹配数量 available_quantity = candidate.quantity - candidate.already_matched if available_quantity <= 0: continue matched_quantity = min(remaining_procurement, available_quantity) # 创建匹配记录 match = self._create_fifo_match( confirmation, candidate, matched_quantity ) if match: matches.append(match) remaining_procurement -= matched_quantity logger.info( f"FIFO matched {matched_quantity} units of {confirmation.product_key} " f"from order {candidate.order_id} to procurement {confirmation_id}" ) # 更新采购确认的剩余数量 confirmation.remaining_quantity = remaining_procurement self.db.commit() logger.info( f"FIFO matching completed for procurement {confirmation_id}. " f"Created {len(matches)} matches, {remaining_procurement} units remaining" ) return matches except Exception as e: logger.error(f"Error in FIFO matching: {e}") self.db.rollback() raise def _get_fifo_candidates(self, product_key: str) -> List[MatchCandidate]: """ 获取FIFO候选订单 按订单日期升序排列（先进先出） """ # 子查询：计算每个订单已匹配的数量 matched_subquery = ( self.db.query( FIFOMatch.normalized_order_id, func.coalesce(func.sum(FIFOMatch.matched_quantity), 0).label('matched_qty') ) .filter( and_( FIFOMatch.product_key == product_key, FIFOMatch.status == 'active' ) ) .group_by(FIFOMatch.normalized_order_id) .subquery() ) # 主查询：获取候选订单 query = ( self.db.query( OrderItemNorm.id, OrderItemNorm.订单日期, OrderItemNorm.product_key, OrderItemNorm.数量, OrderItemNorm.单价, func.coalesce(matched_subquery.c.matched_qty, 0).label('already_matched') ) .outerjoin( matched_subquery, OrderItemNorm.id == matched_subquery.c.normalized_order_id ) .filter( and_( OrderItemNorm.product_key == product_key, OrderItemNorm.数量 > 0, # 只匹配正数量的订单（排除退款） # 只匹配未被退款检测标记的订单 ~OrderItemNorm.id.in_( self.db.query(RefundDetection.original_order_id) .filter(RefundDetection.status.in_(['detected', 'confirmed'])) ) ) ) .order_by(asc(OrderItemNorm.订单日期), asc(OrderItemNorm.id)) ) candidates = [] for row in query.all(): candidates.append(MatchCandidate( order_id=row.id, order_date=row.订单日期, product_key=row.product_key, quantity=row.数量, unit_price=row.单价 or Decimal('0'), already_matched=row.already_matched )) return candidates def _create_fifo_match( self, confirmation: ProcurementConfirmation, candidate: MatchCandidate, matched_quantity: int ) -> Optional[FIFOMatch]: """创建FIFO匹配记录""" try: # 计算匹配评分（考虑时间因素、价格一致性等） match_score = self._calculate_match_score(confirmation, candidate) match = FIFOMatch( normalized_order_id=candidate.order_id, procurement_confirmation_id=confirmation.id, product_key=candidate.product_key, matched_quantity=matched_quantity, unit_cost=confirmation.unit_cost, match_score=match_score, order_date=candidate.order_date, procurement_date=confirmation.confirmed_at, match_type='auto', status='active' ) self.db.add(match) return match except Exception as e: logger.error(f"Error creating FIFO match: {e}") return None def _calculate_match_score( self, confirmation: ProcurementConfirmation, candidate: MatchCandidate ) -> Decimal: """ 计算匹配评分 考虑因素：时间间隔、价格差异、数量匹配度等 """ score = Decimal('1.0') # 时间因素：订单越早，评分越高 if confirmation.confirmed_at and candidate.order_date: time_diff = (confirmation.confirmed_at - candidate.order_date).days if time_diff > 0: # 时间间隔越短，评分越高 time_score = max(Decimal('0.5'), Decimal('1.0') - Decimal(str(time_diff)) / Decimal('365')) score *= time_score # 价格因素：采购价格与订单价格的合理性 if confirmation.unit_cost and candidate.unit_price: price_ratio = confirmation.unit_cost / candidate.unit_price # 采购价格应该合理（不能过高或过低） if Decimal('0.5') <= price_ratio <= Decimal('2.0'): price_score = Decimal('1.0') else: price_score = Decimal('0.8') score *= price_score return min(score, Decimal('1.0')) def get_procurement_summary(self, product_key: str) -> Dict: """获取产品的采购匹配摘要""" try: # 总采购确认数量 total_confirmed = self.db.query( func.coalesce(func.sum(ProcurementConfirmation.confirmed_quantity), 0) ).filter( and_( ProcurementConfirmation.product_key == product_key, ProcurementConfirmation.status.in_(['confirmed', 'shipped', 'arrived']) ) ).scalar() # 总匹配数量 total_matched = self.db.query( func.coalesce(func.sum(FIFOMatch.matched_quantity), 0) ).filter( and_( FIFOMatch.product_key == product_key, FIFOMatch.status == 'active' ) ).scalar() # 剩余未匹配采购 remaining_procurement = self.db.query( func.coalesce(func.sum(ProcurementConfirmation.remaining_quantity), 0) ).filter( and_( ProcurementConfirmation.product_key == product_key, ProcurementConfirmation.status.in_(['confirmed', 'shipped', 'arrived']) ) ).scalar() # 未匹配订单数量 unmatched_orders = self._get_unmatched_order_quantity(product_key) return { 'product_key': product_key, 'total_confirmed': int(total_confirmed), 'total_matched': int(total_matched), 'remaining_procurement': int(remaining_procurement), 'unmatched_orders': unmatched_orders, 'match_rate': float(total_matched / total_confirmed) if total_confirmed > 0 else 0.0 } except Exception as e: logger.error(f"Error getting procurement summary: {e}") return {} def _get_unmatched_order_quantity(self, product_key: str) -> int: """计算未匹配的订单数量""" # 子查询：每个订单的已匹配数量 matched_subquery = ( self.db.query( FIFOMatch.normalized_order_id, func.sum(FIFOMatch.matched_quantity).label('matched_qty') ) .filter( and_( FIFOMatch.product_key == product_key, FIFOMatch.status == 'active' ) ) .group_by(FIFOMatch.normalized_order_id) .subquery() ) # 主查询：计算未匹配数量 unmatched_query = ( self.db.query( func.sum( OrderItemNorm.数量 - func.coalesce(matched_subquery.c.matched_qty, 0) ) ) .outerjoin( matched_subquery, OrderItemNorm.id == matched_subquery.c.normalized_order_id ) .filter( and_( OrderItemNorm.product_key == product_key, OrderItemNorm.数量 > 0, OrderItemNorm.数量 > func.coalesce(matched_subquery.c.matched_qty, 0) ) ) ) result = unmatched_query.scalar() return int(result) if result else 0 def reprocess_matches_for_product(self, product_key: str) -> Dict: """重新处理产品的所有匹配（用于修正或重新计算）""" try: # 删除现有的自动匹配记录 deleted_count = self.db.query(FIFOMatch).filter( and_( FIFOMatch.product_key == product_key, FIFOMatch.match_type == 'auto' ) ).delete() # 重置采购确认的剩余数量 confirmations = self.db.query(ProcurementConfirmation).filter( ProcurementConfirmation.product_key == product_key ).all() for conf in confirmations: conf.remaining_quantity = conf.confirmed_quantity self.db.flush() # 重新执行匹配 total_matches = 0 for conf in confirmations: matches = self.execute_fifo_matching(conf.id) total_matches += len(matches) return { 'product_key': product_key, 'deleted_matches': deleted_count, 'new_matches': total_matches, 'processed_confirmations': len(confirmations) } except Exception as e: logger.error(f"Error reprocessing matches: {e}") self.db.rollback() raise def get_fifo_matches_for_order(self, order_id: int) -> List[Dict]: """获取订单的FIFO匹配记录""" matches = self.db.query(FIFOMatch).filter( FIFOMatch.normalized_order_id == order_id ).all() return [ { 'id': match.id, 'product_key': match.product_key, 'matched_quantity': match.matched_quantity, 'unit_cost': float(match.unit_cost) if match.unit_cost else None, 'match_score': float(match.match_score), 'match_type': match.match_type, 'status': match.status, 'procurement_date': match.procurement_date.isoformat() if match.procurement_date else None, 'matched_at': match.matched_at.isoformat() if match.matched_at else None } for match in matches ]