205 lines
6.6 KiB
Python
205 lines
6.6 KiB
Python
"""Material alias resolution service.
|
|
|
|
Used from Celery tasks (sync context) to resolve raw material names
|
|
(from Excel / user input) to HARTOMAT 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 HARTOMAT 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 HARTOMAT 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 HARTOMAT 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 hartomat_code
|
|
library_mats = [m for m in mat_rows if m.hartomat_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.hartomat_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,
|
|
"hartomat_code": str(m.hartomat_code),
|
|
}
|
|
for _, m in scored[:5]
|
|
]
|
|
|
|
unmapped.append({"raw_name": raw_name, "suggestions": suggestions})
|
|
|
|
return unmapped
|