50 lines
1.6 KiB
Python
50 lines
1.6 KiB
Python
"""工艺/步骤设备多选：逗号分隔 ID 与名称快照解析"""
|
|
from __future__ import annotations
|
|
|
|
from typing import List, Optional, Tuple, Union
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.models import BizEquipment
|
|
|
|
|
|
def parse_equipment_ids(value: Optional[Union[str, List[str]]]) -> List[str]:
|
|
if not value:
|
|
return []
|
|
if isinstance(value, list):
|
|
return [(x or "").strip() for x in value if (x or "").strip()]
|
|
return [x.strip() for x in str(value).split(",") if x.strip()]
|
|
|
|
|
|
def join_equipment_ids(ids: Optional[List[str]]) -> Optional[str]:
|
|
clean = parse_equipment_ids(ids)
|
|
return ",".join(clean) if clean else None
|
|
|
|
|
|
def resolve_equipment_list_snapshot(
|
|
db: Session,
|
|
equipment_ids: Optional[Union[str, List[str]]],
|
|
tenant_no: str,
|
|
) -> Tuple[Optional[str], Optional[str]]:
|
|
"""校验设备列表，返回 (equipment_ids 逗号串, equipment_name 逗号串快照)。"""
|
|
id_list = parse_equipment_ids(equipment_ids)
|
|
if not id_list:
|
|
return None, None
|
|
validated_ids: List[str] = []
|
|
names: List[str] = []
|
|
for eid in id_list:
|
|
obj = (
|
|
db.query(BizEquipment)
|
|
.filter(
|
|
BizEquipment.id == eid,
|
|
BizEquipment.tenant_no == tenant_no,
|
|
BizEquipment.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
if not obj:
|
|
raise ValueError("所选设备不存在或已删除")
|
|
validated_ids.append(obj.id)
|
|
names.append((obj.equipment_name or "").strip() or obj.equipment_code or obj.id)
|
|
return ",".join(validated_ids), ",".join(names)
|