# app/modules/cms/services/content_page_service.py """ Content Page Service Business logic for managing content pages with three-tier hierarchy: 1. Platform Marketing Pages (is_platform_page=True, store_id=NULL) - Platform's own pages (homepage, pricing, about) - Describe the platform/business offering itself 2. Store Default Pages (is_platform_page=False, store_id=NULL) - Fallback pages for stores who haven't customized - About Us, Shipping Policy, Return Policy, etc. 3. Store Override/Custom Pages (is_platform_page=False, store_id=set) - Store-specific customizations - Either overrides a default or is a completely custom page Lookup Strategy for Store Storefronts: 1. Check for store override (platform_id + store_id + slug + published) 2. If not found, check for store default (platform_id + store_id=NULL + is_platform_page=False + slug) 3. If neither exists, return None/404 """ import logging from datetime import UTC, datetime from typing import Any from sqlalchemy import and_ from sqlalchemy.orm import Session from app.modules.cms.exceptions import ( ContentPageNotFoundException, UnauthorizedContentPageAccessException, ) from app.modules.cms.models import ContentPage logger = logging.getLogger(__name__) class ContentPageService: """Service for content page operations with multi-platform support.""" # ========================================================================= # Platform Resolution # ========================================================================= @staticmethod def resolve_platform_id(db: Session, store_id: int) -> int | None: """ Resolve platform_id from store's first active StorePlatform. Resolution order: 1. Primary StorePlatform for the store 2. Any active StorePlatform for the store (fallback) Args: db: Database session store_id: Store ID Returns: Platform ID or None if no platform association found """ from app.modules.tenancy.services.platform_service import platform_service return platform_service.get_first_active_platform_id_for_store(db, store_id) @staticmethod def resolve_platform_id_or_raise(db: Session, store_id: int) -> int: """ Resolve platform_id or raise NoPlatformSubscriptionException. Args: db: Database session store_id: Store ID Returns: Platform ID Raises: NoPlatformSubscriptionException: If no platform found """ from app.modules.cms.exceptions import NoPlatformSubscriptionException platform_id = ContentPageService.resolve_platform_id(db, store_id) if platform_id is None: raise NoPlatformSubscriptionException(store_id=store_id) return platform_id # ========================================================================= # Three-Tier Resolution Methods (for store storefronts) # ========================================================================= @staticmethod def get_page_for_store( db: Session, platform_id: int, slug: str, store_id: int | None = None, include_unpublished: bool = False, ) -> ContentPage | None: """ Get content page with three-tier resolution. Resolution order: 1. Store override (platform_id + store_id + slug) 2. Store default (platform_id + store_id=NULL + is_platform_page=False + slug) Args: db: Database session platform_id: Platform ID (required for multi-platform support) slug: Page slug (about, faq, contact, etc.) store_id: Store ID (None for defaults only) include_unpublished: Include draft pages (for preview) Returns: ContentPage or None """ base_filters = [ ContentPage.platform_id == platform_id, ContentPage.slug == slug, ] if not include_unpublished: base_filters.append(ContentPage.is_published == True) # Tier 1: Try store-specific override first if store_id: store_page = ( db.query(ContentPage) .filter(and_(ContentPage.store_id == store_id, *base_filters)) .first() ) if store_page: logger.debug( f"[CMS] Found store override: {slug} for store_id={store_id}, platform_id={platform_id}" ) return store_page # Tier 2: Fallback to store default (not platform page) store_default_page = ( db.query(ContentPage) .filter( and_( ContentPage.store_id.is_(None), ContentPage.is_platform_page.is_(False), *base_filters, ) ) .first() ) if store_default_page: logger.debug(f"[CMS] Using store default page: {slug} for platform_id={platform_id}") return store_default_page logger.debug(f"[CMS] No page found for slug: {slug}, platform_id={platform_id}") return None @staticmethod def get_platform_page( db: Session, platform_id: int, slug: str, include_unpublished: bool = False, ) -> ContentPage | None: """ Get a platform marketing page. Platform marketing pages are pages that describe the platform itself (homepage, pricing, about, features, etc.). Args: db: Database session platform_id: Platform ID slug: Page slug include_unpublished: Include draft pages Returns: ContentPage or None """ filters = [ ContentPage.platform_id == platform_id, ContentPage.slug == slug, ContentPage.store_id.is_(None), ContentPage.is_platform_page.is_(True), ] if not include_unpublished: filters.append(ContentPage.is_published.is_(True)) page = db.query(ContentPage).filter(and_(*filters)).first() if page: logger.debug(f"[CMS] Found platform page: {slug} for platform_id={platform_id}") else: logger.debug(f"[CMS] No platform page found: {slug} for platform_id={platform_id}") return page @staticmethod def list_pages_for_store( db: Session, platform_id: int, store_id: int | None = None, include_unpublished: bool = False, footer_only: bool = False, header_only: bool = False, legal_only: bool = False, ) -> list[ContentPage]: """ List all available pages for a store storefront. Merges store overrides with store defaults, prioritizing overrides. Does NOT include platform marketing pages. Args: db: Database session platform_id: Platform ID store_id: Store ID (None for store defaults only) include_unpublished: Include draft pages footer_only: Only pages marked for footer display header_only: Only pages marked for header display legal_only: Only pages marked for legal/bottom bar display Returns: List of ContentPage objects """ base_filters = [ContentPage.platform_id == platform_id] if not include_unpublished: base_filters.append(ContentPage.is_published == True) if footer_only: base_filters.append(ContentPage.show_in_footer == True) if header_only: base_filters.append(ContentPage.show_in_header == True) if legal_only: base_filters.append(ContentPage.show_in_legal == True) # Get store-specific pages store_pages = [] if store_id: store_pages = ( db.query(ContentPage) .filter(and_(ContentPage.store_id == store_id, *base_filters)) .order_by(ContentPage.display_order, ContentPage.title) .all() ) # Get store defaults (not platform marketing pages) store_default_pages = ( db.query(ContentPage) .filter( and_( ContentPage.store_id.is_(None), ContentPage.is_platform_page.is_(False), *base_filters, ) ) .order_by(ContentPage.display_order, ContentPage.title) .all() ) # Merge: store overrides take precedence store_slugs = {page.slug for page in store_pages} all_pages = store_pages + [ page for page in store_default_pages if page.slug not in store_slugs ] # Sort by display_order all_pages.sort(key=lambda p: (p.display_order, p.title)) return all_pages @staticmethod def list_platform_pages( db: Session, platform_id: int, include_unpublished: bool = False, footer_only: bool = False, header_only: bool = False, legal_only: bool = False, ) -> list[ContentPage]: """ List platform marketing pages. Args: db: Database session platform_id: Platform ID include_unpublished: Include draft pages footer_only: Only pages marked for footer display header_only: Only pages marked for header display legal_only: Only legal pages (privacy, terms) — not in header or footer nav Returns: List of platform marketing ContentPage objects """ filters = [ ContentPage.platform_id == platform_id, ContentPage.store_id.is_(None), ContentPage.is_platform_page.is_(True), ] if not include_unpublished: filters.append(ContentPage.is_published.is_(True)) if footer_only: filters.append(ContentPage.show_in_footer == True) if header_only: filters.append(ContentPage.show_in_header == True) if legal_only: filters.append(ContentPage.slug.in_(["privacy", "terms"])) return ( db.query(ContentPage) .filter(and_(*filters)) .order_by(ContentPage.display_order, ContentPage.title) .all() ) @staticmethod def get_store_default_page( db: Session, platform_id: int, slug: str, include_unpublished: bool = False, ) -> ContentPage | None: """ Get a single store default page by slug (fallback for stores who haven't customized). These are non-platform-marketing pages with store_id=NULL. Args: db: Database session platform_id: Platform ID slug: Page slug include_unpublished: Include draft pages Returns: ContentPage or None """ filters = [ ContentPage.platform_id == platform_id, ContentPage.slug == slug, ContentPage.store_id.is_(None), ContentPage.is_platform_page.is_(False), ] if not include_unpublished: filters.append(ContentPage.is_published.is_(True)) page = db.query(ContentPage).filter(and_(*filters)).first() if page: logger.debug(f"[CMS] Found store default page: {slug} for platform_id={platform_id}") else: logger.debug(f"[CMS] No store default page found: {slug} for platform_id={platform_id}") return page @staticmethod def list_store_defaults( db: Session, platform_id: int, include_unpublished: bool = False, ) -> list[ContentPage]: """ List store default pages (fallbacks for stores who haven't customized). Args: db: Database session platform_id: Platform ID include_unpublished: Include draft pages Returns: List of store default ContentPage objects """ filters = [ ContentPage.platform_id == platform_id, ContentPage.store_id.is_(None), ContentPage.is_platform_page.is_(False), ] if not include_unpublished: filters.append(ContentPage.is_published.is_(True)) return ( db.query(ContentPage) .filter(and_(*filters)) .order_by(ContentPage.display_order, ContentPage.title) .all() ) @staticmethod def list_all_platform_pages( db: Session, include_unpublished: bool = False, ) -> list[ContentPage]: """ List all platform marketing pages across all platforms (for admin use). Args: db: Database session include_unpublished: Include draft pages Returns: List of all platform marketing ContentPage objects """ filters = [ ContentPage.store_id.is_(None), ContentPage.is_platform_page.is_(True), ] if not include_unpublished: filters.append(ContentPage.is_published.is_(True)) return ( db.query(ContentPage) .filter(and_(*filters)) .order_by(ContentPage.platform_id, ContentPage.display_order, ContentPage.title) .all() ) @staticmethod def list_all_store_defaults( db: Session, include_unpublished: bool = False, ) -> list[ContentPage]: """ List all store default pages across all platforms (for admin use). Args: db: Database session include_unpublished: Include draft pages Returns: List of all store default ContentPage objects """ filters = [ ContentPage.store_id.is_(None), ContentPage.is_platform_page.is_(False), ] if not include_unpublished: filters.append(ContentPage.is_published.is_(True)) return ( db.query(ContentPage) .filter(and_(*filters)) .order_by(ContentPage.platform_id, ContentPage.display_order, ContentPage.title) .all() ) # ========================================================================= # CRUD Methods # ========================================================================= @staticmethod def create_page( db: Session, platform_id: int, slug: str, title: str, content: str, store_id: int | None = None, is_platform_page: bool = False, content_format: str = "html", template: str = "default", meta_description: str | None = None, meta_keywords: str | None = None, is_published: bool = False, show_in_footer: bool = True, show_in_header: bool = False, show_in_legal: bool = False, display_order: int = 0, created_by: int | None = None, ) -> ContentPage: """ Create a new content page. Args: db: Database session platform_id: Platform ID (required) slug: URL-safe identifier title: Page title content: HTML or Markdown content store_id: Store ID (None for platform/default pages) is_platform_page: True for platform marketing pages content_format: "html" or "markdown" template: Template name for landing pages meta_description: SEO description meta_keywords: SEO keywords is_published: Publish immediately show_in_footer: Show in footer navigation show_in_header: Show in header navigation show_in_legal: Show in legal/bottom bar navigation display_order: Sort order created_by: User ID who created it Returns: Created ContentPage """ page = ContentPage( platform_id=platform_id, store_id=store_id, is_platform_page=is_platform_page, slug=slug, title=title, content=content, content_format=content_format, template=template, meta_description=meta_description, meta_keywords=meta_keywords, is_published=is_published, published_at=datetime.now(UTC) if is_published else None, show_in_footer=show_in_footer, show_in_header=show_in_header, show_in_legal=show_in_legal, display_order=display_order, created_by=created_by, updated_by=created_by, ) db.add(page) db.flush() db.refresh(page) page_type = "platform" if is_platform_page else ("store" if store_id else "default") logger.info( f"[CMS] Created {page_type} page: {slug} (platform_id={platform_id}, store_id={store_id}, id={page.id})" ) return page @staticmethod def update_page( db: Session, page_id: int, title: str | None = None, content: str | None = None, content_format: str | None = None, template: str | None = None, meta_description: str | None = None, meta_keywords: str | None = None, is_published: bool | None = None, show_in_footer: bool | None = None, show_in_header: bool | None = None, show_in_legal: bool | None = None, display_order: int | None = None, updated_by: int | None = None, ) -> ContentPage | None: """ Update an existing content page. Args: db: Database session page_id: Page ID ... other fields Returns: Updated ContentPage or None if not found """ page = db.query(ContentPage).filter(ContentPage.id == page_id).first() if not page: logger.warning(f"[CMS] Content page not found: id={page_id}") return None # Update fields if provided if title is not None: page.title = title if content is not None: page.content = content if content_format is not None: page.content_format = content_format if template is not None: page.template = template if meta_description is not None: page.meta_description = meta_description if meta_keywords is not None: page.meta_keywords = meta_keywords if is_published is not None: page.is_published = is_published if is_published and not page.published_at: page.published_at = datetime.now(UTC) if show_in_footer is not None: page.show_in_footer = show_in_footer if show_in_header is not None: page.show_in_header = show_in_header if show_in_legal is not None: page.show_in_legal = show_in_legal if display_order is not None: page.display_order = display_order if updated_by is not None: page.updated_by = updated_by db.flush() db.refresh(page) logger.info(f"[CMS] Updated content page: id={page_id}, slug={page.slug}") return page @staticmethod def delete_page(db: Session, page_id: int) -> bool: """ Delete a content page. Args: db: Database session page_id: Page ID Returns: True if deleted, False if not found """ page = db.query(ContentPage).filter(ContentPage.id == page_id).first() if not page: logger.warning(f"[CMS] Content page not found for deletion: id={page_id}") return False db.delete(page) logger.info(f"[CMS] Deleted content page: id={page_id}, slug={page.slug}") return True @staticmethod def get_page_by_id(db: Session, page_id: int) -> ContentPage | None: """Get content page by ID.""" return db.query(ContentPage).filter(ContentPage.id == page_id).first() @staticmethod def get_page_by_id_or_raise(db: Session, page_id: int) -> ContentPage: """ Get content page by ID or raise ContentPageNotFoundException. """ page = db.query(ContentPage).filter(ContentPage.id == page_id).first() if not page: raise ContentPageNotFoundException(identifier=page_id) return page @staticmethod def get_page_for_store_or_raise( db: Session, platform_id: int, slug: str, store_id: int | None = None, include_unpublished: bool = False, ) -> ContentPage: """ Get content page for a store with three-tier resolution. Raises ContentPageNotFoundException if not found. """ page = ContentPageService.get_page_for_store( db, platform_id=platform_id, slug=slug, store_id=store_id, include_unpublished=include_unpublished, ) if not page: raise ContentPageNotFoundException(identifier=slug) return page @staticmethod def get_platform_page_or_raise( db: Session, platform_id: int, slug: str, include_unpublished: bool = False, ) -> ContentPage: """ Get platform marketing page or raise ContentPageNotFoundException. """ page = ContentPageService.get_platform_page( db, platform_id=platform_id, slug=slug, include_unpublished=include_unpublished, ) if not page: raise ContentPageNotFoundException(identifier=slug) return page # ========================================================================= # Store Page Management (with ownership checks) # ========================================================================= @staticmethod def update_store_page( db: Session, page_id: int, store_id: int, title: str | None = None, content: str | None = None, content_format: str | None = None, meta_description: str | None = None, meta_keywords: str | None = None, is_published: bool | None = None, show_in_footer: bool | None = None, show_in_header: bool | None = None, show_in_legal: bool | None = None, display_order: int | None = None, updated_by: int | None = None, ) -> ContentPage: """ Update a store-specific content page with ownership check. Raises: ContentPageNotFoundException: If page not found UnauthorizedContentPageAccessException: If page doesn't belong to store """ page = ContentPageService.get_page_by_id_or_raise(db, page_id) if page.store_id != store_id: raise UnauthorizedContentPageAccessException(action="edit") return ContentPageService.update_page( db, page_id=page_id, title=title, content=content, content_format=content_format, meta_description=meta_description, meta_keywords=meta_keywords, is_published=is_published, show_in_footer=show_in_footer, show_in_header=show_in_header, show_in_legal=show_in_legal, display_order=display_order, updated_by=updated_by, ) @staticmethod def delete_store_page(db: Session, page_id: int, store_id: int) -> None: """ Delete a store-specific content page with ownership check. Raises: ContentPageNotFoundException: If page not found UnauthorizedContentPageAccessException: If page doesn't belong to store """ page = ContentPageService.get_page_by_id_or_raise(db, page_id) if page.store_id != store_id: raise UnauthorizedContentPageAccessException(action="delete") ContentPageService.delete_page(db, page_id) @staticmethod def create_store_override( db: Session, platform_id: int, store_id: int, slug: str, title: str, content: str, content_format: str = "html", meta_description: str | None = None, meta_keywords: str | None = None, is_published: bool = False, show_in_footer: bool = True, show_in_header: bool = False, show_in_legal: bool = False, display_order: int = 0, created_by: int | None = None, ) -> ContentPage: """ Create a store override page (store-specific customization of a default). Args: db: Database session platform_id: Platform ID store_id: Store ID slug: Page slug (typically matches a default page) ... other fields Returns: Created ContentPage """ return ContentPageService.create_page( db, platform_id=platform_id, slug=slug, title=title, content=content, store_id=store_id, is_platform_page=False, content_format=content_format, meta_description=meta_description, meta_keywords=meta_keywords, is_published=is_published, show_in_footer=show_in_footer, show_in_header=show_in_header, show_in_legal=show_in_legal, display_order=display_order, created_by=created_by, ) @staticmethod def revert_to_default(db: Session, page_id: int, store_id: int) -> None: """ Revert a store override to the default by deleting the override. After deletion, the store storefront will use the store default page. Raises: ContentPageNotFoundException: If page not found UnauthorizedContentPageAccessException: If page doesn't belong to store """ ContentPageService.delete_store_page(db, page_id, store_id) # ========================================================================= # Admin Methods (for listing all pages) # ========================================================================= @staticmethod def list_all_pages( db: Session, platform_id: int | None = None, store_id: int | None = None, include_unpublished: bool = False, page_tier: str | None = None, ) -> list[ContentPage]: """ List all content pages for admin management. Args: db: Database session platform_id: Optional filter by platform ID store_id: Optional filter by store ID include_unpublished: Include draft pages page_tier: Optional filter by tier ("platform", "store_default", "store_override") Returns: List of ContentPage objects """ filters = [] if platform_id: filters.append(ContentPage.platform_id == platform_id) if store_id is not None: filters.append(ContentPage.store_id == store_id) if not include_unpublished: filters.append(ContentPage.is_published == True) if page_tier == "platform": filters.append(ContentPage.is_platform_page.is_(True)) filters.append(ContentPage.store_id.is_(None)) elif page_tier == "store_default": filters.append(ContentPage.is_platform_page.is_(False)) filters.append(ContentPage.store_id.is_(None)) elif page_tier == "store_override": filters.append(ContentPage.store_id.isnot(None)) return ( db.query(ContentPage) .filter(and_(*filters) if filters else True) .order_by( ContentPage.platform_id, ContentPage.store_id, ContentPage.display_order, ContentPage.title, ) .all() ) @staticmethod def list_all_store_pages( db: Session, store_id: int, platform_id: int | None = None, include_unpublished: bool = False, ) -> list[ContentPage]: """ List only store-specific pages (overrides and custom pages). Args: db: Database session store_id: Store ID platform_id: Optional filter by platform include_unpublished: Include draft pages Returns: List of store-specific ContentPage objects """ filters = [ContentPage.store_id == store_id] if platform_id: filters.append(ContentPage.platform_id == platform_id) if not include_unpublished: filters.append(ContentPage.is_published == True) return ( db.query(ContentPage) .filter(and_(*filters)) .order_by(ContentPage.display_order, ContentPage.title) .all() ) # ========================================================================= # Helper Methods for raising exceptions # ========================================================================= @staticmethod def update_page_or_raise( db: Session, page_id: int, title: str | None = None, content: str | None = None, content_format: str | None = None, template: str | None = None, meta_description: str | None = None, meta_keywords: str | None = None, is_published: bool | None = None, show_in_footer: bool | None = None, show_in_header: bool | None = None, show_in_legal: bool | None = None, display_order: int | None = None, updated_by: int | None = None, ) -> ContentPage: """ Update an existing content page or raise exception. Raises: ContentPageNotFoundException: If page not found """ page = ContentPageService.update_page( db, page_id=page_id, title=title, content=content, content_format=content_format, template=template, meta_description=meta_description, meta_keywords=meta_keywords, is_published=is_published, show_in_footer=show_in_footer, show_in_header=show_in_header, show_in_legal=show_in_legal, display_order=display_order, updated_by=updated_by, ) if not page: raise ContentPageNotFoundException(identifier=page_id) return page @staticmethod def delete_page_or_raise(db: Session, page_id: int) -> None: """ Delete a content page or raise exception. Raises: ContentPageNotFoundException: If page not found """ success = ContentPageService.delete_page(db, page_id) if not success: raise ContentPageNotFoundException(identifier=page_id) # ========================================================================= # Placeholder Resolution (for store default pages) # ========================================================================= @staticmethod def resolve_placeholders(content: str, store) -> str: """ Replace {{store_name}}, {{store_email}}, {{store_phone}} placeholders in store default page content with actual store values. Args: content: HTML content with placeholders store: Store object with name, contact_email, phone attributes Returns: Content with placeholders replaced """ if not content or not store: return content or "" replacements = { "{{store_name}}": store.name or "Our Store", "{{store_email}}": getattr(store, "contact_email", "") or "", "{{store_phone}}": getattr(store, "phone", "") or "", } for placeholder, value in replacements.items(): content = content.replace(placeholder, value) return content @staticmethod def resolve_placeholders_deep(data, store) -> Any: """ Recursively resolve {{store_name}} etc. in a nested data structure (dicts, lists, strings). Used for sections JSON in store default pages. """ if not data or not store: return data if isinstance(data, str): return ContentPageService.resolve_placeholders(data, store) if isinstance(data, dict): return { k: ContentPageService.resolve_placeholders_deep(v, store) for k, v in data.items() } if isinstance(data, list): return [ ContentPageService.resolve_placeholders_deep(item, store) for item in data ] return data # ========================================================================= # Homepage Sections Management # ========================================================================= @staticmethod def update_homepage_sections( db: Session, page_id: int, sections: dict, updated_by: int | None = None, ) -> ContentPage: """ Update homepage sections with validation. Args: db: Database session page_id: Content page ID sections: Homepage sections dict (validated against HomepageSections schema) updated_by: User ID making the update Returns: Updated ContentPage Raises: ContentPageNotFoundException: If page not found ValidationError: If sections schema invalid """ from app.modules.cms.schemas import HomepageSections page = ContentPageService.get_page_by_id_or_raise(db, page_id) # Validate sections against schema validated = HomepageSections(**sections) # Update page page.sections = validated.model_dump() page.updated_by = updated_by db.flush() db.refresh(page) logger.info(f"[CMS] Updated homepage sections for page_id={page_id}") return page @staticmethod def update_single_section( db: Session, page_id: int, section_name: str, section_data: dict, updated_by: int | None = None, ) -> ContentPage: """ Update a single section within homepage sections. Args: db: Database session page_id: Content page ID section_name: Section to update (hero, features, pricing, cta) section_data: Section configuration dict updated_by: User ID making the update Returns: Updated ContentPage Raises: ContentPageNotFoundException: If page not found ValueError: If section name is invalid """ from app.modules.cms.schemas import ( CTASection, FeaturesSection, HeroSection, PricingSection, ProductsSection, ) SECTION_SCHEMAS = { "hero": HeroSection, "products": ProductsSection, "features": FeaturesSection, "pricing": PricingSection, "cta": CTASection, } if section_name not in SECTION_SCHEMAS: raise ValueError(f"Invalid section name: {section_name}. Must be one of: {list(SECTION_SCHEMAS.keys())}") page = ContentPageService.get_page_by_id_or_raise(db, page_id) # Validate section data against its schema schema = SECTION_SCHEMAS[section_name] validated_section = schema(**section_data) # Initialize sections if needed current_sections = page.sections or {} current_sections[section_name] = validated_section.model_dump() page.sections = current_sections page.updated_by = updated_by db.flush() db.refresh(page) logger.info(f"[CMS] Updated section '{section_name}' for page_id={page_id}") return page @staticmethod def get_default_sections(languages: list[str] | None = None) -> dict: """ Get empty sections structure for new homepage. Args: languages: List of language codes from platform.supported_languages. Defaults to ['fr', 'de', 'en'] if not provided. Returns: Empty sections dict with language placeholders """ from app.modules.cms.schemas import HomepageSections if languages is None: languages = ["fr", "de", "en"] return HomepageSections.get_empty_structure(languages).model_dump() # Singleton instance content_page_service = ContentPageService()