from datetime import datetime, time, timedelta from typing import Any from sqlalchemy import ColumnElement, Select, false, 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 from module_business.entity.do.master_do import ( BizCustomer, BizCustomerSite, BizPallet, BizProductionLine, 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, ) 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() @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 [] @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, ) @classmethod async def get_pallet_inventory_list( cls, db: AsyncSession, query_object: PalletInventoryReportQueryModel, 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') active_filling = aliased(BizFillingRecord, name='active_filling') conditions = 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'), ) .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) .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, warehouse_ids: list[int] | None = None, ) -> PageModel[FillingReportModel]: operator_user = aliased(SysUser, name='operator_user') conditions = [] 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, 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 = [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, site_ids: list[int] | None = None, ) -> PageModel[ConsumptionReportModel]: operator_user = aliased(SysUser, name='operator_user') conditions = [] 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(BizFillingRecord.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'), BizFillingRecord.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'), ) .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) .join(BizProductionLine, BizProductionLine.line_id == BizUsageRecord.line_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, 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') conditions = 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'), ) .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) .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, 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') operator_user = aliased(SysUser, name='operator_user') conditions = cls._or_scope_conditions( (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 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'), ) .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(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)