shanxia2.0/backend/app/services/column_facet.py

"""列头漏斗可选值（去重 facet）公共拼装。"""
from __future__ import annotations

from typing import Any, Callable, Dict, List, Optional

from sqlalchemy import Column, func
from sqlalchemy.orm import Query, Session


def list_column_facet_values(
    db: Session,
    model: Any,
    column: Column,
    *,
    tenant_no: str,
    keyword: Optional[str] = None,
    limit: int = 500,
    apply_scope: Optional[Callable[[Query], Query]] = None,
    extra_filters: Optional[List[Any]] = None,
) -> List[Dict[str, Any]]:
    """对业务表某列做去重可选值查询。

    约束：tenant_no + del_flag=0；空值剔除；按列值升序；limit 上限 500。
    apply_scope：可选，注入数据权限（如 apply_crm_create_by_scope 包装后的 q 变换）。
    extra_filters：额外 SQLAlchemy 条件（如 status=='on'、公海 view_mode）。
    """
    lim = min(max(1, int(limit or 500)), 500)
    q = db.query(column.label("value"), func.count(model.id).label("cnt")).filter(
        model.tenant_no == tenant_no,
        model.del_flag == 0,
        column.isnot(None),
        column != "",
    )
    if extra_filters:
        for expr in extra_filters:
            if expr is not None:
                q = q.filter(expr)
    if apply_scope is not None:
        q = apply_scope(q)

    kw = (keyword or "").strip()
    if kw:
        q = q.filter(column.like(f"%{kw}%"))

    rows = q.group_by(column).order_by(column.asc()).limit(lim).all()
    out: List[Dict[str, Any]] = []
    for row in rows:
        value = str(row.value).strip() if row.value is not None else ""
        if not value:
            continue
        out.append({"value": value, "label": value, "count": int(row.cnt or 0)})
    return out