""" 库存登记服务层 """ import uuid from datetime import date from typing import Optional, List, Dict, Any from sqlalchemy import select, func, and_, or_, cast, String, desc from sqlalchemy.ext.asyncio import AsyncSession from app.models.inventory_registrations import InventoryBatch, InventoryRegistration, beijing_now from app.models.pending_shipments import PendingShipment from app.models.products_master import ProductMaster from app.models.procurement_orders import ProcurementOrder class InventoryRegistrationService: """库存登记服务""" async def scan_match_product( self, db: AsyncSession, scanned_code: str ) -> Optional[Dict[str, Any]]: """ 扫码匹配商品 - 从历史发货数据中查找 """ if not scanned_code: return None # 搜索待发货订单中的商品扫码记录 search_term = f"%{scanned_code}%" query = select(PendingShipment).where( cast(PendingShipment.商品扫码记录, String).like(search_term) ).order_by(desc(PendingShipment.id)).limit(1) result = await db.execute(query) shipment = result.scalar_one_or_none() if shipment: # 通过 procurement_order_id 获取商品信息 if shipment.procurement_order_id: proc_query = select(ProcurementOrder).where( ProcurementOrder.id == shipment.procurement_order_id ) proc_result = await db.execute(proc_query) proc_order = proc_result.scalar_one_or_none() if proc_order: return { 'found': True, 'source': 'shipment', 'source_order_id': shipment.id, 'brand': proc_order.品牌, 'product_code': proc_order.货号, 'color': proc_order.颜色, 'size': proc_order.尺寸, 'product_name': proc_order.线上宝贝名称, 'scanned_code': scanned_code } # 如果没有关联采购订单，使用待发货订单的商品标题 return { 'found': True, 'source': 'shipment', 'source_order_id': shipment.id, 'brand': None, 'product_code': shipment.商家编码, 'color': None, 'size': None, 'product_name': shipment.商品标题, 'scanned_code': scanned_code } # 补充查询：从库存登记记录中查找条码（精确匹配） reg_query = select(InventoryRegistration).where( InventoryRegistration.scanned_code == scanned_code ).order_by(desc(InventoryRegistration.id)).limit(1) reg_result = await db.execute(reg_query) registration = reg_result.scalar_one_or_none() if registration: return { 'found': True, 'source': 'inventory', 'source_order_id': registration.id, 'brand': registration.brand, 'product_code': registration.product_code, 'color': registration.color, 'size': registration.size, 'product_name': registration.product_name, 'scanned_code': scanned_code } return {'found': False, 'scanned_code': scanned_code} async def search_products( self, db: AsyncSession, keyword: str, limit: int = 20 ) -> List[Dict[str, Any]]: """ 搜索商品列表 """ if not keyword: return [] search_term = f"%{keyword}%" # 从 products_master 搜索（使用中文字段名） query = select(ProductMaster).where( or_( ProductMaster.品牌.like(search_term), ProductMaster.货号.like(search_term), ProductMaster.线上宝贝名称.like(search_term), ProductMaster.颜色.like(search_term) ) ).limit(limit) result = await db.execute(query) products = result.scalars().all() return [ { 'id': p.id, 'brand': p.品牌, 'product_code': p.货号, 'color': p.颜色, 'size': p.尺寸, 'product_name': p.线上宝贝名称 } for p in products ] async def get_or_create_active_batch(self, db: AsyncSession) -> InventoryBatch: """ 获取或创建活跃批次（不限日期，获取最新的pending批次） """ # 查找最新的pending批次（按创建时间倒序） query = select(InventoryBatch).where( InventoryBatch.status == 'pending' ).order_by(desc(InventoryBatch.created_at)).limit(1) result = await db.execute(query) batch = result.scalar_one_or_none() if not batch: # 创建新批次 batch = InventoryBatch( batch_id=str(uuid.uuid4()), batch_date=date.today(), status='pending' ) db.add(batch) await db.commit() await db.refresh(batch) return batch async def add_registration( self, db: AsyncSession, brand: Optional[str] = None, product_code: Optional[str] = None, color: Optional[str] = None, size: Optional[str] = None, product_name: Optional[str] = None, quantity: int = 1, scanned_code: Optional[str] = None, source_type: str = 'scan', source_order_id: Optional[int] = None, operator: Optional[str] = None, remark: Optional[str] = None ) -> Dict[str, Any]: """ 添加库存登记记录 """ # 获取活跃批次 batch = await self.get_or_create_active_batch(db) # 创建登记记录 registration = InventoryRegistration( batch_id=batch.batch_id, scanned_code=scanned_code, brand=brand, product_code=product_code, color=color, size=size, product_name=product_name, quantity=quantity, source_type=source_type, source_order_id=source_order_id, operator=operator, remark=remark ) db.add(registration) # 更新批次统计 batch.total_items += 1 batch.total_quantity += quantity await db.commit() await db.refresh(registration) return registration.to_dict() async def get_active_list( self, db: AsyncSession ) -> Dict[str, Any]: """ 获取当前活跃批次的登记列表 """ batch = await self.get_or_create_active_batch(db) query = select(InventoryRegistration).where( InventoryRegistration.batch_id == batch.batch_id ).order_by(desc(InventoryRegistration.id)) result = await db.execute(query) registrations = result.scalars().all() return { 'batch': batch.to_dict(), 'registrations': [r.to_dict() for r in registrations] } async def delete_registration( self, db: AsyncSession, registration_id: int ) -> bool: """ 删除登记记录 """ query = select(InventoryRegistration).where( InventoryRegistration.id == registration_id ) result = await db.execute(query) registration = result.scalar_one_or_none() if not registration: return False # 更新批次统计 batch_query = select(InventoryBatch).where( InventoryBatch.batch_id == registration.batch_id ) batch_result = await db.execute(batch_query) batch = batch_result.scalar_one_or_none() if batch and batch.status == 'pending': batch.total_items -= 1 batch.total_quantity -= registration.quantity await db.delete(registration) await db.commit() return True async def update_quantity( self, db: AsyncSession, registration_id: int, quantity: int ) -> Optional[Dict[str, Any]]: """ 更新登记数量 """ query = select(InventoryRegistration).where( InventoryRegistration.id == registration_id ) result = await db.execute(query) registration = result.scalar_one_or_none() if not registration: return None # 更新批次统计 batch_query = select(InventoryBatch).where( InventoryBatch.batch_id == registration.batch_id ) batch_result = await db.execute(batch_query) batch = batch_result.scalar_one_or_none() if batch and batch.status == 'pending': diff = quantity - registration.quantity batch.total_quantity += diff registration.quantity = quantity await db.commit() await db.refresh(registration) return registration.to_dict() async def confirm_batch( self, db: AsyncSession, batch_id: str, confirmed_by: Optional[str] = None ) -> Optional[Dict[str, Any]]: """ 确认批次 """ query = select(InventoryBatch).where( InventoryBatch.batch_id == batch_id ) result = await db.execute(query) batch = result.scalar_one_or_none() if not batch: return None batch.status = 'confirmed' batch.confirmed_at = beijing_now() batch.confirmed_by = confirmed_by await db.commit() await db.refresh(batch) return batch.to_dict() async def get_batches( self, db: AsyncSession, status: Optional[str] = None, page: int = 1, page_size: int = 20 ) -> Dict[str, Any]: """ 获取历史批次列表（排除当前活跃的pending批次，按创建时间倒序） 当前活跃批次在"登记列表"中显示，避免重复 """ # 先找到当前活跃的pending批次（最新的pending批次） active_query = select(InventoryBatch.batch_id).where( InventoryBatch.status == 'pending' ).order_by(desc(InventoryBatch.created_at)).limit(1) active_result = await db.execute(active_query) active_batch_id = active_result.scalar_one_or_none() # 构建排除条件 exclude_condition = True # 默认不排除 if active_batch_id: exclude_condition = InventoryBatch.batch_id != active_batch_id # 统计总数（排除活跃批次） count_query = select(func.count(InventoryBatch.id)).where(exclude_condition) if status: count_query = count_query.where(InventoryBatch.status == status) total_result = await db.execute(count_query) total = total_result.scalar() or 0 # 分页查询（排除活跃批次） query = select(InventoryBatch).where(exclude_condition) if status: query = query.where(InventoryBatch.status == status) # 按创建时间倒序 query = query.order_by(desc(InventoryBatch.created_at)).offset( (page - 1) * page_size ).limit(page_size) result = await db.execute(query) batches = result.scalars().all() return { 'total': total, 'page': page, 'page_size': page_size, 'batches': [b.to_dict() for b in batches] } async def get_batch_detail( self, db: AsyncSession, batch_id: str ) -> Optional[Dict[str, Any]]: """ 获取批次详情 """ # 获取批次 batch_query = select(InventoryBatch).where( InventoryBatch.batch_id == batch_id ) batch_result = await db.execute(batch_query) batch = batch_result.scalar_one_or_none() if not batch: return None # 获取登记记录（按品牌、货号、颜色、尺码排序） reg_query = select(InventoryRegistration).where( InventoryRegistration.batch_id == batch_id ).order_by( InventoryRegistration.brand, InventoryRegistration.product_code, InventoryRegistration.color, InventoryRegistration.size ) reg_result = await db.execute(reg_query) registrations = reg_result.scalars().all() return { 'batch': batch.to_dict(), 'registrations': [r.to_dict() for r in registrations] } async def export_batch_data( self, db: AsyncSession, batch_id: str ) -> Optional[List[Dict[str, Any]]]: """ 导出批次数据（用于Excel导出） """ detail = await self.get_batch_detail(db, batch_id) if not detail: return None return detail['registrations'] async def delete_batch( self, db: AsyncSession, batch_id: str ) -> bool: """ 删除批次及其所有登记记录 """ # 获取批次 batch_query = select(InventoryBatch).where( InventoryBatch.batch_id == batch_id ) batch_result = await db.execute(batch_query) batch = batch_result.scalar_one_or_none() if not batch: return False # 删除所有关联的登记记录 reg_query = select(InventoryRegistration).where( InventoryRegistration.batch_id == batch_id ) reg_result = await db.execute(reg_query) registrations = reg_result.scalars().all() for reg in registrations: await db.delete(reg) # 删除批次 await db.delete(batch) await db.commit() return True async def unconfirm_batch( self, db: AsyncSession, batch_id: str ) -> Dict[str, Any]: """ 取消确认批次（将状态改回pending，并使其成为活跃批次） """ query = select(InventoryBatch).where( InventoryBatch.batch_id == batch_id ) result = await db.execute(query) batch = result.scalar_one_or_none() if not batch: return {'success': False, 'error': '批次不存在'} batch.status = 'pending' batch.confirmed_at = None batch.confirmed_by = None # 更新 created_at 使其成为最新的 pending 批次（活跃批次） batch.created_at = beijing_now() await db.commit() await db.refresh(batch) return {'success': True, 'data': batch.to_dict()} async def mark_listed( self, db: AsyncSession, batch_id: str ) -> Dict[str, Any]: """ 标记批次已上架 """ query = select(InventoryBatch).where( InventoryBatch.batch_id == batch_id ) result = await db.execute(query) batch = result.scalar_one_or_none() if not batch: return {'success': False, 'error': '批次不存在'} if batch.status != 'confirmed': return {'success': False, 'error': '只有已确认的批次才能标记为已上架'} batch.is_listed = True batch.listed_at = beijing_now() await db.commit() await db.refresh(batch) return {'success': True, 'data': batch.to_dict()} async def get_all_registrations( self, db: AsyncSession, batch_id: Optional[str] = None, brand: Optional[str] = None, product_code: Optional[str] = None, source_type: Optional[str] = None, start_date: Optional[date] = None, end_date: Optional[date] = None, page: int = 1, page_size: int = 20 ) -> Dict[str, Any]: """ 获取所有库存登记记录（带筛选和分页） """ # 构建基础查询 conditions = [] if batch_id: conditions.append(InventoryRegistration.batch_id == batch_id) if brand: conditions.append(InventoryRegistration.brand.like(f"%{brand}%")) if product_code: conditions.append(InventoryRegistration.product_code.like(f"%{product_code}%")) if source_type: conditions.append(InventoryRegistration.source_type == source_type) if start_date: conditions.append(InventoryRegistration.created_at >= start_date) if end_date: # 结束日期需要加一天，以包含当天的记录 from datetime import timedelta end_datetime = end_date + timedelta(days=1) conditions.append(InventoryRegistration.created_at < end_datetime) # 统计总数 count_query = select(func.count(InventoryRegistration.id)) if conditions: count_query = count_query.where(and_(*conditions)) total_result = await db.execute(count_query) total = total_result.scalar() or 0 # 查询记录（关联批次信息） query = select(InventoryRegistration, InventoryBatch).join( InventoryBatch, InventoryRegistration.batch_id == InventoryBatch.batch_id ) if conditions: query = query.where(and_(*conditions)) # 按创建时间倒序 query = query.order_by(desc(InventoryRegistration.created_at)).offset( (page - 1) * page_size ).limit(page_size) result = await db.execute(query) rows = result.all() # 组装返回数据 items = [] for reg, batch in rows: item = reg.to_dict() item['batch_date'] = batch.batch_date.isoformat() if batch.batch_date else None item['batch_status'] = batch.status item['is_listed'] = batch.is_listed items.append(item) return { 'total': total, 'page': page, 'page_size': page_size, 'items': items } async def get_batch_options(self, db: AsyncSession) -> List[Dict[str, Any]]: """ 获取批次选项列表（用于下拉筛选） """ query = select(InventoryBatch).order_by(desc(InventoryBatch.created_at)).limit(100) result = await db.execute(query) batches = result.scalars().all() return [ { 'batch_id': b.batch_id, 'batch_date': b.batch_date.isoformat() if b.batch_date else None, 'status': b.status, 'total_items': b.total_items, 'total_quantity': b.total_quantity } for b in batches ]