zkxw/backend/app/crud/qc_inspection_item_template.py

"""质检检测项目模版 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()