shanxia2.0/backend/app/api/crm.py

"""CRM 管理接口（产品分类 + 产品 + Excel 导入导出）"""
from __future__ import annotations

import io
from datetime import datetime
from typing import Any, Dict, List
from urllib.parse import quote

from fastapi import APIRouter, Depends, File, Form, UploadFile
from sqlalchemy import or_
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.core.menu_guard import require_admin_or_super_admin
from app.core.oper_log_decorator import oper_log
from app.core.response import error_response, success_response
from app.core.tenant_context import get_tenant_no
from app.crud.crm.data_scope_helpers import (
    crm_user_info,
    ensure_crm_create_by_row_in_scope,
    ensure_crm_contact_row_in_scope,
    ensure_crm_customer_linked_row_in_scope,
    ensure_crm_owner_row_in_scope,
)
from app.utils.crm_export_columns import (
    CLUE_EXPORT_COLUMNS,
    CONTACT_EXPORT_COLUMNS,
    CUSTOMER_EXPORT_COLUMNS,
    CUSTOMER_FOLLOW_EXPORT_COLUMNS,
    OPPORTUNITY_EXPORT_COLUMNS,
    PRODUCT_EXPORT_COLUMNS,
    SUPPLIER_INFO_EXPORT_COLUMNS,
)
from app.utils.excel_export import export_rows_to_xlsx_response
from app.crud.crm.product import (
    _category_row_to_dict,
    create_category,
    create_product,
    get_category,
    get_product,
    list_categories,
    list_category_tree,
    list_product_options,
    list_products,
    product_no_exists,
    product_to_dict,
    soft_delete_category,
    soft_delete_product,
    total_pages,
    update_category,
    update_product,
)
from app.schemas.crm.product import (
    CategoryDeleteBody,
    ProductCategoryCreateBody,
    ProductCategoryTreeBody,
    ProductCategoryUpdateBody,
    ProductCreateBody,
    ProductDeleteBody,
    ProductDetailBody,
    ProductListBody,
    ProductOptionsBody,
    ProductUpdateBody,
    ProductUpdateStatusBody,
)
from app.crud.crm.customer import (
    _row_to_dict as customer_row_to_dict,
    _enrich_user_names as enrich_customer_user_names,
    assign_customer,
    batch_claim_customers,
    check_duplicate_customer,
    claim_customer,
    create_customer,
    customer_name_exists,
    deal_customer,
    generate_customer_no,
    get_customer,
    get_park_area_list,
    into_pool,
    list_customer_options,
    list_customers,
    list_pool_customers,
    soft_delete_customer,
    total_pages as customer_total_pages,
    transfer_customer,
    update_customer,
)
from app.crud.crm.contact import (
    _row_to_dict as contact_row_to_dict,
    create_contact,
    get_contact,
    list_contacts,
    list_contacts_by_customer,
    set_primary_contact,
    soft_delete_contact,
    total_pages as contact_total_pages,
    update_contact,
)
from app.crud.crm.invoice_info import (
    _row_to_dict as invoice_info_row_to_dict,
    create_invoice_info,
    list_invoice_info_by_customer,
    set_default_invoice_info,
    soft_delete_invoice_info,
    update_invoice_info,
)
from app.crud.crm.supplier_info import (
    _row_to_dict as supplier_info_row_to_dict,
    create_supplier_info,
    get_supplier_info,
    list_all_supplier_info,
    list_supplier_info,
    soft_delete_supplier_info,
    total_pages as supplier_info_total_pages,
    update_supplier_info,
)
from app.crud.crm.customer_pool import (
    get_pool_record,
    list_pool_records,
    total_pages as pool_record_total_pages,
)
from app.crud.crm.pool_rule import (
    create_pool_rule,
    delete_pool_rules,
    execute_pool_rules,
    get_pool_rule,
    list_pool_rules,
    total_pages as pool_rule_total_pages,
    update_pool_rule,
    update_pool_rule_status,
)
from app.crud.crm.opportunity import (
    assign_opportunities,
    batch_update_expect_date,
    batch_update_stage,
    change_stage,
    create_opportunity,
    get_kanban_data,
    get_opportunity,
    get_sales_funnel,
    get_stage_log_list,
    get_statistics_by_owner,
    get_statistics_by_stage,
    get_statistics_overview,
    get_win_rate_statistics,
    list_opportunities,
    mark_lost,
    mark_won,
    soft_delete_opportunities,
    total_pages as opportunity_total_pages,
    update_opportunity,
)
from app.crud.crm.opportunity_follow import (
    create_opportunity_follow,
    get_opportunity_follow,
    list_follows_by_opportunity,
    list_opportunity_follows,
    soft_delete_opportunity_follow,
    total_pages as opp_follow_total_pages,
    update_opportunity_follow,
)
from app.crud.crm.opportunity_product import (
    batch_add_opportunity_products,
    create_opportunity_product,
    get_opportunity_product,
    list_opportunity_products,
    list_products_by_opportunity,
    soft_delete_opportunity_products,
    total_pages as opp_product_total_pages,
    update_opportunity_product,
)
from app.crud.crm.customer_follow import (
    _row_to_dict as follow_row_to_dict,
    create_follow,
    get_follow,
    list_follows,
    list_follows_by_customer,
    soft_delete_follow,
    total_pages as follow_total_pages,
    update_follow,
)
from app.crud.crm.statistics import (
    get_follow_statistics,
    get_industry_statistics,
    get_level_statistics,
    get_owner_statistics,
    get_overview_statistics,
    get_source_statistics,
    get_trend_statistics,
)
from app.schemas.crm.customer_pool import (
    PoolAssignBody,
    PoolClaimBody,
    PoolCustomerListBody,
    PoolRecordDetailBody,
    PoolRecordListBody,
)
from app.schemas.crm.pool_rule import (
    PoolRuleAddBody,
    PoolRuleDeleteBody,
    PoolRuleDetailBody,
    PoolRuleExecuteBody,
    PoolRuleListBody,
    PoolRuleUpdateBody,
    PoolRuleUpdateStatusBody,
)
from app.schemas.crm.customer import (
    CustomerCreateBody,
    CustomerDeleteBody,
    CustomerListBody,
    CustomerOptionsBody,
    CustomerUpdateBody,
)
from app.schemas.crm.contact import (
    ContactCreateBody,
    ContactDeleteBody,
    ContactListBody,
    ContactUpdateBody,
)
from app.schemas.crm.invoice_info import (
    InvoiceInfoByCustomerBody,
    InvoiceInfoCreateBody,
    InvoiceInfoDeleteBody,
    InvoiceInfoSetDefaultBody,
    InvoiceInfoUpdateBody,
)
from app.schemas.crm.supplier_info import (
    SupplierInfoCreateBody,
    SupplierInfoDeleteBody,
    SupplierInfoDetailBody,
    SupplierInfoListAllBody,
    SupplierInfoListBody,
    SupplierInfoUpdateBody,
)
from app.schemas.crm.opportunity import (
    OpportunityAssignBody,
    OpportunityBatchUpdateExpectDateBody,
    OpportunityBatchUpdateStageBody,
    OpportunityChangeStageBody,
    OpportunityCreateBody,
    OpportunityDeleteBody,
    OpportunityDetailBody,
    OpportunityFollowCreateBody,
    OpportunityFollowDeleteBody,
    OpportunityFollowDetailBody,
    OpportunityFollowListBody,
    OpportunityFollowUpdateBody,
    OpportunityKanbanBody,
    OpportunityListBody,
    OpportunityMarkLostBody,
    OpportunityMarkWonBody,
    OpportunityProductBatchAddBody,
    OpportunityProductByOpportunityBody,
    OpportunityProductCreateBody,
    OpportunityProductDeleteBody,
    OpportunityProductDetailBody,
    OpportunityProductListBody,
    OpportunityProductUpdateBody,
    OpportunityStageLogListBody,
    OpportunityStatisticsBody,
    OpportunityUpdateBody,
)
from app.schemas.crm.customer_follow import (
    CustomerFollowCreateBody as FollowCreateBody,
    CustomerFollowDeleteBody as FollowDeleteBody,
    CustomerFollowListBody as FollowListBody,
    CustomerFollowUpdateBody as FollowUpdateBody,
)
from app.schemas.crm.statistics import StatisticsBody
from app.crud.crm.clue import (
    _row_to_dict as clue_row_to_dict,
    _enrich_user_names as enrich_clue_user_names,
    assign_clues,
    check_convert,
    check_duplicate_clue,
    convert_clue,
    create_clue,
    get_clue,
    get_park_area_list as get_clue_park_area_list,
    import_clues,
    invalid_clue,
    list_clues,
    soft_delete_clue,
    total_pages as clue_total_pages,
    update_clue,
    update_clue_level,
    update_clue_status,
)
from app.crud.crm.clue_follow import (
    _row_to_dict as clue_follow_row_to_dict,
    create_clue_follow,
    delete_clue_follow,
    get_clue_follow,
    get_follow_statistics as get_clue_follow_statistics,
    list_clue_follows,
    total_pages as clue_follow_total_pages,
    update_clue_follow,
)
from app.crud.crm.clue_statistics import (
    get_clue_level_statistics,
    get_clue_overview_statistics,
    get_clue_owner_statistics,
    get_clue_source_statistics,
    get_clue_trend_statistics,
)
from app.schemas.crm.clue import (
    ClueAssignBody,
    ClueConvertBody,
    ClueCreateBody,
    ClueDeleteBody,
    ClueFollowCreateBody,
    ClueFollowDeleteBody,
    ClueFollowListBody,
    ClueFollowUpdateBody,
    ClueInvalidBody,
    ClueLevelBody,
    ClueListBody,
    ClueStatisticsBody,
    ClueStatusBody,
    ClueUpdateBody,
)

router = APIRouter(prefix="/web/crm", tags=["CRM管理"])


# ── 产品分类 ──────────────────────────────────────────────────


@router.post("/product-categories/tree")
def category_tree(
    body: ProductCategoryTreeBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """获取产品分类树（名称/状态由服务端过滤）"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        q = body or ProductCategoryTreeBody()
        rows = list_category_tree(
            db,
            tenant_no,
            category_name=q.category_name,
            status=q.status,
            user_info=user_info,
        )
        return success_response(rows)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


# ── 客户管理 ──────────────────────────────────────────────────


@router.post("/customer/list")
def customer_list(
    body: CustomerListBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """客户列表（支持多种视图模式）"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")

        rows, total = list_customers(
            db,
            tenant_no,
            user_id,
            user_info=user_info,
            page=body.page,
            page_size=body.pageSize,
            **body.to_list_kwargs(),
        )
        return success_response({
            "list": rows,
            "total": total,
            "page": body.page,
            "pageSize": body.pageSize,
            "totalPages": customer_total_pages(total, body.pageSize),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/detail")
def customer_detail(
    body: dict,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """客户详情"""
    try:
        customer_id = body.get("id")
        if not customer_id:
            return error_response(400, "缺少客户ID")
        row = get_customer(db, customer_id)
        if not row:
            return error_response(404, "客户不存在")
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        ensure_crm_owner_row_in_scope(db, row, user_info, tenant_no, message="无权访问该客户")
        detail = customer_row_to_dict(row)
        enrich_customer_user_names(db, [detail])
        return success_response(detail)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/options")
def customer_options(
    body: CustomerOptionsBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """客户下拉选项（租户内只读，不做 owner 数据权限）"""
    try:
        tenant_no = get_tenant_no(db, _user)
        keyword = (body.keyword or body.customer_name or "").strip() or None
        options = list_customer_options(
            db,
            tenant_no,
            keyword=keyword,
            limit=body.limit,
        )
        return success_response(options)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/add")
@oper_log(title="CRM客户管理", business_type=1, summary="新增客户【{customerName}】")
def customer_create(
    body: CustomerCreateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """新增客户"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        dept_id = _user.get("dept_id", "")

        if customer_name_exists(db, body.customer_name, tenant_no):
            return error_response(400, "客户名称已存在")

        obj = create_customer(
            db,
            tenant_no=tenant_no,
            user_id=user_id,
            dept_id=dept_id,
            **body.model_dump(exclude_unset=True),
        )
        return success_response(customer_row_to_dict(obj), message="创建成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/update")
@oper_log(title="CRM客户管理", business_type=2, summary="修改客户【{customerName}】")
def customer_update(
    body: CustomerUpdateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """编辑客户"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")

        data = body.model_dump(exclude_unset=True)
        customer_id = data.pop("id")
        if "customer_name" in data:
            if customer_name_exists(db, data["customer_name"], tenant_no, exclude_id=customer_id):
                return error_response(400, "客户名称已存在")

        obj = update_customer(db, customer_id, user_id=user_id, tenant_no=tenant_no, user_info=user_info, **data)
        return success_response(customer_row_to_dict(obj), message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/delete")
@oper_log(title="CRM客户管理", business_type=3, business_id_field="ids", summary="删除客户")
def customer_delete(
    body: CustomerDeleteBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """删除客户（支持批量）"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        success_ids = []
        fail_items = []
        for cid in body.ids:
            try:
                soft_delete_customer(db, cid, user_info=user_info, tenant_no=tenant_no)
                success_ids.append(cid)
            except ValueError as e:
                fail_items.append({"id": cid, "reason": str(e)})
        result = {
            "successCount": len(success_ids),
            "failCount": len(fail_items),
            "failItems": fail_items,
        }
        return success_response(result, message=f"删除完成，成功{len(success_ids)}个")
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/generate-no")
def customer_generate_no(
    body: dict = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """生成客户编号"""
    try:
        tenant_no = get_tenant_no(db, _user)
        no = generate_customer_no(db, tenant_no)
        return success_response({"customerNo": no})
    except Exception as e:
        return error_response(500, str(e))



@router.post("/product-categories/add")
def category_add(
    body: ProductCategoryCreateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """新增产品分类"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        data = body.model_dump(exclude_unset=True)
        obj = create_category(db, tenant_no=tenant_no, user_id=user_id, **data)
        return success_response(_category_row_to_dict(obj), message="新增成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/product-categories/update")
def category_update(
    body: ProductCategoryUpdateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """更新产品分类"""
    try:
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        data = body.model_dump(exclude_unset=True)
        category_id = data.pop("id")
        obj = update_category(db, category_id, user_id=user_id, user_info=user_info, **data)
        return success_response(_category_row_to_dict(obj), message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/product-categories/delete")
def category_delete(
    body: CategoryDeleteBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """删除产品分类"""
    try:
        user_info = crm_user_info(_user)
        soft_delete_category(db, body.id, user_info=user_info)
        return success_response(None, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


# ── 产品管理 ──────────────────────────────────────────────────


@router.post("/products/list")
def product_list(
    body: ProductListBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """分页查询产品列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        rows, total = list_products(
            db,
            tenant_no,
            page=body.page,
            page_size=body.pageSize,
            keyword=body.keyword,
            category_id=body.category_id,
            status=body.status,
            supplier_id=body.supplier_id,
            user_info=user_info,
        )
        tp = total_pages(total, body.pageSize)
        return success_response(
            {
                "list": rows,
                "total": total,
                "page": body.page,
                "pageSize": body.pageSize,
                "totalPages": tp,
            }
        )
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/products/detail")
def product_detail(
    body: ProductDetailBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """获取产品详情"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        obj = get_product(db, body.id)
        if not obj:
            return error_response(404, "产品不存在")
        ensure_crm_create_by_row_in_scope(db, obj, user_info, tenant_no, message="无权访问该产品")
        return success_response(product_to_dict(db, obj))
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/products/add")
def product_add(
    body: ProductCreateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """新增产品"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        obj = create_product(
            db,
            tenant_no=tenant_no,
            user_id=user_id,
            **body.model_dump(exclude_unset=True),
        )
        return success_response(product_to_dict(db, obj), message="创建成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/products/update")
def product_update(
    body: ProductUpdateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """更新产品"""
    try:
        user_id = _user.get("user_id", "") or _user.get("id", "")
        data = body.model_dump(exclude_unset=True)
        product_id = data.pop("id")
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        obj = update_product(db, product_id, user_id=user_id, user_info=user_info, **data)
        return success_response(product_to_dict(db, obj), message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/products/delete")
def product_delete(
    body: ProductDeleteBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """删除产品（支持单条 body.id 和批量 body.ids）"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        ids_to_delete: List[str] = []
        if body.ids:
            ids_to_delete = body.ids
        elif body.id:
            ids_to_delete = [body.id]
        else:
            return error_response(400, "请传入 id 或 ids")

        for pid in ids_to_delete:
            soft_delete_product(db, pid, user_info=user_info)

        return success_response(None, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/products/update-status")
def product_update_status(
    body: ProductUpdateStatusBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """批量上架/下架"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        status_label = "上架" if body.status == "on" else "下架"
        for pid in body.ids:
            update_product(db, pid, user_id=user_id, user_info=user_info, status=body.status)
        return success_response(None, message=f"批量{status_label}成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/products/options")
def product_options(
    body: ProductOptionsBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """获取产品下拉选项"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        options = list_product_options(
            db,
            tenant_no,
            keyword=body.keyword,
            category_id=body.category_id,
            user_info=user_info,
            limit=body.limit,
        )
        return success_response(options)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


# ── Excel 导入导出 ────────────────────────────────────────────


@router.post("/products/import")
def product_import(
    file: UploadFile = File(...),
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """Excel 导入产品"""
    try:
        from openpyxl import load_workbook

        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")

        content = file.file.read()
        wb = load_workbook(io.BytesIO(content), read_only=True)
        ws = wb.active

        # 构建分类名→ID映射
        all_categories = list_categories(db, tenant_no)
        cat_name_to_id: Dict[str, str] = {}
        for cat in all_categories:
            cat_name_to_id[cat["categoryName"]] = cat["id"]

        success_count = 0
        fail_count = 0
        errors: List[Dict[str, Any]] = []

        rows = list(ws.iter_rows(min_row=2, values_only=True))
        for idx, row in enumerate(rows, start=2):
            row_errors: List[str] = []
            try:
                # 列: 产品编号、产品名称、分类名称、规格型号、状态、备注
                product_no = str(row[0]).strip() if row[0] else None
                product_name = str(row[1]).strip() if row[1] else ""
                category_name = str(row[2]).strip() if row[2] else None
                model = str(row[3]).strip() if row[3] else None
                status = str(row[4]).strip() if row[4] else "on"
                if status == "1":
                    status = "on"
                elif status == "0":
                    status = "off"
                remark = str(row[5]).strip() if row[5] else None

                # 必填校验
                if not product_name:
                    row_errors.append("产品名称不能为空")

                # 编号唯一性
                if product_no and product_no_exists(db, product_no, tenant_no):
                    row_errors.append(f"产品编号 {product_no} 已存在")

                # 分类名→ID
                category_id = None
                if category_name:
                    category_id = cat_name_to_id.get(category_name)
                    if not category_id:
                        row_errors.append(f"分类名称 '{category_name}' 不存在")

                if row_errors:
                    fail_count += 1
                    errors.append({"row": idx, "errors": row_errors})
                    continue

                create_product(
                    db,
                    tenant_no=tenant_no,
                    user_id=user_id,
                    product_no=product_no or None,
                    product_name=product_name,
                    category_id=category_id,
                    category_name=category_name,
                    model=model,
                    status=status,
                    remark=remark,
                )
                success_count += 1
            except Exception as e:
                fail_count += 1
                errors.append({"row": idx, "errors": [str(e)]})

        wb.close()
        return success_response(
            {"success": success_count, "fail": fail_count, "errors": errors}
        )
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/products/export")
def product_export(
    body: ProductListBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """Excel 导出产品列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        rows, _ = list_products(
            db,
            tenant_no,
            page=1,
            page_size=10000,
            keyword=body.keyword,
            category_id=body.category_id,
            status=body.status,
            supplier_id=body.supplier_id,
        )
        date_str = datetime.now().strftime("%Y%m%d")
        return export_rows_to_xlsx_response(
            db,
            tenant_no=tenant_no,
            sheet_title="产品列表",
            columns=PRODUCT_EXPORT_COLUMNS,
            rows=rows,
            filename_stem=f"产品列表_{date_str}",
        )
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/products/import-template")
def product_import_template(
    _user: dict = Depends(require_admin_or_super_admin),
):
    """下载产品导入模板"""
    try:
        from fastapi.responses import StreamingResponse
        from openpyxl import Workbook

        wb = Workbook()
        ws = wb.active
        ws.title = "产品导入模板"

        headers = [
            "产品编号",
            "产品名称",
            "分类名称",
            "规格型号",
            "状态（on上架/off下架）",
            "备注",
        ]
        ws.append(headers)

        output = io.BytesIO()
        wb.save(output)
        output.seek(0)

        filename = quote("产品导入模板.xlsx")

        return StreamingResponse(
            output,
            media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
            headers={
                "Content-Disposition": f"attachment; filename={filename}; filename*=UTF-8''{filename}"
            },
        )
    except Exception as e:
        return error_response(500, str(e))


# ── 客户管理（扩展路由） ────────────────────────────────────────


@router.post("/customer/transfer")
@oper_log(title="CRM客户管理", business_type=2, business_id_field="customerId", summary="转移客户")
def customer_transfer(
    body: dict,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """批量转移客户"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        ids = body.get("ids", [])
        owner_user_id = body.get("ownerUserId", "")
        owner_user_name = body.get("ownerUserName", "")
        if not ids or not owner_user_id:
            return error_response(400, "缺少必要参数 ids 或 ownerUserId")
        for cid in ids:
            transfer_customer(db, cid, owner_user_id, owner_user_name, user_id=user_id, user_info=user_info, tenant_no=tenant_no)
        return success_response(None, message="转移成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/intoPool")
@oper_log(title="CRM客户管理", business_type=2, business_id_field="customerId", summary="客户放入公海")
def customer_into_pool(
    body: dict,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """批量放入公海池"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        ids = body.get("ids", [])
        reason = body.get("reason", "")
        if not ids:
            return error_response(400, "缺少客户ID列表")
        for cid in ids:
            into_pool(db, cid, reason=reason, user_id=user_id, user_info=user_info, tenant_no=tenant_no)
        return success_response(None, message="已放入公海池")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/deal/{customer_id}")
@oper_log(title="CRM客户管理", business_type=2, path_param="customer_id", summary="客户成交")
def customer_deal(
    customer_id: str,
    body: dict = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """客户成交"""
    try:
        user_id = _user.get("user_id", "") or _user.get("id", "")
        deal_amount = body.get("dealAmount") if body else None
        deal_time = body.get("dealTime") if body else None
        remark = body.get("remark", "") if body else ""
        obj = deal_customer(
            db,
            customer_id,
            deal_amount=deal_amount,
            deal_time=deal_time,
            user_id=user_id,
            user_info=crm_user_info(_user),
            tenant_no=get_tenant_no(db, _user),
        )
        return success_response(customer_row_to_dict(obj), message="成交操作成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/checkDuplicate")
def customer_check_duplicate(
    body: dict = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """客户重复检测"""
    try:
        tenant_no = get_tenant_no(db, _user)
        mobile = body.get("mobile", "") if body else ""
        customer_name = body.get("customerName", "") if body else ""
        exclude_id = body.get("excludeId", "") if body else ""
        duplicates = check_duplicate_customer(
            db, tenant_no, mobile=mobile, customer_name=customer_name, exclude_id=exclude_id
        )
        return success_response({"duplicates": duplicates, "isDuplicate": len(duplicates) > 0})
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/export")
@oper_log(title="CRM客户管理", business_type=5, summary="导出客户列表")
def customer_export(
    body: CustomerListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """导出客户列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        list_body = body or CustomerListBody()
        rows, _ = list_customers(
            db,
            tenant_no,
            user_id,
            user_info=user_info,
            page=1,
            page_size=10000,
            **list_body.to_list_kwargs(),
        )
        date_str = datetime.now().strftime("%Y%m%d")
        return export_rows_to_xlsx_response(
            db,
            tenant_no=tenant_no,
            sheet_title="客户列表",
            columns=CUSTOMER_EXPORT_COLUMNS,
            rows=rows,
            filename_stem=f"客户列表_{date_str}",
        )
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/importTemplate")
def customer_import_template(
    _user: dict = Depends(require_admin_or_super_admin),
):
    """下载客户导入模板"""
    try:
        from fastapi.responses import StreamingResponse
        from openpyxl import Workbook

        wb = Workbook()
        ws = wb.active
        ws.title = "客户导入模板"

        headers = [
            "客户名称", "客户类型", "级别", "来源", "行业",
            "手机号", "电话", "邮箱", "地址", "备注",
        ]
        ws.append(headers)

        output = io.BytesIO()
        wb.save(output)
        output.seek(0)

        filename = quote("客户导入模板.xlsx")

        return StreamingResponse(
            output,
            media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
            headers={
                "Content-Disposition": f"attachment; filename={filename}; filename*=UTF-8''{filename}"
            },
        )
    except Exception as e:
        return error_response(500, str(e))


# ── 客户视图 ────────────────────────────────────────────────────


@router.post("/customer/mine")
def customer_mine(
    body: CustomerListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """我的客户列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        list_body = body or CustomerListBody()
        rows, total = list_customers(
            db,
            tenant_no,
            user_id,
            user_info=user_info,
            page=list_body.page,
            page_size=list_body.pageSize,
            **list_body.to_list_kwargs(view_mode_override="mine"),
        )
        return success_response({
            "list": rows,
            "total": total,
            "page": list_body.page,
            "pageSize": list_body.pageSize,
            "totalPages": customer_total_pages(total, list_body.pageSize),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/toFollow")
def customer_to_follow(
    body: CustomerListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """待跟进客户列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        list_body = body or CustomerListBody()
        rows, total = list_customers(
            db,
            tenant_no,
            user_id,
            user_info=user_info,
            page=list_body.page,
            page_size=list_body.pageSize,
            **list_body.to_list_kwargs(view_mode_override="toFollow"),
        )
        return success_response({
            "list": rows,
            "total": total,
            "page": list_body.page,
            "pageSize": list_body.pageSize,
            "totalPages": customer_total_pages(total, list_body.pageSize),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/key")
def customer_key(
    body: CustomerListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """重点客户列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        list_body = body or CustomerListBody()
        rows, total = list_customers(
            db,
            tenant_no,
            user_id,
            user_info=user_info,
            page=list_body.page,
            page_size=list_body.pageSize,
            **list_body.to_list_kwargs(view_mode_override="key"),
        )
        return success_response({
            "list": rows,
            "total": total,
            "page": list_body.page,
            "pageSize": list_body.pageSize,
            "totalPages": customer_total_pages(total, list_body.pageSize),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/dealed")
def customer_dealed(
    body: CustomerListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """已成交客户列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        page = body.page if body else 1
        page_size = body.pageSize if body else 10
        rows, total = list_customers(
            db, tenant_no, user_id, page=page, page_size=page_size,
            user_info=user_info,
            view_mode="dealed",
            keyword=body.keyword if body else None,
        )
        return success_response({
            "list": rows,
            "total": total,
            "page": page,
            "pageSize": page_size,
            "totalPages": customer_total_pages(total, page_size),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/weekNew")
def customer_week_new(
    body: CustomerListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """本周新增客户列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        page = body.page if body else 1
        page_size = body.pageSize if body else 10
        rows, total = list_customers(
            db, tenant_no, user_id, page=page, page_size=page_size,
            user_info=user_info,
            view_mode="weekNew",
            keyword=body.keyword if body else None,
        )
        return success_response({
            "list": rows,
            "total": total,
            "page": page,
            "pageSize": page_size,
            "totalPages": customer_total_pages(total, page_size),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/longTimeNoFollow")
def customer_long_time_no_follow(
    body: CustomerListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """长期未跟进客户列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        page = body.page if body else 1
        page_size = body.pageSize if body else 10
        rows, total = list_customers(
            db, tenant_no, user_id, page=page, page_size=page_size,
            user_info=user_info,
            view_mode="longTimeNoFollow",
            keyword=body.keyword if body else None,
        )
        return success_response({
            "list": rows,
            "total": total,
            "page": page,
            "pageSize": page_size,
            "totalPages": customer_total_pages(total, page_size),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/parkAreaList")
def customer_park_area_list(
    body: dict = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """获取园区列表（去重）"""
    try:
        tenant_no = get_tenant_no(db, _user)
        areas = get_park_area_list(db, tenant_no)
        return success_response(areas)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


# ── 公海池管理 ──────────────────────────────────────────────────


@router.post("/customer/pool/list")
def pool_customer_list(
    body: PoolCustomerListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = get_tenant_no(db, _user)
        page = body.page if body else 1
        page_size = body.pageSize if body else 10
        rows, total = list_pool_customers(
            db, tenant_no,
            page=page, page_size=page_size,
            keyword=body.keyword if body else None,
            level=body.level if body else None,
            industry=body.industry if body else None,
            park_area=body.parkArea if body else None,
        )
        return success_response({
            "list": rows,
            "total": total,
            "page": page,
            "pageSize": page_size,
            "totalPages": customer_total_pages(total, page_size),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/pool/assign")
@oper_log(title="CRM客户管理", business_type=2, business_id_field="customerIds", summary="公海分配客户")
def pool_customer_assign(
    body: PoolAssignBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        assign_customer(db, body.ids, body.ownerUserId, body.ownerUserName, user_id=user_id, user_info=user_info, tenant_no=tenant_no)
        return success_response(message="分配成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/claim")
@oper_log(title="CRM客户管理", business_type=2, business_id_field="customerId", summary="认领公海客户")
def pool_customer_claim(
    body: PoolClaimBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = get_tenant_no(db, _user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        user_name = _user.get("nick_name", "") or _user.get("name", "")
        batch_claim_customers(db, body.ids, user_id, user_name)
        return success_response(message="领取成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/pool/record/list")
def pool_record_list(
    body: PoolRecordListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = get_tenant_no(db, _user)
        page = body.page if body else 1
        page_size = body.pageSize if body else 10
        rows, total = list_pool_records(
            db, tenant_no,
            page=page, page_size=page_size,
            customer_id=body.customerId if body else None,
            operation_type=body.operationType if body else None,
            keyword=body.keyword if body else None,
        )
        return success_response({
            "list": rows,
            "total": total,
            "page": page,
            "pageSize": page_size,
            "totalPages": pool_record_total_pages(total, page_size),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/pool/record/detail")
def pool_record_detail(
    body: PoolRecordDetailBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = get_tenant_no(db, _user)
        from app.crud.crm.customer_pool import _row_to_dict as pool_record_row_to_dict
        row = get_pool_record(db, body.id, tenant_no)
        if not row:
            return error_response(404, "公海池记录不存在")
        return success_response(pool_record_row_to_dict(row))
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


# ── 联系人管理 ──────────────────────────────────────────────────


@router.post("/contact/list")
def contact_list(
    body: ContactListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """联系人分页列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        page = body.page if body else 1
        page_size = body.pageSize if body else 10
        rows, total = list_contacts(
            db, tenant_no, page=page, page_size=page_size, user_info=user_info,
            keyword=body.keyword if body else None,
            contact_name=body.contact_name if body else None,
            mobile=body.mobile if body else None,
            position=body.position if body else None,
            customer_id=body.customer_id if body else None,
            is_key=body.is_key if body else None,
            is_primary=body.is_primary if body else None,
        )
        tp = customer_total_pages(total, page_size)
        return success_response({
            "list": rows,
            "total": total,
            "page": page,
            "pageSize": page_size,
            "totalPages": tp,
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/contact/detail")
def contact_detail(
    body: dict,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """联系人详情"""
    try:
        contact_id = body.get("id")
        if not contact_id:
            return error_response(400, "缺少联系人ID")
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        row = get_contact(db, contact_id)
        if not row:
            return error_response(404, "联系人不存在")
        ensure_crm_contact_row_in_scope(db, row, user_info, tenant_no, message="无权访问该联系人")
        return success_response(contact_row_to_dict(row))
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/contact/byCustomer")
def contact_by_customer(
    body: dict,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """按客户ID查询联系人"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        customer_id = body.get("customerId", "")
        if not customer_id:
            return error_response(400, "缺少客户ID")
        rows = list_contacts_by_customer(db, tenant_no, customer_id, user_info=user_info)
        return success_response(rows)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/contact/add")
def contact_add(
    body: ContactCreateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """新增联系人"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        user_info = crm_user_info(_user)
        obj = create_contact(
            db, tenant_no=tenant_no, user_id=user_id, user_info=user_info,
            **body.model_dump(exclude_unset=True),
        )
        return success_response(contact_row_to_dict(obj), message="创建成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/contact/update")
def contact_update(
    body: ContactUpdateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """更新联系人"""
    try:
        user_id = _user.get("user_id", "") or _user.get("id", "")
        data = body.model_dump(exclude_unset=True)
        contact_id = data.pop("id")
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        obj = update_contact(db, contact_id, user_id=user_id, tenant_no=tenant_no, user_info=user_info, **data)
        return success_response(contact_row_to_dict(obj), message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/contact/delete")
def contact_delete(
    body: ContactDeleteBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """删除联系人"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        for cid in body.ids:
            soft_delete_contact(db, cid, user_id=user_id, tenant_no=tenant_no, user_info=user_info)
        return success_response(None, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/contact/setPrimary")
def contact_set_primary(
    body: dict,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """设置主要联系人"""
    try:
        contact_id = body.get("id")
        if not contact_id:
            return error_response(400, "缺少联系人ID")
        user_id = _user.get("user_id", "") or _user.get("id", "")
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        obj = set_primary_contact(db, contact_id, user_id=user_id, tenant_no=tenant_no, user_info=user_info)
        return success_response(contact_row_to_dict(obj), message="设置成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/contact/export")
def contact_export(
    body: ContactListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """导出联系人列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        rows, _ = list_contacts(
            db, tenant_no, page=1, page_size=10000, user_info=user_info,
            keyword=body.keyword if body else None,
            customer_id=body.customer_id if body else None,
        )
        date_str = datetime.now().strftime("%Y%m%d")
        return export_rows_to_xlsx_response(
            db,
            tenant_no=tenant_no,
            sheet_title="联系人列表",
            columns=CONTACT_EXPORT_COLUMNS,
            rows=rows,
            filename_stem=f"联系人列表_{date_str}",
        )
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/contact/decision")
def contact_decision(
    body: ContactListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """决策人联系人列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        page = body.page if body else 1
        page_size = body.pageSize if body else 10
        rows, total = list_contacts(
            db, tenant_no, page=page, page_size=page_size, user_info=user_info,
            is_decision=True,
            keyword=body.keyword if body else None,
        )
        tp = customer_total_pages(total, page_size)
        return success_response({
            "list": rows,
            "total": total,
            "page": page,
            "pageSize": page_size,
            "totalPages": tp,
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/contact/primary")
def contact_primary(
    body: ContactListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """主联系人列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        page = body.page if body else 1
        page_size = body.pageSize if body else 10
        rows, total = list_contacts(
            db, tenant_no, page=page, page_size=page_size, user_info=user_info,
            is_primary=True,
            keyword=body.keyword if body else None,
        )
        tp = customer_total_pages(total, page_size)
        return success_response({
            "list": rows,
            "total": total,
            "page": page,
            "pageSize": page_size,
            "totalPages": tp,
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/contact/birthday")
def contact_birthday(
    body: ContactListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """本月生日联系人列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        page = body.page if body else 1
        page_size = body.pageSize if body else 10
        rows, total = list_contacts(
            db, tenant_no, page=page, page_size=page_size, user_info=user_info,
            birthday_month=datetime.now().month,
            keyword=body.keyword if body else None,
        )
        tp = customer_total_pages(total, page_size)
        return success_response({
            "list": rows,
            "total": total,
            "page": page,
            "pageSize": page_size,
            "totalPages": tp,
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


# ── 跟进记录 ────────────────────────────────────────────────────


@router.post("/customer/follow/list")
def follow_list(
    body: FollowListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """跟进记录分页列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        page = body.page if body else 1
        page_size = body.pageSize if body else 10
        rows, total = list_follows(
            db, tenant_no, page=page, page_size=page_size, user_info=user_info,
            keyword=body.keyword if body else None,
            customer_id=body.customer_id if body else None,
            follow_type=body.follow_type if body else None,
        )
        tp = customer_total_pages(total, page_size)
        return success_response({
            "list": rows,
            "total": total,
            "page": page,
            "pageSize": page_size,
            "totalPages": tp,
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/follow/detail")
def follow_detail(
    body: dict,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """跟进记录详情"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        follow_id = body.get("id")
        if not follow_id:
            return error_response(400, "缺少跟进记录ID")
        row = get_follow(db, follow_id)
        if not row:
            return error_response(404, "跟进记录不存在")
        ensure_crm_create_by_row_in_scope(
            db, row, user_info, row.tenant_no, message="无权访问该跟进记录"
        )
        return success_response(follow_row_to_dict(row))
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/follow/byCustomer")
def follow_by_customer(
    body: dict,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """按客户ID查询跟进记录"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        customer_id = body.get("customerId", "")
        if not customer_id:
            return error_response(400, "缺少客户ID")
        rows = list_follows_by_customer(db, tenant_no, customer_id, user_info=user_info)
        return success_response(rows)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/follow/add")
def follow_add(
    body: FollowCreateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """新增跟进记录"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        user_name = _user.get("nick_name", "") or _user.get("name", "")
        obj = create_follow(
            db, tenant_no=tenant_no, user_id=user_id, user_name=user_name,
            user_info=user_info,
            **body.model_dump(exclude_unset=True),
        )
        return success_response(follow_row_to_dict(obj), message="创建成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/follow/update")
def follow_update(
    body: FollowUpdateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """更新跟进记录"""
    try:
        user_id = _user.get("user_id", "") or _user.get("id", "")
        data = body.model_dump(exclude_unset=True)
        follow_id = data.pop("id")
        obj = update_follow(db, follow_id, user_id=user_id, tenant_no=get_tenant_no(db, _user), user_info=crm_user_info(_user), **data)
        return success_response(follow_row_to_dict(obj), message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/follow/delete")
def follow_delete(
    body: FollowDeleteBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """删除跟进记录"""
    try:
        for fid in body.ids:
            soft_delete_follow(db, fid, user_info=crm_user_info(_user), tenant_no=get_tenant_no(db, _user))
        return success_response(None, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/follow/export")
def follow_export(
    body: FollowListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """导出跟进记录"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        rows, _ = list_follows(
            db, tenant_no, page=1, page_size=10000,
            keyword=body.keyword if body else None,
            customer_id=body.customer_id if body else None,
        )
        date_str = datetime.now().strftime("%Y%m%d")
        return export_rows_to_xlsx_response(
            db,
            tenant_no=tenant_no,
            sheet_title="跟进记录",
            columns=CUSTOMER_FOLLOW_EXPORT_COLUMNS,
            rows=rows,
            filename_stem=f"跟进记录_{date_str}",
        )
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


# ── 公海池规则管理 ─────────────────────────────────────────────────


@router.post("/customer/pool/rule/list")
def pool_rule_list(
    body: PoolRuleListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        page = body.page if body else 1
        page_size = body.pageSize if body else 10
        rows, total = list_pool_rules(
            db, tenant_no, user_info=user_info,
            page=page, page_size=page_size,
            pool_name=body.poolName if body else None,
            rule_type=body.ruleType if body else None,
            status=body.status if body else None,
        )
        return success_response({
            "list": rows,
            "total": total,
            "page": page,
            "pageSize": page_size,
            "totalPages": pool_rule_total_pages(total, page_size),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/pool/rule/detail")
def pool_rule_detail(
    body: PoolRuleDetailBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        from app.crud.crm.pool_rule import _row_to_dict as pool_rule_row_to_dict
        row = get_pool_rule(db, body.id, tenant_no)
        if not row:
            return error_response(404, "公海池规则不存在")
        ensure_crm_create_by_row_in_scope(db, row, user_info, tenant_no, message="无权访问该公海规则")
        return success_response(pool_rule_row_to_dict(row))
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/pool/rule/add")
def pool_rule_add(
    body: PoolRuleAddBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = get_tenant_no(db, _user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        row = create_pool_rule(
            db, tenant_no,
            user_id=user_id,
            pool_name=body.poolName,
            rule_type=body.ruleType,
            timeout_days=body.timeoutDays,
            protect_days=body.protectDays,
            status=body.status or "enable",
            sort=body.sort or 0,
            remark=body.remark,
        )
        return success_response({"id": row.id}, message="创建成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/pool/rule/update")
def pool_rule_update(
    body: PoolRuleUpdateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        update_pool_rule(
            db, body.id, tenant_no,
            user_id=user_id,
            user_info=user_info,
            pool_name=body.poolName,
            rule_type=body.ruleType,
            timeout_days=body.timeoutDays,
            protect_days=body.protectDays,
            status=body.status,
            sort=body.sort,
            remark=body.remark,
        )
        return success_response(message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/pool/rule/delete")
def pool_rule_delete(
    body: PoolRuleDeleteBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        count = delete_pool_rules(db, body.ids, tenant_no, user_info=user_info)
        return success_response({"count": count}, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/pool/rule/updateStatus")
def pool_rule_update_status(
    body: PoolRuleUpdateStatusBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        update_pool_rule_status(
            db, body.id, tenant_no, body.status, user_id=user_id, user_info=user_info
        )
        return success_response(message="状态更新成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/pool/rule/execute")
def pool_rule_execute(
    body: PoolRuleExecuteBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = get_tenant_no(db, _user)
        count = execute_pool_rules(db, tenant_no)
        return success_response({"recycledCount": count}, message="规则执行完成")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


# ── 统计分析 ────────────────────────────────────────────────────


@router.post("/customer/statistics/overview")
def statistics_overview(
    body: StatisticsBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """客户统计概览"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        data = get_overview_statistics(
            db, tenant_no, user_id=body.user_id if body else None, user_info=user_info
        )
        return success_response(data)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/statistics/source")
def statistics_source(
    body: StatisticsBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """客户来源统计"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        data = get_source_statistics(db, tenant_no, user_info=user_info)
        return success_response(data)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/statistics/level")
def statistics_level(
    body: StatisticsBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """客户级别分布"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        data = get_level_statistics(db, tenant_no, user_info=user_info)
        return success_response(data)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/statistics/industry")
def statistics_industry(
    body: StatisticsBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """客户行业分布"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        data = get_industry_statistics(db, tenant_no, user_info=user_info)
        return success_response(data)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/statistics/trend")
def statistics_trend(
    body: StatisticsBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """成交趋势"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        data = get_trend_statistics(
            db, tenant_no,
            start_date=body.start_date if body else None,
            end_date=body.end_date if body else None,
            user_info=user_info,
        )
        return success_response(data)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/statistics/owner")
def statistics_owner(
    body: StatisticsBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """负责人统计"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        data = get_owner_statistics(db, tenant_no, user_info=user_info)
        return success_response(data)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/customer/statistics/follow")
def statistics_follow(
    body: StatisticsBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """跟进统计"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        data = get_follow_statistics(
            db, tenant_no,
            start_date=body.start_date if body else None,
            end_date=body.end_date if body else None,
            user_info=user_info,
        )
        return success_response(data)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


# ── 商机管理 ──────────────────────────────────────────────────


def _opp_list_kwargs(body: OpportunityListBody, user_id: str, view_type: str = None):
    create_start = body.begin_time
    create_end = body.end_time
    if body.create_time_range:
        if len(body.create_time_range) > 0 and body.create_time_range[0]:
            create_start = body.create_time_range[0]
        if len(body.create_time_range) > 1 and body.create_time_range[1]:
            create_end = body.create_time_range[1]
    return dict(
        page=body.page,
        page_size=body.pageSize,
        view_type=view_type or body.view_type,
        keyword=body.keyword,
        opportunity_no=body.opportunity_no,
        opportunity_name=body.opportunity_name,
        customer_name=body.customer_name,
        customer_id=body.customer_id,
        stage=body.stage,
        status=body.status,
        source=body.source,
        owner_user_id=body.owner_user_id,
        create_time_start=create_start,
        create_time_end=create_end,
        amount_min=body.amount_min or body.min_amount,
        amount_max=body.amount_max or body.max_amount,
    )


@router.post("/opportunity/list")
def opportunity_list(
    body: OpportunityListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """商机分页列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        list_body = body or OpportunityListBody()
        user_id = _user.get("user_id", "") or _user.get("id", "")
        rows, total = list_opportunities(
            db,
            tenant_no,
            user_id,
            user_info=user_info,
            **_opp_list_kwargs(list_body, user_id),
        )
        return success_response({
            "list": rows,
            "rows": rows,
            "total": total,
            "page": list_body.page,
            "pageSize": list_body.pageSize,
            "totalPages": opportunity_total_pages(total, list_body.pageSize),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/mine")
def opportunity_mine_list(
    body: OpportunityListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """我的商机列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        list_body = body or OpportunityListBody()
        user_id = _user.get("user_id", "") or _user.get("id", "")
        rows, total = list_opportunities(
            db,
            tenant_no,
            user_id,
            user_info=user_info,
            **_opp_list_kwargs(list_body, user_id, view_type="mine"),
        )
        return success_response({
            "list": rows,
            "rows": rows,
            "total": total,
            "page": list_body.page,
            "pageSize": list_body.pageSize,
            "totalPages": opportunity_total_pages(total, list_body.pageSize),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/detail")
def opportunity_detail(
    body: OpportunityDetailBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """商机详情"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        from app.models.crm_models import CrmOpportunity

        row = db.query(CrmOpportunity).filter(
            CrmOpportunity.id == body.id,
            CrmOpportunity.del_flag == 0,
            CrmOpportunity.tenant_no == tenant_no,
        ).first()
        if not row:
            return error_response(404, "商机不存在")
        ensure_crm_owner_row_in_scope(db, row, user_info, tenant_no, message="无权访问该商机")
        detail = get_opportunity(db, body.id, tenant_no=tenant_no)
        if not detail:
            return error_response(404, "商机不存在")
        return success_response(detail)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/add")
def opportunity_add(
    body: OpportunityCreateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """新增商机"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        row = create_opportunity(
            db,
            tenant_no,
            body.model_dump(by_alias=False, exclude_unset=True),
            user_id=user_id,
        )
        return success_response(row, message="创建成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/update")
def opportunity_update(
    body: OpportunityUpdateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """编辑商机"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        data = body.model_dump(by_alias=False, exclude_unset=True)
        oid = data.pop("id")
        row = update_opportunity(db, oid, data, user_id=user_id, tenant_no=tenant_no, user_info=user_info)
        return success_response(row, message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/delete")
def opportunity_delete(
    body: OpportunityDeleteBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """删除商机（软删）"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        count = soft_delete_opportunities(
            db, body.ids, user_id=user_id, tenant_no=tenant_no, user_info=user_info
        )
        return success_response({"count": count}, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/assign")
def opportunity_assign(
    body: OpportunityAssignBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """分配商机"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        count = assign_opportunities(
            db,
            body.ids,
            body.owner_user_id,
            user_id=user_id,
            remark=body.remark,
            tenant_no=tenant_no,
            user_info=user_info,
        )
        return success_response({"count": count}, message="分配成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/stage")
def opportunity_change_stage(
    body: OpportunityChangeStageBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """变更商机阶段"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        row = change_stage(
            db,
            body.id,
            body.stage,
            body.stage_percent,
            body.win_rate,
            change_reason=body.change_reason,
            remark=body.remark,
            user_id=user_id,
            tenant_no=tenant_no,
            user_info=user_info,
        )
        return success_response(row, message="阶段已更新")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/win")
def opportunity_mark_won(
    body: OpportunityMarkWonBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """标记赢单"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        row = mark_won(
            db,
            body.id,
            actual_deal_date=body.actual_deal_date,
            actual_amount=body.actual_amount,
            remark=body.remark,
            user_id=user_id,
            tenant_no=tenant_no,
            user_info=user_info,
        )
        return success_response(row, message="已标记赢单")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/lost")
def opportunity_mark_lost(
    body: OpportunityMarkLostBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """标记输单"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        row = mark_lost(
            db,
            body.id,
            body.lost_reason,
            remark=body.remark,
            competitor_info=body.competitor_info,
            improvement=body.improvement,
            user_id=user_id,
            tenant_no=tenant_no,
            user_info=user_info,
        )
        return success_response(row, message="已标记输单")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/batchUpdateStage")
def opportunity_batch_update_stage(
    body: OpportunityBatchUpdateStageBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """批量变更阶段"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        result = batch_update_stage(
            db,
            body.ids,
            body.stage,
            body.stage_percent,
            body.win_rate if body.win_rate is not None else body.stage_percent,
            body.change_reason,
            tenant_no=tenant_no,
            user_id=user_id,
            user_info=user_info,
        )
        return success_response(result)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/batchUpdateExpectDate")
def opportunity_batch_update_expect_date(
    body: OpportunityBatchUpdateExpectDateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """批量修改预计成交日期"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        result = batch_update_expect_date(
            db,
            body.ids,
            body.update_type,
            expect_deal_date=body.expect_deal_date,
            days=body.days,
            tenant_no=tenant_no,
            user_id=user_id,
            user_info=user_info,
        )
        return success_response(result)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/stageLog/list")
def opportunity_stage_log_list(
    body: OpportunityStageLogListBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """商机阶段变更日志"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        logs = get_stage_log_list(db, body.opportunity_id, tenant_no=tenant_no, user_info=user_info)
        return success_response(logs)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/kanban")
def opportunity_kanban(
    body: OpportunityKanbanBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """看板数据"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        list_body = body or OpportunityKanbanBody()
        list_kwargs = _opp_list_kwargs(list_body, "")
        data = get_kanban_data(
            db,
            tenant_no,
            filters={
                "customer_id": list_body.customer_id,
                "source": list_body.source,
                "owner_user_id": list_body.owner_user_id,
                "create_time_start": list_kwargs["create_time_start"],
                "create_time_end": list_kwargs["create_time_end"],
            },
            user_info=user_info,
        )
        return success_response(data)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/statistics/overview")
def opportunity_statistics_overview(
    body: OpportunityStatisticsBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """商机统计概览"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        stat_body = body or OpportunityStatisticsBody()
        data = get_statistics_overview(
            db,
            tenant_no,
            user_info=user_info,
            owner_user_id=stat_body.owner_user_id,
            begin_time=stat_body.begin_time,
            end_time=stat_body.end_time,
        )
        return success_response(data)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/statistics/stage")
def opportunity_statistics_stage(
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """按阶段统计"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        return success_response(get_statistics_by_stage(db, tenant_no, user_info=user_info))
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/statistics/winRate")
def opportunity_statistics_win_rate(
    body: OpportunityStatisticsBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """赢单率趋势"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        stat_body = body or OpportunityStatisticsBody()
        data = get_win_rate_statistics(
            db,
            tenant_no,
            stat_type=stat_body.type or "month",
            begin_time=stat_body.begin_time,
            end_time=stat_body.end_time,
            user_info=user_info,
        )
        return success_response(data)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/statistics/owner")
def opportunity_statistics_owner(
    body: OpportunityStatisticsBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """按负责人统计"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        stat_body = body or OpportunityStatisticsBody()
        data = get_statistics_by_owner(
            db,
            tenant_no,
            begin_time=stat_body.begin_time,
            end_time=stat_body.end_time,
            user_info=user_info,
        )
        return success_response(data)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/statistics/funnel")
def opportunity_statistics_funnel(
    body: OpportunityStatisticsBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """销售漏斗"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        stat_body = body or OpportunityStatisticsBody()
        data = get_sales_funnel(
            db,
            tenant_no,
            owner_user_id=stat_body.owner_user_id,
            begin_time=stat_body.begin_time,
            end_time=stat_body.end_time,
            user_info=user_info,
        )
        return success_response(data)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/follow/list")
def opportunity_follow_list(
    body: OpportunityFollowListBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """商机跟进列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        rows, total = list_opportunity_follows(
            db,
            opportunity_id=body.opportunity_id,
            page=body.page,
            page_size=body.pageSize,
            tenant_no=tenant_no,
            user_info=user_info,
        )
        return success_response({
            "list": rows,
            "rows": rows,
            "total": total,
            "page": body.page,
            "pageSize": body.pageSize,
            "totalPages": opp_follow_total_pages(total, body.pageSize),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/follow/add")
def opportunity_follow_add(
    body: OpportunityFollowCreateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """新增商机跟进"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        row = create_opportunity_follow(
            db,
            tenant_no,
            body.model_dump(by_alias=False, exclude_unset=True),
            user_id=user_id,
            user_info=user_info,
        )
        return success_response(row, message="创建成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/follow/update")
def opportunity_follow_update(
    body: OpportunityFollowUpdateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """编辑商机跟进"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        data = body.model_dump(by_alias=False, exclude_unset=True)
        fid = data.pop("id")
        row = update_opportunity_follow(
            db, fid, data, user_id=user_id, tenant_no=tenant_no, user_info=user_info
        )
        return success_response(row, message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/follow/delete")
def opportunity_follow_delete(
    body: OpportunityFollowDeleteBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """删除商机跟进"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        count = 0
        for fid in body.ids:
            if soft_delete_opportunity_follow(
                db, fid, user_id=user_id, tenant_no=tenant_no, user_info=user_info
            ):
                count += 1
        return success_response({"count": count}, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/follow/detail")
def opportunity_follow_detail(
    body: OpportunityFollowDetailBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """商机跟进详情"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        row = get_opportunity_follow(db, body.id, tenant_no=tenant_no, user_info=user_info)
        if not row:
            return error_response(404, "跟进记录不存在")
        return success_response(row)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/product/list")
def opportunity_product_list(
    body: OpportunityProductListBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """商机产品列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        rows, total = list_opportunity_products(
            db,
            opportunity_id=body.opportunity_id,
            product_id=body.product_id,
            page=body.page,
            page_size=body.pageSize,
            tenant_no=tenant_no,
            user_info=user_info,
        )
        return success_response({
            "list": rows,
            "rows": rows,
            "total": total,
            "page": body.page,
            "pageSize": body.pageSize,
            "totalPages": opp_product_total_pages(total, body.pageSize),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/product/listByOpportunity")
def opportunity_product_list_by_opportunity(
    body: OpportunityProductByOpportunityBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """按商机查询产品列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        opp_id = (body.opportunity_id or body.id) if body else None
        if not opp_id:
            return error_response(400, "缺少商机ID")
        rows = list_products_by_opportunity(db, opp_id, tenant_no=tenant_no, user_info=user_info)
        return success_response(rows)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/product/add")
def opportunity_product_add(
    body: OpportunityProductCreateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """新增商机产品"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        row = create_opportunity_product(
            db,
            tenant_no,
            body.model_dump(by_alias=False, exclude_unset=True),
            user_id=user_id,
            user_info=user_info,
        )
        return success_response(row, message="创建成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/product/batchAdd")
def opportunity_product_batch_add(
    body: OpportunityProductBatchAddBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """批量添加商机产品"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        product_list = [
            p.model_dump(by_alias=False, exclude_unset=True)
            for p in body.product_list
        ]
        count = batch_add_opportunity_products(
            db,
            tenant_no,
            body.opportunity_id,
            product_list,
            user_id=user_id,
            user_info=user_info,
        )
        return success_response({"count": count}, message="批量添加成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/product/update")
def opportunity_product_update(
    body: OpportunityProductUpdateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """编辑商机产品"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        data = body.model_dump(by_alias=False, exclude_unset=True)
        pid = data.pop("id")
        row = update_opportunity_product(
            db, pid, data, user_id=user_id, tenant_no=tenant_no, user_info=user_info
        )
        return success_response(row, message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/product/delete")
def opportunity_product_delete(
    body: OpportunityProductDeleteBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """删除商机产品"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        count = soft_delete_opportunity_products(
            db, body.ids, user_id=user_id, tenant_no=tenant_no, user_info=user_info
        )
        return success_response({"count": count}, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/product/detail")
def opportunity_product_detail(
    body: OpportunityProductDetailBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """商机产品详情"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        row = get_opportunity_product(db, body.id, tenant_no=tenant_no, user_info=user_info)
        if not row:
            return error_response(404, "商机产品不存在")
        return success_response(row)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/opportunity/export")
def opportunity_export(
    body: OpportunityListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """导出商机列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        list_body = body or OpportunityListBody()
        user_id = _user.get("user_id", "") or _user.get("id", "")
        rows, _ = list_opportunities(
            db,
            tenant_no,
            user_id,
            user_info=user_info,
            **_opp_list_kwargs(list_body, user_id),
            page=1,
            page_size=10000,
        )
        return export_rows_to_xlsx_response(
            db,
            tenant_no=tenant_no,
            sheet_title="商机列表",
            columns=OPPORTUNITY_EXPORT_COLUMNS,
            rows=rows,
            filename_stem=f"商机列表_{datetime.now().strftime('%Y%m%d%H%M%S')}",
            include_legacy_filename=False,
        )
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


        return error_response(500, str(e))


# ── 供方信息管理（P0，报价单前置） ─────────────────────────────


@router.post("/supplierInfo/list")
def supplier_info_list(
    body: SupplierInfoListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """分页查询供方列表 — Java: CrmSupplierInfoController.list"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        list_body = body or SupplierInfoListBody()
        rows, total = list_supplier_info(
            db,
            tenant_no,
            user_info=user_info,
            page=list_body.page,
            page_size=list_body.pageSize,
            supplier_name=list_body.supplier_name,
            status=list_body.status,
        )
        return success_response(
            {
                "list": rows,
                "total": total,
                "page": list_body.page,
                "pageSize": list_body.pageSize,
                "totalPages": supplier_info_total_pages(total, list_body.pageSize),
            }
        )
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/supplierInfo/listAll")
def supplier_info_list_all(
    body: SupplierInfoListAllBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """供方全量列表（下拉）— Java: CrmSupplierInfoController.listAll"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        list_body = body or SupplierInfoListAllBody()
        rows = list_all_supplier_info(
            db,
            tenant_no,
            user_info=user_info,
            supplier_name=list_body.supplier_name,
            status=list_body.status,
        )
        return success_response(rows)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/supplierInfo/detail")
def supplier_info_detail(
    body: SupplierInfoDetailBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """供方详情 — Java: CrmSupplierInfoController.getInfo"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        obj = get_supplier_info(db, body.id, tenant_no=tenant_no)
        if not obj:
            return error_response(404, "供方信息不存在")
        ensure_crm_create_by_row_in_scope(
            db, obj, user_info, tenant_no, message="无权访问该供方信息"
        )
        return success_response(supplier_info_row_to_dict(obj))
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/supplierInfo/add")
def supplier_info_add(
    body: SupplierInfoCreateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """新增供方 — Java: CrmSupplierInfoController.add"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        dept_id = _user.get("dept_id", "") or _user.get("create_dept_id", "") or ""
        obj = create_supplier_info(
            db,
            tenant_no=tenant_no,
            user_id=user_id,
            dept_id=dept_id,
            **body.model_dump(exclude_unset=True),
        )
        return success_response(supplier_info_row_to_dict(obj), message="新增成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/supplierInfo/update")
def supplier_info_update(
    body: SupplierInfoUpdateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """修改供方 — Java: CrmSupplierInfoController.edit"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        data = body.model_dump(exclude_unset=True)
        supplier_id = data.pop("id")
        obj = update_supplier_info(
            db,
            supplier_id,
            tenant_no=tenant_no,
            user_id=user_id,
            user_info=user_info,
            **data,
        )
        return success_response(supplier_info_row_to_dict(obj), message="修改成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/supplierInfo/delete")
def supplier_info_delete(
    body: SupplierInfoDeleteBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """删除供方 — Java: CrmSupplierInfoController.remove"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        ids_to_delete: List[str] = []
        if body.ids:
            ids_to_delete = body.ids
        elif body.id:
            ids_to_delete = [body.id]
        else:
            return error_response(400, "请传入 id 或 ids")

        for sid in ids_to_delete:
            soft_delete_supplier_info(db, sid, tenant_no=tenant_no, user_id=user_id)

        return success_response(None, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/supplierInfo/export")
def supplier_info_export(
    body: SupplierInfoListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """导出供方列表 — Java: CrmSupplierInfoController.export"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        list_body = body or SupplierInfoListBody()
        rows, _ = list_supplier_info(
            db,
            tenant_no,
            user_info=user_info,
            page=1,
            page_size=10000,
            supplier_name=list_body.supplier_name,
            status=list_body.status,
        )
        return export_rows_to_xlsx_response(
            db,
            tenant_no=tenant_no,
            sheet_title="供方信息",
            columns=SUPPLIER_INFO_EXPORT_COLUMNS,
            rows=rows,
            filename_stem=f"供方信息_{datetime.now().strftime('%Y%m%d%H%M%S')}",
            include_legacy_filename=False,
        )
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


# ── 客户开票信息 ──────────────────────────────────────────────────


def _invoice_info_by_customer_impl(
    body: InvoiceInfoByCustomerBody,
    _user: dict,
    db: Session,
):
    tenant_no = get_tenant_no(db, _user)
    user_info = crm_user_info(_user)
    rows = list_invoice_info_by_customer(
        db, tenant_no, body.customer_id, user_info=user_info
    )
    return success_response(rows)


@router.post("/invoice-info/list")
def invoice_info_list(
    body: InvoiceInfoByCustomerBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """按客户ID查询开票信息（与 by-customer 等价）"""
    try:
        return _invoice_info_by_customer_impl(body, _user, db)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/invoice-info/by-customer")
def invoice_info_by_customer(
    body: InvoiceInfoByCustomerBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """按客户ID查询开票信息"""
    try:
        return _invoice_info_by_customer_impl(body, _user, db)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/invoice-info/add")
def invoice_info_add(
    body: InvoiceInfoCreateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """新增开票信息"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        obj = create_invoice_info(
            db,
            tenant_no=tenant_no,
            user_id=user_id,
            user_info=user_info,
            **body.model_dump(exclude_unset=True),
        )
        return success_response(invoice_info_row_to_dict(obj), message="创建成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/invoice-info/update")
def invoice_info_update(
    body: InvoiceInfoUpdateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """更新开票信息"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        data = body.model_dump(exclude_unset=True)
        invoice_id = data.pop("id")
        obj = update_invoice_info(
            db,
            invoice_id,
            tenant_no=tenant_no,
            user_id=user_id,
            user_info=user_info,
            **data,
        )
        return success_response(invoice_info_row_to_dict(obj), message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/invoice-info/delete")
def invoice_info_delete(
    body: InvoiceInfoDeleteBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """删除开票信息"""
    try:
        user_id = _user.get("user_id", "") or _user.get("id", "")
        ids = body.ids or []
        if body.id:
            ids.append(body.id)
        if not ids:
            return error_response(400, "缺少开票信息ID")
        for iid in ids:
            soft_delete_invoice_info(db, iid, user_id=user_id)
        return success_response(None, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/invoice-info/set-default")
def invoice_info_set_default(
    body: InvoiceInfoSetDefaultBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """设置默认开票信息"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        obj = set_default_invoice_info(
            db,
            body.id,
            body.customer_id,
            tenant_no=tenant_no,
            user_id=user_id,
        )
        return success_response(invoice_info_row_to_dict(obj), message="设置成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


# ── 线索管理 ──────────────────────────────────────────────────


@router.post("/clue/list")
def clue_list(
    body: ClueListBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """线索列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")

        rows, total = list_clues(
            db,
            tenant_no,
            user_id,
            user_info=user_info,
            page=body.page,
            page_size=body.pageSize,
            keyword=body.keyword or body.search_keyword,
            view_type=body.view_type,
            view_mode=body.view_mode,
            clue_name=body.clue_name,
            company=body.company,
            mobile=body.mobile,
            source=body.source,
            level=body.level,
            status=body.status,
            industry=body.industry,
            park_area=body.park_area,
            owner_user_id=body.owner_user_id,
            create_time_start=(body.create_time_range[0] if body.create_time_range and len(body.create_time_range) > 0 else body.begin_time),
            create_time_end=(body.create_time_range[1] if body.create_time_range and len(body.create_time_range) > 1 else body.end_time),
        )
        return success_response({
            "list": rows,
            "total": total,
            "page": body.page,
            "pageSize": body.pageSize,
            "totalPages": clue_total_pages(total, body.pageSize),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/detail")
def clue_detail(
    body: dict,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """线索详情"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        clue_id = body.get("id")
        if not clue_id:
            return error_response(400, "缺少线索ID")
        user_info = crm_user_info(_user)
        row = get_clue(db, clue_id)
        if not row or row.tenant_no != tenant_no:
            return error_response(404, "线索不存在")
        ensure_crm_owner_row_in_scope(db, row, user_info, tenant_no, message="无权访问该线索")
        detail = clue_row_to_dict(row)
        enrich_clue_user_names(db, [detail])
        return success_response(detail)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/add")
def clue_create(
    body: ClueCreateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """新增线索"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        obj = create_clue(
            db,
            tenant_no=tenant_no,
            data=body.model_dump(by_alias=False, exclude_unset=True),
            user_id=user_id,
        )
        return success_response(clue_row_to_dict(obj), message="创建成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/update")
def clue_update(
    body: ClueUpdateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """编辑线索"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        clue_id = body.id
        data = body.model_dump(by_alias=False, exclude_unset=True)
        data.pop("id", None)
        obj = update_clue(db, clue_id, data, user_id, tenant_no=tenant_no, user_info=user_info)
        return success_response(clue_row_to_dict(obj), message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/delete")
def clue_delete(
    body: ClueDeleteBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """删除线索（支持批量）"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        success_ids = []
        fail_items = []
        for cid in body.ids:
            try:
                soft_delete_clue(db, cid, user_id=user_id, tenant_no=tenant_no, user_info=user_info)
                success_ids.append(cid)
            except ValueError as e:
                fail_items.append({"id": cid, "reason": str(e)})
        result = {
            "successCount": len(success_ids),
            "failCount": len(fail_items),
            "failItems": fail_items,
        }
        if not success_ids and fail_items:
            return error_response(400, fail_items[0]["reason"], data=result)
        return success_response(result, message=f"删除完成，成功{len(success_ids)}个")
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/assign")
def clue_assign(
    body: ClueAssignBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """分配线索"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        count = assign_clues(
            db,
            body.ids,
            body.owner_user_id,
            "",
            current_user_id=user_id,
            remark=body.remark,
            tenant_no=tenant_no,
            user_info=user_info,
        )
        return success_response({"count": count}, message=f"分配成功{count}条")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/convert")
def clue_convert(
    body: ClueConvertBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """线索转化"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        customer_data = body.model_dump(by_alias=False, exclude_unset=True)
        customer_data.pop("id", None)
        result = convert_clue(
            db,
            body.id,
            customer_data,
            current_user_id=user_id,
            tenant_no=tenant_no,
            user_info=user_info,
        )
        return success_response(result, message="转化成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/checkConvert")
def clue_check_convert(
    body: dict,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """检查线索是否可转化"""
    try:
        tenant_no = get_tenant_no(db, _user)
        clue_id = body.get("id")
        if not clue_id:
            return error_response(400, "缺少线索ID")
        result = check_convert(db, clue_id, tenant_no=tenant_no)
        return success_response(result)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/updateStatus")
def clue_update_status(
    body: ClueStatusBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """更新线索状态"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        obj = update_clue_status(
            db, body.id, body.status, user_id, tenant_no=tenant_no, user_info=user_info
        )
        return success_response(clue_row_to_dict(obj), message="状态更新成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/invalid")
def clue_invalid(
    body: ClueInvalidBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """线索标记无效"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        id_list = list(body.ids or [])
        if body.id:
            id_list.append(body.id)
        id_list = list(dict.fromkeys(id_list))
        if not id_list:
            return error_response(400, "缺少线索ID")
        results = []
        for cid in id_list:
            obj = invalid_clue(
                db, cid, user_id, remark=body.remark, tenant_no=tenant_no, user_info=user_info
            )
            results.append(clue_row_to_dict(obj))
        data = results[0] if len(results) == 1 else {"list": results, "count": len(results)}
        return success_response(data, message="标记无效成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/updateLevel")
def clue_update_level(
    body: ClueLevelBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """批量更新线索级别"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        count = update_clue_level(
            db, body.ids, body.level, user_id, tenant_no=tenant_no, user_info=user_info
        )
        return success_response({"count": count}, message=f"更新成功{count}条")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/mine")
def clue_mine(
    body: ClueListBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """我的线索"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")

        rows, total = list_clues(
            db,
            tenant_no,
            user_id,
            user_info=user_info,
            page=body.page,
            page_size=body.pageSize,
            keyword=body.keyword or body.search_keyword,
            view_mode="mine",
            clue_name=body.clue_name,
            company=body.company,
            mobile=body.mobile,
            source=body.source,
            level=body.level,
            status=body.status,
            industry=body.industry,
            park_area=body.park_area,
            owner_user_id=body.owner_user_id,
            create_time_start=(body.create_time_range[0] if body.create_time_range and len(body.create_time_range) > 0 else body.begin_time),
            create_time_end=(body.create_time_range[1] if body.create_time_range and len(body.create_time_range) > 1 else body.end_time),
        )
        return success_response({
            "list": rows,
            "total": total,
            "page": body.page,
            "pageSize": body.pageSize,
            "totalPages": clue_total_pages(total, body.pageSize),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/toFollow")
def clue_to_follow(
    body: ClueListBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """待跟进线索"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")

        rows, total = list_clues(
            db,
            tenant_no,
            user_id,
            user_info=user_info,
            page=body.page,
            page_size=body.pageSize,
            keyword=body.keyword or body.search_keyword,
            view_type="toFollow",
            clue_name=body.clue_name,
            company=body.company,
            mobile=body.mobile,
            source=body.source,
            level=body.level,
            status=body.status,
            industry=body.industry,
            park_area=body.park_area,
            owner_user_id=body.owner_user_id,
            create_time_start=(body.create_time_range[0] if body.create_time_range and len(body.create_time_range) > 0 else body.begin_time),
            create_time_end=(body.create_time_range[1] if body.create_time_range and len(body.create_time_range) > 1 else body.end_time),
        )
        return success_response({
            "list": rows,
            "total": total,
            "page": body.page,
            "pageSize": body.pageSize,
            "totalPages": clue_total_pages(total, body.pageSize),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/todayNew")
def clue_today_new(
    body: ClueListBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """今日新增线索"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")

        rows, total = list_clues(
            db,
            tenant_no,
            user_id,
            user_info=user_info,
            page=body.page,
            page_size=body.pageSize,
            keyword=body.keyword or body.search_keyword,
            view_type="todayNew",
            clue_name=body.clue_name,
            company=body.company,
            mobile=body.mobile,
            source=body.source,
            level=body.level,
            status=body.status,
            industry=body.industry,
            park_area=body.park_area,
            owner_user_id=body.owner_user_id,
            create_time_start=(body.create_time_range[0] if body.create_time_range and len(body.create_time_range) > 0 else body.begin_time),
            create_time_end=(body.create_time_range[1] if body.create_time_range and len(body.create_time_range) > 1 else body.end_time),
        )
        return success_response({
            "list": rows,
            "total": total,
            "page": body.page,
            "pageSize": body.pageSize,
            "totalPages": clue_total_pages(total, body.pageSize),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/parkAreaList")
def clue_park_area_list(
    body: dict = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """线索所在园区列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        rows = get_clue_park_area_list(db, tenant_no)
        return success_response(rows)
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/follow/list")
def clue_follow_list(
    body: ClueFollowListBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """线索跟进记录列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        rows, total = list_clue_follows(
            db,
            body.clue_id,
            page=body.page,
            page_size=body.pageSize,
            user_info=user_info,
            tenant_no=tenant_no,
        )
        enrich_clue_user_names(db, rows)
        return success_response({
            "list": rows,
            "total": total,
            "page": body.page,
            "pageSize": body.pageSize,
            "totalPages": clue_follow_total_pages(total, body.pageSize),
        })
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/follow/add")
def clue_follow_create(
    body: ClueFollowCreateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """新增线索跟进记录"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        user_name = (
            _user.get("real_name", "")
            or _user.get("nick_name", "")
            or _user.get("user_name", "")
            or _user.get("username", "")
        )
        result = create_clue_follow(
            db,
            tenant_no,
            body,
            user_id,
            user_name,
            user_info=user_info,
        )
        return success_response(result, message="跟进记录创建成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/follow/delete")
def clue_follow_delete(
    body: ClueFollowDeleteBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """删除线索跟进记录（支持批量）"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        success_ids = []
        fail_items = []
        for fid in body.ids:
            try:
                ok = delete_clue_follow(db, fid, user_info=user_info, tenant_no=tenant_no)
                if ok:
                    success_ids.append(fid)
                else:
                    fail_items.append({"id": fid, "reason": "记录不存在"})
            except ValueError as e:
                fail_items.append({"id": fid, "reason": str(e)})
        result = {
            "successCount": len(success_ids),
            "failCount": len(fail_items),
            "failItems": fail_items,
        }
        return success_response(result, message=f"删除完成，成功{len(success_ids)}个")
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/follow/statistics")
def clue_follow_statistics(
    body: dict,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """线索跟进统计"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        clue_id = body.get("clueId")
        if not clue_id:
            return error_response(400, "缺少线索ID")
        result = get_clue_follow_statistics(
            db, clue_id, user_info=user_info, tenant_no=tenant_no
        )
        return success_response(result)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


# ── 线索 Phase B：导入导出 / 查重 / 统计 / 跟进详情 ──


_CLUE_IMPORT_HEADERS = [
    "线索姓名",
    "手机号",
    "固定电话",
    "公司名称",
    "客户类型",
    "邮箱",
    "职位",
    "线索来源",
    "线索等级",
    "所属行业",
    "所在园区",
    "省",
    "市",
    "区",
    "详细地址",
    "客户网站",
    "意向产品",
    "备注",
]

_CLUE_IMPORT_FIELDS = [
    "clue_name",
    "mobile",
    "telephone",
    "company",
    "customer_type",
    "email",
    "position",
    "source",
    "level",
    "industry",
    "park_area",
    "province",
    "city",
    "district",
    "address",
    "website",
    "intent_products",
    "remark",
]


@router.post("/clue/export")
def clue_export(
    body: ClueListBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """导出线索列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        list_body = body or ClueListBody()
        rows, _ = list_clues(
            db,
            tenant_no,
            user_id,
            user_info=user_info,
            page=1,
            page_size=10000,
            keyword=list_body.keyword or list_body.search_keyword,
            view_type=list_body.view_type,
            view_mode=list_body.view_mode,
            clue_name=list_body.clue_name,
            company=list_body.company,
            mobile=list_body.mobile,
            source=list_body.source,
            level=list_body.level,
            status=list_body.status,
            industry=list_body.industry,
            park_area=list_body.park_area,
            owner_user_id=list_body.owner_user_id,
            create_time_start=(
                list_body.create_time_range[0]
                if list_body.create_time_range and len(list_body.create_time_range) > 0
                else list_body.begin_time
            ),
            create_time_end=(
                list_body.create_time_range[1]
                if list_body.create_time_range and len(list_body.create_time_range) > 1
                else list_body.end_time
            ),
        )
        date_str = datetime.now().strftime("%Y%m%d")
        return export_rows_to_xlsx_response(
            db,
            tenant_no=tenant_no,
            sheet_title="线索列表",
            columns=CLUE_EXPORT_COLUMNS,
            rows=rows,
            filename_stem=f"线索列表_{date_str}",
        )
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/importTemplate")
def clue_import_template(
    _user: dict = Depends(require_admin_or_super_admin),
):
    """下载线索导入模板"""
    try:
        from fastapi.responses import StreamingResponse
        from openpyxl import Workbook

        wb = Workbook()
        ws = wb.active
        ws.title = "线索导入模板"
        ws.append(_CLUE_IMPORT_HEADERS)
        ws.append([
            "张三", "13800138000", "", "示例公司有限公司", "company", "zhangsan@example.com",
            "技术总监", "website", "A", "互联网", "杭州高新技术产业开发区",
            "浙江省", "杭州市", "西湖区", "文三路xxx号", "https://www.example.com",
            "产品A,产品B", "这是一条示例数据，请参考填写",
        ])

        output = io.BytesIO()
        wb.save(output)
        output.seek(0)
        filename = quote("线索导入模板.xlsx")
        return StreamingResponse(
            output,
            media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
            headers={
                "Content-Disposition": f"attachment; filename={filename}; filename*=UTF-8''{filename}"
            },
        )
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/import")
def clue_import(
    file: UploadFile = File(...),
    updateSupport: bool = Form(False),
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """Excel 导入线索"""
    try:
        from openpyxl import load_workbook

        tenant_no = get_tenant_no(db, _user)
        user_id = _user.get("user_id", "") or _user.get("id", "")

        content = file.file.read()
        wb = load_workbook(io.BytesIO(content), read_only=True)
        ws = wb.active
        all_rows = list(ws.iter_rows(values_only=True))
        wb.close()
        if len(all_rows) < 2:
            return error_response(400, "导入数据为空，请检查文件格式")

        header_row = [str(c).strip() if c is not None else "" for c in all_rows[0]]
        field_map = {}
        for idx, header in enumerate(header_row):
            if header in _CLUE_IMPORT_HEADERS:
                field_map[idx] = _CLUE_IMPORT_FIELDS[_CLUE_IMPORT_HEADERS.index(header)]

        parsed_rows: List[Dict[str, Any]] = []
        for row in all_rows[1:]:
            if not row or all(cell is None or str(cell).strip() == "" for cell in row):
                continue
            item: Dict[str, Any] = {}
            for idx, field in field_map.items():
                val = row[idx] if idx < len(row) else None
                item[field] = str(val).strip() if val is not None else ""
            parsed_rows.append(item)

        result = import_clues(
            db,
            tenant_no,
            parsed_rows,
            user_id=user_id,
            update_support=updateSupport,
        )
        return success_response(result, message=result.get("message", "导入完成"))
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/checkDuplicate")
def clue_check_duplicate(
    body: dict = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """线索重复检测"""
    try:
        tenant_no = get_tenant_no(db, _user)
        payload = body or {}
        duplicates = check_duplicate_clue(
            db,
            tenant_no,
            mobile=payload.get("mobile"),
            company=payload.get("company"),
            exclude_id=payload.get("excludeId"),
        )
        return success_response({"duplicates": duplicates, "isDuplicate": len(duplicates) > 0})
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/follow/detail")
def clue_follow_detail(
    body: dict,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """线索跟进记录详情"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        follow_id = body.get("id")
        if not follow_id:
            return error_response(400, "缺少跟进记录ID")
        detail = get_clue_follow(db, follow_id, user_info=user_info, tenant_no=tenant_no)
        if not detail:
            return error_response(404, "跟进记录不存在")
        enrich_clue_user_names(db, [detail])
        return success_response(detail)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/follow/update")
def clue_follow_update(
    body: ClueFollowUpdateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """更新线索跟进记录"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        result = update_clue_follow(
            db, body.id, body, user_id, user_info=user_info, tenant_no=tenant_no
        )
        return success_response(result, message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/statistics/overview")
def clue_statistics_overview(
    body: ClueStatisticsBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """线索统计概览"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        payload = body or ClueStatisticsBody()
        data = get_clue_overview_statistics(
            db,
            tenant_no,
            user_info=user_info,
            user_id=user_id,
            owner_user_id=payload.owner_user_id,
            begin_time=payload.begin_time,
            end_time=payload.end_time,
        )
        return success_response(data)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/statistics/source")
def clue_statistics_source(
    body: ClueStatisticsBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """线索来源统计"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        payload = body or ClueStatisticsBody()
        data = get_clue_source_statistics(
            db,
            tenant_no,
            begin_time=payload.begin_time,
            end_time=payload.end_time,
            user_info=user_info,
        )
        return success_response(data)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/statistics/level")
def clue_statistics_level(
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """线索等级分布"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        data = get_clue_level_statistics(db, tenant_no, user_info=user_info)
        return success_response(data)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/statistics/trend")
def clue_statistics_trend(
    body: ClueStatisticsBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """线索转化趋势"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        payload = body or ClueStatisticsBody()
        time_unit = payload.time_unit or payload.type or "day"
        data = get_clue_trend_statistics(
            db,
            tenant_no,
            user_info=user_info,
            begin_time=payload.begin_time,
            end_time=payload.end_time,
            time_unit=time_unit,
        )
        return success_response(data)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/clue/statistics/owner")
def clue_statistics_owner(
    body: ClueStatisticsBody = None,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """负责人线索统计"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        payload = body or ClueStatisticsBody()
        data = get_clue_owner_statistics(
            db,
            tenant_no,
            begin_time=payload.begin_time,
            end_time=payload.end_time,
            user_info=user_info,
        )
        return success_response(data)
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


# ── 调价函管理 ────────────────────────────────────────────────

from app.crud.crm.price_letter import (
    _row_to_dict as price_letter_row_to_dict,
    create_price_letter,
    get_price_letter,
    list_price_letters,
    total_pages as price_letter_total_pages,
    update_price_letter,
    void_price_letter,
)
from app.schemas.crm.price_letter import (
    PriceLetterCreateBody,
    PriceLetterDetailBody,
    PriceLetterListBody,
    PriceLetterUpdateBody,
    PriceLetterVoidBody,
)


@router.post("/price-letters/list")
def price_letter_list(
    body: PriceLetterListBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """分页查询调价函列表"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        rows, total = list_price_letters(
            db,
            tenant_no,
            page=body.page,
            page_size=body.pageSize,
            letter_type=body.letter_type,
            customer_name=body.customer_name,
            product_name=body.product_name,
            status=body.status,
            exclude_status=body.exclude_status,
            adjust_date_from=body.adjust_date_from,
            adjust_date_to=body.adjust_date_to,
            keyword=body.keyword,
            user_info=user_info,
        )
        tp = price_letter_total_pages(total, body.pageSize)
        return success_response(
            {
                "list": rows,
                "total": total,
                "page": body.page,
                "pageSize": body.pageSize,
                "totalPages": tp,
            }
        )
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/price-letters/detail")
def price_letter_detail(
    body: PriceLetterDetailBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """调价函详情"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        obj = get_price_letter(db, body.id)
        if not obj or obj.tenant_no != tenant_no:
            return error_response(404, "调价函不存在")
        ensure_crm_create_by_row_in_scope(db, obj, user_info, tenant_no, message="无权访问该调价函")
        return success_response(price_letter_row_to_dict(obj))
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/price-letters/add")
@oper_log(title="调价函管理", business_type=1)
def price_letter_add(
    body: PriceLetterCreateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """新增调价函"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        dept_id = _user.get("dept_id", "") or _user.get("create_dept_id", "") or ""
        kwargs = body.model_dump(exclude_unset=True)
        obj = create_price_letter(
            db,
            tenant_no=tenant_no,
            user_id=user_id,
            dept_id=dept_id,
            **kwargs,
        )
        return success_response(price_letter_row_to_dict(obj), message="创建成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/price-letters/update")
@oper_log(title="调价函管理", business_type=2)
def price_letter_update(
    body: PriceLetterUpdateBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """编辑调价函"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        data = body.model_dump(exclude_unset=True)
        letter_id = data.pop("id", None)
        if not letter_id:
            return error_response(400, "缺少调价函ID")
        obj = update_price_letter(
            db,
            letter_id,
            tenant_no=tenant_no,
            user_id=user_id,
            user_info=user_info,
            **data,
        )
        return success_response(price_letter_row_to_dict(obj), message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


@router.post("/price-letters/void")
@oper_log(title="调价函管理", business_type=2)
def price_letter_void(
    body: PriceLetterVoidBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    """作废调价函（不可删除）"""
    try:
        tenant_no = get_tenant_no(db, _user)
        user_info = crm_user_info(_user)
        user_id = _user.get("user_id", "") or _user.get("id", "")
        obj = void_price_letter(
            db,
            body.id,
            tenant_no=tenant_no,
            invalid_reason=body.invalid_reason,
            remark=body.remark or "",
            user_id=user_id,
            user_info=user_info,
        )
        return success_response(price_letter_row_to_dict(obj), message="作废成功")
    except ValueError as e:
        return error_response(400, str(e))
    except Exception as e:
        return error_response(500, str(e))


from app.api.crm_quote_routes import register_quote_routes
from app.api.crm_contract_routes import register_contract_routes
from app.api.crm_contract_sub_routes import register_contract_sub_routes

register_quote_routes(router)
register_contract_routes(router)
register_contract_sub_routes(router)