""" Ticket validation - flags problems for staff to investigate and (manually, in JIRA, with a reason) cancel if confirmed. This module only ever FLAGS; it never writes anything back to JIRA or changes a ticket's status - that stays a deliberate human step. Two rules right now, per the actual business logic: 1. Company mismatch: a ticket's SKUs resolve to more than one company (SH + OK present together). Simple and unambiguous - every SKU on a ticket should belong to the same company. 2. Return/device mismatch: a "Shipping - Return Label/Box X" SKU's implied device type doesn't match any other device SKU on the same ticket - covers both asset-recovery (box + return label, same device) and break-fix (asset + matching-type return label) flows. Some return types are exempt (see RETURN_DEVICE_EXEMPT_KEYWORDS) - emailed labels, DPS Device, and scheduled pickups don't have (or need) a matching device line item at all. Device-type keywords are shared with app.serial_suggestions (DEVICE_FIELD_SUGGESTIONS) rather than duplicated - same vocabulary, different use. """ from __future__ import annotations from typing import List, NamedTuple, Optional from app import config from app.companies import parse_mapping, resolve_company_by_sku from app.serial_suggestions import get_device_field_suggestions DEFAULT_RETURN_DEVICE_EXEMPT_KEYWORDS = "emailed,dps,scheduled pickup,padded envelope" class TicketIssue(NamedTuple): code: str message: str def get_return_device_exempt_keywords() -> list[str]: raw = config.get( "RETURN_DEVICE_EXEMPT_KEYWORDS", DEFAULT_RETURN_DEVICE_EXEMPT_KEYWORDS ) return [k.strip().lower() for k in raw.split(",") if k.strip()] def _is_shipping_item(item_name: str) -> bool: return item_name.strip().lower().startswith("shipping") def check_company_mismatch(skus: List[str], sku_map: Optional[dict] = None) -> Optional[TicketIssue]: if sku_map is None: sku_map = parse_mapping(config.load_settings().get("COMPANY_SKU_MAP", "")) companies = set() for sku in skus or []: company = resolve_company_by_sku(sku, sku_map) if company != "Unknown": companies.add(company) if len(companies) > 1: return TicketIssue( code="company_mismatch", message=f"SKUs from multiple companies on one ticket: {', '.join(sorted(companies))}", ) return None def check_return_device_mismatch( line_items: List[dict], keyword_map: Optional[dict] = None, exempt_keywords: Optional[list] = None, ) -> List[TicketIssue]: """ A return SKU's implied device type needs to match SOMETHING else on the ticket - either an actual device line item (break-fix: send the asset, return the same type) OR another shipping item of the same type (asset recovery: a box and a label for the same device, e.g. SH002 "Return Box iPad" + SH011 "Return Label iPad" - both are "shipping" items by our is-it-a-device check, and pairing them like this is correct, not a mismatch - confirmed against real examples: SH002/SH011, OK001/OK006, OK011/OK013). """ if keyword_map is None: keyword_map = get_device_field_suggestions() # {keyword: [serial fields]} - keys only, here if exempt_keywords is None: exempt_keywords = get_return_device_exempt_keywords() issues: List[TicketIssue] = [] device_items = [item for item in (line_items or []) if not _is_shipping_item(item.get("item_name", ""))] return_items = [item for item in (line_items or []) if _is_shipping_item(item.get("item_name", ""))] for return_item in return_items: return_text = return_item.get("item_name", "").lower() if any(exempt in return_text for exempt in exempt_keywords): continue matched_keyword = next((kw for kw in keyword_map if kw in return_text), None) if matched_keyword is None: # The return SKU doesn't mention any known device type at all - # nothing to check it against, so nothing to flag here either. continue matches_a_device = any( matched_keyword in item.get("item_name", "").lower() for item in device_items ) matches_another_shipping_item = any( matched_keyword in other.get("item_name", "").lower() for other in return_items if other is not return_item ) if not matches_a_device and not matches_another_shipping_item: issues.append( TicketIssue( code="return_device_mismatch", message=( f"'{return_item.get('sku', '')}' ({return_item.get('item_name', '')}) " f"expects a matching '{matched_keyword}' device or box/label pair, " "but none is on this ticket" ), ) ) return issues def validate_ticket( skus: List[str], line_items: List[dict], sku_map: Optional[dict] = None, keyword_map: Optional[dict] = None, exempt_keywords: Optional[list] = None, ) -> List[TicketIssue]: """ Validates a single ticket. The optional pre-fetched params exist so a caller validating MANY tickets at once (the orders table, refreshing on every Import/Pull Tracking/etc.) can read these settings from disk ONCE for the whole batch, rather than once per ticket - each of these is itself a full .env read, and doing that per-ticket rather than per-batch was a real, measured performance bug (2.7s for 300 tickets, now ~0.1s - see make_validation_context()). """ issues: List[TicketIssue] = [] company_issue = check_company_mismatch(skus, sku_map) if company_issue: issues.append(company_issue) issues.extend(check_return_device_mismatch(line_items, keyword_map, exempt_keywords)) return issues def make_validation_context() -> dict: """Fetches everything validate_ticket() needs from Settings ONCE, for passing into repeated validate_ticket() calls across a batch (e.g. every order in the table on a refresh) instead of re-reading .env for every single ticket.""" settings = config.load_settings() return { "sku_map": parse_mapping(settings.get("COMPANY_SKU_MAP", "")), "keyword_map": get_device_field_suggestions(), "exempt_keywords": get_return_device_exempt_keywords(), }