zkxw/backend/app/crud/customer.py

"""客户 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)))