sqxp/backend/app/utils/batch_lookup.py

"""批量预取商品/分类/单位，消除明细循环内 N+1。"""
from typing import List, Optional, Tuple

from sqlalchemy.orm import Session
from sqlalchemy import and_

from app.models.models import BasProduct, BasProductCategory, BasProductUnit


def prefetch_products_categories_units(
    db: Session,
    product_ids: List[str],
    extra_unit_ids: Optional[List[str]] = None,
) -> Tuple[dict, dict, dict]:
    """返回 (product_map, category_name_map, unit_name_map)，key 均为 str(id)。"""
    clean_pids = [str(x) for x in (product_ids or []) if x]
    products = []
    if clean_pids:
        products = db.query(BasProduct).filter(
            and_(BasProduct.id.in_(clean_pids), BasProduct.is_del == 0)
        ).all()
    product_map = {str(p.id): p for p in products}

    category_ids = list({str(p.category_id) for p in products if getattr(p, "category_id", None)})
    category_map: dict = {}
    if category_ids:
        categories = db.query(BasProductCategory).filter(
            and_(BasProductCategory.id.in_(category_ids), BasProductCategory.is_del == 0)
        ).all()
        category_map = {str(c.id): str(c.name or "") for c in categories}

    unit_ids = {str(p.unit_id) for p in products if getattr(p, "unit_id", None)}
    if extra_unit_ids:
        unit_ids.update(str(x) for x in extra_unit_ids if x)
    unit_map: dict = {}
    clean_uids = [x for x in unit_ids if x]
    if clean_uids:
        units = db.query(BasProductUnit).filter(
            and_(BasProductUnit.id.in_(clean_uids), BasProductUnit.is_del == 0)
        ).all()
        unit_map = {str(u.id): str(u.name or "") for u in units}

    return product_map, category_map, unit_map


def build_category_path_map(db: Session, category_ids: List[str]) -> dict:
    """批量构建分类路径（一级/二级用 / 连接）。"""
    clean_ids = [str(x).strip() for x in (category_ids or []) if x]
    if not clean_ids:
        return {}
    categories = db.query(BasProductCategory).filter(BasProductCategory.is_del == 0).all()
    by_id = {str(c.id): c for c in categories}
    result: dict = {}
    for cid in clean_ids:
        names: List[str] = []
        cur = cid
        seen = set()
        depth = 0
        while cur and depth < 24 and cur not in seen:
            seen.add(cur)
            cat = by_id.get(cur)
            if not cat:
                break
            names.insert(0, (getattr(cat, "name", None) or "").strip())
            pid = getattr(cat, "parent_id", None)
            cur = str(pid).strip() if pid else ""
            depth += 1
        result[cid] = "/".join(n for n in names if n)
    return result