# 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, vendor_id=NULL) - Platform's own pages (homepage, pricing, about) - Describe the platform/business offering itself 2. Vendor Default Pages (is_platform_page=False, vendor_id=NULL) - Fallback pages for vendors who haven't customized - About Us, Shipping Policy, Return Policy, etc. 3. Vendor Override/Custom Pages (is_platform_page=False, vendor_id=set) - Vendor-specific customizations - Either overrides a default or is a completely custom page Lookup Strategy for Vendor Storefronts: 1. Check for vendor override (platform_id + vendor_id + slug + published) 2. If not found, check for vendor default (platform_id + vendor_id=NULL + is_platform_page=False + slug) 3. If neither exists, return None/404 """ import logging from datetime import UTC, datetime 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.""" # ========================================================================= # Three-Tier Resolution Methods (for vendor storefronts) # ========================================================================= @staticmethod def get_page_for_vendor( db: Session, platform_id: int, slug: str, vendor_id: int | None = None, include_unpublished: bool = False, ) -> ContentPage | None: """ Get content page with three-tier resolution. Resolution order: 1. Vendor override (platform_id + vendor_id + slug) 2. Vendor default (platform_id + vendor_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.) vendor_id: Vendor 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 vendor-specific override first if vendor_id: vendor_page = ( db.query(ContentPage) .filter(and_(ContentPage.vendor_id == vendor_id, *base_filters)) .first() ) if vendor_page: logger.debug( f"[CMS] Found vendor override: {slug} for vendor_id={vendor_id}, platform_id={platform_id}" ) return vendor_page # Tier 2: Fallback to vendor default (not platform page) vendor_default_page = ( db.query(ContentPage) .filter( and_( ContentPage.vendor_id == None, ContentPage.is_platform_page == False, *base_filters, ) ) .first() ) if vendor_default_page: logger.debug(f"[CMS] Using vendor default page: {slug} for platform_id={platform_id}") return vendor_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.vendor_id == None, ContentPage.is_platform_page == True, ] if not include_unpublished: filters.append(ContentPage.is_published == 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_vendor( db: Session, platform_id: int, vendor_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 vendor storefront. Merges vendor overrides with vendor defaults, prioritizing overrides. Does NOT include platform marketing pages. Args: db: Database session platform_id: Platform ID vendor_id: Vendor ID (None for vendor 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 vendor-specific pages vendor_pages = [] if vendor_id: vendor_pages = ( db.query(ContentPage) .filter(and_(ContentPage.vendor_id == vendor_id, *base_filters)) .order_by(ContentPage.display_order, ContentPage.title) .all() ) # Get vendor defaults (not platform marketing pages) vendor_default_pages = ( db.query(ContentPage) .filter( and_( ContentPage.vendor_id == None, ContentPage.is_platform_page == False, *base_filters, ) ) .order_by(ContentPage.display_order, ContentPage.title) .all() ) # Merge: vendor overrides take precedence vendor_slugs = {page.slug for page in vendor_pages} all_pages = vendor_pages + [ page for page in vendor_default_pages if page.slug not in vendor_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, ) -> 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 Returns: List of platform marketing ContentPage objects """ filters = [ ContentPage.platform_id == platform_id, ContentPage.vendor_id == None, ContentPage.is_platform_page == True, ] if not include_unpublished: filters.append(ContentPage.is_published == True) if footer_only: filters.append(ContentPage.show_in_footer == True) if header_only: filters.append(ContentPage.show_in_header == True) return ( db.query(ContentPage) .filter(and_(*filters)) .order_by(ContentPage.display_order, ContentPage.title) .all() ) @staticmethod def list_vendor_defaults( db: Session, platform_id: int, include_unpublished: bool = False, ) -> list[ContentPage]: """ List vendor default pages (fallbacks for vendors who haven't customized). Args: db: Database session platform_id: Platform ID include_unpublished: Include draft pages Returns: List of vendor default ContentPage objects """ filters = [ ContentPage.platform_id == platform_id, ContentPage.vendor_id == None, ContentPage.is_platform_page == False, ] 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() ) @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.vendor_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_vendor_defaults( db: Session, include_unpublished: bool = False, ) -> list[ContentPage]: """ List all vendor default pages across all platforms (for admin use). Args: db: Database session include_unpublished: Include draft pages Returns: List of all vendor default ContentPage objects """ filters = [ ContentPage.vendor_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, vendor_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 vendor_id: Vendor 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, vendor_id=vendor_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 ("vendor" if vendor_id else "default") logger.info( f"[CMS] Created {page_type} page: {slug} (platform_id={platform_id}, vendor_id={vendor_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_vendor_or_raise( db: Session, platform_id: int, slug: str, vendor_id: int | None = None, include_unpublished: bool = False, ) -> ContentPage: """ Get content page for a vendor with three-tier resolution. Raises ContentPageNotFoundException if not found. """ page = ContentPageService.get_page_for_vendor( db, platform_id=platform_id, slug=slug, vendor_id=vendor_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 # ========================================================================= # Vendor Page Management (with ownership checks) # ========================================================================= @staticmethod def update_vendor_page( db: Session, page_id: int, vendor_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 vendor-specific content page with ownership check. Raises: ContentPageNotFoundException: If page not found UnauthorizedContentPageAccessException: If page doesn't belong to vendor """ page = ContentPageService.get_page_by_id_or_raise(db, page_id) if page.vendor_id != vendor_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_vendor_page(db: Session, page_id: int, vendor_id: int) -> None: """ Delete a vendor-specific content page with ownership check. Raises: ContentPageNotFoundException: If page not found UnauthorizedContentPageAccessException: If page doesn't belong to vendor """ page = ContentPageService.get_page_by_id_or_raise(db, page_id) if page.vendor_id != vendor_id: raise UnauthorizedContentPageAccessException(action="delete") ContentPageService.delete_page(db, page_id) @staticmethod def create_vendor_override( db: Session, platform_id: int, vendor_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 vendor override page (vendor-specific customization of a default). Args: db: Database session platform_id: Platform ID vendor_id: Vendor 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, vendor_id=vendor_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, vendor_id: int) -> None: """ Revert a vendor override to the default by deleting the override. After deletion, the vendor storefront will use the vendor default page. Raises: ContentPageNotFoundException: If page not found UnauthorizedContentPageAccessException: If page doesn't belong to vendor """ ContentPageService.delete_vendor_page(db, page_id, vendor_id) # ========================================================================= # Admin Methods (for listing all pages) # ========================================================================= @staticmethod def list_all_pages( db: Session, platform_id: int | None = None, vendor_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 vendor_id: Optional filter by vendor ID include_unpublished: Include draft pages page_tier: Optional filter by tier ("platform", "vendor_default", "vendor_override") Returns: List of ContentPage objects """ filters = [] if platform_id: filters.append(ContentPage.platform_id == platform_id) if vendor_id is not None: filters.append(ContentPage.vendor_id == vendor_id) if not include_unpublished: filters.append(ContentPage.is_published == True) if page_tier == "platform": filters.append(ContentPage.is_platform_page == True) filters.append(ContentPage.vendor_id == None) elif page_tier == "vendor_default": filters.append(ContentPage.is_platform_page == False) filters.append(ContentPage.vendor_id == None) elif page_tier == "vendor_override": filters.append(ContentPage.vendor_id != None) return ( db.query(ContentPage) .filter(and_(*filters) if filters else True) .order_by( ContentPage.platform_id, ContentPage.vendor_id, ContentPage.display_order, ContentPage.title, ) .all() ) @staticmethod def list_all_vendor_pages( db: Session, vendor_id: int, platform_id: int | None = None, include_unpublished: bool = False, ) -> list[ContentPage]: """ List only vendor-specific pages (overrides and custom pages). Args: db: Database session vendor_id: Vendor ID platform_id: Optional filter by platform include_unpublished: Include draft pages Returns: List of vendor-specific ContentPage objects """ filters = [ContentPage.vendor_id == vendor_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) # ========================================================================= # 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 models.schema.homepage_sections 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 models.schema.homepage_sections import ( HeroSection, FeaturesSection, PricingSection, CTASection, ) SECTION_SCHEMAS = { "hero": HeroSection, "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 models.schema.homepage_sections import HomepageSections if languages is None: languages = ["fr", "de", "en"] return HomepageSections.get_empty_structure(languages).model_dump() # Singleton instance content_page_service = ContentPageService()