feat(backend): add QC inspection item template CRUD helpers.
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed
1 parent
4b7d026d4e
commit
bc62ac796a
2 files changed
+341
No files matched your search
@@ -0,0 +1,314 @@
|
||||
"""质检检测项目模版 CRUD"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Any, Dict, List, Optional, Set, Tuple
|
||||
|
||||
from sqlalchemy import func
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.utils import generate_uuid
|
||||
from app.models.models import (
|
||||
BizQcInspectionItemTemplate,
|
||||
BizQcLabTestItem,
|
||||
BizQcThirdPartyItem,
|
||||
)
|
||||
|
||||
|
||||
def parse_id_csv(raw: Optional[str]) -> List[str]:
|
||||
if not raw:
|
||||
return []
|
||||
out: List[str] = []
|
||||
seen: Set[str] = set()
|
||||
for part in str(raw).split(","):
|
||||
sid = part.strip()
|
||||
if not sid or sid in seen:
|
||||
continue
|
||||
seen.add(sid)
|
||||
out.append(sid)
|
||||
return out
|
||||
|
||||
|
||||
def join_id_list(ids: List[str]) -> str:
|
||||
cleaned = parse_id_csv(",".join(ids or []))
|
||||
return ",".join(cleaned)
|
||||
|
||||
|
||||
def filter_active_ids(ids: List[str], active_set: Set[str]) -> Tuple[List[str], int]:
|
||||
"""返回 (保留且去重保序的 id 列表, 跳过次数含重复无效)。"""
|
||||
kept: List[str] = []
|
||||
seen: Set[str] = set()
|
||||
skipped = 0
|
||||
for raw in ids or []:
|
||||
sid = (raw or "").strip()
|
||||
if not sid:
|
||||
continue
|
||||
if sid not in active_set:
|
||||
skipped += 1
|
||||
continue
|
||||
if sid in seen:
|
||||
continue
|
||||
seen.add(sid)
|
||||
kept.append(sid)
|
||||
return kept, skipped
|
||||
|
||||
|
||||
def require_nonempty_items(lab_ids: List[str], third_ids: List[str]) -> None:
|
||||
if not lab_ids and not third_ids:
|
||||
raise ValueError("请至少选择一个检测项目")
|
||||
|
||||
|
||||
def _row_to_dict(r: BizQcInspectionItemTemplate) -> Dict[str, Any]:
|
||||
lab_ids = parse_id_csv(r.lab_item_ids)
|
||||
third_ids = parse_id_csv(r.third_party_item_ids)
|
||||
return {
|
||||
"id": r.id,
|
||||
"templateName": r.template_name,
|
||||
"labItemIds": lab_ids,
|
||||
"thirdPartyItemIds": third_ids,
|
||||
"labItemCount": len(lab_ids),
|
||||
"thirdPartyItemCount": len(third_ids),
|
||||
"status": r.status or "1",
|
||||
"orderNum": r.order_num if r.order_num is not None else 0,
|
||||
"remark": r.remark or "",
|
||||
"createTime": r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else "",
|
||||
"updateTime": r.update_time.strftime("%Y-%m-%d %H:%M:%S") if r.update_time else "",
|
||||
}
|
||||
|
||||
|
||||
def template_name_exists(
|
||||
db: Session, name: str, tenant_no: str, exclude_id: Optional[str] = None
|
||||
) -> bool:
|
||||
n = (name or "").strip()
|
||||
if not n:
|
||||
return False
|
||||
q = db.query(BizQcInspectionItemTemplate).filter(
|
||||
BizQcInspectionItemTemplate.template_name == n,
|
||||
BizQcInspectionItemTemplate.tenant_no == tenant_no,
|
||||
BizQcInspectionItemTemplate.del_flag == 0,
|
||||
)
|
||||
if exclude_id:
|
||||
q = q.filter(BizQcInspectionItemTemplate.id != exclude_id)
|
||||
return q.first() is not None
|
||||
|
||||
|
||||
def _assert_item_ids_exist(
|
||||
db: Session, tenant_no: str, lab_ids: List[str], third_ids: List[str]
|
||||
) -> None:
|
||||
if lab_ids:
|
||||
found = {
|
||||
r.id
|
||||
for r in db.query(BizQcLabTestItem.id)
|
||||
.filter(
|
||||
BizQcLabTestItem.tenant_no == tenant_no,
|
||||
BizQcLabTestItem.del_flag == 0,
|
||||
BizQcLabTestItem.id.in_(lab_ids),
|
||||
)
|
||||
.all()
|
||||
}
|
||||
if found != set(lab_ids):
|
||||
raise ValueError("存在无效的检测项目")
|
||||
if third_ids:
|
||||
found = {
|
||||
r.id
|
||||
for r in db.query(BizQcThirdPartyItem.id)
|
||||
.filter(
|
||||
BizQcThirdPartyItem.tenant_no == tenant_no,
|
||||
BizQcThirdPartyItem.del_flag == 0,
|
||||
BizQcThirdPartyItem.id.in_(third_ids),
|
||||
)
|
||||
.all()
|
||||
}
|
||||
if found != set(third_ids):
|
||||
raise ValueError("存在无效的检测项目")
|
||||
|
||||
|
||||
def list_templates(
|
||||
db: Session,
|
||||
*,
|
||||
page: int = 1,
|
||||
page_size: int = 10,
|
||||
keyword: Optional[str] = None,
|
||||
status: Optional[str] = None,
|
||||
tenant_no: Optional[str] = None,
|
||||
user_info=None,
|
||||
) -> Tuple[List[Dict[str, Any]], int]:
|
||||
def _filtered():
|
||||
q0 = db.query(BizQcInspectionItemTemplate).filter(
|
||||
BizQcInspectionItemTemplate.del_flag == 0
|
||||
)
|
||||
if tenant_no:
|
||||
q0 = q0.filter(BizQcInspectionItemTemplate.tenant_no == tenant_no)
|
||||
kw = (keyword or "").strip()
|
||||
if kw:
|
||||
q0 = q0.filter(BizQcInspectionItemTemplate.template_name.like(f"%{kw}%"))
|
||||
st = (status or "").strip()
|
||||
if st in ("0", "1"):
|
||||
q0 = q0.filter(BizQcInspectionItemTemplate.status == st)
|
||||
if user_info:
|
||||
from app.core.data_scope import apply_data_scope
|
||||
|
||||
q0 = apply_data_scope(q0, BizQcInspectionItemTemplate, user_info, db, tenant_no)
|
||||
return q0
|
||||
|
||||
total = (
|
||||
_filtered()
|
||||
.order_by(None)
|
||||
.with_entities(func.count(BizQcInspectionItemTemplate.id))
|
||||
.scalar()
|
||||
or 0
|
||||
)
|
||||
page = max(1, page)
|
||||
page_size = min(max(1, page_size), 100)
|
||||
rows = (
|
||||
_filtered()
|
||||
.order_by(BizQcInspectionItemTemplate.create_time.desc())
|
||||
.offset((page - 1) * page_size)
|
||||
.limit(page_size)
|
||||
.all()
|
||||
)
|
||||
return [_row_to_dict(x) for x in rows], int(total)
|
||||
|
||||
|
||||
def list_active_templates(
|
||||
db: Session, tenant_no: str, *, user_info=None
|
||||
) -> List[Dict[str, Any]]:
|
||||
q = db.query(BizQcInspectionItemTemplate).filter(
|
||||
BizQcInspectionItemTemplate.tenant_no == tenant_no,
|
||||
BizQcInspectionItemTemplate.del_flag == 0,
|
||||
BizQcInspectionItemTemplate.status == "1",
|
||||
)
|
||||
if user_info:
|
||||
from app.core.data_scope import apply_data_scope
|
||||
|
||||
q = apply_data_scope(q, BizQcInspectionItemTemplate, user_info, db, tenant_no)
|
||||
rows = q.order_by(BizQcInspectionItemTemplate.create_time.desc()).all()
|
||||
return [_row_to_dict(x) for x in rows]
|
||||
|
||||
|
||||
def get_template(
|
||||
db: Session, row_id: str, tenant_no: Optional[str] = None
|
||||
) -> Optional[BizQcInspectionItemTemplate]:
|
||||
q = db.query(BizQcInspectionItemTemplate).filter(
|
||||
BizQcInspectionItemTemplate.id == row_id,
|
||||
BizQcInspectionItemTemplate.del_flag == 0,
|
||||
)
|
||||
if tenant_no:
|
||||
q = q.filter(BizQcInspectionItemTemplate.tenant_no == tenant_no)
|
||||
return q.first()
|
||||
|
||||
|
||||
def create_template(
|
||||
db: Session,
|
||||
*,
|
||||
template_name: str,
|
||||
lab_item_ids: Optional[List[str]] = None,
|
||||
third_party_item_ids: Optional[List[str]] = None,
|
||||
status: str = "1",
|
||||
tenant_no: str,
|
||||
order_num: Optional[int] = None,
|
||||
remark: Optional[str] = None,
|
||||
user_id: Optional[str] = None,
|
||||
create_dept_id: str = "",
|
||||
) -> BizQcInspectionItemTemplate:
|
||||
name = (template_name or "").strip()
|
||||
if not name:
|
||||
raise ValueError("模版名称不能为空")
|
||||
if template_name_exists(db, name, tenant_no):
|
||||
raise ValueError("模版名称已存在")
|
||||
lab_ids = parse_id_csv(",".join(lab_item_ids or []))
|
||||
third_ids = parse_id_csv(",".join(third_party_item_ids or []))
|
||||
require_nonempty_items(lab_ids, third_ids)
|
||||
_assert_item_ids_exist(db, tenant_no, lab_ids, third_ids)
|
||||
st = (status or "1").strip()
|
||||
if st not in ("0", "1"):
|
||||
raise ValueError("状态无效")
|
||||
obj = BizQcInspectionItemTemplate(
|
||||
id=generate_uuid(),
|
||||
template_name=name,
|
||||
lab_item_ids=join_id_list(lab_ids),
|
||||
third_party_item_ids=join_id_list(third_ids),
|
||||
status=st,
|
||||
tenant_no=tenant_no,
|
||||
order_num=order_num if order_num is not None else 0,
|
||||
remark=(remark or "").strip() or None,
|
||||
create_by=user_id,
|
||||
create_dept_id=create_dept_id or "",
|
||||
create_time=datetime.now(),
|
||||
update_by=user_id,
|
||||
update_time=datetime.now(),
|
||||
del_flag=0,
|
||||
)
|
||||
db.add(obj)
|
||||
db.commit()
|
||||
db.refresh(obj)
|
||||
return obj
|
||||
|
||||
|
||||
def update_template(
|
||||
db: Session,
|
||||
row_id: str,
|
||||
*,
|
||||
template_name: Optional[str] = None,
|
||||
lab_item_ids: Optional[List[str]] = None,
|
||||
third_party_item_ids: Optional[List[str]] = None,
|
||||
status: Optional[str] = None,
|
||||
order_num: Optional[int] = None,
|
||||
remark: Optional[str] = None,
|
||||
user_id: Optional[str] = None,
|
||||
tenant_no: Optional[str] = None,
|
||||
) -> BizQcInspectionItemTemplate:
|
||||
obj = get_template(db, row_id, tenant_no=tenant_no)
|
||||
if not obj:
|
||||
raise ValueError("模版不存在")
|
||||
if template_name is not None:
|
||||
name = (template_name or "").strip()
|
||||
if not name:
|
||||
raise ValueError("模版名称不能为空")
|
||||
if name != obj.template_name and template_name_exists(
|
||||
db, name, obj.tenant_no, exclude_id=row_id
|
||||
):
|
||||
raise ValueError("模版名称已存在")
|
||||
obj.template_name = name
|
||||
lab_ids = (
|
||||
parse_id_csv(",".join(lab_item_ids))
|
||||
if lab_item_ids is not None
|
||||
else parse_id_csv(obj.lab_item_ids)
|
||||
)
|
||||
third_ids = (
|
||||
parse_id_csv(",".join(third_party_item_ids))
|
||||
if third_party_item_ids is not None
|
||||
else parse_id_csv(obj.third_party_item_ids)
|
||||
)
|
||||
if lab_item_ids is not None or third_party_item_ids is not None:
|
||||
require_nonempty_items(lab_ids, third_ids)
|
||||
_assert_item_ids_exist(db, obj.tenant_no, lab_ids, third_ids)
|
||||
obj.lab_item_ids = join_id_list(lab_ids)
|
||||
obj.third_party_item_ids = join_id_list(third_ids)
|
||||
if status is not None:
|
||||
st = status.strip()
|
||||
if st not in ("0", "1"):
|
||||
raise ValueError("状态无效")
|
||||
obj.status = st
|
||||
if order_num is not None:
|
||||
obj.order_num = order_num
|
||||
if remark is not None:
|
||||
obj.remark = (remark or "").strip() or None
|
||||
obj.update_by = user_id
|
||||
obj.update_time = datetime.now()
|
||||
db.add(obj)
|
||||
db.commit()
|
||||
db.refresh(obj)
|
||||
return obj
|
||||
|
||||
|
||||
def soft_delete_template(
|
||||
db: Session, row_id: str, tenant_no: Optional[str] = None
|
||||
) -> None:
|
||||
obj = get_template(db, row_id, tenant_no=tenant_no)
|
||||
if not obj:
|
||||
raise ValueError("模版不存在")
|
||||
obj.del_flag = 1
|
||||
obj.update_time = datetime.now()
|
||||
db.add(obj)
|
||||
db.commit()
|
||||
@@ -0,0 +1,27 @@
|
||||
import unittest
|
||||
|
||||
from app.crud import qc_inspection_item_template as svc
|
||||
|
||||
|
||||
class QcInspectionItemTemplateHelpersTests(unittest.TestCase):
|
||||
def test_parse_ids_splits_and_dedupes(self):
|
||||
self.assertEqual(svc.parse_id_csv("a,b, a,,c"), ["a", "b", "c"])
|
||||
|
||||
def test_join_ids(self):
|
||||
self.assertEqual(svc.join_id_list(["a", "b"]), "a,b")
|
||||
self.assertEqual(svc.join_id_list([]), "")
|
||||
|
||||
def test_filter_active_ids(self):
|
||||
active = {"1", "2"}
|
||||
kept, skipped = svc.filter_active_ids(["1", "9", "2", "9"], active)
|
||||
self.assertEqual(kept, ["1", "2"])
|
||||
self.assertEqual(skipped, 2)
|
||||
|
||||
def test_require_at_least_one_item(self):
|
||||
with self.assertRaises(ValueError):
|
||||
svc.require_nonempty_items([], [])
|
||||
svc.require_nonempty_items(["1"], [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user