634 lines
20 KiB
Python
634 lines
20 KiB
Python
"""采集地区及下属医院 CRUD"""
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
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 (
|
|
BizCollectionHospital,
|
|
BizCollectionRegion,
|
|
BizCollectionRegionStaff,
|
|
SysDept,
|
|
SysUser,
|
|
)
|
|
|
|
|
|
def staff_display_name(nick_name: Optional[str], user_no: Optional[str]) -> str:
|
|
n = (nick_name or "").strip()
|
|
if n:
|
|
return n
|
|
return (user_no or "").strip()
|
|
|
|
|
|
def build_staff_leader_name_lists(rows: List[Dict[str, Any]]) -> Tuple[List[str], List[str]]:
|
|
"""rows: dicts with nickName/isLeader (already sorted leaders-first preferred)."""
|
|
leaders = [staff_display_name(r.get("nickName"), r.get("userNo")) for r in rows if r.get("isLeader")]
|
|
# Keep leader-first order if caller sorted; otherwise partition
|
|
non_leaders = [staff_display_name(r.get("nickName"), r.get("userNo")) for r in rows if not r.get("isLeader")]
|
|
staff_names = leaders + non_leaders
|
|
return staff_names, leaders
|
|
|
|
|
|
def _region_to_dict(
|
|
r: BizCollectionRegion,
|
|
hospital_names: Optional[List[str]] = None,
|
|
staff_names: Optional[List[str]] = None,
|
|
leader_names: Optional[List[str]] = None,
|
|
) -> Dict[str, Any]:
|
|
return {
|
|
"id": r.id,
|
|
"regionName": r.region_name,
|
|
"hospitalNames": hospital_names or [],
|
|
"staffNames": staff_names or [],
|
|
"leaderNames": leader_names or [],
|
|
"tenantNo": r.tenant_no,
|
|
"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 _hospital_names_by_region_ids(db: Session, region_ids: List[str]) -> Dict[str, List[str]]:
|
|
"""批量查询地区下属医院名称，按 order_num、create_time 排序。"""
|
|
hospital_map: Dict[str, List[str]] = {}
|
|
if not region_ids:
|
|
return hospital_map
|
|
hosp_rows = (
|
|
db.query(BizCollectionHospital.region_id, BizCollectionHospital.hospital_name)
|
|
.filter(
|
|
and_(
|
|
BizCollectionHospital.region_id.in_(region_ids),
|
|
BizCollectionHospital.del_flag == 0,
|
|
)
|
|
)
|
|
.order_by(BizCollectionHospital.order_num.asc(), BizCollectionHospital.create_time.desc())
|
|
.all()
|
|
)
|
|
for h in hosp_rows:
|
|
hospital_map.setdefault(h.region_id, []).append(h.hospital_name)
|
|
return hospital_map
|
|
|
|
|
|
def _staff_names_by_region_ids(
|
|
db: Session, region_ids: List[str]
|
|
) -> Dict[str, Tuple[List[str], List[str]]]:
|
|
"""Return region_id -> (staffNames, leaderNames). Leaders first in staffNames."""
|
|
out: Dict[str, Tuple[List[str], List[str]]] = {rid: ([], []) for rid in region_ids}
|
|
if not region_ids:
|
|
return out
|
|
rows = (
|
|
db.query(BizCollectionRegionStaff, SysUser)
|
|
.outerjoin(SysUser, SysUser.id == BizCollectionRegionStaff.user_id)
|
|
.filter(
|
|
and_(
|
|
BizCollectionRegionStaff.region_id.in_(region_ids),
|
|
BizCollectionRegionStaff.del_flag == 0,
|
|
)
|
|
)
|
|
.order_by(
|
|
BizCollectionRegionStaff.is_leader.desc(),
|
|
BizCollectionRegionStaff.order_num.asc(),
|
|
BizCollectionRegionStaff.create_time.desc(),
|
|
)
|
|
.all()
|
|
)
|
|
buckets: Dict[str, List[Dict[str, Any]]] = {rid: [] for rid in region_ids}
|
|
for staff, user in rows:
|
|
buckets.setdefault(staff.region_id, []).append(
|
|
{
|
|
"nickName": (user.nick_name if user else "") or "",
|
|
"userNo": (user.user_no if user else "") or "",
|
|
"isLeader": bool(staff.is_leader),
|
|
}
|
|
)
|
|
for rid, items in buckets.items():
|
|
out[rid] = build_staff_leader_name_lists(items)
|
|
return out
|
|
|
|
|
|
def _hospital_to_dict(h: BizCollectionHospital) -> Dict[str, Any]:
|
|
return {
|
|
"id": h.id,
|
|
"regionId": h.region_id,
|
|
"hospitalName": h.hospital_name,
|
|
"tenantNo": h.tenant_no,
|
|
"orderNum": h.order_num if h.order_num is not None else 0,
|
|
"remark": h.remark or "",
|
|
"createTime": h.create_time.strftime("%Y-%m-%d %H:%M:%S") if h.create_time else "",
|
|
"updateTime": h.update_time.strftime("%Y-%m-%d %H:%M:%S") if h.update_time else "",
|
|
}
|
|
|
|
|
|
def _staff_to_dict(
|
|
s: BizCollectionRegionStaff,
|
|
*,
|
|
nick_name: str = "",
|
|
user_no: str = "",
|
|
dept_name: str = "",
|
|
) -> Dict[str, Any]:
|
|
return {
|
|
"id": s.id,
|
|
"regionId": s.region_id,
|
|
"userId": s.user_id,
|
|
"userName": staff_display_name(nick_name, user_no),
|
|
"nickName": nick_name or "",
|
|
"userNo": user_no or "",
|
|
"deptName": dept_name or "",
|
|
"isLeader": bool(s.is_leader),
|
|
"tenantNo": s.tenant_no,
|
|
"orderNum": s.order_num if s.order_num is not None else 0,
|
|
"remark": s.remark or "",
|
|
"createTime": s.create_time.strftime("%Y-%m-%d %H:%M:%S") if s.create_time else "",
|
|
"updateTime": s.update_time.strftime("%Y-%m-%d %H:%M:%S") if s.update_time else "",
|
|
}
|
|
|
|
|
|
def list_collection_regions(
|
|
db: Session,
|
|
*,
|
|
page: int = 1,
|
|
page_size: int = 10,
|
|
keyword: Optional[str] = None,
|
|
tenant_no: Optional[str] = None,
|
|
) -> Tuple[List[Dict[str, Any]], int]:
|
|
def _filtered():
|
|
q0 = db.query(BizCollectionRegion).filter(BizCollectionRegion.del_flag == 0)
|
|
if tenant_no:
|
|
q0 = q0.filter(BizCollectionRegion.tenant_no == tenant_no)
|
|
kw = (keyword or "").strip()
|
|
if kw:
|
|
like = f"%{kw}%"
|
|
q0 = q0.filter(or_(BizCollectionRegion.region_name.like(like), BizCollectionRegion.remark.like(like)))
|
|
return q0
|
|
|
|
total = _filtered().order_by(None).with_entities(func.count(BizCollectionRegion.id)).scalar() or 0
|
|
page = max(1, page)
|
|
page_size = min(max(1, page_size), 100)
|
|
rows = (
|
|
_filtered()
|
|
.order_by(BizCollectionRegion.create_time.desc())
|
|
.offset((page - 1) * page_size)
|
|
.limit(page_size)
|
|
.all()
|
|
)
|
|
|
|
region_ids = [r.id for r in rows]
|
|
hospital_map = _hospital_names_by_region_ids(db, region_ids)
|
|
staff_map = _staff_names_by_region_ids(db, region_ids)
|
|
return [
|
|
_region_to_dict(
|
|
x,
|
|
hospital_names=hospital_map.get(x.id, []),
|
|
staff_names=staff_map.get(x.id, ([], []))[0],
|
|
leader_names=staff_map.get(x.id, ([], []))[1],
|
|
)
|
|
for x in rows
|
|
], int(total)
|
|
|
|
|
|
def get_collection_region(db: Session, region_id: str, tenant_no: Optional[str] = None) -> Optional[BizCollectionRegion]:
|
|
q = db.query(BizCollectionRegion).filter(and_(BizCollectionRegion.id == region_id, BizCollectionRegion.del_flag == 0))
|
|
if tenant_no:
|
|
q = q.filter(BizCollectionRegion.tenant_no == tenant_no)
|
|
return q.first()
|
|
|
|
|
|
def create_collection_region(
|
|
db: Session,
|
|
*,
|
|
region_name: str,
|
|
tenant_no: str = "zkxw",
|
|
order_num: Optional[int] = None,
|
|
remark: Optional[str] = None,
|
|
user_id: Optional[str] = None,
|
|
) -> BizCollectionRegion:
|
|
name = (region_name or "").strip()
|
|
if not name:
|
|
raise ValueError("地区名称不能为空")
|
|
obj = BizCollectionRegion(
|
|
id=generate_uuid(),
|
|
region_name=name,
|
|
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_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_collection_region(
|
|
db: Session,
|
|
region_id: str,
|
|
*,
|
|
region_name: Optional[str] = None,
|
|
order_num: Optional[int] = None,
|
|
remark: Optional[str] = None,
|
|
user_id: Optional[str] = None,
|
|
tenant_no: Optional[str] = None,
|
|
) -> BizCollectionRegion:
|
|
obj = get_collection_region(db, region_id, tenant_no=tenant_no)
|
|
if not obj:
|
|
raise ValueError("采集地区不存在")
|
|
if region_name is not None:
|
|
n = region_name.strip()
|
|
if not n:
|
|
raise ValueError("地区名称不能为空")
|
|
obj.region_name = n
|
|
if order_num is not None:
|
|
obj.order_num = order_num
|
|
if remark is not None:
|
|
obj.remark = remark.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_collection_region(db: Session, region_id: str, tenant_no: Optional[str] = None) -> None:
|
|
obj = get_collection_region(db, region_id, tenant_no=tenant_no)
|
|
if not obj:
|
|
raise ValueError("采集地区不存在")
|
|
now = datetime.now()
|
|
obj.del_flag = 1
|
|
obj.update_time = now
|
|
db.add(obj)
|
|
subs = (
|
|
db.query(BizCollectionHospital)
|
|
.filter(
|
|
and_(
|
|
BizCollectionHospital.region_id == region_id,
|
|
BizCollectionHospital.del_flag == 0,
|
|
)
|
|
)
|
|
.all()
|
|
)
|
|
for h in subs:
|
|
h.del_flag = 1
|
|
h.update_time = now
|
|
db.add(h)
|
|
staff_subs = (
|
|
db.query(BizCollectionRegionStaff)
|
|
.filter(
|
|
and_(
|
|
BizCollectionRegionStaff.region_id == region_id,
|
|
BizCollectionRegionStaff.del_flag == 0,
|
|
)
|
|
)
|
|
.all()
|
|
)
|
|
for s in staff_subs:
|
|
s.del_flag = 1
|
|
s.update_time = now
|
|
db.add(s)
|
|
db.commit()
|
|
|
|
|
|
def list_hospitals(
|
|
db: Session,
|
|
*,
|
|
region_id: str,
|
|
page: int = 1,
|
|
page_size: int = 10,
|
|
keyword: Optional[str] = None,
|
|
tenant_no: Optional[str] = None,
|
|
) -> Tuple[List[Dict[str, Any]], int]:
|
|
region = get_collection_region(db, region_id, tenant_no=tenant_no)
|
|
if not region:
|
|
raise ValueError("采集地区不存在")
|
|
|
|
def _filtered():
|
|
q0 = db.query(BizCollectionHospital).filter(
|
|
and_(BizCollectionHospital.region_id == region_id, BizCollectionHospital.del_flag == 0)
|
|
)
|
|
if tenant_no:
|
|
q0 = q0.filter(BizCollectionHospital.tenant_no == tenant_no)
|
|
kw = (keyword or "").strip()
|
|
if kw:
|
|
like = f"%{kw}%"
|
|
q0 = q0.filter(BizCollectionHospital.hospital_name.like(like))
|
|
return q0
|
|
|
|
total = _filtered().order_by(None).with_entities(func.count(BizCollectionHospital.id)).scalar() or 0
|
|
page = max(1, page)
|
|
page_size = min(max(1, page_size), 100)
|
|
rows = (
|
|
_filtered()
|
|
.order_by(BizCollectionHospital.create_time.desc())
|
|
.offset((page - 1) * page_size)
|
|
.limit(page_size)
|
|
.all()
|
|
)
|
|
return [_hospital_to_dict(x) for x in rows], int(total)
|
|
|
|
|
|
def search_hospitals_by_keyword(
|
|
db: Session,
|
|
*,
|
|
keyword: str,
|
|
tenant_no: Optional[str] = None,
|
|
limit: int = 20,
|
|
) -> List[Dict[str, Any]]:
|
|
"""按医院名称模糊检索，返回医院及所属采集地区信息。"""
|
|
kw = (keyword or "").strip()
|
|
if not kw:
|
|
return []
|
|
like = f"%{kw}%"
|
|
q = (
|
|
db.query(BizCollectionHospital, BizCollectionRegion)
|
|
.join(BizCollectionRegion, BizCollectionHospital.region_id == BizCollectionRegion.id)
|
|
.filter(
|
|
BizCollectionHospital.del_flag == 0,
|
|
BizCollectionRegion.del_flag == 0,
|
|
BizCollectionHospital.hospital_name.like(like),
|
|
)
|
|
)
|
|
if tenant_no:
|
|
q = q.filter(BizCollectionHospital.tenant_no == tenant_no)
|
|
rows = (
|
|
q.order_by(BizCollectionHospital.order_num.asc(), BizCollectionHospital.create_time.desc())
|
|
.limit(min(max(1, limit), 50))
|
|
.all()
|
|
)
|
|
items: List[Dict[str, Any]] = []
|
|
for hospital, region in rows:
|
|
item = _hospital_to_dict(hospital)
|
|
item["regionName"] = region.region_name or ""
|
|
items.append(item)
|
|
return items
|
|
|
|
|
|
def get_hospital(db: Session, hospital_id: str, tenant_no: Optional[str] = None) -> Optional[BizCollectionHospital]:
|
|
q = db.query(BizCollectionHospital).filter(and_(BizCollectionHospital.id == hospital_id, BizCollectionHospital.del_flag == 0))
|
|
if tenant_no:
|
|
q = q.filter(BizCollectionHospital.tenant_no == tenant_no)
|
|
return q.first()
|
|
|
|
|
|
def create_hospital(
|
|
db: Session,
|
|
*,
|
|
region_id: str,
|
|
hospital_name: str,
|
|
tenant_no: str = "zkxw",
|
|
order_num: Optional[int] = None,
|
|
remark: Optional[str] = None,
|
|
user_id: Optional[str] = None,
|
|
) -> BizCollectionHospital:
|
|
region = get_collection_region(db, region_id, tenant_no=tenant_no)
|
|
if not region:
|
|
raise ValueError("采集地区不存在")
|
|
hn = (hospital_name or "").strip()
|
|
if not hn:
|
|
raise ValueError("医院名称不能为空")
|
|
obj = BizCollectionHospital(
|
|
id=generate_uuid(),
|
|
region_id=region_id,
|
|
hospital_name=hn,
|
|
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_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_hospital(
|
|
db: Session,
|
|
hospital_id: str,
|
|
*,
|
|
hospital_name: Optional[str] = None,
|
|
order_num: Optional[int] = None,
|
|
remark: Optional[str] = None,
|
|
user_id: Optional[str] = None,
|
|
tenant_no: Optional[str] = None,
|
|
) -> BizCollectionHospital:
|
|
obj = get_hospital(db, hospital_id, tenant_no=tenant_no)
|
|
if not obj:
|
|
raise ValueError("医院不存在")
|
|
if hospital_name is not None:
|
|
hn = hospital_name.strip()
|
|
if not hn:
|
|
raise ValueError("医院名称不能为空")
|
|
obj.hospital_name = hn
|
|
if order_num is not None:
|
|
obj.order_num = order_num
|
|
if remark is not None:
|
|
obj.remark = remark.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_hospital(db: Session, hospital_id: str, tenant_no: Optional[str] = None) -> None:
|
|
obj = get_hospital(db, hospital_id, tenant_no=tenant_no)
|
|
if not obj:
|
|
raise ValueError("医院不存在")
|
|
obj.del_flag = 1
|
|
obj.update_time = datetime.now()
|
|
db.add(obj)
|
|
db.commit()
|
|
|
|
|
|
def list_staff(
|
|
db: Session,
|
|
*,
|
|
region_id: str,
|
|
page: int = 1,
|
|
page_size: int = 10,
|
|
tenant_no: Optional[str] = None,
|
|
) -> Tuple[List[Dict[str, Any]], int]:
|
|
region = get_collection_region(db, region_id, tenant_no=tenant_no)
|
|
if not region:
|
|
raise ValueError("采集地区不存在")
|
|
|
|
def _filtered():
|
|
q0 = db.query(BizCollectionRegionStaff).filter(
|
|
and_(BizCollectionRegionStaff.region_id == region_id, BizCollectionRegionStaff.del_flag == 0)
|
|
)
|
|
if tenant_no:
|
|
q0 = q0.filter(BizCollectionRegionStaff.tenant_no == tenant_no)
|
|
return q0
|
|
|
|
total = _filtered().order_by(None).with_entities(func.count(BizCollectionRegionStaff.id)).scalar() or 0
|
|
page = max(1, page)
|
|
page_size = min(max(1, page_size), 100)
|
|
rows = (
|
|
_filtered()
|
|
.order_by(
|
|
BizCollectionRegionStaff.is_leader.desc(),
|
|
BizCollectionRegionStaff.order_num.asc(),
|
|
BizCollectionRegionStaff.create_time.desc(),
|
|
)
|
|
.offset((page - 1) * page_size)
|
|
.limit(page_size)
|
|
.all()
|
|
)
|
|
user_ids = [r.user_id for r in rows]
|
|
user_map: Dict[str, SysUser] = {}
|
|
dept_map: Dict[str, str] = {}
|
|
if user_ids:
|
|
for u in db.query(SysUser).filter(SysUser.id.in_(user_ids)).all():
|
|
user_map[u.id] = u
|
|
dept_ids = [u.dept_id for u in user_map.values() if u.dept_id]
|
|
if dept_ids:
|
|
for d in db.query(SysDept).filter(SysDept.id.in_(dept_ids), SysDept.del_flag == 0).all():
|
|
dept_map[d.id] = d.dept_name or ""
|
|
items: List[Dict[str, Any]] = []
|
|
for s in rows:
|
|
u = user_map.get(s.user_id)
|
|
items.append(
|
|
_staff_to_dict(
|
|
s,
|
|
nick_name=(u.nick_name if u else "") or "",
|
|
user_no=(u.user_no if u else "") or "",
|
|
dept_name=dept_map.get((u.dept_id if u else "") or "", ""),
|
|
)
|
|
)
|
|
return items, int(total)
|
|
|
|
|
|
def get_staff(db: Session, staff_id: str, tenant_no: Optional[str] = None) -> Optional[BizCollectionRegionStaff]:
|
|
q = db.query(BizCollectionRegionStaff).filter(
|
|
and_(BizCollectionRegionStaff.id == staff_id, BizCollectionRegionStaff.del_flag == 0)
|
|
)
|
|
if tenant_no:
|
|
q = q.filter(BizCollectionRegionStaff.tenant_no == tenant_no)
|
|
return q.first()
|
|
|
|
|
|
def create_staff(
|
|
db: Session,
|
|
*,
|
|
region_id: str,
|
|
user_id: str,
|
|
is_leader: bool = False,
|
|
tenant_no: str = "zkxw",
|
|
order_num: Optional[int] = None,
|
|
remark: Optional[str] = None,
|
|
user_id_actor: Optional[str] = None,
|
|
) -> BizCollectionRegionStaff:
|
|
region = get_collection_region(db, region_id, tenant_no=tenant_no)
|
|
if not region:
|
|
raise ValueError("采集地区不存在")
|
|
uid = (user_id or "").strip()
|
|
if not uid:
|
|
raise ValueError("请选择系统人员")
|
|
user = (
|
|
db.query(SysUser)
|
|
.filter(
|
|
and_(
|
|
SysUser.id == uid,
|
|
SysUser.del_flag == 0,
|
|
SysUser.tenant_no == tenant_no,
|
|
)
|
|
)
|
|
.first()
|
|
)
|
|
if not user:
|
|
raise ValueError("用户不存在")
|
|
if (user.status or "") != "1":
|
|
raise ValueError("用户已停用，无法绑定")
|
|
exists = (
|
|
db.query(BizCollectionRegionStaff)
|
|
.filter(
|
|
and_(
|
|
BizCollectionRegionStaff.region_id == region_id,
|
|
BizCollectionRegionStaff.user_id == uid,
|
|
BizCollectionRegionStaff.del_flag == 0,
|
|
BizCollectionRegionStaff.tenant_no == tenant_no,
|
|
)
|
|
)
|
|
.first()
|
|
)
|
|
if exists:
|
|
raise ValueError("该人员已绑定到此地区")
|
|
obj = BizCollectionRegionStaff(
|
|
id=generate_uuid(),
|
|
region_id=region_id,
|
|
user_id=uid,
|
|
is_leader=1 if is_leader else 0,
|
|
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_actor,
|
|
create_dept_id="",
|
|
create_time=datetime.now(),
|
|
update_by=user_id_actor,
|
|
update_time=datetime.now(),
|
|
del_flag=0,
|
|
)
|
|
db.add(obj)
|
|
db.commit()
|
|
db.refresh(obj)
|
|
return obj
|
|
|
|
|
|
def update_staff(
|
|
db: Session,
|
|
staff_id: str,
|
|
*,
|
|
is_leader: Optional[bool] = None,
|
|
order_num: Optional[int] = None,
|
|
remark: Optional[str] = None,
|
|
user_id: Optional[str] = None,
|
|
tenant_no: Optional[str] = None,
|
|
) -> BizCollectionRegionStaff:
|
|
obj = get_staff(db, staff_id, tenant_no=tenant_no)
|
|
if not obj:
|
|
raise ValueError("绑定记录不存在")
|
|
if is_leader is not None:
|
|
obj.is_leader = 1 if is_leader else 0
|
|
if order_num is not None:
|
|
obj.order_num = order_num
|
|
if remark is not None:
|
|
obj.remark = remark.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_staff(db: Session, staff_id: str, tenant_no: Optional[str] = None) -> None:
|
|
obj = get_staff(db, staff_id, tenant_no=tenant_no)
|
|
if not obj:
|
|
raise ValueError("绑定记录不存在")
|
|
obj.del_flag = 1
|
|
obj.update_time = datetime.now()
|
|
db.add(obj)
|
|
db.commit()
|
|
|
|
|
|
def total_pages(total: int, page_size: int) -> int:
|
|
if total <= 0:
|
|
return 0
|
|
return int(math.ceil(total / float(page_size)))
|