from datetime import datetime, time, timedelta from typing import Any from sqlalchemy import ColumnElement, Select, and_, case, false, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import aliased from sqlalchemy.orm.attributes import InstrumentedAttribute from common.vo import PageModel from module_admin.entity.do.user_do import SysUser from module_business.domain.enums import DocumentStatus, PalletEventType from module_business.domain.test_batch_scope import TestBatchScope, batch_scope_condition from module_business.entity.do.master_do import ( BizCustomer, BizCustomerSite, BizPallet, BizProductionLine, BizTunnel, BizWarehouse, ) from module_business.entity.do.operation_do import ( BizFillingRecord, BizPalletEvent, BizPalletInventory, BizReturnItem, BizReturnOrder, BizShipmentItem, BizShipmentOrder, BizUsageRecord, ) from module_business.entity.vo.report_vo import ( ConsumptionReportModel, ConsumptionReportQueryModel, FillingReportModel, FillingReportQueryModel, PalletInventoryReportModel, PalletInventoryReportQueryModel, PalletTraceReportModel, PalletTraceReportQueryModel, ReturnReportModel, ReturnReportQueryModel, ShipmentReportModel, ShipmentReportQueryModel, TunnelReportQueryModel, ) from utils.page_util import PageUtil class ReportDao: """六类只读业务报表数据库查询。""" @staticmethod def _time_conditions( column: InstrumentedAttribute, begin_time: str | None, end_time: str | None, ) -> list[ColumnElement[bool]]: conditions = [] if begin_time: begin_at = datetime.combine(datetime.strptime(begin_time, '%Y-%m-%d').date(), time.min) conditions.append(column >= begin_at) if end_time: end_at = datetime.combine( datetime.strptime(end_time, '%Y-%m-%d').date() + timedelta(days=1), time.min, ) conditions.append(column < end_at) return conditions @staticmethod def _scope_condition( column: InstrumentedAttribute, scope_ids: list[int] | None, ) -> ColumnElement[bool] | None: if scope_ids is None: return None normalized_ids = sorted(set(scope_ids)) return column.in_(normalized_ids) if normalized_ids else false() @staticmethod def _joined_scope_condition( id_column: InstrumentedAttribute, scope_column: InstrumentedAttribute, scope: TestBatchScope, ) -> ColumnElement[bool]: """保留空的可选 join，同时拒绝已存在的跨批次关联记录。""" return or_(id_column.is_(None), batch_scope_condition(scope_column, scope)) @staticmethod def _tunnel_conditions( query: TunnelReportQueryModel, trace_column: ColumnElement, *tunnels: tuple[ColumnElement, Any], ) -> list[ColumnElement[bool]]: """历史事件可匹配变更前/后隧道；同一组编号、名称须命中同一个隧道。""" conditions = [] if query.trace_no: conditions.append(trace_column == query.trace_no) matches = [] for id_column, tunnel in tunnels: filters = [] if query.tunnel_id: filters.append(id_column == query.tunnel_id) if query.tunnel_code: filters.append(tunnel.tunnel_code.like(f'%{query.tunnel_code}%')) if query.tunnel_name: filters.append(tunnel.tunnel_name.like(f'%{query.tunnel_name}%')) if filters: matches.append(and_(*filters)) if matches: conditions.append(or_(*matches)) return conditions @classmethod def _or_scope_conditions( cls, *column_scopes: tuple[InstrumentedAttribute, list[int] | None], ) -> list[ColumnElement[bool]]: conditions = [ condition for column, scope_ids in column_scopes if (condition := cls._scope_condition(column, scope_ids)) is not None ] return [or_(*conditions)] if conditions else [] @classmethod def _pallet_trace_scope_conditions( cls, warehouse_ids: list[int] | None, site_ids: list[int] | None, ) -> list[ColumnElement[bool]]: participant_scopes = [ condition for column, scope_ids in ( (BizPalletEvent.before_warehouse_id, warehouse_ids), (BizPalletEvent.after_warehouse_id, warehouse_ids), (BizPalletEvent.before_site_id, site_ids), (BizPalletEvent.after_site_id, site_ids), ) if (condition := cls._scope_condition(column, scope_ids)) is not None ] if not participant_scopes: return [] shipment_scopes = [ condition for column, scope_ids in ( (BizPalletEvent.before_warehouse_id, warehouse_ids), (BizPalletEvent.after_site_id, site_ids), ) if (condition := cls._scope_condition(column, scope_ids)) is not None ] shipment_type = PalletEventType.SHIPMENT_CONFIRMED.value return [ or_( and_(BizPalletEvent.event_type == shipment_type, *shipment_scopes), and_(BizPalletEvent.event_type != shipment_type, or_(*participant_scopes)), ) ] @staticmethod async def _paginate( db: AsyncSession, statement: Select, query_object: Any, ) -> PageModel: return await PageUtil.paginate( db, statement, query_object.page_num, query_object.page_size, True, exclude_fields={'test_batch_id'}, ) @classmethod async def get_pallet_inventory_list( cls, db: AsyncSession, query_object: PalletInventoryReportQueryModel, scope: TestBatchScope, warehouse_ids: list[int] | None = None, site_ids: list[int] | None = None, ) -> PageModel[PalletInventoryReportModel]: current_warehouse = aliased(BizWarehouse, name='current_warehouse') current_site = aliased(BizCustomerSite, name='current_site') current_line = aliased(BizProductionLine, name='current_line') current_tunnel = aliased(BizTunnel, name='current_tunnel') active_filling = aliased(BizFillingRecord, name='active_filling') conditions = [ batch_scope_condition(BizPalletInventory.test_batch_id, scope), batch_scope_condition(BizPallet.test_batch_id, scope), cls._joined_scope_condition(active_filling.filling_id, active_filling.test_batch_id, scope), cls._joined_scope_condition(current_warehouse.warehouse_id, current_warehouse.test_batch_id, scope), cls._joined_scope_condition(current_site.site_id, current_site.test_batch_id, scope), cls._joined_scope_condition(current_line.line_id, current_line.test_batch_id, scope), cls._joined_scope_condition(current_tunnel.tunnel_id, current_tunnel.test_batch_id, scope), ] conditions.extend(cls._tunnel_conditions( query_object, BizPalletInventory.trace_no, (BizPalletInventory.tunnel_id, current_tunnel), )) conditions.extend( cls._or_scope_conditions( (BizPalletInventory.warehouse_id, warehouse_ids), (BizPalletInventory.site_id, site_ids), ) ) if query_object.pallet_code: conditions.append(BizPallet.pallet_code.like(f'%{query_object.pallet_code}%')) if query_object.flow_state: conditions.append(BizPalletInventory.flow_state == query_object.flow_state.value) if query_object.warehouse_id: conditions.append(BizPalletInventory.warehouse_id == query_object.warehouse_id) if query_object.site_id: conditions.append(BizPalletInventory.site_id == query_object.site_id) if query_object.line_id: conditions.append(BizPalletInventory.current_line_id == query_object.line_id) if query_object.strain_id: conditions.append(active_filling.strain_id == query_object.strain_id) if query_object.batch_no: conditions.append(active_filling.batch_no.like(f'%{query_object.batch_no}%')) conditions.extend( cls._time_conditions( BizPalletInventory.updated_at, query_object.begin_time, query_object.end_time, ) ) statement = ( select( BizPalletInventory.pallet_id.label('pallet_id'), BizPallet.pallet_code.label('pallet_code'), BizPalletInventory.flow_state.label('flow_state'), BizPalletInventory.warehouse_id.label('warehouse_id'), current_warehouse.warehouse_name.label('warehouse_name'), BizPalletInventory.site_id.label('site_id'), current_site.site_name.label('site_name'), BizPalletInventory.current_line_id.label('line_id'), current_line.line_name.label('line_name'), active_filling.strain_id.label('strain_id'), active_filling.strain_code_snapshot.label('strain_code'), active_filling.strain_name_snapshot.label('strain_name'), active_filling.batch_no.label('batch_no'), active_filling.initial_packages.label('initial_packages'), BizPalletInventory.remaining_packages.label('remaining_packages'), BizPalletInventory.updated_at.label('updated_at'), BizPalletInventory.tunnel_id.label('tunnel_id'), current_tunnel.tunnel_code.label('tunnel_code'), current_tunnel.tunnel_name.label('tunnel_name'), BizPalletInventory.trace_no.label('trace_no'), active_filling.production_date.label('production_date'), active_filling.expiry_date.label('expiry_date'), ) .select_from(BizPalletInventory) .join(BizPallet, BizPallet.pallet_id == BizPalletInventory.pallet_id) .outerjoin(active_filling, active_filling.filling_id == BizPalletInventory.active_filling_id) .outerjoin(current_warehouse, current_warehouse.warehouse_id == BizPalletInventory.warehouse_id) .outerjoin(current_site, current_site.site_id == BizPalletInventory.site_id) .outerjoin(current_line, current_line.line_id == BizPalletInventory.current_line_id) .outerjoin(current_tunnel, current_tunnel.tunnel_id == BizPalletInventory.tunnel_id) .where(*conditions) .order_by(BizPalletInventory.pallet_id) ) return await cls._paginate(db, statement, query_object) @classmethod async def get_filling_report_list( cls, db: AsyncSession, query_object: FillingReportQueryModel, scope: TestBatchScope, warehouse_ids: list[int] | None = None, ) -> PageModel[FillingReportModel]: operator_user = aliased(SysUser, name='operator_user') conditions = [ batch_scope_condition(BizFillingRecord.test_batch_id, scope), batch_scope_condition(BizPallet.test_batch_id, scope), cls._joined_scope_condition(BizWarehouse.warehouse_id, BizWarehouse.test_batch_id, scope), ] warehouse_scope = cls._scope_condition(BizFillingRecord.warehouse_id, warehouse_ids) if warehouse_scope is not None: conditions.append(warehouse_scope) if query_object.warehouse_id: conditions.append(BizFillingRecord.warehouse_id == query_object.warehouse_id) if query_object.pallet_code: conditions.append(BizPallet.pallet_code.like(f'%{query_object.pallet_code}%')) if query_object.strain_id: conditions.append(BizFillingRecord.strain_id == query_object.strain_id) if query_object.batch_no: conditions.append(BizFillingRecord.batch_no.like(f'%{query_object.batch_no}%')) conditions.extend( cls._time_conditions( BizFillingRecord.operated_at, query_object.begin_time, query_object.end_time, ) ) statement = ( select( BizFillingRecord.filling_id.label('filling_id'), BizFillingRecord.filling_no.label('filling_no'), BizFillingRecord.operated_at.label('operated_at'), BizFillingRecord.warehouse_id.label('warehouse_id'), BizWarehouse.warehouse_name.label('warehouse_name'), BizFillingRecord.pallet_id.label('pallet_id'), BizPallet.pallet_code.label('pallet_code'), BizFillingRecord.strain_id.label('strain_id'), BizFillingRecord.strain_code_snapshot.label('strain_code_snapshot'), BizFillingRecord.strain_name_snapshot.label('strain_name_snapshot'), BizFillingRecord.batch_no.label('batch_no'), BizFillingRecord.package_spec.label('package_spec'), BizFillingRecord.initial_packages.label('initial_packages'), BizFillingRecord.operator_id.label('operator_id'), operator_user.user_name.label('operator_name'), ) .select_from(BizFillingRecord) .join(BizPallet, BizPallet.pallet_id == BizFillingRecord.pallet_id) .outerjoin(BizWarehouse, BizWarehouse.warehouse_id == BizFillingRecord.warehouse_id) .outerjoin(operator_user, operator_user.user_id == BizFillingRecord.operator_id) .where(*conditions) .order_by(BizFillingRecord.filling_id) ) return await cls._paginate(db, statement, query_object) @classmethod async def get_shipment_report_list( cls, db: AsyncSession, query_object: ShipmentReportQueryModel, scope: TestBatchScope, warehouse_ids: list[int] | None = None, site_ids: list[int] | None = None, ) -> PageModel[ShipmentReportModel]: source_warehouse = aliased(BizWarehouse, name='source_warehouse') target_site = aliased(BizCustomerSite, name='target_site') target_customer = aliased(BizCustomer, name='target_customer') conditions = [ batch_scope_condition(BizShipmentOrder.test_batch_id, scope), batch_scope_condition(BizShipmentItem.test_batch_id, scope), batch_scope_condition(BizPallet.test_batch_id, scope), cls._joined_scope_condition(source_warehouse.warehouse_id, source_warehouse.test_batch_id, scope), cls._joined_scope_condition(target_site.site_id, target_site.test_batch_id, scope), cls._joined_scope_condition(target_customer.customer_id, target_customer.test_batch_id, scope), BizShipmentOrder.status == DocumentStatus.CONFIRMED.value, ] warehouse_scope = cls._scope_condition(BizShipmentOrder.source_warehouse_id, warehouse_ids) site_scope = cls._scope_condition(BizShipmentOrder.target_site_id, site_ids) if warehouse_scope is not None: conditions.append(warehouse_scope) if site_scope is not None: conditions.append(site_scope) if query_object.shipment_no: conditions.append(BizShipmentOrder.shipment_no.like(f'%{query_object.shipment_no}%')) if query_object.source_warehouse_id: conditions.append(BizShipmentOrder.source_warehouse_id == query_object.source_warehouse_id) if query_object.target_site_id: conditions.append(BizShipmentOrder.target_site_id == query_object.target_site_id) if query_object.pallet_code: conditions.append(BizPallet.pallet_code.like(f'%{query_object.pallet_code}%')) if query_object.status: conditions.append(BizShipmentOrder.status == query_object.status.value) conditions.extend( cls._time_conditions( BizShipmentOrder.confirmed_at, query_object.begin_time, query_object.end_time, ) ) # 报表行契约要求确认时间和正数包数，仅确认后的事实明细可形成报表行。 conditions.extend( ( BizShipmentOrder.confirmed_at.is_not(None), BizShipmentItem.package_qty.is_not(None), ) ) statement = ( select( BizShipmentOrder.shipment_id.label('shipment_id'), BizShipmentOrder.shipment_no.label('shipment_no'), BizShipmentOrder.confirmed_at.label('confirmed_at'), BizShipmentOrder.source_warehouse_id.label('source_warehouse_id'), source_warehouse.warehouse_name.label('source_warehouse_name'), BizShipmentOrder.target_site_id.label('target_site_id'), target_site.site_name.label('target_site_name'), target_site.customer_id.label('customer_id'), target_customer.customer_name.label('customer_name'), BizShipmentItem.pallet_id.label('pallet_id'), BizPallet.pallet_code.label('pallet_code'), BizShipmentItem.active_filling_id.label('active_filling_id'), BizShipmentItem.package_qty.label('package_qty'), ) .select_from(BizShipmentOrder) .join(BizShipmentItem, BizShipmentItem.shipment_id == BizShipmentOrder.shipment_id) .join(BizPallet, BizPallet.pallet_id == BizShipmentItem.pallet_id) .outerjoin(source_warehouse, source_warehouse.warehouse_id == BizShipmentOrder.source_warehouse_id) .outerjoin(target_site, target_site.site_id == BizShipmentOrder.target_site_id) .outerjoin(target_customer, target_customer.customer_id == target_site.customer_id) .where(*conditions) .order_by(BizShipmentOrder.shipment_id, BizShipmentItem.shipment_item_id) ) return await cls._paginate(db, statement, query_object) @classmethod async def get_consumption_report_list( cls, db: AsyncSession, query_object: ConsumptionReportQueryModel, scope: TestBatchScope, site_ids: list[int] | None = None, ) -> PageModel[ConsumptionReportModel]: operator_user = aliased(SysUser, name='operator_user') usage_tunnel = aliased(BizTunnel, name='usage_tunnel') # 新记录优先使用业务快照；历史空值只回查该次使用引用的装填记录。 batch_no = func.coalesce(BizUsageRecord.batch_no_snapshot, BizFillingRecord.batch_no) conditions = [ batch_scope_condition(BizUsageRecord.test_batch_id, scope), batch_scope_condition(BizPallet.test_batch_id, scope), batch_scope_condition(BizFillingRecord.test_batch_id, scope), batch_scope_condition(BizCustomerSite.test_batch_id, scope), cls._joined_scope_condition(BizProductionLine.line_id, BizProductionLine.test_batch_id, scope), cls._joined_scope_condition(usage_tunnel.tunnel_id, usage_tunnel.test_batch_id, scope), cls._joined_scope_condition(BizCustomer.customer_id, BizCustomer.test_batch_id, scope), ] conditions.extend(cls._tunnel_conditions( query_object, BizUsageRecord.trace_no, (BizUsageRecord.tunnel_id, usage_tunnel), )) site_scope = cls._scope_condition(BizUsageRecord.site_id, site_ids) if site_scope is not None: conditions.append(site_scope) if query_object.customer_id: conditions.append(BizCustomerSite.customer_id == query_object.customer_id) if query_object.site_id: conditions.append(BizUsageRecord.site_id == query_object.site_id) if query_object.line_id: conditions.append(BizUsageRecord.line_id == query_object.line_id) if query_object.pallet_code: conditions.append(BizPallet.pallet_code.like(f'%{query_object.pallet_code}%')) if query_object.strain_id: conditions.append(BizFillingRecord.strain_id == query_object.strain_id) if query_object.batch_no: conditions.append(batch_no.like(f'%{query_object.batch_no}%')) conditions.extend( cls._time_conditions( BizUsageRecord.occurred_at, query_object.begin_time, query_object.end_time, ) ) statement = ( select( BizUsageRecord.usage_id.label('usage_id'), BizUsageRecord.usage_no.label('usage_no'), BizUsageRecord.occurred_at.label('occurred_at'), BizCustomerSite.customer_id.label('customer_id'), BizCustomer.customer_name.label('customer_name'), BizUsageRecord.site_id.label('site_id'), BizCustomerSite.site_name.label('site_name'), BizUsageRecord.line_id.label('line_id'), BizProductionLine.line_name.label('line_name'), BizUsageRecord.pallet_id.label('pallet_id'), BizPallet.pallet_code.label('pallet_code'), BizFillingRecord.strain_id.label('strain_id'), BizFillingRecord.strain_code_snapshot.label('strain_code'), BizFillingRecord.strain_name_snapshot.label('strain_name'), batch_no.label('batch_no'), BizUsageRecord.used_packages.label('used_packages'), BizUsageRecord.remaining_packages_after.label('remaining_packages_after'), BizUsageRecord.operator_id.label('operator_id'), operator_user.user_name.label('operator_name'), BizUsageRecord.tunnel_id.label('tunnel_id'), usage_tunnel.tunnel_code.label('tunnel_code'), usage_tunnel.tunnel_name.label('tunnel_name'), BizUsageRecord.trace_no.label('trace_no'), BizUsageRecord.batch_no_snapshot.label('batch_no_snapshot'), BizFillingRecord.production_date.label('production_date'), BizFillingRecord.expiry_date.label('expiry_date'), ) .select_from(BizUsageRecord) .join(BizPallet, BizPallet.pallet_id == BizUsageRecord.pallet_id) .join(BizFillingRecord, BizFillingRecord.filling_id == BizUsageRecord.active_filling_id) .join(BizCustomerSite, BizCustomerSite.site_id == BizUsageRecord.site_id) .outerjoin(BizProductionLine, BizProductionLine.line_id == BizUsageRecord.line_id) .outerjoin(usage_tunnel, usage_tunnel.tunnel_id == BizUsageRecord.tunnel_id) .outerjoin(BizCustomer, BizCustomer.customer_id == BizCustomerSite.customer_id) .outerjoin(operator_user, operator_user.user_id == BizUsageRecord.operator_id) .where(*conditions) .order_by(BizUsageRecord.usage_id) ) return await cls._paginate(db, statement, query_object) @classmethod async def get_return_report_list( cls, db: AsyncSession, query_object: ReturnReportQueryModel, scope: TestBatchScope, site_ids: list[int] | None = None, warehouse_ids: list[int] | None = None, ) -> PageModel[ReturnReportModel]: source_site = aliased(BizCustomerSite, name='source_site') target_warehouse = aliased(BizWarehouse, name='target_warehouse') return_tunnel = aliased(BizTunnel, name='return_tunnel') return_filling = aliased(BizFillingRecord, name='return_filling') batch_no = func.coalesce(BizReturnItem.batch_no_snapshot, return_filling.batch_no) conditions = [ batch_scope_condition(BizReturnOrder.test_batch_id, scope), batch_scope_condition(BizReturnItem.test_batch_id, scope), batch_scope_condition(BizPallet.test_batch_id, scope), cls._joined_scope_condition(source_site.site_id, source_site.test_batch_id, scope), cls._joined_scope_condition(target_warehouse.warehouse_id, target_warehouse.test_batch_id, scope), cls._joined_scope_condition(return_tunnel.tunnel_id, return_tunnel.test_batch_id, scope), cls._joined_scope_condition(return_filling.filling_id, return_filling.test_batch_id, scope), ] conditions.extend(cls._tunnel_conditions( query_object, BizReturnItem.trace_no_snapshot, (BizReturnItem.tunnel_id_snapshot, return_tunnel), )) if query_object.batch_no: conditions.append(batch_no.like(f'%{query_object.batch_no}%')) conditions.extend( cls._or_scope_conditions( (BizReturnOrder.source_site_id, site_ids), (BizReturnOrder.target_warehouse_id, warehouse_ids), ) ) if query_object.return_no: conditions.append(BizReturnOrder.return_no.like(f'%{query_object.return_no}%')) if query_object.source_site_id: conditions.append(BizReturnOrder.source_site_id == query_object.source_site_id) if query_object.target_warehouse_id: conditions.append(BizReturnOrder.target_warehouse_id == query_object.target_warehouse_id) if query_object.pallet_code: conditions.append(BizPallet.pallet_code.like(f'%{query_object.pallet_code}%')) if query_object.status: conditions.append(BizReturnOrder.status == query_object.status.value) conditions.extend( cls._time_conditions( BizReturnOrder.requested_at, query_object.begin_time, query_object.end_time, ) ) statement = ( select( BizReturnOrder.return_id.label('return_id'), BizReturnOrder.return_no.label('return_no'), BizReturnOrder.status.label('status'), BizReturnOrder.requested_at.label('requested_at'), BizReturnOrder.confirmed_at.label('confirmed_at'), BizReturnOrder.source_site_id.label('source_site_id'), source_site.site_name.label('source_site_name'), BizReturnOrder.target_warehouse_id.label('target_warehouse_id'), target_warehouse.warehouse_name.label('target_warehouse_name'), BizReturnItem.pallet_id.label('pallet_id'), BizPallet.pallet_code.label('pallet_code'), BizReturnItem.active_filling_id_snapshot.label('active_filling_id_snapshot'), BizReturnItem.remaining_packages_snapshot.label('remaining_packages_snapshot'), BizReturnItem.tunnel_id_snapshot.label('tunnel_id'), return_tunnel.tunnel_code.label('tunnel_code'), return_tunnel.tunnel_name.label('tunnel_name'), BizReturnItem.trace_no_snapshot.label('trace_no'), BizReturnItem.batch_no_snapshot.label('batch_no_snapshot'), batch_no.label('batch_no'), return_filling.production_date.label('production_date'), return_filling.expiry_date.label('expiry_date'), ) .select_from(BizReturnOrder) .join(BizReturnItem, BizReturnItem.return_id == BizReturnOrder.return_id) .join(BizPallet, BizPallet.pallet_id == BizReturnItem.pallet_id) .outerjoin(source_site, source_site.site_id == BizReturnOrder.source_site_id) .outerjoin(target_warehouse, target_warehouse.warehouse_id == BizReturnOrder.target_warehouse_id) .outerjoin(return_tunnel, return_tunnel.tunnel_id == BizReturnItem.tunnel_id_snapshot) .outerjoin(return_filling, return_filling.filling_id == BizReturnItem.active_filling_id_snapshot) .where(*conditions) .order_by(BizReturnOrder.return_id, BizReturnItem.return_item_id) ) return await cls._paginate(db, statement, query_object) @classmethod async def get_pallet_trace_list( cls, db: AsyncSession, query_object: PalletTraceReportQueryModel, scope: TestBatchScope, warehouse_ids: list[int] | None = None, site_ids: list[int] | None = None, ) -> PageModel[PalletTraceReportModel]: before_warehouse = aliased(BizWarehouse, name='before_warehouse') after_warehouse = aliased(BizWarehouse, name='after_warehouse') before_site = aliased(BizCustomerSite, name='before_site') after_site = aliased(BizCustomerSite, name='after_site') before_line = aliased(BizProductionLine, name='before_line') after_line = aliased(BizProductionLine, name='after_line') before_tunnel = aliased(BizTunnel, name='before_tunnel') after_tunnel = aliased(BizTunnel, name='after_tunnel') trace_return_item = aliased(BizReturnItem, name='trace_return_item') trace_usage = aliased(BizUsageRecord, name='trace_usage') trace_shipment_item = aliased(BizShipmentItem, name='trace_shipment_item') trace_filling = aliased(BizFillingRecord, name='trace_filling') prior_filling_event = aliased(BizPalletEvent, name='prior_filling_event') # 入隧事件来源是隧道而非装填记录，按托盘事件序号回查当时已发生的最后一次装填。 # 其他事件只使用来源单据的装填引用；引用缺失时保留空日期，不借用后续重装填。 prior_filling_id = ( select(prior_filling_event.source_id) .where( prior_filling_event.pallet_id == BizPalletEvent.pallet_id, prior_filling_event.event_seq < BizPalletEvent.event_seq, prior_filling_event.event_type == PalletEventType.FILLING_CONFIRMED.value, prior_filling_event.source_type == 'FILLING', batch_scope_condition(prior_filling_event.test_batch_id, scope), ) .order_by(prior_filling_event.event_seq.desc()) .limit(1) .correlate(BizPalletEvent) .scalar_subquery() ) trace_filling_id = case( (BizPalletEvent.source_type == 'FILLING', BizPalletEvent.source_id), (BizPalletEvent.source_type == 'USAGE', trace_usage.active_filling_id), (BizPalletEvent.source_type == 'RETURN', trace_return_item.active_filling_id_snapshot), (BizPalletEvent.source_type == 'SHIPMENT', trace_shipment_item.active_filling_id), (BizPalletEvent.event_type == PalletEventType.TUNNEL_ENTERED.value, prior_filling_id), else_=None, ) # 部分退回事件未保存隧道；只从同单同托盘的冻结退回项补读，不读取当前库存。 before_tunnel_id = func.coalesce(BizPalletEvent.before_tunnel_id, trace_return_item.tunnel_id_snapshot) trace_no = func.coalesce(BizPalletEvent.trace_no, trace_return_item.trace_no_snapshot) batch_no = func.coalesce(BizPalletEvent.batch_no_snapshot, trace_return_item.batch_no_snapshot) operator_user = aliased(SysUser, name='operator_user') conditions = [ batch_scope_condition(BizPalletEvent.test_batch_id, scope), batch_scope_condition(BizPallet.test_batch_id, scope), cls._joined_scope_condition(before_warehouse.warehouse_id, before_warehouse.test_batch_id, scope), cls._joined_scope_condition(after_warehouse.warehouse_id, after_warehouse.test_batch_id, scope), cls._joined_scope_condition(before_site.site_id, before_site.test_batch_id, scope), cls._joined_scope_condition(after_site.site_id, after_site.test_batch_id, scope), cls._joined_scope_condition(before_line.line_id, before_line.test_batch_id, scope), cls._joined_scope_condition(after_line.line_id, after_line.test_batch_id, scope), cls._joined_scope_condition(before_tunnel.tunnel_id, before_tunnel.test_batch_id, scope), cls._joined_scope_condition(after_tunnel.tunnel_id, after_tunnel.test_batch_id, scope), cls._joined_scope_condition(trace_return_item.return_item_id, trace_return_item.test_batch_id, scope), cls._joined_scope_condition(trace_usage.usage_id, trace_usage.test_batch_id, scope), cls._joined_scope_condition(trace_shipment_item.shipment_item_id, trace_shipment_item.test_batch_id, scope), cls._joined_scope_condition(trace_filling.filling_id, trace_filling.test_batch_id, scope), ] conditions.extend(cls._tunnel_conditions( query_object, trace_no, (before_tunnel_id, before_tunnel), (BizPalletEvent.after_tunnel_id, after_tunnel), )) if query_object.batch_no: conditions.append(batch_no.like(f'%{query_object.batch_no}%')) conditions.extend(cls._pallet_trace_scope_conditions(warehouse_ids, site_ids)) if query_object.pallet_code: conditions.append(BizPallet.pallet_code.like(f'%{query_object.pallet_code}%')) if query_object.event_type: conditions.append(BizPalletEvent.event_type == query_object.event_type.value) conditions.extend( cls._time_conditions( BizPalletEvent.occurred_at, query_object.begin_time, query_object.end_time, ) ) statement = ( select( BizPalletEvent.event_id.label('event_id'), BizPalletEvent.event_seq.label('event_seq'), BizPalletEvent.event_type.label('event_type'), BizPalletEvent.source_type.label('source_type'), BizPalletEvent.source_id.label('source_id'), BizPalletEvent.pallet_id.label('pallet_id'), BizPallet.pallet_code.label('pallet_code'), BizPalletEvent.before_state.label('before_state'), BizPalletEvent.after_state.label('after_state'), BizPalletEvent.before_warehouse_id.label('before_warehouse_id'), before_warehouse.warehouse_name.label('before_warehouse_name'), BizPalletEvent.before_site_id.label('before_site_id'), before_site.site_name.label('before_site_name'), BizPalletEvent.before_line_id.label('before_line_id'), before_line.line_name.label('before_line_name'), BizPalletEvent.after_warehouse_id.label('after_warehouse_id'), after_warehouse.warehouse_name.label('after_warehouse_name'), BizPalletEvent.after_site_id.label('after_site_id'), after_site.site_name.label('after_site_name'), BizPalletEvent.after_line_id.label('after_line_id'), after_line.line_name.label('after_line_name'), BizPalletEvent.before_remaining_packages.label('before_remaining_packages'), BizPalletEvent.after_remaining_packages.label('after_remaining_packages'), BizPalletEvent.operator_id.label('operator_id'), operator_user.user_name.label('operator_name'), BizPalletEvent.occurred_at.label('occurred_at'), func.coalesce(BizPalletEvent.after_tunnel_id, before_tunnel_id).label('tunnel_id'), func.coalesce(after_tunnel.tunnel_code, before_tunnel.tunnel_code).label('tunnel_code'), func.coalesce(after_tunnel.tunnel_name, before_tunnel.tunnel_name).label('tunnel_name'), trace_no.label('trace_no'), batch_no.label('batch_no_snapshot'), before_tunnel_id.label('before_tunnel_id'), before_tunnel.tunnel_code.label('before_tunnel_code'), before_tunnel.tunnel_name.label('before_tunnel_name'), BizPalletEvent.after_tunnel_id.label('after_tunnel_id'), after_tunnel.tunnel_code.label('after_tunnel_code'), after_tunnel.tunnel_name.label('after_tunnel_name'), trace_filling.production_date.label('production_date'), trace_filling.expiry_date.label('expiry_date'), ) .select_from(BizPalletEvent) .join(BizPallet, BizPallet.pallet_id == BizPalletEvent.pallet_id) .outerjoin(before_warehouse, before_warehouse.warehouse_id == BizPalletEvent.before_warehouse_id) .outerjoin(after_warehouse, after_warehouse.warehouse_id == BizPalletEvent.after_warehouse_id) .outerjoin(before_site, before_site.site_id == BizPalletEvent.before_site_id) .outerjoin(after_site, after_site.site_id == BizPalletEvent.after_site_id) .outerjoin(before_line, before_line.line_id == BizPalletEvent.before_line_id) .outerjoin(after_line, after_line.line_id == BizPalletEvent.after_line_id) .outerjoin(trace_return_item, and_( BizPalletEvent.source_type == 'RETURN', trace_return_item.return_id == BizPalletEvent.source_id, trace_return_item.pallet_id == BizPalletEvent.pallet_id, )) .outerjoin(trace_usage, and_( BizPalletEvent.source_type == 'USAGE', trace_usage.usage_id == BizPalletEvent.source_id, trace_usage.pallet_id == BizPalletEvent.pallet_id, )) .outerjoin(trace_shipment_item, and_( BizPalletEvent.source_type == 'SHIPMENT', trace_shipment_item.shipment_id == BizPalletEvent.source_id, trace_shipment_item.pallet_id == BizPalletEvent.pallet_id, )) .outerjoin(trace_filling, and_( trace_filling.filling_id == trace_filling_id, trace_filling.pallet_id == BizPalletEvent.pallet_id, )) .outerjoin(before_tunnel, before_tunnel.tunnel_id == before_tunnel_id) .outerjoin(after_tunnel, after_tunnel.tunnel_id == BizPalletEvent.after_tunnel_id) .outerjoin(operator_user, operator_user.user_id == BizPalletEvent.operator_id) .where(*conditions) .order_by(BizPalletEvent.occurred_at, BizPalletEvent.event_id) ) return await cls._paginate(db, statement, query_object)