"""Provision Magento collection categories and push PieHS SEO intersection metadata. CLI: python -m app.jobs.magento_collection_landing_repair --connection-id 2 python -m app.jobs.magento_collection_landing_repair --connection-id 2 --apply python -m app.jobs.magento_collection_landing_repair --connection-id 2 --apply --path-slug kitchen-cabinets/anna-natural-rift-white-oak """ from __future__ import annotations import argparse import json import logging import sys from typing import Any, Dict, Optional from db.session import get_session logger = logging.getLogger(__name__) def run_magento_collection_landing_repair( *, connection_id: int, dry_run: bool = True, path_slug: Optional[str] = None, category_l1: Optional[str] = "Kitchen Cabinets", limit: int = 500, baseline_first: bool = False, bootstrap_first: bool = False, ensure_shopping_categories: bool = False, provision_missing_remote: bool = False, only_with_landing_content: bool = False, sync_products: bool = False, sync_shopping_icons: bool = True, apply_listing_paths: bool = True, sync_sample_hub: bool = False, ) -> Dict[str, Any]: """Repair Magento collection landings against locked master taxonomy. Remote category create defaults OFF. While MASTER_TAXONOMY_AUTO_INVENT is frozen, provision flags are coerced off even if the caller opts in. """ from db.collection_landing_push import ( bootstrap_collection_landing_channel, ensure_magento_hub_shopping_categories, push_collection_landing_channel, ) from db.taxonomy_invent_policy import can_provision_remotes, coerce_remote_provision_when_frozen provision_ok = can_provision_remotes() requested_provision = bool(provision_missing_remote) requested_shopping = bool(ensure_shopping_categories) provision_missing_remote, provision_coerced = coerce_remote_provision_when_frozen(requested_provision) if requested_shopping and not provision_ok: ensure_shopping_categories = False if sync_sample_hub and not provision_ok: sync_sample_hub = False result: Dict[str, Any] = { "channel_code": "magento", "connection_id": connection_id, "dry_run": dry_run, "path_slug": path_slug, "category_l1": category_l1, "remote_provision_allowed": provision_ok, } if provision_coerced or (requested_shopping and not provision_ok): result["remote_provision_note"] = ( "Remote category provision is frozen (MASTER_TAXONOMY_AUTO_INVENT=false); " "create/ensure flags coerced off. Link/update existing remotes only." ) with get_session() as session: if baseline_first and not dry_run: from app.jobs.magento_baseline_sync import run_baseline_sync logger.info("Running Magento baseline sync for connection %s", connection_id) result["baseline"] = run_baseline_sync(connection_id, force=True, session=session) if result["baseline"].get("status") == "failed": raise RuntimeError(f"Baseline sync failed: {result['baseline']}") if bootstrap_first: logger.info("Checking Magento collection landing attributes for connection %s", connection_id) result["bootstrap"] = bootstrap_collection_landing_channel( session, channel_code="magento", connection_id=connection_id, dry_run=dry_run, ) if ensure_shopping_categories: hub = category_l1 or "Kitchen Cabinets" logger.info("Ensuring Magento hub shopping categories under %s", hub) result["shopping_categories"] = ensure_magento_hub_shopping_categories( session, connection_id=connection_id, hub_path_slug=hub, dry_run=dry_run, ) logger.info("Pushing collection landing metadata for connection %s", connection_id) result["collections"] = push_collection_landing_channel( session, channel_code="magento", connection_id=connection_id, path_slug=path_slug, category_l1=category_l1, only_with_landing_content=only_with_landing_content, dry_run=dry_run, provision_missing_remote=provision_missing_remote, apply_listing_paths=apply_listing_paths, sync_products=sync_products, sync_shopping_icons=sync_shopping_icons, limit=limit, ) if sync_sample_hub: from db.sample_category import sync_magento_sample_hub logger.info("Ensuring Magento Samples hub and assigning sample products") result["sample_hub"] = sync_magento_sample_hub( session, connection_id=connection_id, dry_run=dry_run, assign_products=True, create_missing_magento=provision_ok, ) if not dry_run: session.commit() return result def _had_errors(result: Dict[str, Any]) -> bool: for key in ("shopping_categories", "collections", "bootstrap", "baseline", "sample_hub"): block = result.get(key) if not isinstance(block, dict): continue if block.get("errors"): return True if block.get("status") == "failed": return True return False def main() -> int: logging.basicConfig(level=logging.INFO) parser = argparse.ArgumentParser( description="Repair Magento collection categories and PieHS SEO intersection metadata", ) parser.add_argument("--connection-id", type=int, required=True, help="Native Magento connection id") parser.add_argument("--dry-run", dest="dry_run", action="store_true", default=True) parser.add_argument("--apply", dest="dry_run", action="store_false", help="Write changes to Magento and DB") parser.add_argument("--path-slug", default=None, help="Optional single collection path slug") parser.add_argument("--category-l1", default="Kitchen Cabinets", help="Hub category label/path") parser.add_argument("--limit", type=int, default=500) parser.add_argument("--baseline-first", action="store_true", help="Pull Magento categories before repair") parser.add_argument("--bootstrap-first", action="store_true", help="Check required hs_* category attributes") parser.add_argument( "--ensure-shopping-categories", dest="ensure_shopping_categories", action="store_true", default=False, help="Create/update hub facet categories (blocked while taxonomy invent is frozen)", ) parser.add_argument( "--provision-missing-remote", dest="provision_missing_remote", action="store_true", default=False, help="Create missing Magento collection categories (blocked while taxonomy invent is frozen)", ) parser.add_argument( "--no-shopping-categories", dest="ensure_shopping_categories", action="store_false", help="Skip hub facet category ensure (default)", ) parser.add_argument( "--no-provision-missing-remote", dest="provision_missing_remote", action="store_false", help="Do not create missing collection categories (default)", ) parser.add_argument( "--only-with-landing-content", action="store_true", help="Skip inferred collection pages with no authored landing content", ) parser.add_argument( "--sync-products", action="store_true", help="Also assign collection SKUs to the Magento category", ) parser.add_argument( "--no-shopping-icons", dest="sync_shopping_icons", action="store_false", help="Skip pushing category icons to shopping facet categories", ) parser.add_argument( "--no-listing-paths", dest="apply_listing_paths", action="store_false", help="Skip applying collection listing path assignments", ) parser.add_argument( "--sync-sample-hub", action="store_true", help="Ensure Magento Samples root category and assign all sample-door SKUs", ) args = parser.parse_args() result = run_magento_collection_landing_repair( connection_id=args.connection_id, dry_run=args.dry_run, path_slug=args.path_slug, category_l1=args.category_l1, limit=args.limit, baseline_first=args.baseline_first, bootstrap_first=args.bootstrap_first, ensure_shopping_categories=args.ensure_shopping_categories, provision_missing_remote=args.provision_missing_remote, only_with_landing_content=args.only_with_landing_content, sync_products=args.sync_products, sync_shopping_icons=args.sync_shopping_icons, apply_listing_paths=args.apply_listing_paths, sync_sample_hub=args.sync_sample_hub, ) print(json.dumps(result, indent=2, default=str)) return 1 if _had_errors(result) else 0 if __name__ == "__main__": sys.exit(main())