Hartmut/HartOMat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: extract workflow material services phase 3

Browse Source
This commit is contained in:
Hartmut Nörenberg
2026-04-07 09:22:24 +02:00
parent e3cda1c9f7
commit 8f8d2e68b7
5 changed files with 324 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/domains/pipeline/tasks/extract_metadata.py
+3 -66
View File
@@ -176,80 +176,17 @@ def _auto_populate_materials_for_cad(cad_file_id: str, tenant_id: str | None = N
Only fills products where cad_part_materials is empty or all-blank,
preventing overwrites of manually assigned materials.
"""
from sqlalchemy import create_engine, select as sql_select, update as sql_update
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from app.config import settings as app_settings
from app.models.cad_file import CadFile
from app.models.product import Product
from app.api.routers.products import build_materials_from_excel
from app.services.step_processor import build_part_colors
from app.core.tenant_context import set_tenant_context_sync
from app.domains.rendering.workflow_runtime_services import auto_populate_materials_for_cad
sync_url = app_settings.database_url.replace("+asyncpg", "")
eng = create_engine(sync_url)
with Session(eng) as session:
set_tenant_context_sync(session, tenant_id)
# Load the CAD file to get parsed objects
cad_file = session.execute(
sql_select(CadFile).where(CadFile.id == cad_file_id)
).scalar_one_or_none()
if cad_file is None:
return
parsed_objects = cad_file.parsed_objects or {}
cad_parts: list[str] = parsed_objects.get("objects", [])
if not cad_parts:
return
# Find products linked to this CAD file that have Excel components
products = session.execute(
sql_select(Product).where(
Product.cad_file_id == cad_file.id,
Product.is_active.is_(True),
)
).scalars().all()
final_part_colors = None
for product in products:
excel_components: list[dict] = product.components or []
if not excel_components:
continue
# Only auto-fill when cad_part_materials is empty or all-blank
existing = product.cad_part_materials or []
if existing and any(m.get("material", "").strip() for m in existing):
continue # has at least one real material — don't overwrite
new_materials = build_materials_from_excel(cad_parts, excel_components)
session.execute(
sql_update(Product)
.where(Product.id == product.id)
.values(cad_part_materials=new_materials)
)
session.flush()
# Compute part colors; thumbnail queued once after the loop
try:
final_part_colors = build_part_colors(cad_parts, new_materials)
except Exception:
logger.exception(f"Part colors build failed for product {product.id}")
logger.info(
f"Auto-populated {len(new_materials)} materials for product {product.id} "
f"from {len(excel_components)} Excel components"
)
session.commit()
# Queue exactly ONE thumbnail regeneration per CAD file regardless of how many
# products were auto-populated. Queuing once-per-product multiplies the task
# count needlessly and causes the Redis queue depth to grow instead of shrink.
if final_part_colors is not None:
try:
from app.domains.pipeline.tasks.render_thumbnail import regenerate_thumbnail
regenerate_thumbnail.delay(str(cad_file_id), final_part_colors)
except Exception:
logger.exception(f"Thumbnail regen queue failed for cad_file {cad_file_id}")
auto_populate_materials_for_cad(session, cad_file_id)
eng.dispose()
backend/app/domains/rendering/workflow_runtime_services.py
+169 -30
View File
@@ -26,6 +26,7 @@ logger = logging.getLogger(__name__)
EmitFn = Callable[..., None] | None
SetupStatus = Literal["ready", "skip", "failed", "missing"]
QueueThumbnailFn = Callable[[str, dict[str, str]], None] | None
@dataclass(slots=True)
@@ -68,6 +69,24 @@ class TemplateResolutionResult:
output_type_id: str | None
@dataclass(slots=True)
class MaterialResolutionResult:
material_map: dict[str, str] | None
use_materials: bool
override_material: str | None
source_material_count: int = 0
resolved_material_count: int = 0
@dataclass(slots=True)
class AutoPopulateMaterialsResult:
cad_file_id: str
updated_product_ids: list[str] = field(default_factory=list)
queued_thumbnail_regeneration: bool = False
part_colors: dict[str, str] | None = None
cad_parts: list[str] = field(default_factory=list)
def _emit(emit: EmitFn, order_line_id: str, message: str, level: str | None = None) -> None:
if emit is None:
return
@@ -270,34 +289,14 @@ def resolve_order_line_template_context(
output_type_id=output_type_id,
)
material_library = get_material_library_path_for_session(session)
material_map = None
use_materials = bool(material_library and materials_source)
if template and not template.material_replace_enabled:
use_materials = False
if use_materials:
material_map = {
material["part_name"]: material["material"]
for material in materials_source
if material.get("part_name") and material.get("material")
}
material_map = resolve_material_map(material_map)
line_override = getattr(line, "material_override", None)
output_override = line.output_type.material_override if line.output_type else None
override_material = line_override or output_override
if override_material:
override_keys = set(material_map.keys()) if material_map else set()
if cad_file and cad_file.parsed_objects:
for part_name in cad_file.parsed_objects.get("objects", []):
override_keys.add(part_name)
material_map = {key: override_material for key in override_keys}
use_materials = True
_emit(
emit,
str(line.id),
f"Material override active: {len(material_map)} parts → {override_material}",
)
material_resolution = resolve_order_line_material_map(
line,
cad_file,
materials_source,
material_library=material_library,
template=template,
emit=emit,
)
if template:
_emit(
@@ -325,11 +324,151 @@ def resolve_order_line_template_context(
return TemplateResolutionResult(
template=template,
material_library=material_library,
material_map=material_resolution.material_map,
use_materials=material_resolution.use_materials,
override_material=material_resolution.override_material,
category_key=category_key,
output_type_id=output_type_id,
)
def resolve_order_line_material_map(
line: OrderLine,
cad_file: CadFile | None,
materials_source: list[dict[str, Any]],
*,
material_library: str | None,
template: RenderTemplate | None,
emit: EmitFn = None,
) -> MaterialResolutionResult:
"""Resolve the effective order-line material map with legacy precedence rules."""
material_map = None
raw_material_count = 0
use_materials = bool(material_library and materials_source)
if template and not template.material_replace_enabled:
use_materials = False
if use_materials:
material_map = {
material["part_name"]: material["material"]
for material in materials_source
if material.get("part_name") and material.get("material")
}
raw_material_count = len(material_map)
material_map = resolve_material_map(material_map)
line_override = getattr(line, "material_override", None)
output_override = line.output_type.material_override if line.output_type else None
override_material = line_override or output_override
if override_material:
override_keys = set(material_map.keys()) if material_map else set()
if cad_file and cad_file.parsed_objects:
for part_name in cad_file.parsed_objects.get("objects", []):
override_keys.add(part_name)
material_map = {key: override_material for key in override_keys}
use_materials = True
_emit(
emit,
str(line.id),
f"Material override active: {len(material_map)} parts → {override_material}",
)
return MaterialResolutionResult(
material_map=material_map,
use_materials=use_materials,
override_material=override_material,
category_key=category_key,
output_type_id=output_type_id,
source_material_count=raw_material_count,
resolved_material_count=len(material_map or {}),
)
def auto_populate_materials_for_cad(
session: Session,
cad_file_id: str,
*,
enqueue_thumbnail: QueueThumbnailFn = None,
) -> AutoPopulateMaterialsResult:
"""Auto-fill empty CAD material mappings from Excel component data.
This preserves the legacy rules:
- only products with empty/all-blank `cad_part_materials` are updated
- thumbnail regeneration is queued at most once per CAD file
- the queued part-color map comes from the last updated product
"""
from app.api.routers.products import build_materials_from_excel
cad_file = session.execute(
select(CadFile).where(CadFile.id == cad_file_id)
).scalar_one_or_none()
if cad_file is None:
return AutoPopulateMaterialsResult(cad_file_id=str(cad_file_id))
parsed_objects = cad_file.parsed_objects or {}
cad_parts: list[str] = parsed_objects.get("objects", [])
if not cad_parts:
return AutoPopulateMaterialsResult(
cad_file_id=str(cad_file_id),
cad_parts=[],
)
products = session.execute(
select(Product).where(
Product.cad_file_id == cad_file.id,
Product.is_active.is_(True),
)
).scalars().all()
updated_product_ids: list[str] = []
final_part_colors: dict[str, str] | None = None
for product in products:
excel_components: list[dict[str, Any]] = product.components or []
if not excel_components:
continue
existing = product.cad_part_materials or []
if existing and any((entry.get("material") or "").strip() for entry in existing):
continue
new_materials = build_materials_from_excel(cad_parts, excel_components)
session.execute(
sql_update(Product)
.where(Product.id == product.id)
.values(cad_part_materials=new_materials)
)
session.flush()
updated_product_ids.append(str(product.id))
try:
final_part_colors = build_part_colors(cad_parts, new_materials)
except Exception:
logger.exception("Part colors build failed for product %s", product.id)
logger.info(
"Auto-populated %d materials for product %s from %d Excel components",
len(new_materials),
product.id,
len(excel_components),
)
session.commit()
queued_thumbnail_regeneration = False
if final_part_colors is not None:
if enqueue_thumbnail is None:
from app.domains.pipeline.tasks.render_thumbnail import regenerate_thumbnail
enqueue_thumbnail = lambda current_cad_file_id, part_colors: regenerate_thumbnail.delay( # noqa: E731
current_cad_file_id,
part_colors,
)
enqueue_thumbnail(str(cad_file_id), final_part_colors)
queued_thumbnail_regeneration = True
return AutoPopulateMaterialsResult(
cad_file_id=str(cad_file_id),
updated_product_ids=updated_product_ids,
queued_thumbnail_regeneration=queued_thumbnail_regeneration,
part_colors=final_part_colors,
cad_parts=cad_parts,
)
backend/tests/domains/test_workflow_runtime_services.py
+149
View File
@@ -16,7 +16,9 @@ from app.domains.orders.models import Order, OrderLine, OrderStatus
from app.domains.products.models import CadFile, Product
from app.domains.rendering.models import OutputType, RenderTemplate
from app.domains.rendering.workflow_runtime_services import (
auto_populate_materials_for_cad,
prepare_order_line_render_context,
resolve_order_line_material_map,
resolve_order_line_template_context,
)
@@ -216,3 +218,150 @@ def test_resolve_order_line_template_context_uses_exact_template_and_override(sy
"InnerRing": "HARTOMAT_OVERRIDE",
"OuterRing": "HARTOMAT_OVERRIDE",
}
def test_resolve_order_line_material_map_disables_materials_when_template_blocks_replacement(
sync_session,
tmp_path,
monkeypatch,
):
from app.config import settings
monkeypatch.setattr(settings, "upload_dir", str(tmp_path / "uploads"))
line = _seed_order_line_graph(sync_session, tmp_path)
template = RenderTemplate(
id=uuid.uuid4(),
name="Lighting Only",
category_key="bearings",
blend_file_path="/templates/lighting-only.blend",
original_filename="lighting-only.blend",
target_collection="Product",
material_replace_enabled=False,
lighting_only=True,
is_active=True,
)
result = resolve_order_line_material_map(
line,
line.product.cad_file,
line.product.cad_part_materials,
material_library="/libraries/materials.blend",
template=template,
)
assert result.use_materials is False
assert result.material_map is None
assert result.override_material is None
def test_resolve_order_line_material_map_prefers_line_override_over_output_override(
sync_session,
tmp_path,
monkeypatch,
):
from app.config import settings
monkeypatch.setattr(settings, "upload_dir", str(tmp_path / "uploads"))
line = _seed_order_line_graph(sync_session, tmp_path)
line.material_override = "LINE_OVERRIDE"
line.output_type.material_override = "OUTPUT_OVERRIDE"
sync_session.commit()
result = resolve_order_line_material_map(
line,
line.product.cad_file,
line.product.cad_part_materials,
material_library="/libraries/materials.blend",
template=None,
)
assert result.override_material == "LINE_OVERRIDE"
assert result.use_materials is True
assert result.material_map == {
"InnerRing": "LINE_OVERRIDE",
"OuterRing": "LINE_OVERRIDE",
}
def test_auto_populate_materials_for_cad_updates_only_blank_products_and_queues_once(sync_session, tmp_path):
line = _seed_order_line_graph(sync_session, tmp_path)
cad_file = line.product.cad_file
assert cad_file is not None
line.product.components = [
{"part_name": "InnerRing", "material": "Steel"},
{"part_name": "OuterRing", "material": "Rubber"},
]
line.product.cad_part_materials = []
second_product = Product(
id=uuid.uuid4(),
pim_id="P-2000",
name="Bearing B",
category_key="bearings",
cad_file_id=cad_file.id,
cad_file=cad_file,
components=[
{"part_name": "InnerRing", "material": "Brass"},
{"part_name": "OuterRing", "material": "Copper"},
],
cad_part_materials=[{"part_name": "InnerRing", "material": "Existing"}],
)
sync_session.add(second_product)
sync_session.commit()
queued: list[tuple[str, dict[str, str]]] = []
result = auto_populate_materials_for_cad(
sync_session,
str(cad_file.id),
enqueue_thumbnail=lambda current_cad_file_id, part_colors: queued.append(
(current_cad_file_id, part_colors)
),
)
sync_session.refresh(line.product)
sync_session.refresh(second_product)
assert result.updated_product_ids == [str(line.product.id)]
assert result.queued_thumbnail_regeneration is True
assert result.part_colors == {"InnerRing": "Steel", "OuterRing": "Rubber"}
assert queued == [(str(cad_file.id), {"InnerRing": "Steel", "OuterRing": "Rubber"})]
assert line.product.cad_part_materials == [
{"part_name": "InnerRing", "material": "Steel"},
{"part_name": "OuterRing", "material": "Rubber"},
]
assert second_product.cad_part_materials == [{"part_name": "InnerRing", "material": "Existing"}]
def test_auto_populate_materials_for_cad_skips_when_materials_already_present(sync_session, tmp_path):
line = _seed_order_line_graph(sync_session, tmp_path)
cad_file = line.product.cad_file
assert cad_file is not None
line.product.components = [
{"part_name": "InnerRing", "material": "Steel"},
{"part_name": "OuterRing", "material": "Rubber"},
]
sync_session.commit()
queued: list[tuple[str, dict[str, str]]] = []
result = auto_populate_materials_for_cad(
sync_session,
str(cad_file.id),
enqueue_thumbnail=lambda current_cad_file_id, part_colors: queued.append(
(current_cad_file_id, part_colors)
),
)
sync_session.refresh(line.product)
assert result.updated_product_ids == []
assert result.queued_thumbnail_regeneration is False
assert result.part_colors is None
assert queued == []
assert line.product.cad_part_materials == [
{"part_name": "InnerRing", "material": "Steel raw"},
{"part_name": "OuterRing", "material": "Steel raw"},
]