Hartmut
b87df4a3e5
Move all models/schemas/services/routers into app/domains/. Keep backward-compat shims in old locations for imports. Preserves domains/rendering/tasks.py from Phase A. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
141 lines
4.6 KiB
Python
141 lines
4.6 KiB
Python
"""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 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}
|