""" Repair Variation Builder parent product names and Magento URL keys. Use after a bad Variation Builder run polluted configurable parent titles with composite axis labels such as variation_option_label. Examples: python -m app.jobs.repair_variation_parent_names --dry-run python -m app.jobs.repair_variation_parent_names --apply-db python -m app.jobs.repair_variation_parent_names --apply-db --push-magento --connection-id 1 python -m app.jobs.repair_variation_parent_names --push-current-db-names --push-magento --connection-id 1 python -m app.jobs.repair_variation_parent_names --dry-run --push-current-db-names --include-magento-current --connection-id 1 --skus SKU1,SKU2 """ from __future__ import annotations import argparse import json import logging import sys import time from dataclasses import asdict, dataclass from datetime import datetime, timezone from typing import Any, Dict, Iterable, List, Optional, Sequence from sqlalchemy import select from sqlalchemy.orm import Session from channel.url_canonical import build_pdp_slug from db.channel_sku_mapping import channel_sku_for_master from db.models import ChannelSkuMapping, MagentoCatalogState, MasterProduct, MasterProductRelation from magento.payload_mapper import _compute_parent_neutral_name, get_axis_attribute_codes logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) NON_AXIS_CODES = {"variation_option_label", "item_size"} BAD_NAME_TOKENS = ( "variation_option_label", "item_size", "variation option label", ) @dataclass(frozen=True) class ParentNameRepairCandidate: sku: str old_name: str new_name: str old_url_key: str new_url_key: str child_count: int reason: str def to_dict(self) -> Dict[str, Any]: return asdict(self) def find_parent_name_repair_candidates( session: Session, *, skus: Optional[Sequence[str]] = None, only_bad: bool = True, limit: Optional[int] = None, ) -> List[ParentNameRepairCandidate]: """Return configurable parent shell rows whose clean title differs from the stored one.""" wanted = {str(s).strip() for s in skus or [] if str(s).strip()} relations = _relations_by_parent(session, wanted or None) if not relations: return [] stmt = select(MasterProduct).where(MasterProduct.sku.in_(relations.keys())) if wanted: stmt = stmt.where(MasterProduct.sku.in_(wanted)) stmt = stmt.order_by(MasterProduct.sku) products = list(session.scalars(stmt).all()) candidates: List[ParentNameRepairCandidate] = [] for product in products: parent_relations = relations.get(product.sku) or [] if not parent_relations: continue if not _looks_like_variation_builder_parent(product, parent_relations): continue old_name = str(product.name or product.sku).strip() new_name = clean_variation_parent_name(product, parent_relations) if not new_name or new_name == old_name: continue reason = _candidate_reason(old_name, new_name) if only_bad and reason == "title_normalized": continue candidates.append( ParentNameRepairCandidate( sku=product.sku, old_name=old_name, new_name=new_name, old_url_key=build_pdp_slug(old_name, product.sku), new_url_key=build_pdp_slug(new_name, product.sku), child_count=len(parent_relations), reason=reason, ) ) if limit and len(candidates) >= limit: break return candidates def find_parent_name_push_candidates( session: Session, *, skus: Optional[Sequence[str]] = None, limit: Optional[int] = None, ) -> List[ParentNameRepairCandidate]: """ Return variation parent shells as Magento push candidates using current DB names. This is for the common follow-up case where master_product already contains clean titles, but Magento still has stale names from a previous bad push. """ wanted = {str(s).strip() for s in skus or [] if str(s).strip()} relations = _relations_by_parent(session, wanted or None) if not relations: return [] stmt = select(MasterProduct).where(MasterProduct.sku.in_(relations.keys())) if wanted: stmt = stmt.where(MasterProduct.sku.in_(wanted)) stmt = stmt.order_by(MasterProduct.sku) products = list(session.scalars(stmt).all()) candidates: List[ParentNameRepairCandidate] = [] for product in products: parent_relations = relations.get(product.sku) or [] if not parent_relations: continue if not _looks_like_variation_builder_parent(product, parent_relations): continue current_name = str(product.name or product.sku).strip() clean_name = clean_variation_parent_name(product, parent_relations) or current_name candidates.append( ParentNameRepairCandidate( sku=product.sku, old_name=current_name, new_name=clean_name, old_url_key=build_pdp_slug(current_name, product.sku), new_url_key=build_pdp_slug(clean_name, product.sku), child_count=len(parent_relations), reason="push_current_db_name" if clean_name == current_name else _candidate_reason(current_name, clean_name), ) ) if limit and len(candidates) >= limit: break return candidates def clean_variation_parent_name( product: MasterProduct, relations: Sequence[MasterProductRelation], ) -> str: """Compute the current canonical configurable parent title for a DB parent shell.""" base_name = _base_parent_name(str(product.name or product.sku).strip()) row = { "sku": product.sku, "name": base_name, "configurable_attributes": _configurable_attributes(relations), "configurable_variations": _configurable_variations(relations), } axis_codes = get_axis_attribute_codes(row) rows_by_sku = {product.sku: row} for relation in relations: rows_by_sku[relation.child_sku] = {"sku": relation.child_sku} neutral = _compute_parent_neutral_name(row, axis_codes, rows_by_sku) return neutral or base_name or product.sku def apply_db_parent_name_repairs( session: Session, candidates: Sequence[ParentNameRepairCandidate], *, dry_run: bool = True, ) -> Dict[str, Any]: if dry_run: return {"status": "dry_run", "would_update": len(candidates)} if not candidates: return {"status": "success", "updated": 0} now = datetime.now(timezone.utc) by_sku = {candidate.sku: candidate for candidate in candidates} products = session.scalars(select(MasterProduct).where(MasterProduct.sku.in_(by_sku))).all() updated = 0 for product in products: candidate = by_sku.get(product.sku) if not candidate: continue product.name = candidate.new_name product.updated_at = now raw_payload = dict(product.raw_payload or {}) repairs = list(raw_payload.get("name_repairs") or []) repairs.append( { "source": "repair_variation_parent_names", "old_name": candidate.old_name, "new_name": candidate.new_name, "repaired_at": now.isoformat(), } ) raw_payload["name_repairs"] = repairs[-5:] product.raw_payload = raw_payload updated += 1 session.commit() return {"status": "success", "updated": updated} def push_magento_parent_name_repairs( session: Session, connection_id: int, candidates: Sequence[ParentNameRepairCandidate], *, dry_run: bool = True, delay_s: float = 0.0, ) -> Dict[str, Any]: if dry_run: return {"status": "dry_run", "would_push": len(candidates)} if not candidates: return {"status": "success", "pushed": 0, "errors": []} from db.magento_repositories import SqlAlchemyMagentoConnectionRepository from magento.magento_api import MagentoRestClient from magento.oauth_client import MagentoOAuthClient, build_magento_oauth_kwargs conn = SqlAlchemyMagentoConnectionRepository(session).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) pushed = 0 errors: List[Dict[str, Any]] = [] sku_cache: Dict[str, str] = {} for candidate in candidates: lookup_method = "not_attempted" channel_sku = channel_sku_for_master( session, candidate.sku, "magento", connection_id=connection_id, cache=sku_cache, ) or candidate.sku payload = { "sku": channel_sku, "name": candidate.new_name, "custom_attributes": [ {"attribute_code": "url_key", "value": candidate.new_url_key}, {"attribute_code": "meta_title", "value": candidate.new_name}, ], } try: product, resolved_sku, lookup_method = _resolve_magento_product_for_parent( session, api, connection_id=connection_id, master_sku=candidate.sku, channel_sku=channel_sku, ) target_sku = resolved_sku or channel_sku if product and target_sku != channel_sku: payload["sku"] = target_sku api.update_product(target_sku, payload) pushed += 1 if delay_s > 0: time.sleep(delay_s) except Exception as exc: # pragma: no cover - depends on remote Magento errors.append( { "sku": candidate.sku, "channel_sku": channel_sku, "lookup_method": lookup_method, "error": str(exc), } ) logger.warning("Magento parent name repair failed for %s (%s): %s", candidate.sku, channel_sku, exc) return { "status": "partial" if errors else "success", "pushed": pushed, "errors": errors, } def fetch_magento_current_parent_names( session: Session, connection_id: int, candidates: Sequence[ParentNameRepairCandidate], ) -> Dict[str, Any]: """Fetch current Magento name/url_key for dry-run comparison.""" if not candidates: return {"status": "success", "fetched": 0, "products": []} from db.magento_repositories import SqlAlchemyMagentoConnectionRepository from magento.magento_api import MagentoRestClient from magento.oauth_client import MagentoOAuthClient, build_magento_oauth_kwargs conn = SqlAlchemyMagentoConnectionRepository(session).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) products: List[Dict[str, Any]] = [] errors: List[Dict[str, Any]] = [] sku_cache: Dict[str, str] = {} for candidate in candidates: channel_sku = channel_sku_for_master( session, candidate.sku, "magento", connection_id=connection_id, cache=sku_cache, ) or candidate.sku try: product, resolved_sku, lookup_method = _resolve_magento_product_for_parent( session, api, connection_id=connection_id, master_sku=candidate.sku, channel_sku=channel_sku, ) if not product: products.append( { "sku": candidate.sku, "channel_sku": channel_sku, "resolved_sku": resolved_sku, "lookup_method": lookup_method, "magento_found": False, "db_name": candidate.new_name, "would_push_url_key": candidate.new_url_key, } ) continue magento_url_key = _custom_attribute_value(product, "url_key") magento_meta_title = _custom_attribute_value(product, "meta_title") products.append( { "sku": candidate.sku, "channel_sku": channel_sku, "resolved_sku": resolved_sku, "lookup_method": lookup_method, "magento_found": True, "magento_name": product.get("name"), "magento_url_key": magento_url_key, "magento_meta_title": magento_meta_title, "db_name": candidate.new_name, "would_push_url_key": candidate.new_url_key, "name_differs": str(product.get("name") or "").strip() != candidate.new_name, "url_key_differs": str(magento_url_key or "").strip() != candidate.new_url_key, } ) except Exception as exc: # pragma: no cover - depends on remote Magento errors.append({"sku": candidate.sku, "error": str(exc)}) return { "status": "partial" if errors else "success", "fetched": len(products), "products": products, "errors": errors, } def _resolve_magento_product_for_parent( session: Session, api: Any, *, connection_id: int, master_sku: str, channel_sku: str, ) -> tuple[Optional[Dict[str, Any]], Optional[str], str]: direct = api.get_product(channel_sku) if direct: resolved_sku = str(direct.get("sku") or channel_sku).strip() or channel_sku return direct, resolved_sku, "direct_sku" status, items, _ = api.search_products(sku=channel_sku, page_size=5) if status == 200 and items: for item in items: item_sku = str((item or {}).get("sku") or "").strip() if item_sku == channel_sku: return item, item_sku, "search_sku" remote_id = _remote_product_id_for_master( session, connection_id=connection_id, master_sku=master_sku, ) if remote_id and str(remote_id).isdigit(): status, items, _ = api.search_products(entity_id=int(remote_id), page_size=5) if status == 200 and items: item = items[0] resolved_sku = str((item or {}).get("sku") or channel_sku).strip() or channel_sku return item, resolved_sku, "search_entity_id" return None, channel_sku, "not_found" def _remote_product_id_for_master( session: Session, *, connection_id: int, master_sku: str, ) -> Optional[str]: remote_id = session.scalar( select(ChannelSkuMapping.remote_id) .where(ChannelSkuMapping.connection_id == connection_id) .where(ChannelSkuMapping.master_sku == master_sku) .where(ChannelSkuMapping.mapping_status == "active") .where(ChannelSkuMapping.remote_id.is_not(None)) ) if str(remote_id or "").strip(): return str(remote_id).strip() catalog_id = session.scalar( select(MagentoCatalogState.magento_product_id) .where(MagentoCatalogState.connection_id == connection_id) .where(MagentoCatalogState.sku == master_sku) ) if catalog_id is not None: return str(catalog_id).strip() return None def run_repair( *, skus: Optional[Sequence[str]] = None, only_bad: bool = True, limit: Optional[int] = None, apply_db: bool = False, push_magento: bool = False, push_current_db_names: bool = False, include_magento_current: bool = False, connection_id: Optional[int] = None, magento_delay_s: float = 0.0, ) -> Dict[str, Any]: from db.session import get_session if push_magento and not connection_id: return {"status": "failed", "error": "--connection-id is required with --push-magento"} with get_session() as session: candidates = find_parent_name_repair_candidates( session, skus=skus, only_bad=only_bad, limit=limit, ) magento_candidates = ( find_parent_name_push_candidates(session, skus=skus, limit=limit) if push_current_db_names else candidates ) db_result = apply_db_parent_name_repairs(session, candidates, dry_run=not apply_db) magento_result = ( push_magento_parent_name_repairs( session, int(connection_id), magento_candidates, dry_run=not push_magento, delay_s=magento_delay_s, ) if connection_id else {"status": "skipped", "reason": "no_connection_id"} ) current_magento = ( fetch_magento_current_parent_names(session, int(connection_id), magento_candidates) if include_magento_current and connection_id else {"status": "skipped", "reason": "not_requested"} ) return { "status": "success", "candidate_count": len(candidates), "candidates": [candidate.to_dict() for candidate in candidates[:100]], "truncated": len(candidates) > 100, "magento_candidate_count": len(magento_candidates), "magento_candidates": [candidate.to_dict() for candidate in magento_candidates[:100]], "magento_truncated": len(magento_candidates) > 100, "db": db_result, "magento": magento_result, "current_magento": current_magento, } def _relations_by_parent( session: Session, parent_skus: Optional[Iterable[str]] = None, ) -> Dict[str, List[MasterProductRelation]]: stmt = select(MasterProductRelation).order_by( MasterProductRelation.parent_sku, MasterProductRelation.sort_order, MasterProductRelation.child_sku, ) wanted = {str(s).strip() for s in parent_skus or [] if str(s).strip()} if wanted: stmt = stmt.where(MasterProductRelation.parent_sku.in_(wanted)) rows = list(session.scalars(stmt).all()) grouped: Dict[str, List[MasterProductRelation]] = {} for row in rows: grouped.setdefault(row.parent_sku, []).append(row) return grouped def _looks_like_variation_builder_parent( product: MasterProduct, relations: Sequence[MasterProductRelation], ) -> bool: raw_payload = dict(product.raw_payload or {}) if raw_payload.get("generated_by") == "variation_builder": return True if any(str(rel.source_label or "").startswith("variation_builder") for rel in relations): return True sku = str(product.sku or "").upper() return sku.endswith("-PARENT") or sku.endswith("-CONFIG") def _base_parent_name(name: str) -> str: text = str(name or "").strip() if not text: return "" lower = text.lower() if any(token in lower for token in BAD_NAME_TOKENS): return text.split("(", 1)[0].strip() or text if "(" in text and ")" in text: suffix = text.split("(", 1)[1].lower() if any(code in suffix for code in ("width", "height", "depth")) and "," in suffix: return text.split("(", 1)[0].strip() or text return text def _candidate_reason(old_name: str, new_name: str) -> str: lower = old_name.lower() for token in BAD_NAME_TOKENS: if token in lower: return f"removed_{token.replace(' ', '_')}" if len(old_name) > len(new_name) + 40: return "shortened_noisy_title" return "title_normalized" def _configurable_attributes(relations: Sequence[MasterProductRelation]) -> str: codes: List[str] = [] seen: set[str] = set() for relation in relations: for code in dict(relation.option_mapping or {}): normalized = str(code or "").strip().lower() if not normalized or normalized in NON_AXIS_CODES or normalized in seen: continue seen.add(normalized) codes.append(normalized) return ",".join(codes) def _configurable_variations(relations: Sequence[MasterProductRelation]) -> str: segments: List[str] = [] for relation in relations: parts = [f"sku={relation.child_sku}"] for code, value in dict(relation.option_mapping or {}).items(): normalized = str(code or "").strip().lower() if not normalized or normalized in NON_AXIS_CODES: continue val = str(value or "").strip() if val: parts.append(f"{normalized}={val}") segments.append(",".join(parts)) return "|".join(segments) def _parse_skus(value: Optional[str]) -> Optional[List[str]]: if not value: return None return [part.strip() for part in value.split(",") if part.strip()] def _custom_attribute_value(product: Dict[str, Any], attribute_code: str) -> Optional[Any]: for attr in product.get("custom_attributes") or []: if isinstance(attr, dict) and attr.get("attribute_code") == attribute_code: return attr.get("value") return None def main() -> int: parser = argparse.ArgumentParser(description="Repair bad Variation Builder parent product titles") parser.add_argument("--skus", help="Comma-separated parent SKUs to inspect") parser.add_argument("--limit", type=int, help="Maximum candidates to process") parser.add_argument( "--all-normalized", action="store_true", help="Also include clean-but-stale parent titles, not only obviously bad titles", ) parser.add_argument("--apply-db", action="store_true", help="Update master_product.name") parser.add_argument("--push-magento", action="store_true", help="Push name/url_key/meta_title to Magento") parser.add_argument( "--push-current-db-names", action="store_true", help="Use current DB parent names as Magento push candidates even when no DB repair is needed", ) parser.add_argument( "--include-magento-current", action="store_true", help="Fetch current Magento name/url_key for dry-run comparison", ) parser.add_argument("--connection-id", type=int, help="Magento connection id for --push-magento") parser.add_argument("--magento-delay-ms", type=int, default=0, help="Delay between Magento updates") parser.add_argument("--dry-run", action="store_true", help="Explicit no-op mode; default without apply flags") args = parser.parse_args() result = run_repair( skus=_parse_skus(args.skus), only_bad=not args.all_normalized, limit=args.limit, apply_db=args.apply_db and not args.dry_run, push_magento=args.push_magento and not args.dry_run, push_current_db_names=args.push_current_db_names, include_magento_current=args.include_magento_current, connection_id=args.connection_id, magento_delay_s=max(0, args.magento_delay_ms) / 1000.0, ) print(json.dumps(result, indent=2, default=str)) return 0 if result.get("status") != "failed" else 1 if __name__ == "__main__": sys.exit(main())