diff --git a/shierfeng-fastapi-backend/module_business/controller/report_controller.py b/shierfeng-fastapi-backend/module_business/controller/report_controller.py new file mode 100644 index 0000000..e88fb76 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/controller/report_controller.py @@ -0,0 +1,131 @@ +"""业务报表接口。""" + +from typing import Annotated + +from fastapi import Query, Request, Response +from sqlalchemy.ext.asyncio import AsyncSession + +from common.aspect.db_seesion import DBSessionDependency +from common.aspect.interface_auth import UserInterfaceAuthDependency +from common.aspect.pre_auth import CurrentUserDependency, PreAuthDependency +from common.router import APIRouterPro +from common.vo import PageResponseModel +from module_admin.entity.vo.user_vo import CurrentUserModel +from module_business.entity.vo.report_vo import ( + ConsumptionReportModel, + ConsumptionReportQueryModel, + FillingReportModel, + FillingReportQueryModel, + PalletInventoryReportModel, + PalletInventoryReportQueryModel, + PalletTraceReportModel, + PalletTraceReportQueryModel, + ReturnReportModel, + ReturnReportQueryModel, + ShipmentReportModel, + ShipmentReportQueryModel, +) +from module_business.service.report_service import ReportService +from utils.response_util import ResponseUtil + +report_controller = APIRouterPro( + prefix='/report', + tags=['业务管理-只读报表'], + dependencies=[PreAuthDependency()], +) + + +@report_controller.get( + '/pallet-inventory/list', + summary='获取托盘库存报表分页列表接口', + response_model=PageResponseModel[PalletInventoryReportModel], + dependencies=[UserInterfaceAuthDependency('report:palletinventory:list')], +) +async def get_pallet_inventory_report_list( + request: Request, + query_object: Annotated[PalletInventoryReportQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ReportService.get_pallet_inventory_list_services(query_db, query_object, current_user) + return ResponseUtil.success(model_content=result) + + +@report_controller.get( + '/filling/list', + summary='获取装填记录报表分页列表接口', + response_model=PageResponseModel[FillingReportModel], + dependencies=[UserInterfaceAuthDependency('report:filling:list')], +) +async def get_filling_report_list( + request: Request, + query_object: Annotated[FillingReportQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ReportService.get_filling_report_list_services(query_db, query_object, current_user) + return ResponseUtil.success(model_content=result) + + +@report_controller.get( + '/shipment/list', + summary='获取发货出库报表分页列表接口', + response_model=PageResponseModel[ShipmentReportModel], + dependencies=[UserInterfaceAuthDependency('report:shipment:list')], +) +async def get_shipment_report_list( + request: Request, + query_object: Annotated[ShipmentReportQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ReportService.get_shipment_report_list_services(query_db, query_object, current_user) + return ResponseUtil.success(model_content=result) + + +@report_controller.get( + '/consumption/list', + summary='获取产线消耗报表分页列表接口', + response_model=PageResponseModel[ConsumptionReportModel], + dependencies=[UserInterfaceAuthDependency('report:consumption:list')], +) +async def get_consumption_report_list( + request: Request, + query_object: Annotated[ConsumptionReportQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ReportService.get_consumption_report_list_services(query_db, query_object, current_user) + return ResponseUtil.success(model_content=result) + + +@report_controller.get( + '/return/list', + summary='获取退回仓库报表分页列表接口', + response_model=PageResponseModel[ReturnReportModel], + dependencies=[UserInterfaceAuthDependency('report:return:list')], +) +async def get_return_report_list( + request: Request, + query_object: Annotated[ReturnReportQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ReportService.get_return_report_list_services(query_db, query_object, current_user) + return ResponseUtil.success(model_content=result) + + +@report_controller.get( + '/pallet-trace/list', + summary='获取托盘追溯报表分页列表接口', + response_model=PageResponseModel[PalletTraceReportModel], + dependencies=[UserInterfaceAuthDependency('report:pallettrace:list')], +) +async def get_pallet_trace_report_list( + request: Request, + query_object: Annotated[PalletTraceReportQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ReportService.get_pallet_trace_list_services(query_db, query_object, current_user) + return ResponseUtil.success(model_content=result) diff --git a/shierfeng-fastapi-backend/module_business/dao/report_dao.py b/shierfeng-fastapi-backend/module_business/dao/report_dao.py new file mode 100644 index 0000000..77dc438 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/dao/report_dao.py @@ -0,0 +1,494 @@ +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.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 = [] + 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.pallet_id, BizPalletEvent.event_seq, BizPalletEvent.event_id) + ) + return await cls._paginate(db, statement, query_object) diff --git a/shierfeng-fastapi-backend/module_business/service/report_service.py b/shierfeng-fastapi-backend/module_business/service/report_service.py new file mode 100644 index 0000000..d995dd9 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/service/report_service.py @@ -0,0 +1,196 @@ +from typing import Any, TypeVar + +from pydantic import BaseModel +from sqlalchemy.ext.asyncio import AsyncSession + +from common.vo import PageModel +from exceptions.exception import ServiceException +from module_admin.entity.vo.user_vo import CurrentUserModel +from module_business.dao.access_dao import AccessDao +from module_business.dao.report_dao import ReportDao +from module_business.entity.vo.report_vo import ( + ConsumptionReportModel, + ConsumptionReportQueryModel, + FillingReportModel, + FillingReportQueryModel, + PalletInventoryReportModel, + PalletInventoryReportQueryModel, + PalletTraceReportModel, + PalletTraceReportQueryModel, + ReturnReportModel, + ReturnReportQueryModel, + ShipmentReportModel, + ShipmentReportQueryModel, +) + +RowModelT = TypeVar('RowModelT', bound=BaseModel) + + +class ReportService: + """六类只读业务报表服务。""" + + @staticmethod + def _is_admin(current_user: CurrentUserModel) -> bool: + return bool(current_user.user and current_user.user.admin) + + @classmethod + def _user_id(cls, current_user: CurrentUserModel) -> int: + if current_user.user is None or current_user.user.user_id is None: + raise ServiceException(message='当前用户信息无效') + return current_user.user.user_id + + @classmethod + async def _warehouse_scope( + cls, + query_db: AsyncSession, + current_user: CurrentUserModel, + ) -> list[int] | None: + if cls._is_admin(current_user): + return None + warehouse_ids = await AccessDao.list_user_warehouse_ids(query_db, cls._user_id(current_user)) + return sorted(set(warehouse_ids)) + + @classmethod + async def _site_scope( + cls, + query_db: AsyncSession, + current_user: CurrentUserModel, + ) -> list[int] | None: + if cls._is_admin(current_user): + return None + site_ids = await AccessDao.list_user_site_ids(query_db, cls._user_id(current_user)) + return sorted(set(site_ids)) + + @classmethod + async def _scopes( + cls, + query_db: AsyncSession, + current_user: CurrentUserModel, + ) -> tuple[list[int] | None, list[int] | None]: + if cls._is_admin(current_user): + return None, None + return ( + await cls._warehouse_scope(query_db, current_user), + await cls._site_scope(query_db, current_user), + ) + + @staticmethod + def _empty_page(query_object: Any) -> PageModel: + return PageModel( + rows=[], + pageNum=query_object.page_num, + pageSize=query_object.page_size, + total=0, + hasNext=False, + ) + + @staticmethod + def _convert_page(page: PageModel, row_model: type[RowModelT]) -> PageModel[RowModelT]: + rows = [row_model.model_validate(row) for row in page.rows] + return page.model_copy(update={'rows': rows}) + + @classmethod + async def get_pallet_inventory_list_services( + cls, + query_db: AsyncSession, + query_object: PalletInventoryReportQueryModel, + current_user: CurrentUserModel, + ) -> PageModel[PalletInventoryReportModel]: + warehouse_ids, site_ids = await cls._scopes(query_db, current_user) + if warehouse_ids == [] and site_ids == []: + return cls._empty_page(query_object) + page = await ReportDao.get_pallet_inventory_list( + query_db, + query_object, + warehouse_ids=warehouse_ids, + site_ids=site_ids, + ) + return cls._convert_page(page, PalletInventoryReportModel) + + @classmethod + async def get_filling_report_list_services( + cls, + query_db: AsyncSession, + query_object: FillingReportQueryModel, + current_user: CurrentUserModel, + ) -> PageModel[FillingReportModel]: + warehouse_ids = await cls._warehouse_scope(query_db, current_user) + if warehouse_ids == []: + return cls._empty_page(query_object) + page = await ReportDao.get_filling_report_list( + query_db, + query_object, + warehouse_ids=warehouse_ids, + ) + return cls._convert_page(page, FillingReportModel) + + @classmethod + async def get_shipment_report_list_services( + cls, + query_db: AsyncSession, + query_object: ShipmentReportQueryModel, + current_user: CurrentUserModel, + ) -> PageModel[ShipmentReportModel]: + warehouse_ids, site_ids = await cls._scopes(query_db, current_user) + if warehouse_ids == [] or site_ids == []: + return cls._empty_page(query_object) + page = await ReportDao.get_shipment_report_list( + query_db, + query_object, + warehouse_ids=warehouse_ids, + site_ids=site_ids, + ) + return cls._convert_page(page, ShipmentReportModel) + + @classmethod + async def get_consumption_report_list_services( + cls, + query_db: AsyncSession, + query_object: ConsumptionReportQueryModel, + current_user: CurrentUserModel, + ) -> PageModel[ConsumptionReportModel]: + site_ids = await cls._site_scope(query_db, current_user) + if site_ids == []: + return cls._empty_page(query_object) + page = await ReportDao.get_consumption_report_list( + query_db, + query_object, + site_ids=site_ids, + ) + return cls._convert_page(page, ConsumptionReportModel) + + @classmethod + async def get_return_report_list_services( + cls, + query_db: AsyncSession, + query_object: ReturnReportQueryModel, + current_user: CurrentUserModel, + ) -> PageModel[ReturnReportModel]: + warehouse_ids, site_ids = await cls._scopes(query_db, current_user) + if warehouse_ids == [] and site_ids == []: + return cls._empty_page(query_object) + page = await ReportDao.get_return_report_list( + query_db, + query_object, + site_ids=site_ids, + warehouse_ids=warehouse_ids, + ) + return cls._convert_page(page, ReturnReportModel) + + @classmethod + async def get_pallet_trace_list_services( + cls, + query_db: AsyncSession, + query_object: PalletTraceReportQueryModel, + current_user: CurrentUserModel, + ) -> PageModel[PalletTraceReportModel]: + warehouse_ids, site_ids = await cls._scopes(query_db, current_user) + if warehouse_ids == [] and site_ids == []: + return cls._empty_page(query_object) + page = await ReportDao.get_pallet_trace_list( + query_db, + query_object, + warehouse_ids=warehouse_ids, + site_ids=site_ids, + ) + return cls._convert_page(page, PalletTraceReportModel) diff --git a/shierfeng-fastapi-backend/tests/module_business/test_report_service.py b/shierfeng-fastapi-backend/tests/module_business/test_report_service.py new file mode 100644 index 0000000..29c9e4e --- /dev/null +++ b/shierfeng-fastapi-backend/tests/module_business/test_report_service.py @@ -0,0 +1,989 @@ +import ast +import importlib.util +import inspect +import unittest +from collections.abc import Callable +from datetime import datetime +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +from fastapi.routing import APIRoute +from pydantic import BaseModel +from sqlalchemy import Select +from sqlalchemy.dialects import mysql, postgresql +from sqlalchemy.ext.asyncio import AsyncSession + +from common.aspect.interface_auth import CheckUserInterfaceAuth +from common.aspect.pre_auth import PreAuth +from common.vo import PageModel, PageResponseModel +from module_admin.entity.vo.user_vo import CurrentUserModel, UserInfoModel +from module_admin.service.login_service import LoginService +from module_business.controller import report_controller as report_controller_module +from module_business.dao import report_dao as report_dao_module +from module_business.dao.access_dao import AccessDao +from module_business.domain.enums import DocumentStatus, PalletEventType, PalletFlowState, ReturnStatus +from module_business.entity.vo.report_vo import ( + ConsumptionReportModel, + ConsumptionReportQueryModel, + FillingReportModel, + FillingReportQueryModel, + PalletInventoryReportModel, + PalletInventoryReportQueryModel, + PalletTraceReportModel, + PalletTraceReportQueryModel, + ReturnReportModel, + ReturnReportQueryModel, + ShipmentReportModel, + ShipmentReportQueryModel, +) +from module_business.service import report_service as report_service_module +from server import create_app +from utils.page_util import PageUtil + +ReportDao = getattr(report_dao_module, 'ReportDao', None) +ReportService = getattr(report_service_module, 'ReportService', None) +report_controller = getattr(report_controller_module, 'report_controller', None) + + +def make_user(user_id: int = 7) -> CurrentUserModel: + return CurrentUserModel( + permissions=[], + roles=[], + user=UserInfoModel(userId=user_id, userName='admin' if user_id == 1 else f'user-{user_id}'), + ) + + +def report_row(model_type: type[BaseModel]) -> dict[str, Any]: + now = datetime(2026, 7, 26, 10, 30) + rows = { + PalletInventoryReportModel: { + 'pallet_id': 1, + 'pallet_code': 'P-001', + 'flow_state': PalletFlowState.FILLED_IN_WAREHOUSE, + 'warehouse_id': 7, + 'warehouse_name': '一号仓', + 'site_id': None, + 'site_name': None, + 'line_id': None, + 'line_name': None, + 'strain_id': 11, + 'strain_code': 'S-011', + 'strain_name': '香菇 11', + 'batch_no': 'B-202607', + 'initial_packages': 20, + 'remaining_packages': 12, + 'updated_at': now, + }, + FillingReportModel: { + 'filling_id': 2, + 'filling_no': 'FL-002', + 'operated_at': now, + 'warehouse_id': 7, + 'warehouse_name': '一号仓', + 'pallet_id': 1, + 'pallet_code': 'P-001', + 'strain_id': 11, + 'strain_code_snapshot': 'S-011', + 'strain_name_snapshot': '香菇 11', + 'batch_no': 'B-202607', + 'package_spec': '20kg', + 'initial_packages': 20, + 'operator_id': 7, + 'operator_name': 'user-7', + }, + ShipmentReportModel: { + 'shipment_id': 3, + 'shipment_no': 'SH-003', + 'confirmed_at': now, + 'source_warehouse_id': 7, + 'source_warehouse_name': '一号仓', + 'target_site_id': 9, + 'target_site_name': '一号现场', + 'customer_id': 3, + 'customer_name': '一号客户', + 'pallet_id': 1, + 'pallet_code': 'P-001', + 'active_filling_id': 2, + 'package_qty': 12, + }, + ConsumptionReportModel: { + 'usage_id': 4, + 'usage_no': 'US-004', + 'occurred_at': now, + 'customer_id': 3, + 'customer_name': '一号客户', + 'site_id': 9, + 'site_name': '一号现场', + 'line_id': 4, + 'line_name': '一号线', + 'pallet_id': 1, + 'pallet_code': 'P-001', + 'strain_id': 11, + 'strain_code': 'S-011', + 'strain_name': '香菇 11', + 'batch_no': 'B-202607', + 'used_packages': 3, + 'remaining_packages_after': 9, + 'operator_id': 7, + 'operator_name': 'user-7', + }, + ReturnReportModel: { + 'return_id': 5, + 'return_no': 'RT-005', + 'status': ReturnStatus.REQUESTED, + 'requested_at': now, + 'confirmed_at': None, + 'source_site_id': 9, + 'source_site_name': '一号现场', + 'target_warehouse_id': 7, + 'target_warehouse_name': '一号仓', + 'pallet_id': 1, + 'pallet_code': 'P-001', + 'active_filling_id_snapshot': 2, + 'remaining_packages_snapshot': 9, + }, + PalletTraceReportModel: { + 'event_id': 6, + 'event_seq': 3, + 'event_type': PalletEventType.USAGE_RECORDED, + 'source_type': 'USAGE', + 'source_id': 4, + 'pallet_id': 1, + 'pallet_code': 'P-001', + 'before_state': PalletFlowState.AT_SITE_UNUSED, + 'after_state': PalletFlowState.AT_SITE_PARTIAL, + 'before_warehouse_id': None, + 'before_warehouse_name': None, + 'before_site_id': 9, + 'before_site_name': '一号现场', + 'before_line_id': None, + 'before_line_name': None, + 'after_warehouse_id': None, + 'after_warehouse_name': None, + 'after_site_id': 9, + 'after_site_name': '一号现场', + 'after_line_id': 4, + 'after_line_name': '一号线', + 'before_remaining_packages': 12, + 'after_remaining_packages': 9, + 'operator_id': 7, + 'operator_name': 'user-7', + 'occurred_at': now, + }, + } + return rows[model_type] + + +class TestReportBootstrap(unittest.TestCase): + """报表三层模块必须能够被项目自动发现。""" + + def test_report_modules_exist(self) -> None: + module_names = ( + 'module_business.dao.report_dao', + 'module_business.service.report_service', + 'module_business.controller.report_controller', + ) + + for module_name in module_names: + with self.subTest(module_name=module_name): + self.assertIsNotNone(importlib.util.find_spec(module_name), f'{module_name} 尚未实现') + + def test_report_controller_router_exists(self) -> None: + self.assertIsNotNone(report_controller, 'report_controller 路由尚未实现') + + +class TestReportPublicContract(unittest.TestCase): + def test_dao_exposes_exact_public_methods_and_signatures(self) -> None: + report_dao = getattr(report_dao_module, 'ReportDao', None) + self.assertIsNotNone(report_dao, 'ReportDao 尚未实现') + expected_signatures = { + 'get_pallet_inventory_list': ('db', 'query_object', 'warehouse_ids', 'site_ids'), + 'get_filling_report_list': ('db', 'query_object', 'warehouse_ids'), + 'get_shipment_report_list': ('db', 'query_object', 'warehouse_ids', 'site_ids'), + 'get_consumption_report_list': ('db', 'query_object', 'site_ids'), + 'get_return_report_list': ('db', 'query_object', 'site_ids', 'warehouse_ids'), + 'get_pallet_trace_list': ('db', 'query_object', 'warehouse_ids', 'site_ids'), + } + + for method_name, expected_parameters in expected_signatures.items(): + with self.subTest(method_name=method_name): + method = getattr(report_dao, method_name, None) + self.assertIsNotNone(method, f'ReportDao.{method_name} 尚未实现') + self.assertEqual(expected_parameters, tuple(inspect.signature(method).parameters)) + + def test_service_exposes_exact_public_methods_and_signatures(self) -> None: + report_service = getattr(report_service_module, 'ReportService', None) + self.assertIsNotNone(report_service, 'ReportService 尚未实现') + method_names = ( + 'get_pallet_inventory_list_services', + 'get_filling_report_list_services', + 'get_shipment_report_list_services', + 'get_consumption_report_list_services', + 'get_return_report_list_services', + 'get_pallet_trace_list_services', + ) + + for method_name in method_names: + with self.subTest(method_name=method_name): + method = getattr(report_service, method_name, None) + self.assertIsNotNone(method, f'ReportService.{method_name} 尚未实现') + self.assertEqual( + ('query_db', 'query_object', 'current_user'), + tuple(inspect.signature(method).parameters), + ) + + def test_dao_and_service_have_no_write_calls(self) -> None: + forbidden_calls = { + 'add', + 'add_all', + 'commit', + 'delete', + 'execute_delete', + 'execute_update', + 'flush', + 'rollback', + 'update', + } + for module in (report_dao_module, report_service_module): + with self.subTest(module=module.__name__): + tree = ast.parse(inspect.getsource(module)) + called_attributes = { + node.func.attr + for node in ast.walk(tree) + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) + } + self.assertTrue(forbidden_calls.isdisjoint(called_attributes)) + + +@unittest.skipUnless(ReportDao is not None, 'ReportDao 尚未实现') +class TestReportDaoQueries(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + + async def _capture_statement( + self, + method: Callable[..., Any], + query_object: Any, + **scope: Any, + ) -> Select: + page = PageModel( + rows=[], + pageNum=query_object.page_num, + pageSize=query_object.page_size, + total=0, + hasNext=False, + ) + paginate = AsyncMock(return_value=page) + with patch.object(PageUtil, 'paginate', new=paginate): + try: + result = await method(self.db, query_object, **scope) + except NotImplementedError: + self.fail(f'{method.__name__} 尚未构造只读分页查询') + + self.assertIs(page, result) + paginate.assert_awaited_once() + args = paginate.await_args.args + self.assertIs(self.db, args[0]) + self.assertEqual(query_object.page_num, args[2]) + self.assertEqual(query_object.page_size, args[3]) + self.assertIs(True, args[4]) + return args[1] + + def _assert_compiles_for_supported_databases(self, statement: Select) -> tuple[str, str]: + compiled = tuple( + ' '.join(str(statement.compile(dialect=dialect, compile_kwargs={'literal_binds': True})).lower().split()) + for dialect in (mysql.dialect(), postgresql.dialect()) + ) + self.assertTrue(all(sql.startswith('select ') for sql in compiled)) + return compiled + + def _assert_labels(self, statement: Select, row_model: type[BaseModel]) -> None: + self.assertEqual(tuple(row_model.model_fields), tuple(column.key for column in statement.selected_columns)) + + async def test_inventory_query_uses_current_facts_snapshots_all_filters_and_or_scope(self) -> None: + query = PalletInventoryReportQueryModel( + pageNum=2, + pageSize=7, + palletCode='P-001', + flowState=PalletFlowState.AT_SITE_PARTIAL, + warehouseId=7, + siteId=9, + lineId=4, + strainId=11, + batchNo='B-202607', + beginTime='2026-07-01', + endTime='2026-07-31', + ) + statement = await self._capture_statement( + ReportDao.get_pallet_inventory_list, + query, + warehouse_ids=[8, 7, 7], + site_ids=[10, 9, 9], + ) + + self._assert_labels(statement, PalletInventoryReportModel) + for sql in self._assert_compiles_for_supported_databases(statement): + self.assertIn('from biz_pallet_inventory', sql) + self.assertIn('join biz_pallet ', sql) + self.assertIn('left outer join biz_filling_record as active_filling', sql) + self.assertIn('active_filling.strain_code_snapshot as strain_code', sql) + self.assertIn('active_filling.strain_name_snapshot as strain_name', sql) + self.assertNotIn('biz_strain', sql) + self.assertIn( + '(biz_pallet_inventory.warehouse_id in (7, 8) or biz_pallet_inventory.site_id in (9, 10))', + sql, + ) + for column in ( + 'pallet_code', + 'flow_state', + 'warehouse_id', + 'site_id', + 'current_line_id', + 'strain_id', + 'batch_no', + ): + self.assertIn(column, sql) + self.assertIn('biz_pallet_inventory.updated_at >=', sql) + self.assertIn('biz_pallet_inventory.updated_at <', sql) + self.assertIn('order by biz_pallet_inventory.pallet_id', sql) + + async def test_filling_query_uses_snapshot_columns_all_filters_and_warehouse_scope(self) -> None: + query = FillingReportQueryModel( + pageNum=3, + pageSize=5, + warehouseId=7, + palletCode='P-001', + strainId=11, + batchNo='B-202607', + beginTime='2026-07-01', + endTime='2026-07-31', + ) + statement = await self._capture_statement( + ReportDao.get_filling_report_list, + query, + warehouse_ids=[8, 7, 7], + ) + + self._assert_labels(statement, FillingReportModel) + for sql in self._assert_compiles_for_supported_databases(statement): + self.assertIn('from biz_filling_record', sql) + self.assertIn('join biz_pallet ', sql) + self.assertIn('join biz_warehouse ', sql) + self.assertIn('strain_code_snapshot', sql) + self.assertIn('strain_name_snapshot', sql) + self.assertNotIn('biz_strain', sql) + self.assertIn('biz_filling_record.warehouse_id in (7, 8)', sql) + for column in ('warehouse_id', 'pallet_code', 'strain_id', 'batch_no'): + self.assertIn(column, sql) + self.assertIn('biz_filling_record.operated_at >=', sql) + self.assertIn('biz_filling_record.operated_at <', sql) + self.assertIn('order by biz_filling_record.filling_id', sql) + + async def test_shipment_query_is_one_row_per_item_uses_snapshots_and_and_scope(self) -> None: + query = ShipmentReportQueryModel( + shipmentNo='SH-001', + sourceWarehouseId=7, + targetSiteId=9, + palletCode='P-001', + status=DocumentStatus.CONFIRMED, + beginTime='2026-07-01', + endTime='2026-07-31', + ) + statement = await self._capture_statement( + ReportDao.get_shipment_report_list, + query, + warehouse_ids=[8, 7], + site_ids=[10, 9], + ) + + self._assert_labels(statement, ShipmentReportModel) + for sql in self._assert_compiles_for_supported_databases(statement): + self.assertIn('from biz_shipment_order', sql) + self.assertIn('join biz_shipment_item', sql) + self.assertIn('biz_shipment_item.active_filling_id as active_filling_id', sql) + self.assertIn('biz_shipment_item.package_qty as package_qty', sql) + self.assertIn('biz_shipment_order.source_warehouse_id in (7, 8)', sql) + self.assertIn('biz_shipment_order.target_site_id in (9, 10)', sql) + self.assertNotIn( + 'source_warehouse_id in (7, 8) or biz_shipment_order.target_site_id', + sql, + ) + for column in ('shipment_no', 'source_warehouse_id', 'target_site_id', 'pallet_code', 'status'): + self.assertIn(column, sql) + self.assertIn('biz_shipment_order.confirmed_at >=', sql) + self.assertIn('biz_shipment_order.confirmed_at <', sql) + self.assertIn( + 'order by biz_shipment_order.shipment_id, biz_shipment_item.shipment_item_id', + sql, + ) + + async def test_consumption_query_uses_usage_filling_snapshots_all_filters_and_site_scope(self) -> None: + query = ConsumptionReportQueryModel( + customerId=3, + siteId=9, + lineId=4, + palletCode='P-001', + strainId=11, + batchNo='B-202607', + beginTime='2026-07-01', + endTime='2026-07-31', + ) + statement = await self._capture_statement( + ReportDao.get_consumption_report_list, + query, + site_ids=[10, 9, 9], + ) + + self._assert_labels(statement, ConsumptionReportModel) + for sql in self._assert_compiles_for_supported_databases(statement): + self.assertIn('from biz_usage_record', sql) + self.assertIn( + 'join biz_filling_record on biz_filling_record.filling_id = biz_usage_record.active_filling_id', + sql, + ) + self.assertIn('biz_filling_record.strain_code_snapshot as strain_code', sql) + self.assertIn('biz_filling_record.strain_name_snapshot as strain_name', sql) + self.assertNotIn('biz_strain', sql) + self.assertIn('biz_usage_record.site_id in (9, 10)', sql) + for column in ('customer_id', 'site_id', 'line_id', 'pallet_code', 'strain_id', 'batch_no'): + self.assertIn(column, sql) + self.assertIn('biz_usage_record.occurred_at >=', sql) + self.assertIn('biz_usage_record.occurred_at <', sql) + self.assertIn('order by biz_usage_record.usage_id', sql) + + async def test_return_query_is_one_row_per_item_uses_snapshots_and_or_scope(self) -> None: + query = ReturnReportQueryModel( + returnNo='RT-001', + sourceSiteId=9, + targetWarehouseId=7, + palletCode='P-001', + status=ReturnStatus.REQUESTED, + beginTime='2026-07-01', + endTime='2026-07-31', + ) + statement = await self._capture_statement( + ReportDao.get_return_report_list, + query, + site_ids=[10, 9], + warehouse_ids=[8, 7], + ) + + self._assert_labels(statement, ReturnReportModel) + for sql in self._assert_compiles_for_supported_databases(statement): + self.assertIn('from biz_return_order', sql) + self.assertIn('join biz_return_item', sql) + self.assertIn('biz_return_item.active_filling_id_snapshot as active_filling_id_snapshot', sql) + self.assertIn('biz_return_item.remaining_packages_snapshot as remaining_packages_snapshot', sql) + self.assertIn( + '(biz_return_order.source_site_id in (9, 10) or biz_return_order.target_warehouse_id in (7, 8))', + sql, + ) + for column in ('return_no', 'source_site_id', 'target_warehouse_id', 'pallet_code', 'status'): + self.assertIn(column, sql) + self.assertIn('biz_return_order.requested_at >=', sql) + self.assertIn('biz_return_order.requested_at <', sql) + self.assertIn('order by biz_return_order.return_id, biz_return_item.return_item_id', sql) + + async def test_trace_query_reads_event_history_aliases_all_filters_and_event_scope(self) -> None: + query = PalletTraceReportQueryModel( + palletCode='P-001', + eventType=PalletEventType.USAGE_RECORDED, + beginTime='2026-07-01', + endTime='2026-07-31', + ) + statement = await self._capture_statement( + ReportDao.get_pallet_trace_list, + query, + warehouse_ids=[8, 7], + site_ids=[10, 9], + ) + + self._assert_labels(statement, PalletTraceReportModel) + for sql in self._assert_compiles_for_supported_databases(statement): + self.assertIn('from biz_pallet_event', sql) + self.assertNotIn('biz_pallet_inventory', sql) + self.assertNotIn('sys_oper_log', sql) + for column in ( + 'before_state', + 'after_state', + 'before_warehouse_id', + 'after_warehouse_id', + 'before_site_id', + 'after_site_id', + 'before_line_id', + 'after_line_id', + 'before_remaining_packages', + 'after_remaining_packages', + 'event_seq', + 'event_type', + 'source_type', + 'source_id', + 'operator_id', + 'occurred_at', + ): + self.assertIn(f'biz_pallet_event.{column}', sql) + self.assertIn('biz_pallet_event.before_warehouse_id in (7, 8)', sql) + self.assertIn('biz_pallet_event.after_warehouse_id in (7, 8)', sql) + self.assertIn('biz_pallet_event.before_site_id in (9, 10)', sql) + self.assertIn('biz_pallet_event.after_site_id in (9, 10)', sql) + self.assertIn('biz_pallet_event.occurred_at >=', sql) + self.assertIn('biz_pallet_event.occurred_at <', sql) + self.assertIn( + 'order by biz_pallet_event.pallet_id, biz_pallet_event.event_seq, biz_pallet_event.event_id', + sql, + ) + + +@unittest.skipUnless(ReportService is not None, 'ReportService 尚未实现') +class TestReportServices(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + self.user = make_user() + + async def _assert_scoped_conversion( + self, + service_name: str, + dao_name: str, + query_object: Any, + row_model: type[BaseModel], + expected_scope: dict[str, list[int] | None], + ) -> None: + page = PageModel( + rows=[report_row(row_model)], + pageNum=query_object.page_num, + pageSize=query_object.page_size, + total=17, + hasNext=True, + ) + dao_method = AsyncMock(return_value=page) + warehouse_scope = AsyncMock(return_value=[8, 7, 7]) + site_scope = AsyncMock(return_value=[10, 9, 9]) + with ( + patch.object(ReportDao, dao_name, new=dao_method), + patch.object(AccessDao, 'list_user_warehouse_ids', new=warehouse_scope), + patch.object(AccessDao, 'list_user_site_ids', new=site_scope), + ): + try: + result = await getattr(ReportService, service_name)(self.db, query_object, self.user) + except NotImplementedError: + self.fail(f'ReportService.{service_name} 尚未实现数据范围和行转换') + + dao_method.assert_awaited_once_with(self.db, query_object, **expected_scope) + if 'warehouse_ids' in expected_scope: + warehouse_scope.assert_awaited_once_with(self.db, 7) + else: + warehouse_scope.assert_not_awaited() + if 'site_ids' in expected_scope: + site_scope.assert_awaited_once_with(self.db, 7) + else: + site_scope.assert_not_awaited() + self.assertEqual(17, result.total) + self.assertEqual(query_object.page_num, result.page_num) + self.assertEqual(query_object.page_size, result.page_size) + self.assertTrue(result.has_next) + self.assertIsInstance(result.rows[0], row_model) + + async def test_inventory_service_pushes_both_scopes_and_converts_page(self) -> None: + await self._assert_scoped_conversion( + 'get_pallet_inventory_list_services', + 'get_pallet_inventory_list', + PalletInventoryReportQueryModel(pageNum=2, pageSize=7), + PalletInventoryReportModel, + {'warehouse_ids': [7, 8], 'site_ids': [9, 10]}, + ) + + async def test_filling_service_pushes_warehouse_scope_and_converts_page(self) -> None: + await self._assert_scoped_conversion( + 'get_filling_report_list_services', + 'get_filling_report_list', + FillingReportQueryModel(pageNum=2, pageSize=7), + FillingReportModel, + {'warehouse_ids': [7, 8]}, + ) + + async def test_shipment_service_pushes_both_scopes_and_converts_page(self) -> None: + await self._assert_scoped_conversion( + 'get_shipment_report_list_services', + 'get_shipment_report_list', + ShipmentReportQueryModel(pageNum=2, pageSize=7), + ShipmentReportModel, + {'warehouse_ids': [7, 8], 'site_ids': [9, 10]}, + ) + + async def test_consumption_service_pushes_site_scope_and_converts_page(self) -> None: + await self._assert_scoped_conversion( + 'get_consumption_report_list_services', + 'get_consumption_report_list', + ConsumptionReportQueryModel(pageNum=2, pageSize=7), + ConsumptionReportModel, + {'site_ids': [9, 10]}, + ) + + async def test_return_service_pushes_both_scopes_and_converts_page(self) -> None: + await self._assert_scoped_conversion( + 'get_return_report_list_services', + 'get_return_report_list', + ReturnReportQueryModel(pageNum=2, pageSize=7), + ReturnReportModel, + {'site_ids': [9, 10], 'warehouse_ids': [7, 8]}, + ) + + async def test_trace_service_pushes_both_scopes_and_converts_page(self) -> None: + await self._assert_scoped_conversion( + 'get_pallet_trace_list_services', + 'get_pallet_trace_list', + PalletTraceReportQueryModel(pageNum=2, pageSize=7), + PalletTraceReportModel, + {'warehouse_ids': [7, 8], 'site_ids': [9, 10]}, + ) + + async def test_admin_skips_scope_queries_and_passes_none(self) -> None: + cases = ( + ( + 'get_pallet_inventory_list_services', + 'get_pallet_inventory_list', + PalletInventoryReportQueryModel(), + {'warehouse_ids': None, 'site_ids': None}, + ), + ( + 'get_filling_report_list_services', + 'get_filling_report_list', + FillingReportQueryModel(), + {'warehouse_ids': None}, + ), + ( + 'get_shipment_report_list_services', + 'get_shipment_report_list', + ShipmentReportQueryModel(), + {'warehouse_ids': None, 'site_ids': None}, + ), + ( + 'get_consumption_report_list_services', + 'get_consumption_report_list', + ConsumptionReportQueryModel(), + {'site_ids': None}, + ), + ( + 'get_return_report_list_services', + 'get_return_report_list', + ReturnReportQueryModel(), + {'site_ids': None, 'warehouse_ids': None}, + ), + ( + 'get_pallet_trace_list_services', + 'get_pallet_trace_list', + PalletTraceReportQueryModel(), + {'warehouse_ids': None, 'site_ids': None}, + ), + ) + for service_name, dao_name, query_object, expected_scope in cases: + with self.subTest(service_name=service_name): + dao_method = AsyncMock( + return_value=PageModel( + rows=[], + pageNum=1, + pageSize=10, + total=0, + hasNext=False, + ) + ) + warehouse_scope = AsyncMock() + site_scope = AsyncMock() + with ( + patch.object(ReportDao, dao_name, new=dao_method), + patch.object(AccessDao, 'list_user_warehouse_ids', new=warehouse_scope), + patch.object(AccessDao, 'list_user_site_ids', new=site_scope), + ): + try: + await getattr(ReportService, service_name)(self.db, query_object, make_user(1)) + except NotImplementedError: + self.fail(f'ReportService.{service_name} 尚未实现管理员范围语义') + + warehouse_scope.assert_not_awaited() + site_scope.assert_not_awaited() + dao_method.assert_awaited_once_with(self.db, query_object, **expected_scope) + + async def _assert_empty_page_without_dao( + self, + service_name: str, + dao_name: str, + query_object: Any, + warehouse_ids: list[int] | None, + site_ids: list[int] | None, + ) -> None: + dao_method = AsyncMock() + with ( + patch.object(ReportDao, dao_name, new=dao_method), + patch.object( + AccessDao, + 'list_user_warehouse_ids', + new=AsyncMock(return_value=warehouse_ids), + ), + patch.object( + AccessDao, + 'list_user_site_ids', + new=AsyncMock(return_value=site_ids), + ), + ): + try: + result = await getattr(ReportService, service_name)(self.db, query_object, self.user) + except NotImplementedError: + self.fail(f'ReportService.{service_name} 尚未实现空范围短路') + + dao_method.assert_not_awaited() + self.assertEqual([], result.rows) + self.assertEqual(0, result.total) + self.assertEqual(query_object.page_num, result.page_num) + self.assertEqual(query_object.page_size, result.page_size) + self.assertFalse(result.has_next) + + async def test_inventory_return_and_trace_only_short_circuit_when_both_scopes_empty(self) -> None: + empty_cases = ( + ( + 'get_pallet_inventory_list_services', + 'get_pallet_inventory_list', + PalletInventoryReportQueryModel(pageNum=3, pageSize=4), + ), + ( + 'get_return_report_list_services', + 'get_return_report_list', + ReturnReportQueryModel(pageNum=3, pageSize=4), + ), + ( + 'get_pallet_trace_list_services', + 'get_pallet_trace_list', + PalletTraceReportQueryModel(pageNum=3, pageSize=4), + ), + ) + for service_name, dao_name, query_object in empty_cases: + with self.subTest(service_name=service_name): + await self._assert_empty_page_without_dao( + service_name, + dao_name, + query_object, + [], + [], + ) + + visible_cases = ( + ( + 'get_pallet_inventory_list_services', + 'get_pallet_inventory_list', + PalletInventoryReportQueryModel(), + ), + ('get_return_report_list_services', 'get_return_report_list', ReturnReportQueryModel()), + ( + 'get_pallet_trace_list_services', + 'get_pallet_trace_list', + PalletTraceReportQueryModel(), + ), + ) + for service_name, dao_name, query_object in visible_cases: + with self.subTest(service_name=service_name, one_scope_visible=True): + dao_method = AsyncMock(return_value=PageModel(rows=[], pageNum=1, pageSize=10, total=0, hasNext=False)) + with ( + patch.object(ReportDao, dao_name, new=dao_method), + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[])), + ): + try: + await getattr(ReportService, service_name)(self.db, query_object, self.user) + except NotImplementedError: + self.fail(f'ReportService.{service_name} 尚未实现单侧 OR 范围可见性') + dao_method.assert_awaited_once() + + async def test_filling_and_consumption_empty_applicable_scope_short_circuits(self) -> None: + await self._assert_empty_page_without_dao( + 'get_filling_report_list_services', + 'get_filling_report_list', + FillingReportQueryModel(pageNum=3, pageSize=4), + [], + [9], + ) + await self._assert_empty_page_without_dao( + 'get_consumption_report_list_services', + 'get_consumption_report_list', + ConsumptionReportQueryModel(pageNum=3, pageSize=4), + [7], + [], + ) + + async def test_shipment_either_empty_scope_short_circuits(self) -> None: + for warehouse_ids, site_ids in (([], [9]), ([7], []), ([], [])): + with self.subTest(warehouse_ids=warehouse_ids, site_ids=site_ids): + await self._assert_empty_page_without_dao( + 'get_shipment_report_list_services', + 'get_shipment_report_list', + ShipmentReportQueryModel(pageNum=3, pageSize=4), + warehouse_ids, + site_ids, + ) + + +@unittest.skipUnless(report_controller is not None, 'report_controller 尚未实现') +class TestReportControllerContract(unittest.TestCase): + @staticmethod + def _routes() -> dict[tuple[str, str], APIRoute]: + return { + (route.path, next(iter(route.methods))): route + for route in report_controller.routes + if isinstance(route, APIRoute) + } + + def test_create_app_openapi_has_exact_six_report_get_operations_and_aliases(self) -> None: + schema = create_app().openapi() + report_operations = { + (path, method) + for path, methods in schema['paths'].items() + if path.startswith('/report/') + for method in methods + } + expected_operations = { + ('/report/pallet-inventory/list', 'get'), + ('/report/filling/list', 'get'), + ('/report/shipment/list', 'get'), + ('/report/consumption/list', 'get'), + ('/report/return/list', 'get'), + ('/report/pallet-trace/list', 'get'), + } + self.assertEqual(expected_operations, report_operations) + + expected_parameters = { + '/report/pallet-inventory/list': { + 'pageNum', + 'pageSize', + 'beginTime', + 'endTime', + 'palletCode', + 'flowState', + 'warehouseId', + 'siteId', + 'lineId', + 'strainId', + 'batchNo', + }, + '/report/filling/list': { + 'pageNum', + 'pageSize', + 'beginTime', + 'endTime', + 'warehouseId', + 'palletCode', + 'strainId', + 'batchNo', + }, + '/report/shipment/list': { + 'pageNum', + 'pageSize', + 'beginTime', + 'endTime', + 'shipmentNo', + 'sourceWarehouseId', + 'targetSiteId', + 'palletCode', + 'status', + }, + '/report/consumption/list': { + 'pageNum', + 'pageSize', + 'beginTime', + 'endTime', + 'customerId', + 'siteId', + 'lineId', + 'palletCode', + 'strainId', + 'batchNo', + }, + '/report/return/list': { + 'pageNum', + 'pageSize', + 'beginTime', + 'endTime', + 'returnNo', + 'sourceSiteId', + 'targetWarehouseId', + 'palletCode', + 'status', + }, + '/report/pallet-trace/list': { + 'pageNum', + 'pageSize', + 'beginTime', + 'endTime', + 'palletCode', + 'eventType', + }, + } + for path, parameters in expected_parameters.items(): + with self.subTest(path=path): + actual = {parameter['name'] for parameter in schema['paths'][path]['get']['parameters']} + self.assertEqual(parameters, actual) + self.assertNotIn('mode', actual) + + def test_routes_have_exact_permissions_response_models_current_user_and_no_log(self) -> None: + expected = { + ('/report/pallet-inventory/list', 'GET'): ( + 'report:palletinventory:list', + PageResponseModel[PalletInventoryReportModel], + 'get_pallet_inventory_list_services', + ), + ('/report/filling/list', 'GET'): ( + 'report:filling:list', + PageResponseModel[FillingReportModel], + 'get_filling_report_list_services', + ), + ('/report/shipment/list', 'GET'): ( + 'report:shipment:list', + PageResponseModel[ShipmentReportModel], + 'get_shipment_report_list_services', + ), + ('/report/consumption/list', 'GET'): ( + 'report:consumption:list', + PageResponseModel[ConsumptionReportModel], + 'get_consumption_report_list_services', + ), + ('/report/return/list', 'GET'): ( + 'report:return:list', + PageResponseModel[ReturnReportModel], + 'get_return_report_list_services', + ), + ('/report/pallet-trace/list', 'GET'): ( + 'report:pallettrace:list', + PageResponseModel[PalletTraceReportModel], + 'get_pallet_trace_list_services', + ), + } + routes = self._routes() + self.assertEqual(set(expected), set(routes)) + for key, (permission, response_model, service_method) in expected.items(): + route = routes[key] + permissions = [ + dependency.dependency + for dependency in route.dependencies + if isinstance(dependency.dependency, CheckUserInterfaceAuth) + ] + self.assertEqual(1, len(permissions), key) + self.assertEqual(permission, permissions[0].perm) + self.assertEqual(response_model, route.response_model) + self.assertTrue(any(isinstance(dependency.dependency, PreAuth) for dependency in route.dependencies)) + self.assertTrue( + any( + dependency.name == 'current_user' and dependency.call == LoginService.get_current_user + for dependency in route.dependant.dependencies + ), + key, + ) + self.assertFalse(hasattr(route.endpoint, '__wrapped__'), key) + self.assertIn(service_method, inspect.getsource(route.endpoint)) + + def test_controller_source_has_no_export_write_route_or_log(self) -> None: + source = inspect.getsource(report_controller_module) + self.assertNotIn('@Log(', source) + self.assertNotIn('/export', source) + for decorator in ('.post(', '.put(', '.delete(', '.patch('): + self.assertNotIn(decorator, source) + + +if __name__ == '__main__': + unittest.main()