shanxia2.0/backend/app/crud/crm/contact.py

"""CRM 联系人 CRUD"""
from __future__ import annotations

import math
from datetime import datetime, date
from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy import func, or_
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.crud.crm.data_scope_helpers import (
    apply_crm_create_by_scope,
    ensure_crm_contact_row_in_scope,
    ensure_crm_owner_row_in_scope,
)
from app.models.crm_models import CrmContact


def total_pages(total: int, page_size: int) -> int:
    if page_size <= 0:
        return 0
    return math.ceil(total / page_size)


_FMT = "%Y-%m-%d %H:%M:%S"
_DATE_FMT = "%Y-%m-%d"


def _parse_optional_date(value: Any) -> Optional[date]:
    if value is None:
        return None
    if isinstance(value, date) and not isinstance(value, datetime):
        return value
    if isinstance(value, datetime):
        return value.date()
    text = str(value).strip()
    if not text:
        return None
    for fmt in (_DATE_FMT, _FMT):
        try:
            return datetime.strptime(text, fmt).date()
        except ValueError:
            continue
    raise ValueError(f"日期格式无效: {text}")


def _row_to_dict(row: CrmContact) -> Dict[str, Any]:
    return {
        "id": row.id,
        "customerId": row.customer_id or "",
        "customerName": row.customer_name or "",
        "contactName": row.contact_name or "",
        "gender": row.gender or "",
        "mobile": row.mobile or "",
        "telephone": row.telephone or "",
        "email": row.email or "",
        "wechat": row.wechat or "",
        "qq": row.qq or "",
        "department": row.department or "",
        "position": row.position or "",
        "role": row.role or "",
        "birthday": row.birthday.strftime("%Y-%m-%d") if row.birthday else "",
        "address": row.address or "",
        "isKey": row.is_key if row.is_key is not None else 0,
        "isPrimary": row.is_primary if row.is_primary is not None else 0,
        "status": row.status or "",
        "orderNum": row.order_num if row.order_num is not None else 0,
        "remark": row.remark or "",
        "createBy": row.create_by or "",
        "createTime": row.create_time.strftime(_FMT) if row.create_time else "",
        "updateBy": row.update_by or "",
        "updateTime": row.update_time.strftime(_FMT) if row.update_time else "",
    }


def list_contacts(
    db: Session,
    tenant_no: str,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: str = None,
    contact_name: str = None,
    mobile: str = None,
    position: str = None,
    customer_id: str = None,
    status: str = None,
    is_key: int = None,
    is_primary: int = None,
    is_decision: bool = None,
    birthday_month: int = None,
    role: str = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    q = db.query(CrmContact).filter(
        CrmContact.tenant_no == tenant_no,
        CrmContact.del_flag == 0,
    )
    if user_info:
        q = apply_crm_create_by_scope(q, CrmContact, user_info, db, tenant_no)

    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q = q.filter(
            or_(
                CrmContact.contact_name.like(like),
                CrmContact.mobile.like(like),
                CrmContact.email.like(like),
                CrmContact.position.like(like),
            )
        )

    name_kw = (contact_name or "").strip()
    if name_kw:
        q = q.filter(CrmContact.contact_name.like(f"%{name_kw}%"))
    mobile_kw = (mobile or "").strip()
    if mobile_kw:
        q = q.filter(CrmContact.mobile.like(f"%{mobile_kw}%"))
    position_kw = (position or "").strip()
    if position_kw:
        q = q.filter(CrmContact.position.like(f"%{position_kw}%"))

    if customer_id:
        q = q.filter(CrmContact.customer_id == customer_id)
    if status:
        q = q.filter(CrmContact.status == status)
    if is_key is not None:
        q = q.filter(CrmContact.is_key == is_key)
    if is_primary is not None:
        q = q.filter(CrmContact.is_primary == is_primary)
    if is_decision is not None and is_decision:
        q = q.filter(CrmContact.is_key == 1)
    if birthday_month is not None:
        q = q.filter(func.month(CrmContact.birthday) == birthday_month)
    if role:
        q = q.filter(CrmContact.role == role)

    total = q.order_by(None).with_entities(func.count(CrmContact.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)

    rows = (
        q.order_by(CrmContact.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    return [_row_to_dict(r) for r in rows], int(total)


def list_contacts_by_customer(
    db: Session,
    tenant_no: str,
    customer_id: str,
    *,
    user_info=None,
) -> List[Dict[str, Any]]:
    if user_info:
        from app.crud.crm.customer import get_customer

        customer = get_customer(db, customer_id)
        if not customer or customer.tenant_no != tenant_no:
            raise ValueError("客户不存在")
        ensure_crm_owner_row_in_scope(
            db, customer, user_info, tenant_no, message="无权访问该客户"
        )
    rows = (
        db.query(CrmContact)
        .filter(
            CrmContact.tenant_no == tenant_no,
            CrmContact.customer_id == customer_id,
            CrmContact.del_flag == 0,
        )
        .order_by(CrmContact.is_primary.desc(), CrmContact.create_time.desc())
        .all()
    )
    return [_row_to_dict(r) for r in rows]


def get_contact(db: Session, contact_id: str) -> Optional[CrmContact]:
    return (
        db.query(CrmContact)
        .filter(
            CrmContact.id == contact_id,
            CrmContact.del_flag == 0,
        )
        .first()
    )


def create_contact(
    db: Session,
    *,
    tenant_no: str,
    user_id: str = None,
    dept_id: str = "",
    user_info=None,
    **kwargs,
) -> CrmContact:
    customer_id = kwargs.get("customer_id")
    customer_name = kwargs.get("customer_name")
    if customer_id:
        from app.crud.crm.customer import get_customer

        cust = get_customer(db, customer_id)
        if user_info:
            if not cust or cust.tenant_no != tenant_no:
                raise ValueError("客户不存在")
            ensure_crm_owner_row_in_scope(
                db, cust, user_info, tenant_no, message="无权访问该客户"
            )
        if cust and not customer_name:
            customer_name = cust.customer_name

    obj = CrmContact(
        id=generate_uuid(),
        customer_id=customer_id,
        customer_name=customer_name or "",
        contact_name=kwargs.get("contact_name"),
        gender=kwargs.get("gender"),
        mobile=kwargs.get("mobile"),
        telephone=kwargs.get("telephone"),
        email=kwargs.get("email"),
        wechat=kwargs.get("wechat"),
        qq=kwargs.get("qq"),
        department=kwargs.get("department"),
        position=kwargs.get("position"),
        role=kwargs.get("role"),
        birthday=_parse_optional_date(kwargs.get("birthday")),
        address=kwargs.get("address"),
        is_key=kwargs.get("is_key", 0),
        is_primary=kwargs.get("is_primary", 0),
        status=kwargs.get("status", "active"),
        tenant_no=tenant_no,
        order_num=kwargs.get("order_num"),
        remark=kwargs.get("remark"),
        create_by=user_id,
        create_dept_id=dept_id,
        del_flag=0,
    )
    db.add(obj)
    db.commit()
    db.refresh(obj)

    if customer_id:
        from app.crud.crm.customer import update_contact_count
        update_contact_count(db, customer_id)

    return obj


def update_contact(
    db: Session,
    contact_id: str,
    *,
    user_id: str = None,
    tenant_no: str = None,
    user_info=None,
    **kwargs,
) -> CrmContact:
    obj = get_contact(db, contact_id)
    if not obj:
        raise ValueError("联系人不存在")
    if tenant_no and obj.tenant_no != tenant_no:
        raise ValueError("联系人不存在")
    ensure_crm_contact_row_in_scope(db, obj, user_info, obj.tenant_no)

    _updatable = (
        "contact_name", "gender", "mobile", "telephone", "email",
        "wechat", "qq", "department", "position", "role", "birthday",
        "address", "is_key", "is_primary", "status", "order_num", "remark",
    )
    _date_fields = {"birthday"}
    for field in _updatable:
        if field not in kwargs:
            continue
        val = kwargs[field]
        if field in _date_fields:
            setattr(obj, field, _parse_optional_date(val))
        elif val is not None:
            setattr(obj, field, val)

    if "customer_name" in kwargs and kwargs["customer_name"] is not None:
        obj.customer_name = kwargs["customer_name"]

    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.commit()
    db.refresh(obj)
    return obj


def set_primary_contact(
    db: Session,
    contact_id: str,
    user_id: str = None,
    *,
    tenant_no: str = None,
    user_info=None,
) -> CrmContact:
    obj = get_contact(db, contact_id)
    if not obj:
        raise ValueError("联系人不存在")
    if tenant_no and obj.tenant_no != tenant_no:
        raise ValueError("联系人不存在")
    ensure_crm_contact_row_in_scope(db, obj, user_info, obj.tenant_no)

    db.query(CrmContact).filter(
        CrmContact.customer_id == obj.customer_id,
        CrmContact.del_flag == 0,
        CrmContact.id != contact_id,
    ).update({"is_primary": 0})

    obj.is_primary = 1
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_contact(
    db: Session,
    contact_id: str,
    user_id: str = None,
    *,
    tenant_no: str = None,
    user_info=None,
) -> bool:
    obj = get_contact(db, contact_id)
    if not obj:
        raise ValueError("联系人不存在")
    if tenant_no and obj.tenant_no != tenant_no:
        raise ValueError("联系人不存在")
    ensure_crm_contact_row_in_scope(db, obj, user_info, obj.tenant_no)
    customer_id = obj.customer_id
    obj.del_flag = 1
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.commit()

    if customer_id:
        from app.crud.crm.customer import update_contact_count
        update_contact_count(db, customer_id)

    return True