Autoparts-DB/scripts/index_meilisearch.py

#!/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()