"""第三方检测项目 CRUD""" from __future__ import annotations from datetime import datetime from decimal import Decimal from typing import Any, Dict, List, Optional, Tuple from sqlalchemy import and_, func, or_ from sqlalchemy.orm import Session from app.core.utils import generate_uuid from app.crud.qc_third_party_company import get_company from app.models.models import BizQcThirdPartyCompany, BizQcThirdPartyItem def _dec(v: Any) -> float: if v is None: return 0.0 return float(v) def _row_to_dict(r: BizQcThirdPartyItem, company_name: str = "") -> Dict[str, Any]: return { "id": r.id, "companyId": r.company_id, "companyName": company_name, "itemCode": r.item_code, "itemName": r.item_name, "feeAmount": _dec(r.fee_amount), "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 item_code_exists( db: Session, company_id: str, code: str, tenant_no: str, exclude_id: Optional[str] = None, ) -> bool: c = (code or "").strip() if not c: return False q = db.query(BizQcThirdPartyItem).filter( BizQcThirdPartyItem.company_id == company_id, BizQcThirdPartyItem.item_code == c, BizQcThirdPartyItem.tenant_no == tenant_no, BizQcThirdPartyItem.del_flag == 0, ) if exclude_id: q = q.filter(BizQcThirdPartyItem.id != exclude_id) return q.first() is not None def _company_name_map(db: Session, company_ids: List[str]) -> Dict[str, str]: if not company_ids: return {} rows = ( db.query(BizQcThirdPartyCompany) .filter(BizQcThirdPartyCompany.id.in_(company_ids), BizQcThirdPartyCompany.del_flag == 0) .all() ) return {r.id: r.company_name for r in rows} def list_items( db: Session, *, page: int = 1, page_size: int = 10, keyword: Optional[str] = None, status: Optional[str] = None, company_id: Optional[str] = None, tenant_no: Optional[str] = None, user_info=None, ) -> Tuple[List[Dict[str, Any]], int]: def _filtered(): q0 = db.query(BizQcThirdPartyItem).filter(BizQcThirdPartyItem.del_flag == 0) if tenant_no: q0 = q0.filter(BizQcThirdPartyItem.tenant_no == tenant_no) if company_id: q0 = q0.filter(BizQcThirdPartyItem.company_id == company_id) kw = (keyword or "").strip() if kw: like = f"%{kw}%" q0 = q0.filter( or_( BizQcThirdPartyItem.item_code.like(like), BizQcThirdPartyItem.item_name.like(like), ) ) st = (status or "").strip() if st in ("0", "1"): q0 = q0.filter(BizQcThirdPartyItem.status == st) if user_info: from app.core.data_scope import apply_data_scope q0 = apply_data_scope(q0, BizQcThirdPartyItem, user_info, db, tenant_no) return q0 total = _filtered().order_by(None).with_entities(func.count(BizQcThirdPartyItem.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( _filtered() .order_by(BizQcThirdPartyItem.order_num.asc(), BizQcThirdPartyItem.create_time.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) cids = list({r.company_id for r in rows}) cmap = _company_name_map(db, cids) return [_row_to_dict(x, cmap.get(x.company_id, "")) for x in rows], int(total) def list_active_items( db: Session, tenant_no: str, company_id: Optional[str] = None, *, user_info=None ) -> List[Dict[str, Any]]: q = db.query(BizQcThirdPartyItem).filter( BizQcThirdPartyItem.tenant_no == tenant_no, BizQcThirdPartyItem.del_flag == 0, BizQcThirdPartyItem.status == "1", ) if company_id: q = q.filter(BizQcThirdPartyItem.company_id == company_id) if user_info: from app.core.data_scope import apply_data_scope q = apply_data_scope(q, BizQcThirdPartyItem, user_info, db, tenant_no) rows = q.order_by( BizQcThirdPartyItem.order_num.asc(), BizQcThirdPartyItem.create_time.desc() ).all() cmap = _company_name_map(db, list({r.company_id for r in rows})) return [_row_to_dict(x, cmap.get(x.company_id, "")) for x in rows] def get_item( db: Session, row_id: str, tenant_no: Optional[str] = None ) -> Optional[BizQcThirdPartyItem]: q = db.query(BizQcThirdPartyItem).filter( BizQcThirdPartyItem.id == row_id, BizQcThirdPartyItem.del_flag == 0 ) if tenant_no: q = q.filter(BizQcThirdPartyItem.tenant_no == tenant_no) return q.first() def create_item( db: Session, *, company_id: str, item_code: str, item_name: str, fee_amount: float, status: str = "1", tenant_no: str, order_num: Optional[int] = None, remark: Optional[str] = None, user_id: Optional[str] = None, create_dept_id: str = "", ) -> BizQcThirdPartyItem: comp = get_company(db, company_id, tenant_no=tenant_no) if not comp: raise ValueError("所属公司不存在") code = (item_code or "").strip() if not code: raise ValueError("项目编码不能为空") if item_code_exists(db, company_id, code, tenant_no): raise ValueError("该公司下项目编码已存在") name = (item_name or "").strip() if not name: raise ValueError("项目名称不能为空") st = (status or "1").strip() if st not in ("0", "1"): raise ValueError("状态无效") fee = Decimal(str(fee_amount if fee_amount is not None else 0)) obj = BizQcThirdPartyItem( id=generate_uuid(), company_id=company_id, item_code=code, item_name=name, fee_amount=fee, 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_item( db: Session, row_id: str, *, company_id: Optional[str] = None, item_code: Optional[str] = None, item_name: Optional[str] = None, fee_amount: Optional[float] = None, status: Optional[str] = None, order_num: Optional[int] = None, remark: Optional[str] = None, user_id: Optional[str] = None, tenant_no: Optional[str] = None, ) -> BizQcThirdPartyItem: obj = get_item(db, row_id, tenant_no=tenant_no) if not obj: raise ValueError("项目不存在") cid = company_id or obj.company_id if company_id is not None: comp = get_company(db, company_id, tenant_no=obj.tenant_no) if not comp: raise ValueError("所属公司不存在") obj.company_id = company_id if item_code is not None: code = item_code.strip() if not code: raise ValueError("项目编码不能为空") if code != obj.item_code and item_code_exists(db, cid, code, obj.tenant_no, exclude_id=row_id): raise ValueError("该公司下项目编码已存在") obj.item_code = code if item_name is not None: name = (item_name or "").strip() if not name: raise ValueError("项目名称不能为空") obj.item_name = name if fee_amount is not None: obj.fee_amount = Decimal(str(fee_amount)) 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_item(db: Session, row_id: str, tenant_no: Optional[str] = None) -> None: obj = get_item(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()