"""Push one or more Magento product fields across selected SKUs or the assigned catalog. Examples: python -m app.jobs.push_magento_attributes --field drawer_box --dry-run python -m app.jobs.push_magento_attributes --field drawer_box --apply python -m app.jobs.push_magento_attributes --field drawer_box --sku ACH-B12 --sku ACH-B15 --apply --wait --run-worker python -m app.jobs.push_magento_attributes --field name --field meta_title --apply --connection-id 1 python -m app.jobs.push_magento_attributes --field base_cabinets --where cabinet_type=base cabinets --apply python -m app.jobs.push_magento_attributes --field drawer_box --apply --sync-options python -m app.jobs.push_magento_attributes --bundle essentials --apply """ from __future__ import annotations import argparse import json from datetime import datetime from typing import Dict, Iterable, List, Set from sqlalchemy import select from db.channel_exports import _build_alias_map from db.compat_connections import compat_connection_id from db.models import MagentoConnection from db.session import get_session from db.source_imports import normalize_column_key FIELD_BUNDLES: Dict[str, List[str]] = { "essentials": [ "manufacturer", "family", "collection_style", "depth", "door_style", "color_finish", "overlay", "finish_type", "drawer_slides", "hinges", "box_construction", "shelf_clips", "materials_face_frame", "materials_box", "shelves_included", "warranty", "care_instructions", "accessories", "description", "additional_information", "materials", ], "specs": [ "cabinet_back_material", "cabinet_bottom_material", "cabinet_top_material", "cabinet_carcass_material", "cabinet_toe_kick_construction", "cabinet_interior_finish", "cabinet_corner_support", "cabinet_i_beam_support_braces", "cabinet_shelf_material", "cabinet_exterior_side_finish", "manufacturer", "family", "collection_style", "depth", "cabinet_type", "base_cabinets", "wall_cabinets", "panels_and_fillers", "mouldings", "angle", "drawer_box", "soft_close", "packaged_dimension_height", "packaged_dimension_width", "packaged_dimension_depth", "packaged_gross_weight", "door_style", "color_finish", "overlay", "finish_type", "drawer_slides", "hinges", "box_construction", "shelf_clips", "materials_face_frame", "materials_box", "shelves_included", "warranty", "care_instructions", "accessories", "description", "additional_information", "materials", ], "materials": [ "cabinet_back_material", "cabinet_bottom_material", "cabinet_top_material", "cabinet_carcass_material", "cabinet_toe_kick_construction", "cabinet_interior_finish", "cabinet_corner_support", "cabinet_i_beam_support_braces", "cabinet_shelf_material", "cabinet_exterior_side_finish", "materials_face_frame", "materials_box", "materials", ], "content": [ "description", "short_description", "additional_information", "materials", "warranty", "care_instructions", ], } def expand_requested_field_codes(field_codes: Iterable[str] | None, bundle_names: Iterable[str] | None) -> List[str]: expanded: List[str] = [] seen: Set[str] = set() unknown_bundles: List[str] = [] for bundle_name in bundle_names or []: normalized_bundle = normalize_column_key(bundle_name) if not normalized_bundle: continue bundle_fields = FIELD_BUNDLES.get(normalized_bundle) if not bundle_fields: unknown_bundles.append(str(bundle_name)) continue for code in bundle_fields: normalized_code = normalize_column_key(code) if normalized_code and normalized_code not in seen: seen.add(normalized_code) expanded.append(normalized_code) for code in field_codes or []: normalized_code = normalize_column_key(code) if normalized_code and normalized_code not in seen: seen.add(normalized_code) expanded.append(normalized_code) if unknown_bundles: available = ", ".join(sorted(FIELD_BUNDLES)) requested = ", ".join(unknown_bundles) raise ValueError(f"Unknown bundle(s): {requested}. Available bundles: {available}") if not expanded: raise ValueError("At least one --field or --bundle is required") return expanded def resolve_magento_attribute_selection(session, requested_codes: Iterable[str]) -> Dict[str, List[str]]: alias_map = _build_alias_map(session, "magento") canonical_to_remote: Dict[str, Set[str]] = {} remote_to_canonical: Dict[str, Set[str]] = {} for canonical, remotes in alias_map.items(): canonical_code = normalize_column_key(canonical) if not canonical_code: continue bucket = canonical_to_remote.setdefault(canonical_code, set()) for remote in remotes: remote_code = normalize_column_key(remote) if not remote_code: continue bucket.add(remote_code) remote_to_canonical.setdefault(remote_code, set()).add(canonical_code) requested = { normalize_column_key(code) for code in requested_codes if normalize_column_key(code) } resolved_remote: Set[str] = set() resolved_canonical: Set[str] = set() passthrough: Set[str] = set() for code in requested: matched = False if code in canonical_to_remote: resolved_canonical.add(code) resolved_remote.update(canonical_to_remote[code]) matched = True if code in remote_to_canonical: resolved_remote.add(code) resolved_canonical.update(remote_to_canonical[code]) matched = True if not matched: passthrough.add(code) resolved_remote.add(code) resolved_canonical.add(code) return { "requested_codes": sorted(requested), "remote_codes": sorted(resolved_remote), "canonical_codes": sorted(resolved_canonical), "passthrough_codes": sorted(passthrough), "unresolved_codes": [], } def parse_field_filter(raw: str) -> Dict[str, str]: text = str(raw or "").strip() if not text or "=" not in text: raise ValueError(f"Invalid filter {raw!r}; expected code=value") code, value = text.split("=", 1) normalized_code = normalize_column_key(code) normalized_value = str(value or "").strip() if not normalized_code or not normalized_value: raise ValueError(f"Invalid filter {raw!r}; expected non-empty code=value") return { "code": normalized_code, "value": normalized_value, "normalized_value": _normalized_compare_value(normalized_value), } def resolve_filtered_magento_skus(session, filters: Iterable[str]) -> Dict[str, object]: from db.channel_exports import build_channel_product_payloads parsed = [parse_field_filter(item) for item in filters or [] if str(item or "").strip()] if not parsed: return {"filters": [], "matched_skus": []} payloads = build_channel_product_payloads( session, "magento", only_assigned=True, skip_taxonomy=True, ) matched: List[str] = [] for payload in payloads: canonical_fields = payload.get("canonical_fields") or {} channel_fields = payload.get("fields") or {} if all(_payload_matches_filter(canonical_fields, channel_fields, item) for item in parsed): sku = str(payload.get("sku") or "").strip() if sku: matched.append(sku) return { "filters": [{"code": item["code"], "value": item["value"]} for item in parsed], "matched_skus": matched, } def _normalized_compare_value(value: object) -> str: return " ".join(str(value or "").strip().lower().split()) def _payload_matches_filter(canonical_fields: Dict[str, object], channel_fields: Dict[str, object], item: Dict[str, str]) -> bool: code = item["code"] expected = item["normalized_value"] candidates = [ canonical_fields.get(code), channel_fields.get(code), ] return any(_normalized_compare_value(candidate) == expected for candidate in candidates) def _scope_label(*, has_explicit_skus: bool, has_filters: bool) -> str: if has_explicit_skus and has_filters: return "explicit_skus_filtered" if has_explicit_skus: return "explicit_skus" if has_filters: return "filtered_assigned_magento_skus" return "all_assigned_magento_skus" def enqueue_magento_attribute_push( *, field_codes: List[str], skus: List[str] | None, filters: List[str] | None, dry_run: bool, connection_id: int | None, wait: bool, run_worker: bool, wait_timeout: int, sync_options_from_master: bool, ) -> Dict[str, object]: from app.jobs.push_channel_sample import _wait_for_magento_queue from db.magento_repositories import SqlAlchemyMagentoSyncQueueRepository with get_session() as session: native_connection_id = connection_id if native_connection_id is None: row = session.scalar( select(MagentoConnection).where(MagentoConnection.status == "active").order_by(MagentoConnection.id).limit(1) ) if row is None: row = session.scalar(select(MagentoConnection).order_by(MagentoConnection.id).limit(1)) native_connection_id = int(row.id) if row is not None else None if native_connection_id is None: raise RuntimeError("No Magento connection found") resolved = resolve_magento_attribute_selection(session, field_codes) filter_result = resolve_filtered_magento_skus(session, filters or []) scoped_skus = [str(s).strip() for s in skus or [] if str(s).strip()] if filter_result["matched_skus"]: if scoped_skus: matched_set = set(filter_result["matched_skus"]) scoped_skus = [sku for sku in scoped_skus if sku in matched_set] else: scoped_skus = list(filter_result["matched_skus"]) elif filters: scoped_skus = [] if filters and not scoped_skus: return { "status": "no_matches", "queue_id": None, "label": None, "connection_id": compat_connection_id("magento", int(native_connection_id)), "native_connection_id": int(native_connection_id), "field_selection": resolved, "filters": filter_result["filters"], "matched_filter_sku_count": 0, "dry_run": dry_run, "sku_count": 0, "scope": _scope_label(has_explicit_skus=bool(skus), has_filters=True), "options": {}, } queue_options = { "dry_run": dry_run, "use_master_catalog": True, "products_only": True, "force_upsert": True, "run_pull_first": False, "run_pull_after": False, "expand_relations": False, "sync_options_from_master": bool(sync_options_from_master), "selected_attribute_codes": resolved["remote_codes"], "allow_parallel_with_newer_queue_items": True, } if scoped_skus: queue_options["limit_skus"] = scoped_skus queue_options["force_upsert_skus"] = list(queue_options["limit_skus"]) queue_label = ( "push-magento-attributes-" + "-".join(resolved["remote_codes"][:3]) + "-" + datetime.utcnow().strftime("%Y%m%d-%H%M%S") ) queue_id = SqlAlchemyMagentoSyncQueueRepository(session).enqueue( int(native_connection_id), queue_label, options=queue_options, supersede_queued=False, ) session.commit() result: Dict[str, object] = { "status": "queued", "queue_id": queue_id, "label": queue_label, "connection_id": compat_connection_id("magento", int(native_connection_id)), "native_connection_id": int(native_connection_id), "field_selection": resolved, "filters": filter_result["filters"], "matched_filter_sku_count": len(filter_result["matched_skus"]), "dry_run": dry_run, "sku_count": len(queue_options.get("limit_skus") or []), "scope": _scope_label(has_explicit_skus=bool(skus), has_filters=bool(filters)), "options": queue_options, } if wait: outcome = _wait_for_magento_queue(queue_id, timeout=wait_timeout, run_worker=run_worker) result.update(outcome) result["status"] = outcome.get("status", "queued") return result def main() -> int: parser = argparse.ArgumentParser(description="Push specific Magento product field(s) without a full product re-publish") parser.add_argument("--field", dest="fields", action="append", default=None, help="Magento field/attribute code") parser.add_argument( "--bundle", dest="bundles", action="append", default=None, help=f"Named field bundle: {', '.join(sorted(FIELD_BUNDLES))}", ) parser.add_argument("--sku", dest="skus", action="append", default=None, help="Target a specific SKU; repeatable") parser.add_argument( "--where", dest="filters", action="append", default=None, help="Exact match filter in the form code=value; repeatable", ) parser.add_argument("--connection-id", type=int, default=None, 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="Live push") parser.add_argument("--wait", action="store_true", help="Wait for queue completion") parser.add_argument("--run-worker", action="store_true", help="Process the queue in-process while waiting") parser.add_argument("--wait-timeout", type=int, default=7200, help="Wait timeout in seconds") parser.add_argument( "--sync-options", action="store_true", help="Also sync missing select options from master before pushing attributes", ) args = parser.parse_args() field_codes = expand_requested_field_codes(args.fields, args.bundles) result = enqueue_magento_attribute_push( field_codes=field_codes, skus=args.skus, filters=args.filters, dry_run=args.dry_run, connection_id=args.connection_id, wait=args.wait, run_worker=args.run_worker, wait_timeout=args.wait_timeout, sync_options_from_master=args.sync_options, ) print(json.dumps(result, indent=2, default=str)) return 0 if result.get("status") not in {"failed", "cancelled"} else 1 if __name__ == "__main__": raise SystemExit(main())