"""Completion Wizard for adding new items to a BOM sheet. Three-step guided workflow: 1. Category selection (from schema) 2. Required fields (Description, optional PN) 3. Common fields (Source, Unit Cost, QTY, Sourcing Link, category-specific properties) If a manually entered PN already exists, the PN Conflict Resolution dialog is shown. """ from typing import Any, Dict, List, Optional, Tuple from . import ai_client as _ai from . import dialogs, sync_engine from . import settings as _settings from . import sheet_format as sf from .client import SiloClient # UNO imports try: import uno _HAS_UNO = True _HAS_UNO = True except ImportError: _HAS_UNO = False # Category prefix descriptions for grouping in the picker _PREFIX_GROUPS = { "F": "Fasteners", "C": "Fittings", "R": "Motion", "S": "Structural", "E": "Electrical", "M": "Mechanical", "T": "Tooling", "A": "Assemblies", "P": "Purchased", "X": "Custom Fabricated", } # Default sourcing type by category prefix _DEFAULT_SOURCING = { "A": "M", # assemblies are manufactured "X": "M", # custom fab is manufactured "T": "M", # tooling is manufactured } def _get_categories( client: SiloClient, schema: str = "kindred-rd" ) -> List[Tuple[str, str]]: """Fetch category codes and descriptions from the schema. Returns list of (code, description) tuples sorted by code. """ try: schema_data = client.get_schema(schema) segments = schema_data.get("segments", []) cat_segment = None for seg in segments: if seg.get("name") == "category": cat_segment = seg break if cat_segment and cat_segment.get("values"): return sorted(cat_segment["values"].items()) except RuntimeError: pass return [] def _get_category_properties( client: SiloClient, category: str, schema: str = "kindred-rd" ) -> List[str]: """Fetch property field names relevant to a category. Returns the list of property keys that apply to the category's prefix group. """ try: prop_schema = client.get_property_schema(schema) # prop_schema has global defaults and category-specific overrides defaults = prop_schema.get("defaults", {}) category_props = prop_schema.get("categories", {}).get(category[:1], {}) # Merge: category-specific fields + global defaults all_keys = set(defaults.keys()) all_keys.update(category_props.keys()) return sorted(all_keys) except RuntimeError: return list(sf.PROPERTY_KEY_MAP.values()) # --------------------------------------------------------------------------- # Wizard dialog (UNO) # --------------------------------------------------------------------------- def run_completion_wizard( client: SiloClient, doc, sheet, insert_row: int, project_code: str = "", schema: str = "kindred-rd", ) -> bool: """Run the item completion wizard. Returns True if a row was inserted. Parameters ---------- client : SiloClient doc : XSpreadsheetDocument sheet : XSpreadsheet insert_row : int (0-based row index to insert at) project_code : str (for auto-tagging) schema : str """ if not _HAS_UNO: return False ctx = uno.getComponentContext() smgr = ctx.ServiceManager # -- Step 1: Category selection ----------------------------------------- categories = _get_categories(client, schema) if not categories: dialogs._msgbox( None, "Add Item", "Could not fetch categories from server.", box_type="errorbox", ) return False # Build display list grouped by prefix cat_display = [] for code, desc in categories: prefix = code[0] if code else "?" group = _PREFIX_GROUPS.get(prefix, "Other") cat_display.append(f"{code} - {desc} [{group}]") # Use a simple input box with the category list as hint # (A proper ListBox dialog would be more polished but this is functional) cat_hint = ", ".join(c[0] for c in categories[:20]) if len(categories) > 20: cat_hint += f"... ({len(categories)} total)" category_input = dialogs._input_box( "Add Item - Step 1/3", f"Category code ({cat_hint}):", ) if not category_input: return False category = category_input.strip().upper() # Validate category valid_codes = {c[0] for c in categories} if category not in valid_codes: dialogs._msgbox( None, "Add Item", f"Unknown category: {category}", box_type="errorbox", ) return False # -- Step 2: Required fields -------------------------------------------- description = dialogs._input_box( "Add Item - Step 2/3", "Description (required, leave blank to use AI):", ) # If blank and AI is configured, offer AI generation from seller description if (not description or not description.strip()) and _ai.is_configured(): seller_desc = dialogs._input_box( "Add Item - AI Description", "Paste the seller description for AI generation:", ) if seller_desc and seller_desc.strip(): try: ai_desc = _ai.generate_description( seller_description=seller_desc.strip(), category=category, ) accepted = dialogs.show_ai_description_dialog( seller_desc.strip(), ai_desc ) if accepted: description = accepted except RuntimeError as e: dialogs._msgbox( None, "AI Description Failed", str(e), box_type="errorbox", ) if not description or not description.strip(): dialogs._msgbox( None, "Add Item", "Description is required.", box_type="errorbox" ) return False manual_pn = dialogs._input_box( "Add Item - Step 2/3", "Part number (leave blank for auto-generation):", ) # Check for PN conflict if user entered one use_existing_item = None if manual_pn and manual_pn.strip(): manual_pn = manual_pn.strip() try: existing = client.get_item(manual_pn) # PN exists -- show conflict dialog result = dialogs.show_pn_conflict_dialog(manual_pn, existing) if result == dialogs.PN_USE_EXISTING: use_existing_item = existing elif result == dialogs.PN_CREATE_NEW: manual_pn = "" # will auto-generate else: return False # cancelled except RuntimeError: pass # PN doesn't exist, which is fine # -- Step 3: Common fields ---------------------------------------------- prefix = category[0] if category else "" default_source = _DEFAULT_SOURCING.get(prefix, "P") source = dialogs._input_box( "Add Item - Step 3/3", f"Sourcing type (M=manufactured, P=purchased) [default: {default_source}]:", default=default_source, ) if source is None: return False source = source.strip().upper() or default_source unit_cost_str = dialogs._input_box( "Add Item - Step 3/3", "Unit cost (e.g. 10.50):", default="0", ) unit_cost = 0.0 if unit_cost_str: try: unit_cost = float(unit_cost_str.strip().replace("$", "").replace(",", "")) except ValueError: pass qty_str = dialogs._input_box( "Add Item - Step 3/3", "Quantity [default: 1]:", default="1", ) qty = 1.0 if qty_str: try: qty = float(qty_str.strip()) except ValueError: pass sourcing_link = ( dialogs._input_box( "Add Item - Step 3/3", "Sourcing link (URL, optional):", ) or "" ) # -- Create item or use existing ---------------------------------------- created_item = None if use_existing_item: # Use the existing item's data created_item = use_existing_item final_pn = use_existing_item.get("part_number", manual_pn) elif manual_pn: # Create with the user's manual PN try: created_item = client.create_item( schema=schema, category=category, description=description.strip(), projects=[project_code] if project_code else None, sourcing_type=source, sourcing_link=sourcing_link.strip(), standard_cost=unit_cost if unit_cost else None, ) final_pn = created_item.get("part_number", manual_pn) except RuntimeError as e: dialogs._msgbox(None, "Add Item Failed", str(e), box_type="errorbox") return False else: # Auto-generate PN try: created_item = client.create_item( schema=schema, category=category, description=description.strip(), projects=[project_code] if project_code else None, sourcing_type=source, sourcing_link=sourcing_link.strip(), standard_cost=unit_cost if unit_cost else None, ) final_pn = created_item.get("part_number", "") except RuntimeError as e: dialogs._msgbox(None, "Add Item Failed", str(e), box_type="errorbox") return False if not final_pn: dialogs._msgbox( None, "Add Item", "No part number returned.", box_type="errorbox" ) return False # Auto-tag with project if needed if project_code and created_item and not use_existing_item: try: client.add_item_projects(final_pn, [project_code]) except RuntimeError: pass # -- Insert row into sheet ---------------------------------------------- _insert_bom_row( sheet, insert_row, pn=final_pn, description=created_item.get("description", description.strip()) if created_item else description.strip(), unit_cost=unit_cost, qty=qty, sourcing_link=sourcing_link.strip(), schema=schema, status=sync_engine.STATUS_NEW, parent_pn="", ) return True def _insert_bom_row( sheet, row: int, pn: str, description: str, source: str, unit_cost: float, qty: float, sourcing_link: str, schema: str, status: str, parent_pn: str, ): """Write a single BOM row at the given position with sync tracking.""" from . import pull as _pull # avoid circular import at module level _pull._set_cell_string(sheet, sf.COL_ITEM, row, "") _pull._set_cell_string(sheet, sf.COL_LEVEL, row, "") _pull._set_cell_string(sheet, sf.COL_SOURCE, row, source) _pull._set_cell_string(sheet, sf.COL_PN, row, pn) _pull._set_cell_string(sheet, sf.COL_DESCRIPTION, row, description) _pull._set_cell_string(sheet, sf.COL_SELLER_DESC, row, "") if unit_cost: _pull._set_cell_float(sheet, sf.COL_UNIT_COST, row, unit_cost) _pull._set_cell_float(sheet, sf.COL_QTY, row, qty) # Ext Cost formula ext_formula = f"={sf.col_letter(sf.COL_UNIT_COST)}{row + 1}*{sf.col_letter(sf.COL_QTY)}{row + 1}" _pull._set_cell_formula(sheet, sf.COL_EXT_COST, row, ext_formula) _pull._set_cell_string(sheet, sf.COL_SOURCING_LINK, row, sourcing_link) _pull._set_cell_string(sheet, sf.COL_SCHEMA, row, schema) # Build row cells for hash computation row_cells = [""] * sf.BOM_TOTAL_COLS row_cells[sf.COL_SOURCE] = source row_cells[sf.COL_PN] = pn row_cells[sf.COL_DESCRIPTION] = description row_cells[sf.COL_UNIT_COST] = str(unit_cost) if unit_cost else "" row_cells[sf.COL_QTY] = str(qty) row_cells[sf.COL_SOURCING_LINK] = sourcing_link row_cells[sf.COL_SCHEMA] = schema sync_engine.update_row_sync_state(row_cells, status, parent_pn=parent_pn) _pull._set_cell_string(sheet, sf.COL_ROW_HASH, row, row_cells[sf.COL_ROW_HASH]) _pull._set_cell_string(sheet, sf.COL_ROW_STATUS, row, row_cells[sf.COL_ROW_STATUS]) _pull._set_cell_string(sheet, sf.COL_UPDATED_AT, row, row_cells[sf.COL_UPDATED_AT]) _pull._set_cell_string(sheet, sf.COL_PARENT_PN, row, row_cells[sf.COL_PARENT_PN]) # Colour the row color = _pull._STATUS_COLORS.get(status) if color: _pull._set_row_bg(sheet, row, sf.BOM_TOTAL_COLS, color)