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

"""运输计划产品展示：标准名 / 别名拆分（与前端 applyAliasDisplayMeta 对齐）"""
from __future__ import annotations

from typing import Optional

from sqlalchemy.orm import Session

from app.models.crm_models import CrmProduct, CrmProductAlias


def resolve_alias_from_saved_name(
    saved_product_name: Optional[str],
    standard_name: Optional[str],
    alias_options: list[str] | None,
) -> Optional[str]:
    """从已存 productName 反推别名；等于标准名或不在别名列表 → None。"""
    saved = (saved_product_name or "").strip()
    standard = (standard_name or "").strip()
    if not saved or saved == standard:
        return None
    if saved in (alias_options or []):
        return saved
    return None


def load_crm_product_alias_maps(
    db: Session,
    *,
    tenant_no: str,
    product_ids: set[str] | list[str],
) -> tuple[dict[str, str], dict[str, list[str]]]:
    """批量加载 CRM 标准名与别名选项。返回 (standard_name_by_id, alias_options_by_id)。"""
    ids = [pid for pid in product_ids if (pid or "").strip()]
    if not ids:
        return {}, {}

    standard_by_id: dict[str, str] = {}
    rows = (
        db.query(CrmProduct.id, CrmProduct.product_name)
        .filter(
            CrmProduct.id.in_(ids),
            CrmProduct.tenant_no == tenant_no,
            CrmProduct.del_flag == 0,
        )
        .all()
    )
    for row in rows:
        name = (row.product_name or "").strip()
        if name:
            standard_by_id[row.id] = name

    alias_options_by_id: dict[str, list[str]] = {}
    alias_rows = (
        db.query(CrmProductAlias)
        .filter(
            CrmProductAlias.product_id.in_(ids),
            CrmProductAlias.tenant_no == tenant_no,
            CrmProductAlias.del_flag == 0,
        )
        .order_by(CrmProductAlias.order_num.asc(), CrmProductAlias.create_time.asc())
        .all()
    )
    for row in alias_rows:
        name = (row.alias_name or "").strip()
        if not name:
            continue
        bucket = alias_options_by_id.setdefault(row.product_id, [])
        if name not in bucket:
            bucket.append(name)

    return standard_by_id, alias_options_by_id


def resolve_product_display_names(
    *,
    saved_product_name: Optional[str],
    product_id: Optional[str],
    standard_by_id: dict[str, str],
    alias_options_by_id: dict[str, list[str]],
) -> tuple[str, Optional[str]]:
    """
    返回 (展示用产品名称, 别名)。
    产品名称优先 CRM 标准名；别名仅在已存名命中别名列表时返回。
    """
    saved = (saved_product_name or "").strip()
    pid = (product_id or "").strip()
    standard = (standard_by_id.get(pid) or "").strip()
    alias_options = alias_options_by_id.get(pid) or []
    alias = resolve_alias_from_saved_name(saved, standard or None, alias_options)
    display_name = standard or saved
    return display_name, alias


def apply_product_alias_display(
    item: dict,
    *,
    product_id: Optional[str],
    standard_by_id: dict[str, str],
    alias_options_by_id: dict[str, list[str]],
    saved_name_key: str = "productName",
) -> None:
    """就地写入 item 的 productName（标准名）与 aliasName。"""
    saved = item.get(saved_name_key) or item.get("product_name")
    display_name, alias = resolve_product_display_names(
        saved_product_name=saved,
        product_id=product_id,
        standard_by_id=standard_by_id,
        alias_options_by_id=alias_options_by_id,
    )
    item["productName"] = display_name
    item["aliasName"] = alias