HartOMat/backend/app/domains/materials/service.py

"""Material alias resolution service.

Used from Celery tasks (sync context) to resolve raw material names
(from Excel / user input) to SCHAEFFLER library material names via aliases.

Resolution chain:
1. Alias lookup (case-insensitive) → use alias.material.name
2. Exact Material.name match (case-insensitive) → use it
3. Pass through unchanged → Blender will show FailedMaterial magenta
"""
import logging
from difflib import SequenceMatcher

from sqlalchemy import create_engine, select, func
from sqlalchemy.orm import Session, selectinload
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.materials.models import Material, MaterialAlias

logger = logging.getLogger(__name__)

_engine = None


def _get_engine():
    global _engine
    if _engine is None:
        from app.config import settings as app_settings
        _engine = create_engine(app_settings.database_url_sync)
    return _engine


def resolve_material_map(raw_map: dict[str, str]) -> dict[str, str]:
    """Resolve raw material names to SCHAEFFLER library names via aliases.

    For each value in raw_map:
    1. Alias lookup (case-insensitive) → return alias.material.name
    2. Exact Material.name match (case-insensitive) → use canonical name
    3. Pass through unchanged

    Returns a new dict with the same keys but resolved material names.
    """
    if not raw_map:
        return raw_map

    engine = _get_engine()
    with Session(engine) as session:
        # Load all materials
        materials = session.execute(
            select(Material).options(selectinload(Material.aliases))
        ).scalars().all()

        # Build lookup dicts (case-insensitive)
        # material name (lower) → canonical Material.name
        name_lookup: dict[str, str] = {}
        # alias (lower) → Material.name
        alias_lookup: dict[str, str] = {}

        for mat in materials:
            name_lookup[mat.name.lower()] = mat.name
            for a in mat.aliases:
                alias_lookup[a.alias.lower()] = mat.name

    resolved = {}
    for part_name, raw_material in raw_map.items():
        raw_lower = raw_material.lower()

        # 1. Alias lookup first — aliases explicitly map intermediate/display names
        #    to the canonical SCHAEFFLER library names
        if raw_lower in alias_lookup:
            target = alias_lookup[raw_lower]
            logger.info("resolved '%s' → '%s' (alias match)", raw_material, target)
            resolved[part_name] = target
            continue

        # 2. Exact material name match (canonical name used as-is)
        if raw_lower in name_lookup:
            canonical = name_lookup[raw_lower]
            if canonical != raw_material:
                logger.info("resolved '%s' → '%s' (exact name match)", raw_material, canonical)
            resolved[part_name] = canonical
            continue

        # 3. Pass through unchanged
        logger.warning("no material match for '%s' — will use FailedMaterial fallback", raw_material)
        resolved[part_name] = raw_material

    return resolved


async def seed_material_aliases_from_mappings(
    db: AsyncSession, mappings: list[dict]
) -> dict:
    """Seed material aliases from Excel materialmapping sheet.

    For each {display_name, render_name}:
    - Find or create Material by render_name
    - Add display_name as alias if not already present

    Returns {"created": N, "skipped": N}.
    """
    created = 0
    skipped = 0

    for mapping in mappings:
        display_name = mapping.get("display_name", "").strip()
        render_name = mapping.get("render_name", "").strip()
        if not display_name or not render_name:
            skipped += 1
            continue

        # Find or create Material by render_name
        result = await db.execute(
            select(Material).where(func.lower(Material.name) == render_name.lower())
        )
        material = result.scalar_one_or_none()
        if material is None:
            material = Material(name=render_name, source="excel_mapping")
            db.add(material)
            await db.flush()

        # Check if alias already exists
        alias_result = await db.execute(
            select(MaterialAlias).where(
                func.lower(MaterialAlias.alias) == display_name.lower()
            )
        )
        existing_alias = alias_result.scalar_one_or_none()
        if existing_alias:
            skipped += 1
            continue

        # Create alias
        alias = MaterialAlias(material_id=material.id, alias=display_name)
        db.add(alias)
        created += 1

    if created > 0:
        await db.flush()

    return {"created": created, "skipped": skipped}


async def find_unmapped_materials(
    material_names: list[str], db: AsyncSession
) -> list[dict]:
    """Find material names that have no alias or library match.

    Returns a list of {"raw_name": str, "suggestions": [...]} for each
    unmapped name. Suggestions are the top 5 SCHAEFFLER library materials
    by string similarity.
    """
    if not material_names:
        return []

    # Load all aliases (case-insensitive lookup)
    alias_rows = (await db.execute(select(MaterialAlias))).scalars().all()
    alias_set: set[str] = {a.alias.lower() for a in alias_rows}

    # Load all materials
    mat_rows = (await db.execute(select(Material))).scalars().all()
    # Library materials have a schaeffler_code
    library_mats = [m for m in mat_rows if m.schaeffler_code is not None]
    # All material names (case-insensitive) for exact-match check
    name_lookup: dict[str, Material] = {m.name.lower(): m for m in mat_rows}

    unmapped: list[dict] = []
    seen: set[str] = set()

    for raw_name in material_names:
        raw_lower = raw_name.lower()
        if raw_lower in seen:
            continue
        seen.add(raw_lower)

        # 1. Alias match → mapped
        if raw_lower in alias_set:
            continue

        # 2. Exact name match with a library material → mapped
        matched_mat = name_lookup.get(raw_lower)
        if matched_mat and matched_mat.schaeffler_code is not None:
            continue

        # Unmapped — compute suggestions from library materials
        scored = []
        for lib_mat in library_mats:
            ratio = SequenceMatcher(None, raw_lower, lib_mat.name.lower()).ratio()
            if ratio > 0.3:
                scored.append((ratio, lib_mat))
        scored.sort(key=lambda x: x[0], reverse=True)

        suggestions = [
            {
                "id": str(m.id),
                "name": m.name,
                "schaeffler_code": str(m.schaeffler_code),
            }
            for _, m in scored[:5]
        ]

        unmapped.append({"raw_name": raw_name, "suggestions": suggestions})

    return unmapped