"""Reconcile Magento galleries to current master images, then force re-push. Use when Magento still shows wrong/stale collection vignettes after master is clean (append-only media sync leaves orphans; map-only cleanup can miss unmapped entries). Examples: python -m app.jobs.magento_media_reconcile --connection-id 1 --skus ASW-BEC1224L --dry-run python -m app.jobs.magento_media_reconcile --connection-id 1 --skus ASW-BEC1224L,ASW-BEC1224R --apply --push --wait """ from __future__ import annotations import argparse import json import logging import sys from typing import Any, Dict, List, Optional, Sequence logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def _desired_urls_for_sku(session, sku: str) -> List[str]: from db.master_product_images import image_fields_by_sku fields = image_fields_by_sku(session, [sku]).get(sku) or {} urls: List[str] = [] base = str(fields.get("base_image") or "").strip() addl_raw = str(fields.get("additional_images") or "").strip() if base: urls.append(base) if addl_raw: urls.extend([part.strip() for part in addl_raw.split(",") if part.strip()]) # stable unique seen = set() out: List[str] = [] for url in urls: if url in seen: continue seen.add(url) out.append(url) return out def _parse_skus(raw: Sequence[str]) -> List[str]: skus: List[str] = [] seen = set() for chunk in raw: for part in str(chunk or "").replace(";", ",").split(","): sku = part.strip() if not sku or sku in seen: continue seen.add(sku) skus.append(sku) return skus def reconcile_skus( *, connection_id: int, skus: Sequence[str], dry_run: bool, purge_unmapped: bool = True, wipe_all: bool = False, push: bool = False, wait: bool = False, wait_timeout: int = 1800, ) -> Dict[str, Any]: from db.magento_repositories import ( SqlAlchemyMagentoCatalogStateRepository, SqlAlchemyMagentoConnectionRepository, SqlAlchemyMagentoMediaMapRepository, SqlAlchemyMagentoSyncStateRepository, ) from db.session import get_session from magento.magento_api import MagentoRestClient from magento.media_cleanup import cleanup_orphaned_product_media, wipe_product_media from magento.oauth_client import MagentoOAuthClient, build_magento_oauth_kwargs clean_skus = _parse_skus(skus) if not clean_skus: return {"status": "failed", "error": "No SKUs provided"} results: List[Dict[str, Any]] = [] with get_session() as session: conn_repo = SqlAlchemyMagentoConnectionRepository(session) media_repo = SqlAlchemyMagentoMediaMapRepository(session) catalog_repo = SqlAlchemyMagentoCatalogStateRepository(session) state_repo = SqlAlchemyMagentoSyncStateRepository(session) conn = conn_repo.get_for_sync(connection_id) if not conn: return {"status": "failed", "error": f"Connection {connection_id} not found"} oauth = MagentoOAuthClient(**build_magento_oauth_kwargs(conn)) api = MagentoRestClient(oauth) for sku in clean_skus: if wipe_all: logger.info("Wipe-all gallery for %s dry_run=%s", sku, dry_run) result = wipe_product_media( connection_id=connection_id, sku=sku, api=api, media_repo=media_repo, catalog_repo=catalog_repo, dry_run=dry_run, ) else: desired = _desired_urls_for_sku(session, sku) logger.info( "Reconcile %s desired_urls=%d purge_unmapped=%s dry_run=%s", sku, len(desired), purge_unmapped, dry_run, ) if not desired: results.append( { "sku": sku, "status": "skipped", "error": "No master images for SKU", "desired_urls": [], } ) continue result = cleanup_orphaned_product_media( connection_id=connection_id, sku=sku, desired_urls=desired, api=api, media_repo=media_repo, catalog_repo=catalog_repo, dry_run=dry_run, purge_unmapped=purge_unmapped, ) results.append(result) logger.info( "Reconcile %s status=%s before=%s after=%s deleted=%s errors=%s", sku, result.get("status"), result.get("media_entry_count_before"), result.get("media_entry_count_after"), len(result.get("deleted_entries") or []), len(result.get("delete_errors") or []), ) if not dry_run: # Force next images pass even if hashes previously matched polluted gallery. state_repo.reset_hashes(connection_id, list(clean_skus)) session.commit() push_result: Optional[Dict[str, Any]] = None if push and not dry_run: from app.jobs.push_channel_sample import push_magento_sample from db.compat_connections import compat_connection_id from db.session import get_session as get_session_again with get_session_again() as session: push_result = push_magento_sample( session, skus=list(clean_skus), connection_id=compat_connection_id("magento", connection_id), dry_run=False, images_only=True, products_only=False, force_images=True, force_products=False, force_relations=False, phased=False, wait=wait, wait_timeout=wait_timeout, run_worker=wait, run_pull_after=False, ) session.commit() failed = [r for r in results if r.get("status") in {"failed", "partial"}] return { "status": "ok" if not failed else "partial", "connection_id": connection_id, "dry_run": dry_run, "purge_unmapped": purge_unmapped, "wipe_all": wipe_all, "sku_count": len(clean_skus), "results": results, "push": push_result, } def main(argv: Optional[Sequence[str]] = None) -> int: parser = argparse.ArgumentParser(description="Reconcile Magento media galleries to master images") parser.add_argument("--connection-id", type=int, default=1) parser.add_argument( "--skus", nargs="+", required=True, help="SKU list (space and/or comma separated)", ) parser.add_argument("--dry-run", action="store_true", help="Report delete targets only") parser.add_argument("--apply", action="store_true", help="Apply deletions (required unless --dry-run)") parser.add_argument( "--no-purge-unmapped", action="store_true", help="Only delete stale media-map entries (legacy conservative mode)", ) parser.add_argument( "--wipe-all", action="store_true", help="Delete ALL Magento gallery entries + media_map for the SKUs, then rely on --push to rebuild", ) parser.add_argument( "--push", action="store_true", help="After cleanup, enqueue/run images_only force push for the SKUs", ) parser.add_argument("--wait", action="store_true", help="With --push, run worker until queue completes") parser.add_argument("--wait-timeout", type=int, default=1800) args = parser.parse_args(list(argv) if argv is not None else None) if not args.dry_run and not args.apply: parser.error("Pass --dry-run or --apply") if args.wipe_all and args.no_purge_unmapped: parser.error("--wipe-all and --no-purge-unmapped are mutually exclusive") if args.apply and args.wipe_all and not args.push: logger.warning("Wipe-all without --push leaves products with an empty gallery") result = reconcile_skus( connection_id=args.connection_id, skus=args.skus, dry_run=bool(args.dry_run), purge_unmapped=not bool(args.no_purge_unmapped), wipe_all=bool(args.wipe_all), push=bool(args.push), wait=bool(args.wait), wait_timeout=int(args.wait_timeout), ) print(json.dumps(result, indent=2, default=str)) return 0 if result.get("status") in {"ok", "dry_run"} or args.dry_run else 1 if __name__ == "__main__": sys.exit(main())