#!/usr/bin/env python3 """Index catalog data into Meilisearch for typo-tolerant search. Indexes: - nexus_supplier_catalog (master DB) — supplier SKUs with cross-references - nexus_inventory_{db_name} — per-tenant inventory (optional refresh) Both indexes are configured with: - searchableAttributes including `aliases` (inventory_sku_aliases / supplier interchange numbers → cross-reference search, A6) - a Spanish/English `synonyms` map (curated + services.translations) Usage (as postgres): MASTER_DB_NAME=nexus_autoparts python3 scripts/index_meilisearch.py all python3 scripts/index_meilisearch.py catalog python3 scripts/index_meilisearch.py tenant tenant_capi_autopartes python3 scripts/index_meilisearch.py synonyms # settings only, no reindex Designed to run from a systemd timer after inventory imports/syncs. """ import os import sys sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'pos')) import psycopg2 # noqa: E402 import meilisearch # noqa: E402 from services.meili_search import ( # noqa: E402 MEILI_URL, MEILI_API_KEY, inventory_index_settings, catalog_index_settings, apply_synonyms_to_all, get_synonyms, ) MASTER_DB = os.environ.get('MASTER_DB_NAME', 'nexus_autoparts') CATALOG_INDEX = 'nexus_supplier_catalog' client = meilisearch.Client(MEILI_URL, MEILI_API_KEY) def _catalog_aliases(cur): """catalog_id -> [interchange part numbers] (A6 cross-refs).""" aliases = {} try: cur.execute("SELECT catalog_id, part_number FROM supplier_catalog_interchange") for cid, pn in cur.fetchall(): if cid is not None and pn: aliases.setdefault(cid, []).append(str(pn)) except Exception: cur.connection.rollback() return aliases def index_supplier_catalog(cur): cur.execute(""" SELECT sc.id, sc.sku, sc.name, sc.category, sc.supplier_name, sc.image_url, COALESCE(scp.price, NULL) AS supplier_price FROM supplier_catalog sc LEFT JOIN LATERAL ( SELECT price FROM supplier_catalog_prices p WHERE p.catalog_id = sc.id ORDER BY p.id DESC LIMIT 1 ) scp ON true WHERE sc.is_active """) rows = cur.fetchall() aliases = _catalog_aliases(cur) docs = [ { 'id': r[0], 'sku': r[1], 'name': r[2], 'category': r[3] or '', 'supplier_name': r[4] or '', 'image_url': r[5] or '', 'supplier_price': float(r[6]) if r[6] else None, 'aliases': aliases.get(r[0], []), } for r in rows ] idx = client.index(CATALOG_INDEX) idx.update_settings(catalog_index_settings()) task = idx.add_documents(docs, primary_key='id') client.wait_for_task(task.task_uid) n_alias = sum(1 for d in docs if d['aliases']) print(f'[catalog] {len(docs)} documentos indexados en {CATALOG_INDEX} ' f'({n_alias} con cross-refs)', flush=True) def index_tenant(db_name): conn = psycopg2.connect(dbname=db_name) cur = conn.cursor() cur.execute(""" SELECT i.id, i.part_number, i.name, i.brand, i.price_1, i.image_url, COALESCE(s.stock, 0) FROM inventory i LEFT JOIN ( SELECT inventory_id, SUM(stock) AS stock FROM inventory_stock GROUP BY inventory_id ) s ON s.inventory_id = i.id WHERE i.is_active """) rows = cur.fetchall() aliases = {} try: cur.execute(""" SELECT inventory_id, sku FROM inventory_sku_aliases WHERE is_active AND sku IS NOT NULL """) for inv_id, sku in cur.fetchall(): aliases.setdefault(inv_id, []).append(str(sku)) except Exception: conn.rollback() docs = [ { 'id': r[0], 'part_number': r[1], 'name': r[2], 'brand': r[3] or '', 'price_1': float(r[4]) if r[4] else None, 'image_url': r[5] or '', 'stock': int(r[6] or 0), 'aliases': aliases.get(r[0], []), } for r in rows ] index_name = f"nexus_inventory_{db_name}" idx = client.index(index_name) idx.update_settings(inventory_index_settings()) task = idx.add_documents(docs, primary_key='id') client.wait_for_task(task.task_uid) n_alias = sum(1 for d in docs if d['aliases']) print(f'[{db_name}] {len(docs)} documentos indexados en {index_name} ' f'({n_alias} con aliases)', flush=True) conn.close() def apply_synonyms(): """Push the synonym map to existing indexes without reindexing.""" results = apply_synonyms_to_all() for uid, ok in results.items(): print(f'[synonyms] {uid}: {"ok" if ok is True else ok}', flush=True) print(f'[synonyms] {len(get_synonyms())} términos', flush=True) def main(): mode = sys.argv[1] if len(sys.argv) > 1 else 'all' if mode == 'synonyms': apply_synonyms() return if mode in ('all', 'catalog'): conn = psycopg2.connect(dbname=MASTER_DB) cur = conn.cursor() index_supplier_catalog(cur) cur.close() conn.close() if mode in ('all', 'tenants'): mconn = psycopg2.connect(dbname=MASTER_DB) mcur = mconn.cursor() if len(sys.argv) > 2: dbs = [sys.argv[2]] else: mcur.execute("SELECT db_name FROM tenants WHERE is_active = true") dbs = [r[0] for r in mcur.fetchall()] for db in dbs: try: index_tenant(db) except Exception as e: print(f'[{db}] ERROR: {e}', flush=True) mcur.close() mconn.close() if __name__ == '__main__': main()