Some checks failed
Remove all hardcoded OMS-specific content from platform base template: nav links, contact info, brand name, and footer columns. Everything is now dynamic via platform model and CMS page queries. Wire up legal_pages context (privacy/terms) from database instead of hardcoded fallback. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1123 lines
36 KiB
Python
1123 lines
36 KiB
Python
# 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 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 primary 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_primary_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
|
|
|
|
# =========================================================================
|
|
# 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()
|