86 lines
3.5 KiB
Python
86 lines
3.5 KiB
Python
"""按真实收货时间推荐可用物料，不推断历史到货时间。"""
|
|
|
|
from typing import Any
|
|
|
|
from sqlalchemy import and_, func, or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from module_business.domain.test_batch_scope import TestBatchScope, batch_scope_condition
|
|
from module_business.entity.do.master_do import BizPallet
|
|
from module_business.entity.do.operation_do import BizFillingRecord, BizPalletInventory
|
|
|
|
|
|
class UsageFifoDao:
|
|
@staticmethod
|
|
async def recommendation(
|
|
db: AsyncSession,
|
|
*,
|
|
site_id: int,
|
|
strain_id: int,
|
|
tunnel_id: int | None,
|
|
trace_no: str | None,
|
|
scope: TestBatchScope,
|
|
for_update: bool = False,
|
|
) -> tuple[dict[str, Any] | None, int]:
|
|
inventory, filling, pallet = BizPalletInventory, BizFillingRecord, BizPallet
|
|
unbound = and_(inventory.tunnel_id.is_(None), inventory.trace_no.is_(None))
|
|
conditions = [
|
|
inventory.site_id == site_id,
|
|
inventory.warehouse_id.is_(None),
|
|
inventory.flow_state.in_(['AT_SITE_UNUSED', 'AT_SITE_PARTIAL']),
|
|
inventory.remaining_packages > 0,
|
|
inventory.remaining_packages <= pallet.capacity_packages,
|
|
filling.strain_id == strain_id,
|
|
filling.active_slot == 'ACTIVE',
|
|
filling.closed_at.is_(None),
|
|
pallet.asset_status == '0',
|
|
batch_scope_condition(inventory.test_batch_id, scope),
|
|
batch_scope_condition(filling.test_batch_id, scope),
|
|
batch_scope_condition(pallet.test_batch_id, scope),
|
|
(
|
|
unbound
|
|
if tunnel_id is None
|
|
else or_(unbound, and_(inventory.tunnel_id == tunnel_id, inventory.trace_no == trace_no))
|
|
),
|
|
]
|
|
base = (
|
|
select(
|
|
pallet.pallet_id,
|
|
pallet.pallet_code,
|
|
filling.batch_no,
|
|
inventory.remaining_packages,
|
|
inventory.site_arrived_at,
|
|
)
|
|
.select_from(inventory)
|
|
.join(pallet, pallet.pallet_id == inventory.pallet_id)
|
|
.join(
|
|
filling,
|
|
and_(filling.filling_id == inventory.active_filling_id, filling.pallet_id == inventory.pallet_id),
|
|
)
|
|
.where(*conditions)
|
|
)
|
|
statement = (
|
|
base.where(inventory.site_arrived_at.is_not(None))
|
|
.order_by(
|
|
inventory.site_arrived_at,
|
|
inventory.pallet_id,
|
|
)
|
|
.limit(1)
|
|
)
|
|
if for_update:
|
|
# 写入决策必须用当前读，不能复用权限预检期间建立的 MVCC 旧快照。
|
|
# 不使用 skip_locked，避免把正在操作的早到物料悄悄跳过。
|
|
# 此时已持有所选库存及现场/隧道锁，不能等待其他候选库存造成反向锁依赖。
|
|
# NOWAIT 不会跳过早到托盘：冲突令整个命令回滚，再按原幂等键有限重试。
|
|
statement = statement.with_for_update(nowait=True)
|
|
known = await db.execute(statement)
|
|
first = known.mappings().first()
|
|
if for_update:
|
|
return (dict(first) if first is not None else None, 0)
|
|
unknown = await db.execute(
|
|
select(func.count()).select_from(
|
|
base.where(inventory.site_arrived_at.is_(None)).subquery(),
|
|
)
|
|
)
|
|
return (dict(first) if first is not None else None, unknown.scalar_one())
|