1019 lines
45 KiB
Python
1019 lines
45 KiB
Python
"""客户现场使用服务。"""
|
|
|
|
import hashlib
|
|
import json
|
|
import secrets
|
|
from datetime import datetime
|
|
from typing import Any, NoReturn
|
|
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from common.vo import CrudResponseModel, PageModel
|
|
from exceptions.exception import PermissionException, ServiceException
|
|
from module_admin.entity.vo.user_vo import CurrentUserModel
|
|
from module_business.dao.access_dao import AccessDao
|
|
from module_business.dao.inventory_dao import InventoryDao
|
|
from module_business.dao.master_dao import MasterDao
|
|
from module_business.dao.operation_dao import OperationDao
|
|
from module_business.dao.usage_fifo_dao import UsageFifoDao
|
|
from module_business.domain.enums import PalletEventType, PalletFlowState
|
|
from module_business.domain.fifo_policy import requires_fifo_skip_reason
|
|
from module_business.domain.pallet_state_machine import PalletStateMachine
|
|
from module_business.domain.test_batch_scope import TestBatchScope, assert_record_in_scope
|
|
from module_business.entity.do.operation_do import BizFillingRecord, BizPalletEvent, BizPalletInventory, BizUsageRecord
|
|
from module_business.entity.vo.master_vo import LinePageQueryModel, TunnelPageQueryModel
|
|
from module_business.entity.vo.operation_vo import (
|
|
BusinessOptionModel,
|
|
PalletScanResponseModel,
|
|
TunnelEntryCreateModel,
|
|
TunnelEntrySuccessResponseModel,
|
|
UsageCreateModel,
|
|
UsageModel,
|
|
UsagePageQueryModel,
|
|
UsageSuccessResponseModel,
|
|
)
|
|
from module_business.service.access_service import AccessService
|
|
from module_business.service.test_batch_access_service import TestBatchAccessService
|
|
from module_business.service.usage_transaction import retry_mysql_lock_conflicts
|
|
|
|
|
|
class UsageService:
|
|
"""客户现场使用服务层。"""
|
|
|
|
@staticmethod
|
|
def _current_identity(current_user: CurrentUserModel) -> tuple[int, str]:
|
|
user = current_user.user
|
|
if user is None or user.user_id is None or user.user_id <= 0 or not user.user_name:
|
|
raise ServiceException(message='当前用户信息无效')
|
|
return user.user_id, user.user_name
|
|
|
|
@staticmethod
|
|
def _is_admin(current_user: CurrentUserModel) -> bool:
|
|
return bool(current_user.user and current_user.user.admin)
|
|
|
|
@classmethod
|
|
async def _site_scope(
|
|
cls,
|
|
query_db: AsyncSession,
|
|
current_user: CurrentUserModel,
|
|
) -> list[int] | None:
|
|
if cls._is_admin(current_user):
|
|
return None
|
|
user_id, _ = cls._current_identity(current_user)
|
|
return sorted(set(await AccessDao.list_user_site_ids(query_db, user_id)))
|
|
|
|
@classmethod
|
|
def _raise_pallet_not_accessible(
|
|
cls,
|
|
current_user: CurrentUserModel,
|
|
admin_message: str = '托盘不存在',
|
|
) -> NoReturn:
|
|
if cls._is_admin(current_user):
|
|
raise ServiceException(message=admin_message)
|
|
raise PermissionException(data='', message='无权访问该托盘')
|
|
|
|
@classmethod
|
|
async def _assert_pallet_site_access(
|
|
cls,
|
|
query_db: AsyncSession,
|
|
current_user: CurrentUserModel,
|
|
site_id: int,
|
|
*,
|
|
for_update: bool = False,
|
|
) -> None:
|
|
try:
|
|
if for_update:
|
|
await AccessService.assert_site_access(
|
|
query_db,
|
|
current_user,
|
|
site_id,
|
|
for_update=True,
|
|
)
|
|
else:
|
|
await AccessService.assert_site_access(query_db, current_user, site_id)
|
|
except PermissionException:
|
|
if cls._is_admin(current_user):
|
|
raise
|
|
cls._raise_pallet_not_accessible(current_user)
|
|
|
|
@classmethod
|
|
def _raise_usage_not_accessible(cls, current_user: CurrentUserModel) -> NoReturn:
|
|
if cls._is_admin(current_user):
|
|
raise ServiceException(message='使用记录不存在')
|
|
raise PermissionException(data='', message='无权访问该使用记录')
|
|
|
|
@staticmethod
|
|
def _record_value(record: Any, field: str) -> Any:
|
|
if not isinstance(record, dict):
|
|
return getattr(record, field)
|
|
if field in record:
|
|
return record[field]
|
|
words = field.split('_')
|
|
alias = words[0] + ''.join(word.capitalize() for word in words[1:])
|
|
return record.get(alias)
|
|
|
|
@staticmethod
|
|
def _build_request_hash(page_object: UsageCreateModel) -> str:
|
|
fields = {
|
|
'qr_token': page_object.qr_token.strip(),
|
|
'tunnel_id': page_object.tunnel_id,
|
|
'line_id': page_object.line_id,
|
|
'used_packages': page_object.used_packages,
|
|
}
|
|
# 无原因时保留原摘要，迁移前已成功的请求仍可按原键安全重放。
|
|
if page_object.fifo_skip_reason:
|
|
fields['fifo_skip_reason'] = page_object.fifo_skip_reason
|
|
payload = json.dumps(
|
|
fields,
|
|
sort_keys=True,
|
|
separators=(',', ':'),
|
|
).encode()
|
|
return hashlib.sha256(payload).hexdigest()
|
|
|
|
@staticmethod
|
|
async def _check_fifo_skip(
|
|
query_db: AsyncSession,
|
|
inventory: BizPalletInventory,
|
|
filling: BizFillingRecord,
|
|
tunnel_id: int,
|
|
trace_no: str | None,
|
|
reason: str | None,
|
|
test_scope: TestBatchScope,
|
|
) -> dict[str, Any] | None:
|
|
if inventory.site_arrived_at is None:
|
|
return None
|
|
candidate, _ = await UsageFifoDao.recommendation(
|
|
query_db, site_id=inventory.site_id, strain_id=filling.strain_id,
|
|
tunnel_id=tunnel_id, trace_no=trace_no, scope=test_scope, for_update=True,
|
|
)
|
|
if not requires_fifo_skip_reason(inventory.pallet_id, inventory.site_arrived_at, candidate):
|
|
return None
|
|
if not reason:
|
|
raise ServiceException(
|
|
data='FIFO_SKIP_REASON_REQUIRED',
|
|
message=f'存在先到托盘 {candidate["pallet_code"]}，请填写跳过原因后重新提交',
|
|
)
|
|
return {
|
|
'palletId': candidate['pallet_id'], 'palletCode': candidate['pallet_code'],
|
|
'batchNo': candidate['batch_no'], 'remainingPackages': candidate['remaining_packages'],
|
|
'siteArrivedAt': candidate['site_arrived_at'].isoformat(),
|
|
'selectedSiteArrivedAt': inventory.site_arrived_at.isoformat(),
|
|
}
|
|
|
|
@staticmethod
|
|
def _is_exact_idempotent_result(
|
|
usage: BizUsageRecord,
|
|
page_object: UsageCreateModel,
|
|
request_hash: str,
|
|
) -> bool:
|
|
return usage.idempotency_key == page_object.idempotency_key and usage.request_hash == request_hash
|
|
|
|
@staticmethod
|
|
def _success(usage: BizUsageRecord) -> CrudResponseModel:
|
|
remaining_packages = usage.remaining_packages_after
|
|
flow_state = PalletFlowState.AT_SITE_EMPTY if remaining_packages == 0 else PalletFlowState.AT_SITE_PARTIAL
|
|
return CrudResponseModel(
|
|
is_success=True,
|
|
message='使用成功',
|
|
result=UsageSuccessResponseModel(
|
|
usage_id=usage.usage_id,
|
|
remaining_packages=remaining_packages,
|
|
flow_state=flow_state,
|
|
),
|
|
)
|
|
|
|
@staticmethod
|
|
def _tunnel_entry_success(pallet_id: int, tunnel_id: int, trace_no: str) -> CrudResponseModel:
|
|
return CrudResponseModel(
|
|
is_success=True,
|
|
message='扫码入隧成功',
|
|
result=TunnelEntrySuccessResponseModel(
|
|
pallet_id=pallet_id,
|
|
tunnel_id=tunnel_id,
|
|
trace_no=trace_no,
|
|
),
|
|
)
|
|
|
|
@staticmethod
|
|
def _is_exact_tunnel_entry_result(
|
|
event: BizPalletEvent,
|
|
page_object: TunnelEntryCreateModel,
|
|
request_hash: str,
|
|
) -> bool:
|
|
return (
|
|
event.event_type == PalletEventType.TUNNEL_ENTERED.value
|
|
and event.after_tunnel_id == page_object.tunnel_id
|
|
and event.idempotency_key == page_object.idempotency_key
|
|
and event.request_hash == request_hash
|
|
and bool(event.trace_no)
|
|
)
|
|
|
|
@classmethod
|
|
async def _lock_and_reauthorize_usage_site(
|
|
cls,
|
|
query_db: AsyncSession,
|
|
site_id: int,
|
|
current_user: CurrentUserModel,
|
|
test_scope: TestBatchScope,
|
|
) -> None:
|
|
site = await MasterDao.get_site(query_db, site_id, test_scope, for_update=True)
|
|
try:
|
|
await AccessService.assert_site_access(
|
|
query_db,
|
|
current_user,
|
|
site_id,
|
|
for_update=True,
|
|
)
|
|
except PermissionException:
|
|
if cls._is_admin(current_user):
|
|
raise
|
|
cls._raise_usage_not_accessible(current_user)
|
|
if site is None:
|
|
raise ServiceException(message='客户现场不存在')
|
|
assert_record_in_scope(site, test_scope, '现场')
|
|
|
|
@classmethod
|
|
async def _recover_integrity_error(
|
|
cls,
|
|
query_db: AsyncSession,
|
|
page_object: UsageCreateModel,
|
|
request_hash: str,
|
|
current_user: CurrentUserModel,
|
|
expected_site_id: int,
|
|
test_scope: TestBatchScope,
|
|
) -> CrudResponseModel:
|
|
try:
|
|
await TestBatchAccessService.lock_active_batch(query_db, test_scope)
|
|
site_ids = await cls._site_scope(query_db, current_user)
|
|
if site_ids is not None and expected_site_id not in site_ids:
|
|
cls._raise_pallet_not_accessible(current_user)
|
|
recovered = await OperationDao.get_usage_by_idempotency_key(
|
|
query_db,
|
|
page_object.idempotency_key,
|
|
test_scope,
|
|
site_ids=[expected_site_id],
|
|
)
|
|
if recovered is None:
|
|
raise ServiceException(message='使用失败，数据冲突')
|
|
assert_record_in_scope(recovered, test_scope, '使用记录')
|
|
if recovered.site_id != expected_site_id or not cls._is_exact_idempotent_result(
|
|
recovered, page_object, request_hash
|
|
):
|
|
raise ServiceException(message='使用失败，幂等键已被其他请求使用')
|
|
await cls._lock_and_reauthorize_usage_site(
|
|
query_db,
|
|
recovered.site_id,
|
|
current_user,
|
|
test_scope,
|
|
)
|
|
return cls._success(recovered)
|
|
finally:
|
|
await query_db.rollback()
|
|
|
|
@classmethod
|
|
async def get_usage_list_services(
|
|
cls,
|
|
query_db: AsyncSession,
|
|
query_object: UsagePageQueryModel,
|
|
current_user: CurrentUserModel,
|
|
test_scope: TestBatchScope,
|
|
is_page: bool = True,
|
|
) -> PageModel | list[UsageModel]:
|
|
site_ids = await cls._site_scope(query_db, current_user)
|
|
if site_ids is not None and query_object.site_id is not None and query_object.site_id not in site_ids:
|
|
raise PermissionException(data='', message='无权访问该客户现场')
|
|
if site_ids == []:
|
|
if is_page:
|
|
return PageModel(
|
|
rows=[],
|
|
pageNum=query_object.page_num,
|
|
pageSize=query_object.page_size,
|
|
total=0,
|
|
hasNext=False,
|
|
)
|
|
return []
|
|
result = await OperationDao.get_usage_list(
|
|
query_db,
|
|
query_object,
|
|
site_ids=site_ids,
|
|
scope=test_scope,
|
|
is_page=is_page,
|
|
)
|
|
rows = result.rows if isinstance(result, PageModel) else result
|
|
models = [UsageModel.model_validate(record) for record in rows]
|
|
if isinstance(result, PageModel):
|
|
return result.model_copy(update={'rows': models})
|
|
return models
|
|
|
|
@classmethod
|
|
async def usage_detail_services(
|
|
cls,
|
|
query_db: AsyncSession,
|
|
usage_id: int,
|
|
current_user: CurrentUserModel,
|
|
test_scope: TestBatchScope,
|
|
) -> UsageModel:
|
|
site_ids = await cls._site_scope(query_db, current_user)
|
|
usage = await OperationDao.get_usage_by_id(
|
|
query_db,
|
|
usage_id,
|
|
test_scope,
|
|
site_ids=site_ids,
|
|
)
|
|
if usage is None:
|
|
cls._raise_usage_not_accessible(current_user)
|
|
assert_record_in_scope(usage, test_scope, '使用记录')
|
|
return UsageModel.model_validate(usage)
|
|
|
|
@classmethod
|
|
async def get_line_options_services(
|
|
cls,
|
|
query_db: AsyncSession,
|
|
site_id: int,
|
|
current_user: CurrentUserModel,
|
|
test_scope: TestBatchScope,
|
|
) -> list[BusinessOptionModel]:
|
|
cls._current_identity(current_user)
|
|
await AccessService.assert_site_access(query_db, current_user, site_id)
|
|
site = await MasterDao.get_site(query_db, site_id, test_scope)
|
|
if site is None:
|
|
raise ServiceException(message='客户现场不存在')
|
|
assert_record_in_scope(site, test_scope, '现场')
|
|
if site.status != '0':
|
|
raise ServiceException(message='客户现场已停用')
|
|
lines = await MasterDao.list_line(
|
|
query_db,
|
|
LinePageQueryModel(siteId=site_id, status='0'),
|
|
test_scope,
|
|
False,
|
|
)
|
|
options = [
|
|
BusinessOptionModel(
|
|
id=cls._record_value(line, 'line_id'),
|
|
code=cls._record_value(line, 'line_code'),
|
|
name=cls._record_value(line, 'line_name'),
|
|
)
|
|
for line in lines
|
|
if cls._record_value(line, 'site_id') == site_id and cls._record_value(line, 'status') == '0'
|
|
]
|
|
return sorted(options, key=lambda option: option.id)
|
|
|
|
@classmethod
|
|
async def get_tunnel_options_services(
|
|
cls,
|
|
query_db: AsyncSession,
|
|
site_id: int,
|
|
current_user: CurrentUserModel,
|
|
test_scope: TestBatchScope,
|
|
) -> list[BusinessOptionModel]:
|
|
cls._current_identity(current_user)
|
|
await AccessService.assert_site_access(query_db, current_user, site_id)
|
|
site = await MasterDao.get_site(query_db, site_id, test_scope)
|
|
if site is None:
|
|
raise ServiceException(message='客户现场不存在')
|
|
assert_record_in_scope(site, test_scope, '现场')
|
|
if site.status != '0':
|
|
raise ServiceException(message='客户现场已停用')
|
|
tunnels = await MasterDao.list_tunnel(
|
|
query_db,
|
|
TunnelPageQueryModel(siteId=site_id, status='0'),
|
|
test_scope,
|
|
False,
|
|
)
|
|
options = [
|
|
BusinessOptionModel(
|
|
id=cls._record_value(tunnel, 'tunnel_id'),
|
|
code=cls._record_value(tunnel, 'tunnel_code'),
|
|
name=cls._record_value(tunnel, 'tunnel_name'),
|
|
)
|
|
for tunnel in tunnels
|
|
if cls._record_value(tunnel, 'site_id') == site_id and cls._record_value(tunnel, 'status') == '0'
|
|
]
|
|
return sorted(options, key=lambda option: option.id)
|
|
|
|
@staticmethod
|
|
def _build_tunnel_entry_request_hash(page_object: TunnelEntryCreateModel) -> str:
|
|
fields = {'qr_token': page_object.qr_token.strip(), 'tunnel_id': page_object.tunnel_id}
|
|
if page_object.fifo_skip_reason:
|
|
fields['fifo_skip_reason'] = page_object.fifo_skip_reason
|
|
payload = json.dumps(
|
|
fields,
|
|
sort_keys=True,
|
|
separators=(',', ':'),
|
|
).encode()
|
|
return hashlib.sha256(payload).hexdigest()
|
|
|
|
@classmethod
|
|
async def _recover_tunnel_entry_integrity_error(
|
|
cls,
|
|
query_db: AsyncSession,
|
|
page_object: TunnelEntryCreateModel,
|
|
request_hash: str,
|
|
current_user: CurrentUserModel,
|
|
test_scope: TestBatchScope,
|
|
) -> CrudResponseModel:
|
|
try:
|
|
await TestBatchAccessService.lock_active_batch(query_db, test_scope)
|
|
pallet = await InventoryDao.get_pallet_by_qr_token(
|
|
query_db,
|
|
page_object.qr_token.strip(),
|
|
test_scope,
|
|
)
|
|
if pallet is None:
|
|
cls._raise_pallet_not_accessible(current_user)
|
|
event = await InventoryDao.get_event_by_idempotency_key(
|
|
query_db,
|
|
pallet.pallet_id,
|
|
page_object.idempotency_key,
|
|
test_scope,
|
|
)
|
|
if event is None or not cls._is_exact_tunnel_entry_result(event, page_object, request_hash):
|
|
raise ServiceException(message='扫码入隧失败，幂等键已被其他请求使用')
|
|
return cls._tunnel_entry_success(pallet.pallet_id, page_object.tunnel_id, event.trace_no)
|
|
finally:
|
|
await query_db.rollback()
|
|
|
|
@classmethod
|
|
@retry_mysql_lock_conflicts
|
|
async def enter_tunnel_services( # noqa: PLR0912, PLR0915
|
|
cls,
|
|
query_db: AsyncSession,
|
|
page_object: TunnelEntryCreateModel,
|
|
current_user: CurrentUserModel,
|
|
test_scope: TestBatchScope,
|
|
) -> CrudResponseModel:
|
|
try:
|
|
await TestBatchAccessService.lock_active_batch(query_db, test_scope)
|
|
request_hash = cls._build_tunnel_entry_request_hash(page_object)
|
|
user_id, user_name = cls._current_identity(current_user)
|
|
pallet = await InventoryDao.get_pallet_by_qr_token(
|
|
query_db,
|
|
page_object.qr_token.strip(),
|
|
test_scope,
|
|
)
|
|
if pallet is None:
|
|
cls._raise_pallet_not_accessible(current_user)
|
|
assert_record_in_scope(pallet, test_scope, '托盘')
|
|
tunnel = await MasterDao.get_tunnel(query_db, page_object.tunnel_id, test_scope)
|
|
if tunnel is None:
|
|
raise ServiceException(message='隧道不存在')
|
|
assert_record_in_scope(tunnel, test_scope, '隧道')
|
|
if tunnel.status != '0':
|
|
raise ServiceException(message='隧道已停用')
|
|
preflight_tunnel_site_id = tunnel.site_id
|
|
await cls._assert_pallet_site_access(query_db, current_user, tunnel.site_id)
|
|
|
|
existing = await InventoryDao.get_event_by_idempotency_key(
|
|
query_db,
|
|
pallet.pallet_id,
|
|
page_object.idempotency_key,
|
|
test_scope,
|
|
)
|
|
if existing is not None:
|
|
assert_record_in_scope(existing, test_scope, '托盘事件')
|
|
if not cls._is_exact_tunnel_entry_result(existing, page_object, request_hash):
|
|
raise ServiceException(message='扫码入隧失败，幂等键已被其他请求使用')
|
|
result = cls._tunnel_entry_success(pallet.pallet_id, page_object.tunnel_id, existing.trace_no)
|
|
await query_db.rollback()
|
|
return result
|
|
|
|
# rollback 会令 ORM 属性过期；跨事务只传普通值，不触发隐式异步加载。
|
|
scanned_pallet_id = pallet.pallet_id
|
|
await query_db.rollback()
|
|
await TestBatchAccessService.lock_active_batch(query_db, test_scope)
|
|
inventory = await InventoryDao.get_inventory_for_update(query_db, scanned_pallet_id, test_scope)
|
|
if inventory is None:
|
|
cls._raise_pallet_not_accessible(current_user, '托盘库存不存在')
|
|
assert_record_in_scope(inventory, test_scope, '库存')
|
|
if inventory.site_id is None:
|
|
cls._raise_pallet_not_accessible(current_user, '托盘不在客户现场')
|
|
await cls._assert_pallet_site_access(query_db, current_user, inventory.site_id)
|
|
if inventory.site_id != preflight_tunnel_site_id:
|
|
raise ServiceException(message='隧道不属于托盘当前客户现场')
|
|
|
|
locked_pallets = await InventoryDao.get_pallets_by_ids(
|
|
query_db,
|
|
[inventory.pallet_id],
|
|
test_scope,
|
|
)
|
|
if len(locked_pallets) != 1 or locked_pallets[0].pallet_id != inventory.pallet_id:
|
|
cls._raise_pallet_not_accessible(current_user)
|
|
pallet = locked_pallets[0]
|
|
assert_record_in_scope(pallet, test_scope, '托盘')
|
|
tunnel = await MasterDao.get_tunnel(
|
|
query_db,
|
|
page_object.tunnel_id,
|
|
test_scope,
|
|
for_update=True,
|
|
)
|
|
if tunnel is None:
|
|
raise ServiceException(message='隧道不存在')
|
|
assert_record_in_scope(tunnel, test_scope, '隧道')
|
|
if tunnel.status != '0':
|
|
raise ServiceException(message='隧道已停用')
|
|
if tunnel.site_id != inventory.site_id:
|
|
raise ServiceException(message='隧道不属于托盘当前客户现场')
|
|
existing = await InventoryDao.get_event_by_idempotency_key(
|
|
query_db,
|
|
pallet.pallet_id,
|
|
page_object.idempotency_key,
|
|
test_scope,
|
|
)
|
|
if existing is not None:
|
|
assert_record_in_scope(existing, test_scope, '托盘事件')
|
|
if not cls._is_exact_tunnel_entry_result(existing, page_object, request_hash):
|
|
raise ServiceException(message='扫码入隧失败，幂等键已被其他请求使用')
|
|
result = cls._tunnel_entry_success(pallet.pallet_id, page_object.tunnel_id, existing.trace_no)
|
|
await query_db.rollback()
|
|
return result
|
|
|
|
if pallet.asset_status != '0':
|
|
raise ServiceException(message='托盘资产已停用，不允许入隧')
|
|
if (
|
|
inventory.flow_state
|
|
not in {
|
|
PalletFlowState.AT_SITE_UNUSED.value,
|
|
PalletFlowState.AT_SITE_PARTIAL.value,
|
|
}
|
|
or inventory.warehouse_id is not None
|
|
or inventory.active_filling_id is None
|
|
or type(inventory.remaining_packages) is not int
|
|
or inventory.remaining_packages <= 0
|
|
or inventory.remaining_packages > pallet.capacity_packages
|
|
):
|
|
raise ServiceException(message='托盘当前状态不允许入隧')
|
|
filling = await OperationDao.get_active_filling_for_update(
|
|
query_db,
|
|
inventory.active_filling_id,
|
|
test_scope,
|
|
)
|
|
if filling is None:
|
|
raise ServiceException(message='托盘活动装填周期不存在或已关闭')
|
|
assert_record_in_scope(filling, test_scope, '活动装填')
|
|
if (
|
|
filling.pallet_id != pallet.pallet_id
|
|
or filling.active_slot != 'ACTIVE'
|
|
or filling.closed_at is not None
|
|
):
|
|
raise ServiceException(message='托盘活动装填周期无效')
|
|
if inventory.tunnel_id is not None:
|
|
if inventory.tunnel_id == tunnel.tunnel_id and inventory.trace_no == tunnel.current_trace_no:
|
|
raise ServiceException(message='托盘已入当前隧道')
|
|
raise ServiceException(message='托盘已入其他隧道')
|
|
if inventory.trace_no is not None:
|
|
raise ServiceException(message='托盘隧道溯源状态不一致')
|
|
fifo_snapshot = await cls._check_fifo_skip(
|
|
query_db, inventory, filling, tunnel.tunnel_id, tunnel.current_trace_no,
|
|
page_object.fifo_skip_reason, test_scope,
|
|
)
|
|
if tunnel.current_trace_no is None:
|
|
tunnel.current_trace_no = f'TN-{secrets.token_hex(16)}'
|
|
tunnel.current_trace_started_at = datetime.now()
|
|
elif not tunnel.current_trace_no.strip():
|
|
raise ServiceException(message='隧道当前溯源号无效')
|
|
|
|
current_time = datetime.now()
|
|
before_state = inventory.flow_state
|
|
before_remaining_packages = inventory.remaining_packages
|
|
inventory.tunnel_id = tunnel.tunnel_id
|
|
inventory.trace_no = tunnel.current_trace_no
|
|
inventory.version += 1
|
|
inventory.updated_at = current_time
|
|
event = BizPalletEvent(
|
|
pallet_id=inventory.pallet_id,
|
|
event_type=PalletEventType.TUNNEL_ENTERED.value,
|
|
source_type='TUNNEL',
|
|
source_id=tunnel.tunnel_id,
|
|
fifo_skip_reason=page_object.fifo_skip_reason,
|
|
fifo_recommendation=fifo_snapshot,
|
|
before_state=before_state,
|
|
after_state=inventory.flow_state,
|
|
before_warehouse_id=inventory.warehouse_id,
|
|
before_site_id=inventory.site_id,
|
|
before_tunnel_id=None,
|
|
before_line_id=inventory.current_line_id,
|
|
after_warehouse_id=inventory.warehouse_id,
|
|
after_site_id=inventory.site_id,
|
|
after_tunnel_id=inventory.tunnel_id,
|
|
after_line_id=inventory.current_line_id,
|
|
before_remaining_packages=before_remaining_packages,
|
|
after_remaining_packages=inventory.remaining_packages,
|
|
trace_no=inventory.trace_no,
|
|
batch_no_snapshot=filling.batch_no,
|
|
operator_id=user_id,
|
|
occurred_at=current_time,
|
|
idempotency_key=page_object.idempotency_key,
|
|
request_hash=request_hash,
|
|
test_batch_id=test_scope.batch_id,
|
|
)
|
|
await InventoryDao.add_event(query_db, inventory, event, test_scope)
|
|
tunnel.update_by = user_name
|
|
tunnel.update_time = current_time
|
|
result = cls._tunnel_entry_success(pallet.pallet_id, tunnel.tunnel_id, tunnel.current_trace_no)
|
|
await query_db.commit()
|
|
return result
|
|
except IntegrityError:
|
|
await query_db.rollback()
|
|
return await cls._recover_tunnel_entry_integrity_error(
|
|
query_db,
|
|
page_object,
|
|
cls._build_tunnel_entry_request_hash(page_object),
|
|
current_user,
|
|
test_scope,
|
|
)
|
|
except Exception:
|
|
await query_db.rollback()
|
|
raise
|
|
|
|
@classmethod
|
|
async def scan_usage_pallet_services( # noqa: PLR0912
|
|
cls,
|
|
query_db: AsyncSession,
|
|
qr_token: str,
|
|
current_user: CurrentUserModel,
|
|
test_scope: TestBatchScope,
|
|
) -> PalletScanResponseModel:
|
|
normalized_qr = qr_token.strip()
|
|
pallet = await InventoryDao.get_pallet_by_qr_token(query_db, normalized_qr, test_scope)
|
|
if pallet is None:
|
|
cls._raise_pallet_not_accessible(current_user)
|
|
assert_record_in_scope(pallet, test_scope, '托盘')
|
|
inventory = await MasterDao.get_pallet_inventory(query_db, pallet.pallet_id, test_scope)
|
|
if inventory is None:
|
|
cls._raise_pallet_not_accessible(current_user, '托盘库存不存在')
|
|
assert_record_in_scope(inventory, test_scope, '库存')
|
|
if inventory.site_id is None:
|
|
cls._raise_pallet_not_accessible(current_user, '托盘不在客户现场')
|
|
await cls._assert_pallet_site_access(query_db, current_user, inventory.site_id)
|
|
|
|
if normalized_qr not in {pallet.qr_token, pallet.pallet_code} or pallet.pallet_id != inventory.pallet_id:
|
|
raise ServiceException(message='托盘二维码与库存不一致')
|
|
if pallet.asset_status != '0':
|
|
raise ServiceException(message='托盘资产已停用，不允许使用')
|
|
if (
|
|
inventory.flow_state
|
|
not in {
|
|
PalletFlowState.AT_SITE_UNUSED.value,
|
|
PalletFlowState.AT_SITE_PARTIAL.value,
|
|
}
|
|
or inventory.warehouse_id is not None
|
|
or inventory.active_filling_id is None
|
|
or type(inventory.remaining_packages) is not int
|
|
or inventory.remaining_packages <= 0
|
|
or inventory.remaining_packages > pallet.capacity_packages
|
|
):
|
|
raise ServiceException(message='托盘当前状态不允许使用')
|
|
filling = await OperationDao.get_filling_by_id(query_db, inventory.active_filling_id, test_scope)
|
|
if filling is None:
|
|
raise ServiceException(message='托盘活动装填周期不存在')
|
|
assert_record_in_scope(filling, test_scope, '活动装填')
|
|
if (
|
|
filling.filling_id != inventory.active_filling_id
|
|
or filling.pallet_id != pallet.pallet_id
|
|
or filling.active_slot != 'ACTIVE'
|
|
or filling.closed_at is not None
|
|
):
|
|
raise ServiceException(message='托盘活动装填周期无效')
|
|
allowed_actions = ['usage']
|
|
if inventory.tunnel_id is None:
|
|
allowed_actions = ['enter_tunnel']
|
|
else:
|
|
if inventory.trace_no is None:
|
|
raise ServiceException(message='托盘隧道溯源状态不一致')
|
|
tunnel = await MasterDao.get_tunnel(query_db, inventory.tunnel_id, test_scope)
|
|
if tunnel is None or tunnel.site_id != inventory.site_id:
|
|
raise ServiceException(message='托盘当前隧道不存在或现场不一致')
|
|
assert_record_in_scope(tunnel, test_scope, '隧道')
|
|
if tunnel.status != '0' or tunnel.current_trace_no != inventory.trace_no:
|
|
raise ServiceException(message='托盘当前隧道已停用或溯源状态不一致')
|
|
return PalletScanResponseModel(
|
|
pallet_id=pallet.pallet_id,
|
|
qr_token=pallet.qr_token,
|
|
pallet_code=pallet.pallet_code,
|
|
flow_state=PalletFlowState(inventory.flow_state),
|
|
warehouse_id=None,
|
|
site_id=inventory.site_id,
|
|
tunnel_id=inventory.tunnel_id,
|
|
trace_no=inventory.trace_no,
|
|
current_line_id=inventory.current_line_id,
|
|
active_filling_id=inventory.active_filling_id,
|
|
strain_id=filling.strain_id,
|
|
strain_code=filling.strain_code_snapshot,
|
|
strain_name=filling.strain_name_snapshot,
|
|
batch_no=filling.batch_no,
|
|
production_date=filling.production_date,
|
|
expiry_date=filling.expiry_date,
|
|
package_spec=filling.package_spec,
|
|
capacity_packages=pallet.capacity_packages,
|
|
remaining_packages=inventory.remaining_packages,
|
|
allowed_actions=allowed_actions,
|
|
)
|
|
|
|
@classmethod
|
|
@retry_mysql_lock_conflicts
|
|
async def record_usage_services( # noqa: PLR0912, PLR0915
|
|
cls,
|
|
query_db: AsyncSession,
|
|
page_object: UsageCreateModel,
|
|
current_user: CurrentUserModel,
|
|
test_scope: TestBatchScope,
|
|
) -> CrudResponseModel:
|
|
expected_site_id: int | None = None
|
|
try:
|
|
await TestBatchAccessService.lock_active_batch(query_db, test_scope)
|
|
request_hash = cls._build_request_hash(page_object)
|
|
user_id, user_name = cls._current_identity(current_user)
|
|
preflight_site_ids = await cls._site_scope(query_db, current_user)
|
|
preflight_tunnel = await MasterDao.get_tunnel(query_db, page_object.tunnel_id, test_scope)
|
|
if preflight_tunnel is None:
|
|
raise ServiceException(message='隧道不存在')
|
|
assert_record_in_scope(preflight_tunnel, test_scope, '隧道')
|
|
preflight_site_id = preflight_tunnel.site_id
|
|
if page_object.line_id is not None:
|
|
# 仅在预检阶段读取产线以确定批次范围；状态和实际现场归属在锁定库存后复核。
|
|
preflight_line = await MasterDao.get_line(query_db, page_object.line_id, test_scope)
|
|
if preflight_line is not None:
|
|
assert_record_in_scope(preflight_line, test_scope, '产线')
|
|
existing = None
|
|
if preflight_site_id is not None and (
|
|
preflight_site_ids is None or preflight_site_id in preflight_site_ids
|
|
):
|
|
existing = await OperationDao.get_usage_by_idempotency_key(
|
|
query_db,
|
|
page_object.idempotency_key,
|
|
test_scope,
|
|
site_ids=[preflight_site_id],
|
|
)
|
|
if existing is not None:
|
|
assert_record_in_scope(existing, test_scope, '使用记录')
|
|
if not cls._is_exact_idempotent_result(existing, page_object, request_hash):
|
|
raise ServiceException(message='使用失败，幂等键已被其他请求使用')
|
|
if preflight_site_ids is not None and existing.site_id not in preflight_site_ids:
|
|
cls._raise_usage_not_accessible(current_user)
|
|
await cls._lock_and_reauthorize_usage_site(
|
|
query_db,
|
|
existing.site_id,
|
|
current_user,
|
|
test_scope,
|
|
)
|
|
result = cls._success(existing)
|
|
await query_db.rollback()
|
|
return result
|
|
|
|
scanned_pallet = await InventoryDao.get_pallet_by_qr_token(
|
|
query_db,
|
|
page_object.qr_token.strip(),
|
|
test_scope,
|
|
)
|
|
if scanned_pallet is None:
|
|
cls._raise_pallet_not_accessible(current_user)
|
|
assert_record_in_scope(scanned_pallet, test_scope, '托盘')
|
|
scanned_pallet_id = int(scanned_pallet.pallet_id)
|
|
await query_db.rollback()
|
|
await TestBatchAccessService.lock_active_batch(query_db, test_scope)
|
|
|
|
inventory = await InventoryDao.get_inventory_for_update(
|
|
query_db,
|
|
scanned_pallet_id,
|
|
test_scope,
|
|
)
|
|
if inventory is None:
|
|
cls._raise_pallet_not_accessible(current_user, '托盘库存不存在')
|
|
assert_record_in_scope(inventory, test_scope, '库存')
|
|
if inventory.site_id is None:
|
|
cls._raise_pallet_not_accessible(current_user, '托盘不在客户现场')
|
|
expected_site_id = inventory.site_id
|
|
await cls._assert_pallet_site_access(query_db, current_user, inventory.site_id)
|
|
locked_site_ids = await cls._site_scope(query_db, current_user)
|
|
if locked_site_ids is not None and inventory.site_id not in locked_site_ids:
|
|
cls._raise_pallet_not_accessible(current_user)
|
|
|
|
locked_pallets = await InventoryDao.get_pallets_by_ids(
|
|
query_db,
|
|
[inventory.pallet_id],
|
|
test_scope,
|
|
)
|
|
if len(locked_pallets) != 1 or locked_pallets[0].pallet_id != inventory.pallet_id:
|
|
cls._raise_pallet_not_accessible(current_user)
|
|
pallet = locked_pallets[0]
|
|
assert_record_in_scope(pallet, test_scope, '托盘')
|
|
|
|
filling = (
|
|
await OperationDao.get_active_filling_for_update(
|
|
query_db,
|
|
inventory.active_filling_id,
|
|
test_scope,
|
|
)
|
|
if inventory.active_filling_id is not None
|
|
else None
|
|
)
|
|
locked_existing = await OperationDao.get_usage_by_idempotency_key(
|
|
query_db,
|
|
page_object.idempotency_key,
|
|
test_scope,
|
|
site_ids=[inventory.site_id],
|
|
)
|
|
if locked_existing is not None:
|
|
assert_record_in_scope(locked_existing, test_scope, '使用记录')
|
|
if locked_existing.site_id != inventory.site_id or not cls._is_exact_idempotent_result(
|
|
locked_existing,
|
|
page_object,
|
|
request_hash,
|
|
):
|
|
raise ServiceException(message='使用失败，幂等键已被其他请求使用')
|
|
await cls._lock_and_reauthorize_usage_site(
|
|
query_db,
|
|
locked_existing.site_id,
|
|
current_user,
|
|
test_scope,
|
|
)
|
|
result = cls._success(locked_existing)
|
|
await query_db.rollback()
|
|
return result
|
|
|
|
if inventory.active_filling_id is None:
|
|
raise ServiceException(message='托盘不存在活动装填周期')
|
|
if filling is None:
|
|
raise ServiceException(message='托盘活动装填周期不存在或已关闭')
|
|
assert_record_in_scope(filling, test_scope, '活动装填')
|
|
|
|
site = await MasterDao.get_site(query_db, inventory.site_id, test_scope, for_update=True)
|
|
await cls._assert_pallet_site_access(
|
|
query_db,
|
|
current_user,
|
|
inventory.site_id,
|
|
for_update=True,
|
|
)
|
|
if site is None:
|
|
raise ServiceException(message='客户现场不存在')
|
|
assert_record_in_scope(site, test_scope, '现场')
|
|
if site.status != '0':
|
|
raise ServiceException(message='客户现场已停用')
|
|
tunnel = await MasterDao.get_tunnel(
|
|
query_db,
|
|
page_object.tunnel_id,
|
|
test_scope,
|
|
for_update=True,
|
|
)
|
|
if tunnel is None:
|
|
raise ServiceException(message='隧道不存在')
|
|
assert_record_in_scope(tunnel, test_scope, '隧道')
|
|
if tunnel.status != '0':
|
|
raise ServiceException(message='隧道已停用')
|
|
if tunnel.site_id != inventory.site_id:
|
|
raise ServiceException(message='隧道不属于托盘当前客户现场')
|
|
if inventory.tunnel_id != tunnel.tunnel_id or not inventory.trace_no:
|
|
raise ServiceException(message='托盘尚未扫码入当前隧道')
|
|
if tunnel.current_trace_no != inventory.trace_no:
|
|
raise ServiceException(message='托盘与隧道的物料溯源号不一致')
|
|
|
|
line = None
|
|
if page_object.line_id is not None:
|
|
line = await MasterDao.get_line(query_db, page_object.line_id, test_scope, for_update=True)
|
|
if line is None:
|
|
raise ServiceException(message='产线不存在')
|
|
assert_record_in_scope(line, test_scope, '产线')
|
|
if line.status != '0':
|
|
raise ServiceException(message='产线已停用')
|
|
if line.site_id != inventory.site_id:
|
|
raise ServiceException(message='产线不属于托盘当前客户现场')
|
|
|
|
if pallet.pallet_id != inventory.pallet_id or pallet.qr_token != page_object.qr_token.strip():
|
|
raise ServiceException(message='托盘二维码与库存不一致')
|
|
if pallet.asset_status != '0':
|
|
raise ServiceException(message='托盘资产已停用，不允许使用')
|
|
if inventory.flow_state not in {
|
|
PalletFlowState.AT_SITE_UNUSED.value,
|
|
PalletFlowState.AT_SITE_PARTIAL.value,
|
|
}:
|
|
raise ServiceException(message='托盘当前状态不允许使用')
|
|
if inventory.warehouse_id is not None:
|
|
raise ServiceException(message='托盘当前位置无效')
|
|
if type(inventory.remaining_packages) is not int or inventory.remaining_packages <= 0:
|
|
raise ServiceException(message='托盘当前余量无效')
|
|
if inventory.remaining_packages > pallet.capacity_packages:
|
|
raise ServiceException(message='托盘当前余量超过容量')
|
|
if type(inventory.version) is not int or inventory.version < 0:
|
|
raise ServiceException(message='托盘库存版本无效')
|
|
if (
|
|
filling.filling_id != inventory.active_filling_id
|
|
or filling.pallet_id != pallet.pallet_id
|
|
or filling.active_slot != 'ACTIVE'
|
|
or filling.closed_at is not None
|
|
):
|
|
raise ServiceException(message='托盘活动装填周期无效')
|
|
if (
|
|
line is not None
|
|
and inventory.current_line_id is not None
|
|
and inventory.current_line_id != line.line_id
|
|
):
|
|
raise ServiceException(message='托盘本次驻场周期已绑定其他产线')
|
|
if type(page_object.used_packages) is not int or page_object.used_packages <= 0:
|
|
raise ServiceException(message='使用包数必须为正整数')
|
|
if page_object.used_packages > inventory.remaining_packages:
|
|
raise ServiceException(message='使用包数不能超过当前余量')
|
|
|
|
fifo_snapshot = await cls._check_fifo_skip(
|
|
query_db, inventory, filling, tunnel.tunnel_id, inventory.trace_no,
|
|
page_object.fifo_skip_reason, test_scope,
|
|
)
|
|
|
|
remaining_after = inventory.remaining_packages - page_object.used_packages
|
|
target_state = PalletFlowState.AT_SITE_EMPTY if remaining_after == 0 else PalletFlowState.AT_SITE_PARTIAL
|
|
PalletStateMachine.ensure_transition(
|
|
PalletFlowState(inventory.flow_state),
|
|
target_state,
|
|
)
|
|
|
|
current_time = datetime.now()
|
|
usage = BizUsageRecord(
|
|
fifo_skip_reason=page_object.fifo_skip_reason,
|
|
fifo_recommendation=fifo_snapshot,
|
|
usage_no=f'US-{secrets.token_hex(16)}',
|
|
pallet_id=pallet.pallet_id,
|
|
active_filling_id=filling.filling_id,
|
|
site_id=inventory.site_id,
|
|
tunnel_id=tunnel.tunnel_id,
|
|
trace_no=inventory.trace_no,
|
|
batch_no_snapshot=filling.batch_no,
|
|
line_id=line.line_id if line is not None else inventory.current_line_id,
|
|
used_packages=page_object.used_packages,
|
|
remaining_packages_after=remaining_after,
|
|
idempotency_key=page_object.idempotency_key,
|
|
request_hash=request_hash,
|
|
operator_id=user_id,
|
|
occurred_at=current_time,
|
|
test_batch_id=test_scope.batch_id,
|
|
)
|
|
usage = await OperationDao.add_usage_record(query_db, usage)
|
|
assert_record_in_scope(usage, test_scope, '使用记录')
|
|
usage_id = usage.usage_id
|
|
result = cls._success(usage)
|
|
|
|
before_state = inventory.flow_state
|
|
before_warehouse_id = inventory.warehouse_id
|
|
before_site_id = inventory.site_id
|
|
before_tunnel_id = inventory.tunnel_id
|
|
before_trace_no = inventory.trace_no
|
|
before_line_id = inventory.current_line_id
|
|
before_remaining_packages = inventory.remaining_packages
|
|
if line is not None:
|
|
inventory.current_line_id = line.line_id
|
|
inventory.remaining_packages = remaining_after
|
|
inventory.flow_state = target_state.value
|
|
if target_state is PalletFlowState.AT_SITE_EMPTY:
|
|
filling.active_slot = None
|
|
filling.closed_at = current_time
|
|
filling.update_by = user_name
|
|
filling.update_time = current_time
|
|
await OperationDao.update_filling_record(query_db, filling)
|
|
inventory.active_filling_id = None
|
|
inventory.version += 1
|
|
inventory.updated_at = current_time
|
|
event = BizPalletEvent(
|
|
pallet_id=inventory.pallet_id,
|
|
event_type=PalletEventType.USAGE_RECORDED.value,
|
|
source_type='USAGE',
|
|
source_id=usage_id,
|
|
fifo_skip_reason=page_object.fifo_skip_reason,
|
|
fifo_recommendation=fifo_snapshot,
|
|
before_state=before_state,
|
|
after_state=target_state.value,
|
|
before_warehouse_id=before_warehouse_id,
|
|
before_site_id=before_site_id,
|
|
before_tunnel_id=before_tunnel_id,
|
|
before_line_id=before_line_id,
|
|
after_warehouse_id=inventory.warehouse_id,
|
|
after_site_id=inventory.site_id,
|
|
after_tunnel_id=inventory.tunnel_id,
|
|
after_line_id=inventory.current_line_id,
|
|
before_remaining_packages=before_remaining_packages,
|
|
after_remaining_packages=remaining_after,
|
|
trace_no=before_trace_no,
|
|
batch_no_snapshot=filling.batch_no,
|
|
operator_id=user_id,
|
|
occurred_at=current_time,
|
|
idempotency_key=page_object.idempotency_key,
|
|
request_hash=request_hash,
|
|
test_batch_id=test_scope.batch_id,
|
|
)
|
|
await InventoryDao.add_event(query_db, inventory, event, test_scope)
|
|
await query_db.commit()
|
|
return result
|
|
except IntegrityError:
|
|
await query_db.rollback()
|
|
if expected_site_id is None:
|
|
raise ServiceException(message='使用失败，数据冲突') from None
|
|
return await cls._recover_integrity_error(
|
|
query_db,
|
|
page_object,
|
|
request_hash,
|
|
current_user,
|
|
expected_site_id,
|
|
test_scope,
|
|
)
|
|
except Exception:
|
|
await query_db.rollback()
|
|
raise
|