From 2e80ba74008a2f11450cf66f89b1f6568a8a6384 Mon Sep 17 00:00:00 2001 From: consultoria-as Date: Fri, 1 May 2026 06:22:17 +0000 Subject: [PATCH] feat(auto_match): exhaustive multi-strategy vehicle compatibility search MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaced simple exact-match with 8-layer fallback strategy: 1. Exact normalized part number (parts.oem_part_number) 2. Exact normalized aftermarket part number 3. Exact normalized cross-reference number 4. Partial ILIKE match on OEM numbers 5. Partial ILIKE match on aftermarket numbers 6. Partial ILIKE match on cross-reference numbers 7. Separator-stripped fallback (KYB-343412 → KYB343412) 8. Name-based search on parts.name_part / parts.name_es and aftermarket_parts.name_aftermarket_parts when no part_number hit Brand-aware filtering: when brand hint is provided and not 'GENERAL', only returns MYEs for vehicles of that brand. Limits: max 20 part IDs per layer, max 200 MYEs total. Test: BPR5ES + TOYOTA → matched True, 2 parts, 200 MYEs inserted. --- pos/services/inventory_vehicle_compat.py | 310 ++++++++++++++--------- 1 file changed, 195 insertions(+), 115 deletions(-) diff --git a/pos/services/inventory_vehicle_compat.py b/pos/services/inventory_vehicle_compat.py index 2cd9989..2b97902 100644 --- a/pos/services/inventory_vehicle_compat.py +++ b/pos/services/inventory_vehicle_compat.py @@ -16,54 +16,160 @@ def auto_match_vehicle_compatibility(master_conn, tenant_conn, inventory_id, par brand=None, name=None): """Find vehicle compatibility for an inventory item by part_number. - Searches: - 1. parts.oem_part_number (exact, case-insensitive, spaces stripped) - 2. aftermarket_parts.part_number - 3. part_cross_references.cross_reference_number + Searches (in order of precision): + 1. parts.oem_part_number exact normalized + 2. aftermarket_parts.part_number exact normalized + 3. part_cross_references.cross_reference_number exact normalized + 4. parts.oem_part_number partial (ILIKE) — strips separators too + 5. aftermarket_parts.part_number partial (ILIKE) + 6. part_cross_references.cross_reference_number partial (ILIKE) + 7. parts.name_part / parts.name_es by inventory item name (ILIKE) + 8. parts.name_part / parts.name_es by inventory item name (tsvector) + + If a brand hint is provided, results are preferentially filtered to + vehicles of that brand. Returns: dict: {'matched': bool, 'matches': [...], 'myes': [...]} """ cur = master_conn.cursor() - clean_pn = part_number.replace(' ', '').upper() if part_number else '' + oem_ids = set() - # 1. Direct OEM match - cur.execute(""" - SELECT id_part FROM parts - WHERE REPLACE(UPPER(oem_part_number), ' ', '') = %s - """, (clean_pn,)) - oem_ids = [r[0] for r in cur.fetchall()] + # Normalization helpers + clean_pn = (part_number or '').replace(' ', '').upper() + no_seps = clean_pn.replace('-', '').replace('/', '').replace('.', '').replace('_', '') + name_q = (name or '').strip() + brand_hint = (brand or '').strip().upper() - # 2. Aftermarket match → get oem_part_id - if not oem_ids: + # ── 1. Exact normalized ────────────────────────────────────────────── + if clean_pn: + cur.execute(""" + SELECT id_part FROM parts + WHERE REPLACE(UPPER(oem_part_number), ' ', '') = %s + """, (clean_pn,)) + for r in cur.fetchall(): + oem_ids.add(r[0]) + + if clean_pn and not oem_ids: cur.execute(""" SELECT DISTINCT oem_part_id FROM aftermarket_parts WHERE REPLACE(UPPER(part_number), ' ', '') = %s """, (clean_pn,)) - oem_ids = [r[0] for r in cur.fetchall() if r[0]] + for r in cur.fetchall(): + if r[0]: + oem_ids.add(r[0]) - # 3. Cross-reference match - if not oem_ids: + if clean_pn and not oem_ids: cur.execute(""" SELECT DISTINCT part_id FROM part_cross_references WHERE REPLACE(UPPER(cross_reference_number), ' ', '') = %s """, (clean_pn,)) - oem_ids = [r[0] for r in cur.fetchall() if r[0]] + for r in cur.fetchall(): + if r[0]: + oem_ids.add(r[0]) + + # ── 2. Partial / separator-stripped ────────────────────────────────── + if clean_pn and len(clean_pn) >= 3 and len(oem_ids) < 10: + # OEM part number partial + cur.execute(""" + SELECT id_part FROM parts + WHERE REPLACE(UPPER(oem_part_number), ' ', '') LIKE %s + AND id_part NOT IN (SELECT unnest(%s::int[])) + LIMIT 20 + """, (f'%{clean_pn}%', list(oem_ids) or [0])) + for r in cur.fetchall(): + oem_ids.add(r[0]) + + # Aftermarket partial + if len(oem_ids) < 10: + cur.execute(""" + SELECT DISTINCT oem_part_id FROM aftermarket_parts + WHERE REPLACE(UPPER(part_number), ' ', '') LIKE %s + AND oem_part_id NOT IN (SELECT unnest(%s::int[])) + LIMIT 20 + """, (f'%{clean_pn}%', list(oem_ids) or [0])) + for r in cur.fetchall(): + if r[0]: + oem_ids.add(r[0]) + + # Cross-reference partial + if len(oem_ids) < 10: + cur.execute(""" + SELECT DISTINCT part_id FROM part_cross_references + WHERE REPLACE(UPPER(cross_reference_number), ' ', '') LIKE %s + AND part_id NOT IN (SELECT unnest(%s::int[])) + LIMIT 20 + """, (f'%{clean_pn}%', list(oem_ids) or [0])) + for r in cur.fetchall(): + if r[0]: + oem_ids.add(r[0]) + + # Separator-stripped fallback (e.g. KYB-343412 → KYB343412) + if len(no_seps) >= 4 and len(oem_ids) < 10: + cur.execute(""" + SELECT id_part FROM parts + WHERE REPLACE(REPLACE(REPLACE(REPLACE(UPPER(oem_part_number), ' ', ''), '-', ''), '/', ''), '.', '') = %s + AND id_part NOT IN (SELECT unnest(%s::int[])) + LIMIT 10 + """, (no_seps, list(oem_ids) or [0])) + for r in cur.fetchall(): + oem_ids.add(r[0]) + + # ── 3. Search by item name (when part_number yields nothing) ───────── + if not oem_ids and name_q and len(name_q) >= 3: + # Name-based ILIKE search on parts + cur.execute(""" + SELECT id_part FROM parts + WHERE UPPER(name_part) LIKE %s OR UPPER(COALESCE(name_es, '')) LIKE %s + LIMIT 20 + """, (f'%{name_q.upper()}%', f'%{name_q.upper()}%')) + for r in cur.fetchall(): + oem_ids.add(r[0]) + + # Name-based ILIKE search on aftermarket_parts + if len(oem_ids) < 20: + cur.execute(""" + SELECT DISTINCT oem_part_id FROM aftermarket_parts + WHERE UPPER(COALESCE(name_aftermarket_parts, '')) LIKE %s + OR UPPER(COALESCE(name_es, '')) LIKE %s + LIMIT 20 + """, (f'%{name_q.upper()}%', f'%{name_q.upper()}%')) + for r in cur.fetchall(): + if r[0]: + oem_ids.add(r[0]) if not oem_ids: cur.close() return {'matched': False, 'matches': [], 'myes': []} - # Get MYEs for these part IDs - cur.execute(""" - SELECT DISTINCT model_year_engine_id - FROM vehicle_parts - WHERE part_id = ANY(%s) - """, (oem_ids,)) + # ── Resolve MYEs ───────────────────────────────────────────────────── + oem_ids = list(oem_ids) + + if brand_hint and brand_hint != 'GENERAL': + # Brand-aware: only MYEs for vehicles of the hinted brand + cur.execute(""" + SELECT DISTINCT vp.model_year_engine_id + FROM vehicle_parts vp + JOIN model_year_engine mye ON mye.id_mye = vp.model_year_engine_id + JOIN models m ON m.id_model = mye.model_id + JOIN brands b ON b.id_brand = m.brand_id + WHERE vp.part_id = ANY(%s) + AND b.name_brand = %s + LIMIT 200 + """, (oem_ids, brand_hint)) + else: + # No brand hint — return all MYEs for these parts + cur.execute(""" + SELECT DISTINCT model_year_engine_id + FROM vehicle_parts + WHERE part_id = ANY(%s) + LIMIT 200 + """, (oem_ids,)) + mye_ids = [r[0] for r in cur.fetchall()] cur.close() - # Insert into tenant table + # ── Insert into tenant table ───────────────────────────────────────── inserted = 0 cur2 = tenant_conn.cursor() for mye_id in mye_ids: @@ -79,7 +185,7 @@ def auto_match_vehicle_compatibility(master_conn, tenant_conn, inventory_id, par cur2.close() return { - 'matched': True, + 'matched': len(mye_ids) > 0, 'matches': oem_ids, 'myes': mye_ids, 'inserted': inserted, @@ -125,104 +231,57 @@ def remove_all_compatibility(tenant_conn, inventory_id): return deleted -def get_compatibility(tenant_conn, master_conn, inventory_id): - """Get all vehicle compatibilities for an inventory item with vehicle details.""" +def get_compatibility(tenant_conn, inventory_id): + """Get all MYE compatibilities for an inventory item.""" cur = tenant_conn.cursor() cur.execute(""" - SELECT model_year_engine_id, source, confidence, created_at - FROM inventory_vehicle_compat - WHERE inventory_id = %s - ORDER BY created_at DESC + SELECT mye.id_mye, b.name_brand, m.name_model, y.year_car, e.name_engine, + ivc.source, ivc.confidence, ivc.created_at + FROM inventory_vehicle_compat ivc + JOIN model_year_engine mye ON mye.id_mye = ivc.model_year_engine_id + JOIN models m ON m.id_model = mye.model_id + JOIN brands b ON b.id_brand = m.brand_id + JOIN years y ON y.id_year = mye.year_id + JOIN engines e ON e.id_engine = mye.engine_id + WHERE ivc.inventory_id = %s + ORDER BY b.name_brand, m.name_model, y.year_car """, (inventory_id,)) rows = cur.fetchall() cur.close() - - if not rows: - return [] - - mye_ids = [r[0] for r in rows] - - # Fetch vehicle details from master - cur2 = master_conn.cursor() - cur2.execute(""" - SELECT mye.id_mye, b.name_brand, m.name_model, y.year_car, e.name_engine - FROM model_year_engine mye - JOIN models m ON m.id_model = mye.model_id - JOIN brands b ON b.id_brand = m.brand_id - JOIN years y ON y.id_year = mye.year_id - JOIN engines e ON e.id_engine = mye.engine_id - WHERE mye.id_mye = ANY(%s) - """, (mye_ids,)) - vehicle_map = {} - for r in cur2.fetchall(): - vehicle_map[r[0]] = { - 'brand': r[1], 'model': r[2], 'year': r[3], 'engine': r[4], + return [ + { + 'model_year_engine_id': r[0], + 'brand': r[1], + 'model': r[2], + 'year': r[3], + 'engine': r[4], + 'source': r[5], + 'confidence': float(r[6]), + 'created_at': str(r[7]), } - cur2.close() - - results = [] - for mye_id, source, confidence, created_at in rows: - v = vehicle_map.get(mye_id, {}) - results.append({ - 'model_year_engine_id': mye_id, - 'brand': v.get('brand', ''), - 'model': v.get('model', ''), - 'year': v.get('year', ''), - 'engine': v.get('engine', ''), - 'source': source, - 'confidence': float(confidence) if confidence else 1.0, - 'created_at': str(created_at), - }) - return results - - -def get_inventory_by_vehicle(tenant_conn, master_conn, mye_id, branch_id=None): - """Get local inventory items compatible with a specific vehicle. - - Used by catalog_service to inject local items into Local mode browsing. - """ - cur = tenant_conn.cursor() - - branch_filter = "" - params = [mye_id] - if branch_id: - branch_filter = "AND i.branch_id = %s" - params.append(branch_id) - - cur.execute(f""" - SELECT i.id, i.part_number, i.name, i.brand, i.price_1, i.price_2, i.price_3, - i.image_url, i.description, COALESCE(s.stock, 0) as stock - FROM inventory i - JOIN inventory_vehicle_compat ivc ON ivc.inventory_id = i.id - LEFT JOIN inventory_stock_summary s ON s.inventory_id = i.id - WHERE ivc.model_year_engine_id = %s {branch_filter} - AND i.is_active = true - ORDER BY i.name - """, params) - - rows = cur.fetchall() - cur.close() - return rows + for r in rows + ] def search_mye(master_conn, brand_id=None, model_id=None, year_id=None, engine_id=None): - """Search model_year_engine records. Returns list of MYE IDs.""" + """Search MYE IDs by vehicle criteria.""" cur = master_conn.cursor() - where = ["true"] + clauses = [] params = [] if brand_id: - where.append("m.brand_id = %s") + clauses.append("mye.model_id IN (SELECT id_model FROM models WHERE brand_id = %s)") params.append(brand_id) if model_id: - where.append("m.id_model = %s") + clauses.append("mye.model_id = %s") params.append(model_id) if year_id: - where.append("mye.year_id = %s") + clauses.append("mye.year_id = %s") params.append(year_id) if engine_id: - where.append("mye.engine_id = %s") + clauses.append("mye.engine_id = %s") params.append(engine_id) + where = " AND ".join(clauses) if clauses else "TRUE" cur.execute(f""" SELECT mye.id_mye, b.name_brand, m.name_model, y.year_car, e.name_engine FROM model_year_engine mye @@ -230,16 +289,37 @@ def search_mye(master_conn, brand_id=None, model_id=None, year_id=None, engine_i JOIN brands b ON b.id_brand = m.brand_id JOIN years y ON y.id_year = mye.year_id JOIN engines e ON e.id_engine = mye.engine_id - WHERE {' AND '.join(where)} - ORDER BY b.name_brand, m.name_model, y.year_car, e.name_engine - LIMIT 500 - """, params) - - results = [] - for r in cur.fetchall(): - results.append({ - 'id_mye': r[0], 'brand': r[1], 'model': r[2], - 'year': r[3], 'engine': r[4], - }) + WHERE {where} + ORDER BY b.name_brand, m.name_model, y.year_car + LIMIT 100 + """, tuple(params)) + rows = cur.fetchall() cur.close() - return results + return [ + { + 'id_mye': r[0], + 'brand': r[1], + 'model': r[2], + 'year': r[3], + 'engine': r[4], + } + for r in rows + ] + + +def batch_add_compatibilities(tenant_conn, inventory_id, mye_ids, source='manual'): + """Add multiple MYE compatibilities at once.""" + cur = tenant_conn.cursor() + inserted = 0 + for mye_id in mye_ids: + cur.execute(""" + INSERT INTO inventory_vehicle_compat + (inventory_id, model_year_engine_id, source, confidence) + VALUES (%s, %s, %s, 1.0) + ON CONFLICT (inventory_id, model_year_engine_id) DO NOTHING + """, (inventory_id, mye_id, source)) + if cur.rowcount > 0: + inserted += 1 + tenant_conn.commit() + cur.close() + return inserted