From bc62ac796a259f587051caaefeab33ea41412783 Mon Sep 17 00:00:00 2001
From: zlwu <704541015@qq.com>
Date: Mon, 17 Aug 2026 12:46:59 +0800
Subject: [PATCH] feat(backend): add QC inspection item template CRUD helpers.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../app/crud/qc_inspection_item_template.py   | 314 ++++++++++++++++++
 ...est_qc_inspection_item_template_helpers.py |  27 ++
 2 files changed, 341 insertions(+)
 create mode 100644 backend/app/crud/qc_inspection_item_template.py
 create mode 100644 backend/tests/test_qc_inspection_item_template_helpers.py

diff --git a/backend/app/crud/qc_inspection_item_template.py b/backend/app/crud/qc_inspection_item_template.py
new file mode 100644
index 00000000..18839d36
--- /dev/null
+++ b/backend/app/crud/qc_inspection_item_template.py
@@ -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()
diff --git a/backend/tests/test_qc_inspection_item_template_helpers.py b/backend/tests/test_qc_inspection_item_template_helpers.py
new file mode 100644
index 00000000..50d30bd7
--- /dev/null
+++ b/backend/tests/test_qc_inspection_item_template_helpers.py
@@ -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()