from __future__ import annotations import argparse import html import json import logging import re import sys from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Tuple from bs4 import BeautifulSoup, NavigableString from db.models import MagentoConnection from db.session import get_session from magento.magento_api import MagentoRestClient from magento.oauth_client import MagentoOAuthClient, build_magento_oauth_kwargs logger = logging.getLogger(__name__) DEFAULT_EXPORT_PATH = Path(r"E:\workroom\2026\HSMagentoPie\var\all_pages.json") CMS_PAGE_FIELDS = { "title", "identifier", "content", "content_heading", "page_layout", "meta_title", "meta_keywords", "meta_description", "sort_order", "custom_theme", "custom_root_template", } PAGEBUILDER_ATTRS = { "data-content-type", "data-appearance", "data-element", "data-pb-style", "data-enable-parallax", "data-parallax-speed", "data-background-images", "data-background-type", "data-grid-size", "data-overlay-color", "data-link-type", "data-show-button", "data-show-overlay", "data-same-width", } PAGEBUILDER_ATTR_PREFIXES = ("data-video-",) PAGEBUILDER_CLASS_PREFIXES = ("pagebuilder-",) PAGEBUILDER_CLASSES = { "pagebuilder-column", "pagebuilder-column-group", "pagebuilder-column-line", "pagebuilder-banner-wrapper", "pagebuilder-overlay", "pagebuilder-poster-overlay", "pagebuilder-poster-content", "pagebuilder-button-primary", "pagebuilder-button-secondary", "pagebuilder-button-link", "pagebuilder-mobile-hidden", "pagebuilder-mobile-only", } BASE_STYLE = """""" def load_exported_pages(path: Path) -> List[Dict[str, Any]]: text = path.read_text(encoding="utf-8").strip() if not text: return [] try: data = json.loads(text) except json.JSONDecodeError: data = json.loads(f"[{text.rstrip(',')}]") if isinstance(data, dict): if isinstance(data.get("items"), list): data = data["items"] elif isinstance(data.get("pages"), list): data = data["pages"] else: data = [data] if not isinstance(data, list): raise ValueError(f"Unsupported CMS export shape in {path}") pages = [item for item in data if isinstance(item, dict)] if len(pages) != len(data): raise ValueError("CMS export contains non-object page entries") return pages def _fix_malformed_export_html(content: str) -> str: fixed = content or "" fixed = re.sub(r"<([A-Za-z][\w:-]*)(class|type|id|href|src|alt|title|name|role|style|target|rel|aria-|data-)", r"<\1 \2", fixed) fixed = re.sub(r'"(data-[\w:-]+=)', r'" \1', fixed) fixed = re.sub(r'"(aria-[\w:-]+=)', r'" \1', fixed) fixed = re.sub(r'"(title=)', r'" \1', fixed) return fixed def _background_image_from_attr(raw_value: str) -> Optional[str]: if not raw_value: return None decoded = html.unescape(raw_value).replace(r"\/", "/") try: data = json.loads(decoded) except json.JSONDecodeError: match = re.search(r"{{\s*media\s+url=[^}]+}}", decoded) return match.group(0) if match else None if not isinstance(data, dict): return None for key in ("desktop_image", "mobile_image"): value = data.get(key) if value: return str(value) return None def _append_style(existing: Optional[str], addition: str) -> str: existing = (existing or "").strip() if not existing: return addition if existing.endswith(";"): return f"{existing} {addition}" return f"{existing}; {addition}" def _clean_classes(classes: Iterable[str]) -> List[str]: cleaned: List[str] = [] for cls in classes: if cls == "pagebuilder-column": cleaned.append("cms-column") elif cls == "pagebuilder-column-line": cleaned.append("cms-row") elif cls == "pagebuilder-column-group": cleaned.append("cms-column-group") elif cls == "pagebuilder-button-primary": cleaned.extend(["action", "primary"]) elif cls == "pagebuilder-button-secondary": cleaned.extend(["action", "secondary"]) elif cls == "pagebuilder-mobile-hidden": cleaned.append("cms-mobile-hidden") elif cls == "pagebuilder-mobile-only": cleaned.append("cms-mobile-only") elif cls in PAGEBUILDER_CLASSES or cls.startswith(PAGEBUILDER_CLASS_PREFIXES): continue else: cleaned.append(cls) deduped: List[str] = [] for cls in cleaned: if cls and cls not in deduped: deduped.append(cls) return deduped def _replace_escaped_html_blocks(soup: BeautifulSoup) -> None: for tag in list(soup.find_all(attrs={"data-content-type": "html"})): raw = "".join(str(child) for child in tag.contents).strip() if not raw: continue decoded = html.unescape(raw) if "<" not in decoded or ">" not in decoded: continue fragment = BeautifulSoup(decoded, "html.parser") tag.clear() for child in list(fragment.contents): tag.append(child.extract()) def _unwrap_empty_pagebuilder_links(soup: BeautifulSoup) -> None: for tag in list(soup.find_all(attrs={"data-element": "empty_link"})): tag.unwrap() def clean_page_content(content: str, *, identifier: str, include_base_styles: bool = True) -> str: fixed = _fix_malformed_export_html(content) soup = BeautifulSoup(fixed, "html.parser") _replace_escaped_html_blocks(soup) _unwrap_empty_pagebuilder_links(soup) for tag in list(soup.find_all("style")): tag.decompose() for tag in soup.find_all(True): background = _background_image_from_attr(str(tag.attrs.get("data-background-images", ""))) if background: tag.attrs["style"] = _append_style(tag.attrs.get("style"), f"background-image:url('{background}')") classes = tag.get("class") if classes: cleaned = _clean_classes(classes if isinstance(classes, list) else str(classes).split()) if cleaned: tag.attrs["class"] = cleaned else: tag.attrs.pop("class", None) for attr in list(tag.attrs): if attr in PAGEBUILDER_ATTRS or attr.startswith(PAGEBUILDER_ATTR_PREFIXES): tag.attrs.pop(attr, None) for tag in list(soup.find_all(True)): if tag.name == "div" and not tag.attrs: tag.unwrap() body = "".join(str(child) for child in soup.contents).strip() body = re.sub(r"\s+\n", "\n", body) body = re.sub(r"\n{3,}", "\n\n", body) wrapper_class = f"cms-page-clean cms-page-{re.sub(r'[^a-z0-9_-]+', '-', identifier.lower()).strip('-')}" prefix = f"{BASE_STYLE}\n" if include_base_styles else "" return f'{prefix}
\n{body}\n
' def build_cms_payload(page: Dict[str, Any], *, include_base_styles: bool = True) -> Dict[str, Any]: identifier = str(page.get("identifier") or "").strip() if not identifier: raise ValueError(f"CMS page is missing identifier: {page!r}") payload = {key: page.get(key) for key in CMS_PAGE_FIELDS if key in page and page.get(key) is not None} payload["identifier"] = identifier payload["title"] = str(payload.get("title") or identifier) payload["content"] = clean_page_content(str(page.get("content") or ""), identifier=identifier, include_base_styles=include_base_styles) payload["is_active"] = bool(page.get("is_active", page.get("active", True))) if "sort_order" in payload: try: payload["sort_order"] = int(payload["sort_order"]) except (TypeError, ValueError): payload.pop("sort_order", None) return payload def _magento_api(connection_id: int) -> MagentoRestClient: with get_session() as session: conn = session.get(MagentoConnection, connection_id) if not conn: raise ValueError(f"Magento connection {connection_id} not found") oauth = MagentoOAuthClient(**build_magento_oauth_kwargs(conn)) return MagentoRestClient(oauth) def push_pages( *, connection_id: int, export_path: Path, identifiers: Optional[set[str]] = None, dry_run: bool = False, create_only: bool = False, update_only: bool = False, include_base_styles: bool = True, ) -> List[Dict[str, Any]]: pages = load_exported_pages(export_path) api = None if dry_run else _magento_api(connection_id) results: List[Dict[str, Any]] = [] for page in pages: identifier = str(page.get("identifier") or "").strip() if identifiers and identifier not in identifiers: continue payload = build_cms_payload(page, include_base_styles=include_base_styles) result: Dict[str, Any] = { "identifier": identifier, "title": payload.get("title"), "content_length": len(payload.get("content") or ""), } if dry_run: result["action"] = "dry_run" result["payload"] = payload results.append(result) continue assert api is not None status, existing, err = api.find_cms_page_by_identifier(identifier) if err: result.update({"action": "lookup_failed", "status": status, "error": err}) results.append(result) continue if existing: if create_only: result.update({"action": "skipped_existing", "status": status, "page_id": existing.get("id") or existing.get("page_id")}) results.append(result) continue page_id = int(existing.get("id") or existing.get("page_id")) write_status, body, write_err = api.update_cms_page(page_id, payload) result.update({"action": "updated", "status": write_status, "page_id": page_id, "error": write_err, "response": body}) else: if update_only: result.update({"action": "skipped_missing", "status": status}) results.append(result) continue write_status, body, write_err = api.create_cms_page(payload) result.update({"action": "created", "status": write_status, "error": write_err, "response": body}) results.append(result) return results def main() -> int: logging.basicConfig(level=logging.INFO, format="%(levelname)s:%(name)s:%(message)s") parser = argparse.ArgumentParser(description="Create or update Magento CMS pages from an exported JSON file.") parser.add_argument("--connection-id", type=int, required=True) parser.add_argument("--export-path", type=Path, default=DEFAULT_EXPORT_PATH) parser.add_argument("--identifier", action="append", help="Only push this CMS page identifier. Repeatable.") parser.add_argument("--dry-run", action="store_true") parser.add_argument("--create-only", action="store_true") parser.add_argument("--update-only", action="store_true") parser.add_argument("--no-base-styles", action="store_true", help="Do not prepend the small non-Page-Builder responsive CSS block.") parser.add_argument("--report-path", type=Path) args = parser.parse_args() if args.create_only and args.update_only: parser.error("--create-only and --update-only cannot be used together") results = push_pages( connection_id=args.connection_id, export_path=args.export_path, identifiers=set(args.identifier or []) or None, dry_run=args.dry_run, create_only=args.create_only, update_only=args.update_only, include_base_styles=not args.no_base_styles, ) output = json.dumps(results, indent=2, default=str) if args.report_path: args.report_path.parent.mkdir(parents=True, exist_ok=True) args.report_path.write_text(output, encoding="utf-8") print(f"Wrote report to {args.report_path}") else: print(output) failed = [item for item in results if item.get("error") or str(item.get("action", "")).endswith("failed")] return 1 if failed else 0 if __name__ == "__main__": sys.exit(main())