"""Reconcile live Magento product category links to repaired master taxonomy targets.""" from __future__ import annotations import argparse import json import logging import time from typing import Any, Dict, Iterable, List, Optional, Set from db.channel_sku_mapping import master_sku_for_channel_sku from db.channel_exports import build_channel_product_payloads from db.models import MagentoConnection from db.product_channel_taxonomy import resolve_magento_category_targets from db.session import get_session from magento.magento_api import MagentoRestClient from magento.oauth_client import MagentoOAuthClient, build_magento_oauth_kwargs logger = logging.getLogger(__name__) def _api_for_connection(conn: MagentoConnection) -> MagentoRestClient: # Category link repair hits many DELETE/POST calls against internal Magento. # Force verify off + fail-fast recovery (do not sleep 5m/10m/30m per SKU). kwargs = build_magento_oauth_kwargs(conn, verify_ssl=False) kwargs["max_recovery_rounds"] = 1 return MagentoRestClient(MagentoOAuthClient(**kwargs)) def _current_category_ids(product: Optional[Dict[str, Any]]) -> Set[int]: if not isinstance(product, dict): return set() out: Set[int] = set() links = ((product.get("extension_attributes") or {}).get("category_links") or []) for link in links: if not isinstance(link, dict): continue try: out.add(int(link.get("category_id"))) except (TypeError, ValueError): pass return out def _expected_category_ids(session, *, master_sku: str, connection_id: int) -> Set[int]: ids, _paths = resolve_magento_category_targets( session, master_sku=master_sku, canonical_fields={}, connection_id=connection_id, ) return {int(value) for value in ids if str(value).strip().isdigit() or isinstance(value, int)} def run_repair( *, connection_id: int, skus: Iterable[str], dry_run: bool = True, keep_category_ids: Optional[Iterable[int]] = None, remove_stale: bool = True, add_missing: bool = True, delay_ms: Optional[int] = None, ) -> Dict[str, Any]: from settings import load_magento_sync_runtime_config wanted = [str(s).strip() for s in skus if str(s).strip()] keep = {int(v) for v in (keep_category_ids or [])} runtime_cfg = load_magento_sync_runtime_config() api_delay_s = max(0, int(delay_ms if delay_ms is not None else runtime_cfg.api_delay_ms)) / 1000.0 with get_session() as session: conn = session.get(MagentoConnection, connection_id) if conn is None: return {"status": "failed", "error": f"Magento connection {connection_id} not found"} api = _api_for_connection(conn) items: List[Dict[str, Any]] = [] removed = 0 added = 0 failed: List[Dict[str, Any]] = [] connection_errors = 0 for idx, channel_sku in enumerate(wanted, start=1): if api_delay_s and idx > 1: time.sleep(api_delay_s) if idx == 1 or idx % 25 == 0 or idx == len(wanted): logger.info( "category-link repair progress %d/%d sku=%s", idx, len(wanted), channel_sku, ) master_sku = master_sku_for_channel_sku( session, channel_sku, "magento", connection_id=connection_id, ) or channel_sku try: product = api.get_product(channel_sku) except Exception as exc: connection_errors += 1 error = {"sku": channel_sku, "error": str(exc)} failed.append(error) items.append( { "sku": channel_sku, "master_sku": master_sku, "found": False, "errors": [error], } ) # Magento TLS often collapses under burst load; cool down briefly. time.sleep(max(api_delay_s, 1.0)) continue current = _current_category_ids(product) expected = _expected_category_ids(session, master_sku=master_sku, connection_id=connection_id) stale = sorted(current - expected - keep) missing = sorted(expected - current) item = { "sku": channel_sku, "master_sku": master_sku, "found": product is not None, "current_category_ids": sorted(current), "expected_category_ids": sorted(expected), "keep_category_ids": sorted(keep), "would_remove": stale, "would_add": missing, "removed": [], "added": [], "errors": [], } if product is None: items.append(item) continue if not dry_run and remove_stale: for category_id in stale: if api_delay_s: time.sleep(api_delay_s) status, err = api.remove_product_from_category(category_id, channel_sku) if status in (200, 204): removed += 1 item["removed"].append(category_id) else: error = { "sku": channel_sku, "category_id": category_id, "status": status, "error": err, } failed.append(error) item["errors"].append(error) if status == 0: connection_errors += 1 time.sleep(max(api_delay_s, 1.0)) if not dry_run and add_missing: for position, category_id in enumerate(missing): if api_delay_s: time.sleep(api_delay_s) status, _body, err = api.assign_product_to_category( category_id, channel_sku, position=position ) if status in (200, 201): added += 1 item["added"].append(category_id) else: error = { "sku": channel_sku, "category_id": category_id, "status": status, "error": err, } failed.append(error) item["errors"].append(error) if status == 0: connection_errors += 1 time.sleep(max(api_delay_s, 1.0)) items.append(item) return { "status": "ok" if not failed else "partial", "dry_run": dry_run, "connection_id": connection_id, "sku_count": len(wanted), "api_delay_ms": int(api_delay_s * 1000), "would_remove_count": sum(len(item.get("would_remove") or []) for item in items), "would_add_count": sum(len(item.get("would_add") or []) for item in items), "removed_count": removed, "added_count": added, "failed_count": len(failed), "connection_error_count": connection_errors, "failed": failed[:50], "items": items, } def resolve_assigned_magento_channel_skus(*, connection_id: int) -> List[str]: with get_session() as session: payloads = build_channel_product_payloads( session, "magento", only_assigned=True, connection_id=connection_id, skip_taxonomy=True, ) return sorted( { str(payload.get("channel_sku") or payload.get("sku") or "").strip() for payload in payloads if str(payload.get("channel_sku") or payload.get("sku") or "").strip() } ) def main() -> int: logging.basicConfig(level=logging.INFO) parser = argparse.ArgumentParser(description="Repair Magento product category links for explicit SKUs") parser.add_argument("--connection-id", type=int, required=True) parser.add_argument("--sku", action="append", dest="skus", default=[]) parser.add_argument( "--all-assigned", action="store_true", help="Repair all assigned Magento SKUs for this connection", ) parser.add_argument("--limit", type=int, default=None, help="Process at most N SKUs") parser.add_argument( "--delay-ms", type=int, default=None, help="Pause between Magento API calls (default: MAGENTO_API_DELAY_MS, usually 200)", ) parser.add_argument("--keep-category-id", action="append", type=int, default=[]) parser.add_argument("--apply", action="store_true") parser.add_argument("--dry-run", action="store_true") parser.add_argument("--no-remove-stale", action="store_true") parser.add_argument("--no-add-missing", action="store_true") args = parser.parse_args() skus = [str(s).strip() for s in args.skus if str(s).strip()] if args.all_assigned: skus = resolve_assigned_magento_channel_skus(connection_id=args.connection_id) if args.limit is not None: skus = skus[: max(0, int(args.limit))] if not skus: parser.error("Provide at least one --sku or use --all-assigned") result = run_repair( connection_id=args.connection_id, skus=skus, dry_run=not args.apply or args.dry_run, keep_category_ids=args.keep_category_id, remove_stale=not args.no_remove_stale, add_missing=not args.no_add_missing, delay_ms=args.delay_ms, ) print(json.dumps(result, indent=2, default=str)) return 0 if result.get("status") == "ok" else 1 if __name__ == "__main__": raise SystemExit(main())