270 lines
9.1 KiB
Python
270 lines
9.1 KiB
Python
"""第三方检测公司 CRUD"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
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.models.models import BizQcThirdPartyCompany, BizQcThirdPartyItem
|
|
|
|
|
|
def _row_to_dict(r: BizQcThirdPartyCompany) -> Dict[str, Any]:
|
|
return {
|
|
"id": r.id,
|
|
"companyCode": r.company_code,
|
|
"companyName": r.company_name,
|
|
"contactName": r.contact_name or "",
|
|
"contactPhone": r.contact_phone or "",
|
|
"companyLogo": r.company_logo or "",
|
|
"address": r.address or "",
|
|
"invoiceTitle": r.invoice_title or "",
|
|
"taxId": r.tax_id or "",
|
|
"bankName": r.bank_name or "",
|
|
"bankAccount": r.bank_account or "",
|
|
"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 company_code_exists(
|
|
db: Session, code: str, tenant_no: str, exclude_id: Optional[str] = None
|
|
) -> bool:
|
|
c = (code or "").strip()
|
|
if not c:
|
|
return False
|
|
q = db.query(BizQcThirdPartyCompany).filter(
|
|
BizQcThirdPartyCompany.company_code == c,
|
|
BizQcThirdPartyCompany.tenant_no == tenant_no,
|
|
BizQcThirdPartyCompany.del_flag == 0,
|
|
)
|
|
if exclude_id:
|
|
q = q.filter(BizQcThirdPartyCompany.id != exclude_id)
|
|
return q.first() is not None
|
|
|
|
|
|
def list_companies(
|
|
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(BizQcThirdPartyCompany).filter(BizQcThirdPartyCompany.del_flag == 0)
|
|
if tenant_no:
|
|
q0 = q0.filter(BizQcThirdPartyCompany.tenant_no == tenant_no)
|
|
kw = (keyword or "").strip()
|
|
if kw:
|
|
like = f"%{kw}%"
|
|
q0 = q0.filter(
|
|
or_(
|
|
BizQcThirdPartyCompany.company_code.like(like),
|
|
BizQcThirdPartyCompany.company_name.like(like),
|
|
)
|
|
)
|
|
st = (status or "").strip()
|
|
if st in ("0", "1"):
|
|
q0 = q0.filter(BizQcThirdPartyCompany.status == st)
|
|
if user_info:
|
|
from app.core.data_scope import apply_data_scope
|
|
q0 = apply_data_scope(q0, BizQcThirdPartyCompany, user_info, db, tenant_no)
|
|
return q0
|
|
|
|
total = _filtered().order_by(None).with_entities(func.count(BizQcThirdPartyCompany.id)).scalar() or 0
|
|
page = max(1, page)
|
|
page_size = min(max(1, page_size), 100)
|
|
rows = (
|
|
_filtered()
|
|
.order_by(BizQcThirdPartyCompany.order_num.asc(), BizQcThirdPartyCompany.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_company_options(db: Session, tenant_no: str) -> List[Dict[str, Any]]:
|
|
rows = (
|
|
db.query(BizQcThirdPartyCompany)
|
|
.filter(
|
|
BizQcThirdPartyCompany.tenant_no == tenant_no,
|
|
BizQcThirdPartyCompany.del_flag == 0,
|
|
BizQcThirdPartyCompany.status == "1",
|
|
)
|
|
.order_by(BizQcThirdPartyCompany.order_num.asc(), BizQcThirdPartyCompany.create_time.desc())
|
|
.all()
|
|
)
|
|
return [
|
|
{"id": r.id, "companyCode": r.company_code, "companyName": r.company_name}
|
|
for r in rows
|
|
]
|
|
|
|
|
|
def get_company(
|
|
db: Session, row_id: str, tenant_no: Optional[str] = None
|
|
) -> Optional[BizQcThirdPartyCompany]:
|
|
q = db.query(BizQcThirdPartyCompany).filter(
|
|
BizQcThirdPartyCompany.id == row_id, BizQcThirdPartyCompany.del_flag == 0
|
|
)
|
|
if tenant_no:
|
|
q = q.filter(BizQcThirdPartyCompany.tenant_no == tenant_no)
|
|
return q.first()
|
|
|
|
|
|
def create_company(
|
|
db: Session,
|
|
*,
|
|
company_code: str,
|
|
company_name: str,
|
|
contact_name: Optional[str] = None,
|
|
contact_phone: Optional[str] = None,
|
|
company_logo: Optional[str] = None,
|
|
address: Optional[str] = None,
|
|
invoice_title: Optional[str] = None,
|
|
tax_id: Optional[str] = None,
|
|
bank_name: Optional[str] = None,
|
|
bank_account: Optional[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 = "",
|
|
) -> BizQcThirdPartyCompany:
|
|
code = (company_code or "").strip()
|
|
if not code:
|
|
raise ValueError("公司编码不能为空")
|
|
if company_code_exists(db, code, tenant_no):
|
|
raise ValueError("公司编码已存在")
|
|
name = (company_name or "").strip()
|
|
if not name:
|
|
raise ValueError("公司名称不能为空")
|
|
st = (status or "1").strip()
|
|
if st not in ("0", "1"):
|
|
raise ValueError("状态无效")
|
|
obj = BizQcThirdPartyCompany(
|
|
id=generate_uuid(),
|
|
company_code=code,
|
|
company_name=name,
|
|
contact_name=(contact_name or "").strip() or None,
|
|
contact_phone=(contact_phone or "").strip() or None,
|
|
company_logo=(company_logo or "").strip() or None,
|
|
address=(address or "").strip() or None,
|
|
invoice_title=(invoice_title or "").strip() or None,
|
|
tax_id=(tax_id or "").strip() or None,
|
|
bank_name=(bank_name or "").strip() or None,
|
|
bank_account=(bank_account or "").strip() or None,
|
|
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_company(
|
|
db: Session,
|
|
row_id: str,
|
|
*,
|
|
company_code: Optional[str] = None,
|
|
company_name: Optional[str] = None,
|
|
contact_name: Optional[str] = None,
|
|
contact_phone: Optional[str] = None,
|
|
company_logo: Optional[str] = None,
|
|
address: Optional[str] = None,
|
|
invoice_title: Optional[str] = None,
|
|
tax_id: Optional[str] = None,
|
|
bank_name: Optional[str] = None,
|
|
bank_account: Optional[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,
|
|
) -> BizQcThirdPartyCompany:
|
|
obj = get_company(db, row_id, tenant_no=tenant_no)
|
|
if not obj:
|
|
raise ValueError("公司不存在")
|
|
if company_code is not None:
|
|
code = company_code.strip()
|
|
if not code:
|
|
raise ValueError("公司编码不能为空")
|
|
if code != obj.company_code and company_code_exists(db, code, obj.tenant_no, exclude_id=row_id):
|
|
raise ValueError("公司编码已存在")
|
|
obj.company_code = code
|
|
if company_name is not None:
|
|
name = (company_name or "").strip()
|
|
if not name:
|
|
raise ValueError("公司名称不能为空")
|
|
obj.company_name = name
|
|
if contact_name is not None:
|
|
obj.contact_name = (contact_name or "").strip() or None
|
|
if contact_phone is not None:
|
|
obj.contact_phone = (contact_phone or "").strip() or None
|
|
if company_logo is not None:
|
|
obj.company_logo = (company_logo or "").strip() or None
|
|
if address is not None:
|
|
obj.address = (address or "").strip() or None
|
|
if invoice_title is not None:
|
|
obj.invoice_title = (invoice_title or "").strip() or None
|
|
if tax_id is not None:
|
|
obj.tax_id = (tax_id or "").strip() or None
|
|
if bank_name is not None:
|
|
obj.bank_name = (bank_name or "").strip() or None
|
|
if bank_account is not None:
|
|
obj.bank_account = (bank_account or "").strip() or None
|
|
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_company(db: Session, row_id: str, tenant_no: Optional[str] = None) -> None:
|
|
obj = get_company(db, row_id, tenant_no=tenant_no)
|
|
if not obj:
|
|
raise ValueError("公司不存在")
|
|
child = (
|
|
db.query(BizQcThirdPartyItem)
|
|
.filter(
|
|
BizQcThirdPartyItem.company_id == row_id,
|
|
BizQcThirdPartyItem.tenant_no == obj.tenant_no,
|
|
BizQcThirdPartyItem.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
if child:
|
|
raise ValueError("该公司下仍有检测项目，无法删除")
|
|
obj.del_flag = 1
|
|
obj.update_time = datetime.now()
|
|
db.add(obj)
|
|
db.commit()
|