250 lines
7.8 KiB
Python
250 lines
7.8 KiB
Python
"""客户 CRUD（biz_customer）"""
|
||
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 BizCustomer
|
||
|
||
|
||
def _row_to_dict(c: BizCustomer) -> Dict[str, Any]:
|
||
return {
|
||
"id": c.id,
|
||
"customerName": c.customer_name,
|
||
"sex": (c.sex or "2").strip() if c.sex else "2",
|
||
"idCardNo": c.id_card_no or "",
|
||
"age": c.age,
|
||
"birthYm": c.birth_ym or "",
|
||
"ethnic": c.ethnic or "",
|
||
"email": c.email or "",
|
||
"address": c.address or "",
|
||
"phoneNo": c.phone_no or "",
|
||
"customerTags": c.customer_tags or "",
|
||
"tenantNo": c.tenant_no,
|
||
"orderNum": c.order_num if c.order_num is not None else 0,
|
||
"remark": c.remark or "",
|
||
"createTime": c.create_time.strftime("%Y-%m-%d %H:%M:%S") if c.create_time else "",
|
||
"updateTime": c.update_time.strftime("%Y-%m-%d %H:%M:%S") if c.update_time else "",
|
||
}
|
||
|
||
|
||
def id_card_taken(db: Session, id_card: str, tenant_no: str, exclude_id: Optional[str] = None) -> bool:
|
||
ic = (id_card or "").strip()
|
||
if not ic:
|
||
return False
|
||
q = db.query(BizCustomer).filter(
|
||
and_(BizCustomer.id_card_no == ic, BizCustomer.tenant_no == tenant_no, BizCustomer.del_flag == 0)
|
||
)
|
||
if exclude_id:
|
||
q = q.filter(BizCustomer.id != exclude_id)
|
||
return q.first() is not None
|
||
|
||
|
||
def _normalize_date_bound(value: str, *, end_of_day: bool = False) -> str:
|
||
val = (value or "").strip()
|
||
if len(val) == 10:
|
||
return f"{val} 23:59:59" if end_of_day else f"{val} 00:00:00"
|
||
return val
|
||
|
||
|
||
def list_customers(
|
||
db: Session,
|
||
*,
|
||
page: int = 1,
|
||
page_size: int = 10,
|
||
keyword: Optional[str] = None,
|
||
create_time_start: Optional[str] = None,
|
||
create_time_end: Optional[str] = None,
|
||
tenant_no: Optional[str] = None,
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
def _filtered():
|
||
q0 = db.query(BizCustomer).filter(BizCustomer.del_flag == 0)
|
||
if tenant_no:
|
||
q0 = q0.filter(BizCustomer.tenant_no == tenant_no)
|
||
kw = (keyword or "").strip()
|
||
if kw:
|
||
like = f"%{kw}%"
|
||
q0 = q0.filter(
|
||
or_(
|
||
BizCustomer.customer_name.like(like),
|
||
BizCustomer.phone_no.like(like),
|
||
BizCustomer.id_card_no.like(like),
|
||
BizCustomer.customer_tags.like(like),
|
||
)
|
||
)
|
||
if create_time_start:
|
||
q0 = q0.filter(
|
||
BizCustomer.create_time >= _normalize_date_bound(create_time_start)
|
||
)
|
||
if create_time_end:
|
||
q0 = q0.filter(
|
||
BizCustomer.create_time <= _normalize_date_bound(create_time_end, end_of_day=True)
|
||
)
|
||
return q0
|
||
|
||
total = _filtered().order_by(None).with_entities(func.count(BizCustomer.id)).scalar() or 0
|
||
page = max(1, page)
|
||
page_size = min(max(1, page_size), 100)
|
||
rows = (
|
||
_filtered()
|
||
.order_by(BizCustomer.create_time.desc())
|
||
.offset((page - 1) * page_size)
|
||
.limit(page_size)
|
||
.all()
|
||
)
|
||
return [_row_to_dict(x) for x in rows], int(total)
|
||
|
||
|
||
def get_customer(db: Session, customer_id: str, tenant_no: Optional[str] = None) -> Optional[BizCustomer]:
|
||
q = db.query(BizCustomer).filter(and_(BizCustomer.id == customer_id, BizCustomer.del_flag == 0))
|
||
if tenant_no:
|
||
q = q.filter(BizCustomer.tenant_no == tenant_no)
|
||
return q.first()
|
||
|
||
|
||
def create_customer(
|
||
db: Session,
|
||
*,
|
||
customer_name: str,
|
||
sex: str = "2",
|
||
id_card_no: Optional[str] = None,
|
||
age: Optional[int] = None,
|
||
birth_ym: Optional[str] = None,
|
||
ethnic: Optional[str] = None,
|
||
email: Optional[str] = None,
|
||
address: Optional[str] = None,
|
||
phone_no: Optional[str] = None,
|
||
customer_tags: Optional[str] = None,
|
||
tenant_no: str = "zkxw",
|
||
order_num: Optional[int] = None,
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
commit: bool = True,
|
||
create_dept_id: str = "",
|
||
) -> BizCustomer:
|
||
name = (customer_name or "").strip()
|
||
if not name:
|
||
raise ValueError("客户姓名不能为空")
|
||
ic = (id_card_no or "").strip() or None
|
||
if ic and id_card_taken(db, ic, tenant_no):
|
||
raise ValueError("该身份证号已存在客户，请勿重复创建")
|
||
obj = BizCustomer(
|
||
id=generate_uuid(),
|
||
customer_name=name,
|
||
sex=(sex or "2")[:1] if sex else "2",
|
||
id_card_no=ic,
|
||
age=age,
|
||
birth_ym=(birth_ym or "").strip()[:6] or None,
|
||
ethnic=(ethnic or "").strip() or None,
|
||
email=(email or "").strip() or None,
|
||
address=(address or "").strip() or None,
|
||
phone_no=(phone_no or "").strip() or None,
|
||
customer_tags=(customer_tags or "").strip() or None,
|
||
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,
|
||
create_time=datetime.now(),
|
||
update_by=user_id,
|
||
update_time=datetime.now(),
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
if commit:
|
||
db.commit()
|
||
db.refresh(obj)
|
||
else:
|
||
db.flush()
|
||
return obj
|
||
|
||
|
||
def update_customer(
|
||
db: Session,
|
||
customer_id: str,
|
||
*,
|
||
customer_name: Optional[str] = None,
|
||
sex: Optional[str] = None,
|
||
id_card_no: Optional[str] = None,
|
||
age: Optional[int] = None,
|
||
birth_ym: Optional[str] = None,
|
||
ethnic: Optional[str] = None,
|
||
email: Optional[str] = None,
|
||
address: Optional[str] = None,
|
||
phone_no: Optional[str] = None,
|
||
customer_tags: Optional[str] = None,
|
||
order_num: Optional[int] = None,
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
tenant_no: Optional[str] = None,
|
||
) -> BizCustomer:
|
||
obj = get_customer(db, customer_id, tenant_no=tenant_no)
|
||
if not obj:
|
||
raise ValueError("客户不存在")
|
||
if customer_name is not None:
|
||
n = customer_name.strip()
|
||
if not n:
|
||
raise ValueError("客户姓名不能为空")
|
||
obj.customer_name = n
|
||
if sex is not None:
|
||
obj.sex = (sex or "2")[:1]
|
||
if id_card_no is not None:
|
||
ic = id_card_no.strip() or None
|
||
if ic and id_card_taken(db, ic, obj.tenant_no, exclude_id=customer_id):
|
||
raise ValueError("该身份证号已被其他客户使用")
|
||
obj.id_card_no = ic
|
||
if age is not None:
|
||
obj.age = age
|
||
if birth_ym is not None:
|
||
obj.birth_ym = birth_ym.strip()[:6] or None
|
||
if ethnic is not None:
|
||
obj.ethnic = ethnic.strip() or None
|
||
if email is not None:
|
||
obj.email = email.strip() or None
|
||
if address is not None:
|
||
obj.address = address.strip() or None
|
||
if phone_no is not None:
|
||
obj.phone_no = phone_no.strip() or None
|
||
if customer_tags is not None:
|
||
obj.customer_tags = customer_tags.strip() or None
|
||
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_customer(
|
||
db: Session,
|
||
customer_id: str,
|
||
tenant_no: Optional[str] = None,
|
||
*,
|
||
commit: bool = True,
|
||
) -> None:
|
||
obj = get_customer(db, customer_id, tenant_no=tenant_no)
|
||
if not obj:
|
||
raise ValueError("客户不存在")
|
||
obj.del_flag = 1
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
if commit:
|
||
db.commit()
|
||
else:
|
||
db.flush()
|
||
|
||
|
||
def total_pages(total: int, page_size: int) -> int:
|
||
if total <= 0:
|
||
return 0
|
||
return int(math.ceil(total / float(page_size)))
|