74 lines
2.8 KiB
Python
74 lines
2.8 KiB
Python
from typing import Any
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from common.vo import PageModel
|
|
from exceptions.exception import ServiceException
|
|
from module_admin.dao.log_dao import OperationLogDao
|
|
from module_admin.entity.vo.user_vo import CurrentUserModel
|
|
from module_business.entity.vo.dashboard_vo import ScanExceptionModel
|
|
|
|
|
|
class DashboardService:
|
|
"""首页运营驾驶舱数据服务。"""
|
|
|
|
_SCAN_EXCEPTION_LIMIT = 5
|
|
_SCAN_SOURCE_LABELS = {
|
|
'/business/inbound/scan/{qr_token}': '首次入库扫码',
|
|
'/business/filling/scan/{qr_token}': '托盘装填扫码',
|
|
'/business/shipment/scan/{qr_token}': '整托发货扫码',
|
|
'/business/usage/scan/{qr_token}': '托盘使用扫码',
|
|
'/business/return/scan/{qr_token}': '托盘退回扫码',
|
|
}
|
|
|
|
@staticmethod
|
|
def _row_value(row: Any, *keys: str) -> Any:
|
|
if isinstance(row, dict):
|
|
for key in keys:
|
|
if key in row:
|
|
return row[key]
|
|
return None
|
|
for key in keys:
|
|
value = getattr(row, key, None)
|
|
if value is not None:
|
|
return value
|
|
return None
|
|
|
|
@classmethod
|
|
def _scan_source(cls, oper_url: str | None) -> str:
|
|
if not oper_url:
|
|
return '扫码接口'
|
|
for route_path, label in cls._SCAN_SOURCE_LABELS.items():
|
|
route_prefix = route_path.split('/{', 1)[0]
|
|
if oper_url == route_path or oper_url.startswith(f'{route_prefix}/'):
|
|
return label
|
|
return '扫码接口'
|
|
|
|
@classmethod
|
|
async def get_scan_exception_list_services(
|
|
cls, query_db: AsyncSession, current_user: CurrentUserModel
|
|
) -> PageModel[ScanExceptionModel]:
|
|
"""按用户范围获取首页最近扫码异常。"""
|
|
if current_user.user is None or not current_user.user.user_name:
|
|
raise ServiceException(message='当前用户信息无效')
|
|
|
|
oper_name = None if current_user.user.admin else current_user.user.user_name
|
|
page = await OperationLogDao.get_scan_exception_log_list(query_db, oper_name, cls._SCAN_EXCEPTION_LIMIT)
|
|
rows = [
|
|
ScanExceptionModel(
|
|
oper_id=cls._row_value(row, 'operId', 'oper_id'),
|
|
source=cls._scan_source(cls._row_value(row, 'operUrl', 'oper_url')),
|
|
error_message=cls._row_value(row, 'errorMsg', 'error_msg') or '扫码请求失败',
|
|
operator_name=cls._row_value(row, 'operName', 'oper_name') or '未知用户',
|
|
occurred_at=cls._row_value(row, 'operTime', 'oper_time'),
|
|
)
|
|
for row in page.rows
|
|
]
|
|
return PageModel[ScanExceptionModel](
|
|
rows=rows,
|
|
pageNum=page.page_num,
|
|
pageSize=page.page_size,
|
|
total=page.total,
|
|
hasNext=page.has_next,
|
|
)
|