""" Repair titles where naive dimension spacing turned Galaxy into "Gala x y". A global ``x`` → `` x `` cleanup (meant for ``12x24`` → ``12 x 24``) corrupted product names / SEO titles / URL slugs for Galaxy Horizon SKUs. Collection and category_l3 usually stayed correct because they are derived separately. Examples: python -m app.jobs.repair_gala_x_y_titles --dry-run python -m app.jobs.repair_gala_x_y_titles --apply-db python -m app.jobs.repair_gala_x_y_titles --apply-db --sku FGH-B09FD python -m app.jobs.repair_gala_x_y_titles --apply-db --push-magento --connection-id 1 """ from __future__ import annotations import argparse import json import logging import re import time from dataclasses import asdict, dataclass from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Sequence from sqlalchemy import or_, select from sqlalchemy.orm import Session from channel.url_canonical import build_pdp_slug from db.models import MasterProduct, MasterProductAttributeValue, MasterProductOverride from db.session import get_session logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Display / SEO text: "Gala x y" / "Gala x y" → "Galaxy" _GALA_XY_TEXT_RE = re.compile(r"gala\s+x\s+y", re.IGNORECASE) # Slug fragments already baked into url keys / handles _GALA_XY_SLUG_RE = re.compile(r"gala-x-y", re.IGNORECASE) SEO_ATTR_CODES = ("meta_title", "seo_title", "product_url_slug", "url_key", "handle") RAW_PAYLOAD_KEYS = ( "meta_title", "seo_title", "product_url_slug", "url_key", "handle", "Display Name", "display_name", "Name", "name", "Label", "label", ) @dataclass(frozen=True) class GalaXyRepairCandidate: sku: str old_name: str new_name: str old_url_key: str new_url_key: str fields_touched: tuple[str, ...] reason: str = "gala_x_y_to_galaxy" def to_dict(self) -> Dict[str, Any]: return asdict(self) def fix_gala_x_y_text(value: Optional[str]) -> Optional[str]: """Replace spaced Galaxy corruption in free text. Returns None if unchanged/empty.""" text = str(value or "") if not text or not _GALA_XY_TEXT_RE.search(text): return None fixed = _GALA_XY_TEXT_RE.sub("Galaxy", text) fixed = re.sub(r"\s{2,}", " ", fixed).strip() return fixed if fixed != text else None def fix_gala_x_y_slug(value: Optional[str]) -> Optional[str]: """Replace gala-x-y slug fragments. Returns None if unchanged/empty.""" text = str(value or "").strip() if not text or not _GALA_XY_SLUG_RE.search(text): return None fixed = _GALA_XY_SLUG_RE.sub("galaxy", text) fixed = re.sub(r"-{2,}", "-", fixed).strip("-") return fixed if fixed != text else None def repair_title_and_slug(name: str, sku: str, *, existing_slug: Optional[str] = None) -> Dict[str, str]: """Return cleaned name + rebuilt PDP slug for a product.""" cleaned_name = fix_gala_x_y_text(name) or str(name or "").strip() or sku slug_source = existing_slug if slug_source: fixed_slug = fix_gala_x_y_slug(slug_source) if fixed_slug: slug_source = fixed_slug elif fix_gala_x_y_text(slug_source.replace("-", " ")): slug_source = None # force rebuild from cleaned title new_slug = build_pdp_slug(cleaned_name, sku, existing_slug=None) return {"name": cleaned_name, "url_key": new_slug} def find_gala_x_y_repair_candidates( session: Session, *, skus: Optional[Sequence[str]] = None, limit: Optional[int] = None, ) -> List[GalaXyRepairCandidate]: wanted = {str(s).strip() for s in skus or [] if str(s).strip()} stmt = select(MasterProduct).where(MasterProduct.is_active.is_(True)) if wanted: stmt = stmt.where(MasterProduct.sku.in_(wanted)) else: # Broad prefilter: name or collection Galaxy / FGH style, or literal corruption. stmt = stmt.where( or_( MasterProduct.name.ilike("%gala%x%y%"), MasterProduct.name.ilike("%Gala x y%"), MasterProduct.collection.ilike("%Galaxy%"), MasterProduct.sku.ilike("FGH-%"), ) ) stmt = stmt.order_by(MasterProduct.sku) products = list(session.scalars(stmt).all()) attr_by_sku = _seo_attrs_by_sku(session, [p.sku for p in products]) override_by_sku = _seo_overrides_by_sku(session, [p.sku for p in products]) candidates: List[GalaXyRepairCandidate] = [] for product in products: candidate = _candidate_for_product( product, attrs=attr_by_sku.get(product.sku) or {}, overrides=override_by_sku.get(product.sku) or {}, ) if candidate is None: continue candidates.append(candidate) if limit and len(candidates) >= limit: break return candidates def apply_db_gala_x_y_repairs( session: Session, candidates: Sequence[GalaXyRepairCandidate], *, 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 = {c.sku: c for c in candidates} products = list(session.scalars(select(MasterProduct).where(MasterProduct.sku.in_(by_sku))).all()) attrs = list( session.scalars( select(MasterProductAttributeValue) .where(MasterProductAttributeValue.sku.in_(by_sku)) .where(MasterProductAttributeValue.attribute_code.in_(SEO_ATTR_CODES)) ).all() ) overrides = list( session.scalars( select(MasterProductOverride) .where(MasterProductOverride.sku.in_(by_sku)) .where(MasterProductOverride.attribute_code.in_(SEO_ATTR_CODES + ("name", "title"))) ).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 {}) for key in RAW_PAYLOAD_KEYS: if key not in raw_payload: continue raw_val = raw_payload.get(key) if not isinstance(raw_val, str): continue if key in {"product_url_slug", "url_key", "handle"}: fixed = fix_gala_x_y_slug(raw_val) or ( candidate.new_url_key if fix_gala_x_y_text(raw_val.replace("-", " ")) else None ) else: fixed = fix_gala_x_y_text(raw_val) if fixed: raw_payload[key] = fixed raw_payload["meta_title"] = candidate.new_name raw_payload["product_url_slug"] = candidate.new_url_key repairs = list(raw_payload.get("name_repairs") or []) repairs.append( { "source": "repair_gala_x_y_titles", "old_name": candidate.old_name, "new_name": candidate.new_name, "old_url_key": candidate.old_url_key, "new_url_key": candidate.new_url_key, "repaired_at": now.isoformat(), } ) raw_payload["name_repairs"] = repairs[-5:] product.raw_payload = raw_payload updated += 1 for row in attrs: candidate = by_sku.get(row.sku) if not candidate: continue code = str(row.attribute_code or "").strip().lower() if code in {"meta_title", "seo_title"}: row.value = candidate.new_name row.source_label = "repair_gala_x_y_titles" elif code in {"product_url_slug", "url_key", "handle"}: row.value = candidate.new_url_key row.source_label = "repair_gala_x_y_titles" for row in overrides: candidate = by_sku.get(row.sku) if not candidate: continue code = str(row.attribute_code or "").strip().lower() if code in {"name", "title", "meta_title", "seo_title"}: fixed = fix_gala_x_y_text(str(row.value or "")) if fixed: row.value = fixed row.source_label = "repair_gala_x_y_titles" elif code in {"product_url_slug", "url_key", "handle"}: fixed = fix_gala_x_y_slug(str(row.value or "")) if fixed: row.value = fixed row.source_label = "repair_gala_x_y_titles" # Ensure SEO attrs exist after repair so channel export picks them up. _ensure_seo_attrs(session, products, by_sku, source_label="repair_gala_x_y_titles") session.commit() return {"status": "success", "updated": updated} def push_magento_gala_x_y_repairs( session: Session, connection_id: int, candidates: Sequence[GalaXyRepairCandidate], *, 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]] = [] for candidate in candidates: channel_sku = 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: api.update_product(channel_sku, payload) pushed += 1 if delay_s > 0: time.sleep(delay_s) except Exception as exc: # pragma: no cover - remote Magento errors.append({"sku": channel_sku, "error": str(exc)}) logger.warning("Magento gala-x-y repair failed for %s: %s", channel_sku, exc) return { "status": "partial" if errors else "success", "pushed": pushed, "errors": errors, } def _candidate_for_product( product: MasterProduct, *, attrs: Dict[str, str], overrides: Dict[str, str], ) -> Optional[GalaXyRepairCandidate]: old_name = str(product.name or "").strip() touched: List[str] = [] new_name = fix_gala_x_y_text(old_name) if new_name: touched.append("name") else: new_name = old_name for code, value in {**attrs, **overrides}.items(): if fix_gala_x_y_text(value) or fix_gala_x_y_slug(value): touched.append(code) raw = product.raw_payload if isinstance(product.raw_payload, dict) else {} for key in RAW_PAYLOAD_KEYS: raw_val = raw.get(key) if isinstance(raw_val, str) and (fix_gala_x_y_text(raw_val) or fix_gala_x_y_slug(raw_val)): touched.append(f"raw.{key}") if not touched: return None existing_slug = ( attrs.get("product_url_slug") or attrs.get("url_key") or (str(raw.get("product_url_slug") or "").strip() or None) ) repaired = repair_title_and_slug(new_name or old_name or product.sku, product.sku, existing_slug=existing_slug) old_url = build_pdp_slug(old_name or product.sku, product.sku, existing_slug=existing_slug) return GalaXyRepairCandidate( sku=product.sku, old_name=old_name or product.sku, new_name=repaired["name"], old_url_key=old_url, new_url_key=repaired["url_key"], fields_touched=tuple(sorted(set(touched))), ) def _seo_attrs_by_sku(session: Session, skus: Sequence[str]) -> Dict[str, Dict[str, str]]: if not skus: return {} rows = session.scalars( select(MasterProductAttributeValue) .where(MasterProductAttributeValue.sku.in_(list(skus))) .where(MasterProductAttributeValue.attribute_code.in_(SEO_ATTR_CODES)) ).all() out: Dict[str, Dict[str, str]] = {} for row in rows: out.setdefault(row.sku, {})[str(row.attribute_code)] = str(row.value or "") return out def _seo_overrides_by_sku(session: Session, skus: Sequence[str]) -> Dict[str, Dict[str, str]]: if not skus: return {} rows = session.scalars( select(MasterProductOverride) .where(MasterProductOverride.sku.in_(list(skus))) .where(MasterProductOverride.is_active.is_(True)) .where(MasterProductOverride.attribute_code.in_(SEO_ATTR_CODES + ("name", "title"))) ).all() out: Dict[str, Dict[str, str]] = {} for row in rows: out.setdefault(row.sku, {})[str(row.attribute_code)] = str(row.value or "") return out def _ensure_seo_attrs( session: Session, products: Sequence[MasterProduct], by_sku: Dict[str, GalaXyRepairCandidate], *, source_label: str, ) -> None: existing = { (row.sku, row.attribute_code): row for row in session.scalars( select(MasterProductAttributeValue) .where(MasterProductAttributeValue.sku.in_(by_sku)) .where(MasterProductAttributeValue.attribute_code.in_(("meta_title", "product_url_slug"))) ).all() } for product in products: candidate = by_sku.get(product.sku) if not candidate: continue for code, value in ( ("meta_title", candidate.new_name), ("product_url_slug", candidate.new_url_key), ): key = (product.sku, code) row = existing.get(key) if row is None: row = MasterProductAttributeValue( product_id=product.id, sku=product.sku, attribute_code=code, ) session.add(row) row.value = value row.source_label = source_label def main() -> int: parser = argparse.ArgumentParser(description="Repair Gala x y → Galaxy title/slug corruption") parser.add_argument("--dry-run", dest="dry_run", action="store_true", default=True) parser.add_argument("--apply-db", dest="dry_run", action="store_false", help="Persist master DB fixes") parser.add_argument("--sku", action="append", dest="skus", help="Limit to SKU(s); repeatable") parser.add_argument("--limit", type=int, default=None) parser.add_argument("--push-magento", action="store_true", help="Also PUT name/url_key/meta_title to Magento") parser.add_argument("--connection-id", type=int, default=None, help="Magento connection id for --push-magento") parser.add_argument("--delay", type=float, default=0.0, help="Delay between Magento PUTs") args = parser.parse_args() with get_session() as session: candidates = find_gala_x_y_repair_candidates(session, skus=args.skus, limit=args.limit) summary: Dict[str, Any] = { "dry_run": args.dry_run, "candidate_count": len(candidates), "candidates": [c.to_dict() for c in candidates[:50]], } summary["db"] = apply_db_gala_x_y_repairs(session, candidates, dry_run=args.dry_run) if args.push_magento: if not args.connection_id: summary["magento"] = {"status": "failed", "error": "--connection-id required with --push-magento"} else: summary["magento"] = push_magento_gala_x_y_repairs( session, args.connection_id, candidates, dry_run=args.dry_run, delay_s=args.delay, ) print(json.dumps(summary, indent=2, default=str)) if summary.get("magento", {}).get("status") == "failed": return 1 if summary.get("magento", {}).get("errors"): return 1 return 0 if __name__ == "__main__": raise SystemExit(main())