#!/usr/bin/env python3 """Inventory <-> supplier-catalog SKU matching (A2). Populates ``inventory.catalog_part_id`` and registers matched supplier SKUs as ``inventory_sku_aliases`` by matching ``inventory.part_number`` against the master ``supplier_catalog.sku`` / ``supplier_catalog_interchange`` (normalized) and, as a fallback, fuzzy name similarity. Usage (as root/postgres, from /home/Autopartes): python3 scripts/match_inventory_catalog.py --tenant 45 --limit 500 python3 scripts/match_inventory_catalog.py --tenant 45 --dry-run python3 scripts/match_inventory_catalog.py --all --batch 2000 """ import argparse import json import os import sys import time sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'pos')) from tenant_db import get_tenant_conn, get_master_conn # noqa: E402 from services.sku_matcher import match_inventory, count_links # noqa: E402 DEFAULT_CHECKPOINT = os.path.join( os.path.dirname(os.path.abspath(__file__)), '..', 'backups', 'match_catalog_checkpoint.json' ) def list_tenants(only_id=None): conn = get_master_conn() cur = conn.cursor() if only_id: cur.execute( "SELECT id, name, db_name FROM tenants WHERE is_active = true AND id = %s", (only_id,), ) else: cur.execute( "SELECT id, name, db_name FROM tenants WHERE is_active = true ORDER BY id" ) rows = cur.fetchall() cur.close() conn.close() return rows def load_checkpoint(path): try: with open(path) as f: return json.load(f) except Exception: return {} def save_checkpoint(path, data): try: os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, 'w') as f: json.dump(data, f, indent=2) except Exception as e: print(f" ! no se pudo guardar checkpoint: {e}") def run_tenant(tenant_id, name, db_name, args, checkpoint): print(f"\n=== tenant {tenant_id} — {name} ({db_name}) ===") tenant_conn = get_tenant_conn(tenant_id) master_conn = get_master_conn() key = str(tenant_id) state = checkpoint.get(key, {}) after_id = None if args.reset else state.get('after_id') before_total, before_linked = count_links(tenant_conn) print(f" antes: total={before_total} linked={before_linked} " f"({(before_linked * 100.0 / before_total) if before_total else 0:.2f}%)") total_scanned = total_matched = total_updated = total_aliases = 0 total_by_method = {} try: while True: call_limit = args.batch if args.limit: call_limit = min(args.batch, max(args.limit - total_scanned, 1)) res = match_inventory( tenant_conn, master_conn, limit=call_limit, offset=0, only_unmatched=args.only_unmatched, create_aliases=not args.no_aliases, use_names=not args.no_names, name_threshold=args.name_threshold, apply=not args.dry_run, supplier=args.supplier, after_id=after_id, sample_limit=args.show, ) total_scanned += res['scanned'] total_matched += res['matched'] total_updated += res['updated'] total_aliases += res['aliases_created'] for k, v in res['by_method'].items(): total_by_method[k] = total_by_method.get(k, 0) + v if args.show and res['sample']: for s in res['sample']: print(f" {s['part_number']:<20} -> cat#{s['matched_catalog_id']} " f"{s['matched_sku']:<14} ({s['method']}, {s['score']})") after_id = res['last_id'] checkpoint[key] = { 'after_id': after_id, 'matched': state.get('matched', 0) + total_matched, 'updated_at': time.strftime('%Y-%m-%dT%H:%M:%S'), } if not args.dry_run: save_checkpoint(args.checkpoint, checkpoint) print(f" batch: scanned={res['scanned']} matched={res['matched']} " f"updated={res['updated']} aliases={res['aliases_created']} " f"methods={res['by_method']}") if res['scanned'] < args.batch or res['scanned'] == 0: break if args.limit and total_scanned >= args.limit: break finally: after_total, after_linked = count_links(tenant_conn) print(f" despues: total={after_total} linked={after_linked} " f"({(after_linked * 100.0 / after_total) if after_total else 0:.2f}%)") print(f" total: scanned={total_scanned} matched={total_matched} " f"updated={total_updated} aliases={total_aliases} " f"methods={total_by_method}") tenant_conn.close() master_conn.close() def main(): p = argparse.ArgumentParser(description='Match inventory to supplier catalog (A2).') p.add_argument('--tenant', type=int, help='tenant id (omit with --all)') p.add_argument('--all', action='store_true', help='process every active tenant') p.add_argument('--batch', type=int, default=5000, help='rows per batch') p.add_argument('--limit', type=int, default=0, help='max rows per tenant (0 = all)') p.add_argument('--only-unmatched', action='store_true', default=True, help='only items with catalog_part_id NULL/0 (default)') p.add_argument('--all-items', dest='only_unmatched', action='store_false', help='consider every active item') p.add_argument('--no-aliases', action='store_true', help='do not create aliases') p.add_argument('--no-names', action='store_true', help='disable fuzzy-name fallback') p.add_argument('--name-threshold', type=float, default=0.82) p.add_argument('--supplier', help='restrict matching to one supplier name') p.add_argument('--dry-run', action='store_true', help='do not persist') p.add_argument('--reset', action='store_true', help='ignore saved checkpoint') p.add_argument('--show', type=int, default=5, help='sample rows to print (0 = none)') p.add_argument('--checkpoint', default=DEFAULT_CHECKPOINT) args = p.parse_args() if not args.all and not args.tenant: p.error('--tenant ID o --all requerido') checkpoint = load_checkpoint(args.checkpoint) tenants = list_tenants(None if args.all else args.tenant) if not tenants: print('sin tenants para procesar') return 1 for tid, name, db_name in tenants: run_tenant(tid, name, db_name, args, checkpoint) return 0 if __name__ == '__main__': sys.exit(main())