zkxw/backend/app/services/storage_facility_code.py

"""库位节点编号路径与储物格完整编号计算、快照级联刷新。"""
from __future__ import annotations

import re
from datetime import datetime
from typing import Dict, List, Optional, Union

from sqlalchemy.orm import Session

from app.models.models import (
    BizCellProduct,
    BizOutboundReleaseCell,
    BizStorageFacility,
    BizStorageFacilitySlot,
)

_NODE_CODE_PATTERN = re.compile(r"^[A-Za-z0-9]+$")


def validate_node_code(node_code: str) -> str:
    code = (node_code or "").strip()
    if not code:
        raise ValueError("节点编号不能为空")
    if not _NODE_CODE_PATTERN.match(code):
        raise ValueError("节点编号只能包含字母和数字，不能包含连字符")
    if len(code) > 32:
        raise ValueError("节点编号长度不能超过 32 个字符")
    return code


def assert_sibling_node_code_unique(
    db: Session,
    *,
    parent_id: str,
    node_code: str,
    tenant_no: str,
    exclude_id: Optional[str] = None,
) -> None:
    pid = (parent_id or "0").strip() or "0"
    code = validate_node_code(node_code)
    q = db.query(BizStorageFacility).filter(
        BizStorageFacility.parent_id == pid,
        BizStorageFacility.node_code == code,
        BizStorageFacility.tenant_no == tenant_no,
        BizStorageFacility.del_flag == 0,
    )
    if exclude_id:
        q = q.filter(BizStorageFacility.id != exclude_id)
    if q.first():
        raise ValueError("同级节点编号已存在")


def build_node_code_path(
    node_id: str,
    facility_by_id: Dict[str, Union[BizStorageFacility, dict]],
) -> str:
    parts: List[str] = []
    cur_id = (node_id or "").strip()
    visited: set[str] = set()
    while cur_id and cur_id != "0" and cur_id not in visited:
        visited.add(cur_id)
        node = facility_by_id.get(cur_id)
        if not node:
            break
        if isinstance(node, dict):
            code = (node.get("nodeCode") or node.get("node_code") or "").strip()
            parent_id = (node.get("parentId") or node.get("parent_id") or "0").strip()
        else:
            code = (node.node_code or "").strip()
            parent_id = (node.parent_id or "0").strip()
        if code:
            parts.append(code)
        cur_id = parent_id
    parts.reverse()
    return "-".join(parts)


def build_full_slot_code(
    facility_node_id: str,
    slot_code: Optional[str],
    facility_by_id: Dict[str, Union[BizStorageFacility, dict]],
) -> str:
    path = build_node_code_path(facility_node_id, facility_by_id)
    local = (slot_code or "").strip()
    if path and local:
        return f"{path}-{local}"
    return local or path


def load_facility_map(db: Session, tenant_no: str) -> Dict[str, BizStorageFacility]:
    rows = (
        db.query(BizStorageFacility)
        .filter(
            BizStorageFacility.tenant_no == tenant_no,
            BizStorageFacility.del_flag == 0,
        )
        .all()
    )
    return {r.id: r for r in rows}


def collect_slot_ids_under_nodes(
    db: Session,
    node_ids: List[str],
    tenant_no: str,
    *,
    collect_descendant_ids_fn,
) -> List[str]:
    all_node_ids: set[str] = set()
    for nid in node_ids:
        nid = (nid or "").strip()
        if not nid:
            continue
        all_node_ids.add(nid)
        all_node_ids.update(collect_descendant_ids_fn(db, nid, tenant_no))
    if not all_node_ids:
        return []

    leaves = (
        db.query(BizStorageFacility)
        .filter(
            BizStorageFacility.id.in_(all_node_ids),
            BizStorageFacility.is_leaf == 1,
            BizStorageFacility.tenant_no == tenant_no,
            BizStorageFacility.del_flag == 0,
        )
        .all()
    )
    leaf_ids = [leaf.id for leaf in leaves]
    if not leaf_ids:
        return []

    slots = (
        db.query(BizStorageFacilitySlot)
        .filter(
            BizStorageFacilitySlot.facility_node_id.in_(leaf_ids),
            BizStorageFacilitySlot.tenant_no == tenant_no,
            BizStorageFacilitySlot.del_flag == 0,
        )
        .all()
    )
    return [s.id for s in slots]


def refresh_outbound_release_cell_slot_names(
    db: Session,
    slot_full_map: Dict[str, str],
    tenant_no: str,
) -> int:
    if not slot_full_map:
        return 0

    products = (
        db.query(BizCellProduct)
        .filter(
            BizCellProduct.storage_slot_id.in_(list(slot_full_map.keys())),
            BizCellProduct.tenant_no == tenant_no,
            BizCellProduct.del_flag == 0,
        )
        .all()
    )
    product_name_map = {
        p.id: slot_full_map.get((p.storage_slot_id or "").strip(), "")
        for p in products
        if (p.storage_slot_id or "").strip()
    }
    if not product_name_map:
        return 0

    cells = (
        db.query(BizOutboundReleaseCell)
        .filter(
            BizOutboundReleaseCell.cell_product_id.in_(list(product_name_map.keys())),
            BizOutboundReleaseCell.tenant_no == tenant_no,
            BizOutboundReleaseCell.del_flag == 0,
        )
        .all()
    )
    now = datetime.now()
    updated = 0
    for cell in cells:
        new_name = product_name_map.get(cell.cell_product_id, "")
        if new_name and cell.storage_slot_name != new_name:
            cell.storage_slot_name = new_name
            cell.update_time = now
            updated += 1
    return updated


def cascade_refresh_slot_name_snapshots_for_slots(
    db: Session,
    slot_ids: List[str],
    tenant_no: str,
) -> int:
    ids = [(sid or "").strip() for sid in slot_ids if (sid or "").strip()]
    if not ids:
        return 0

    slots = (
        db.query(BizStorageFacilitySlot)
        .filter(
            BizStorageFacilitySlot.id.in_(ids),
            BizStorageFacilitySlot.tenant_no == tenant_no,
            BizStorageFacilitySlot.del_flag == 0,
        )
        .all()
    )
    if not slots:
        return 0

    facility_by_id = load_facility_map(db, tenant_no)
    slot_full_map = {
        s.id: build_full_slot_code(s.facility_node_id, s.slot_code, facility_by_id)
        for s in slots
    }
    return refresh_outbound_release_cell_slot_names(db, slot_full_map, tenant_no)


def cascade_refresh_slot_name_snapshots_for_nodes(
    db: Session,
    node_ids: List[str],
    tenant_no: str,
    *,
    collect_descendant_ids_fn,
) -> int:
    slot_ids = collect_slot_ids_under_nodes(
        db, node_ids, tenant_no, collect_descendant_ids_fn=collect_descendant_ids_fn
    )
    return cascade_refresh_slot_name_snapshots_for_slots(db, slot_ids, tenant_no)