- Auto-fixed 4,496 lint issues (import sorting, modern syntax, etc.) - Added ignore rules for patterns intentional in this codebase: E402 (late imports), E712 (SQLAlchemy filters), B904 (raise from), SIM108/SIM105/SIM117 (readability preferences) - Added per-file ignores for tests and scripts - Excluded broken scripts/rename_terminology.py (has curly quotes) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
253 lines
7.7 KiB
Python
253 lines
7.7 KiB
Python
# app/modules/cms/services/cms_metrics.py
|
|
"""
|
|
Metrics provider for the CMS module.
|
|
|
|
Provides metrics for:
|
|
- Content pages
|
|
- Media files
|
|
- Themes
|
|
"""
|
|
|
|
import logging
|
|
from typing import TYPE_CHECKING
|
|
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.modules.contracts.metrics import (
|
|
MetricsContext,
|
|
MetricValue,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
pass
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class CMSMetricsProvider:
|
|
"""
|
|
Metrics provider for CMS module.
|
|
|
|
Provides content management metrics for store and platform dashboards.
|
|
"""
|
|
|
|
@property
|
|
def metrics_category(self) -> str:
|
|
return "cms"
|
|
|
|
def get_store_metrics(
|
|
self,
|
|
db: Session,
|
|
store_id: int,
|
|
context: MetricsContext | None = None,
|
|
) -> list[MetricValue]:
|
|
"""
|
|
Get CMS metrics for a specific store.
|
|
|
|
Provides:
|
|
- Total content pages
|
|
- Published pages
|
|
- Media files count
|
|
- Theme status
|
|
"""
|
|
from app.modules.cms.models import ContentPage, MediaFile, StoreTheme
|
|
|
|
try:
|
|
# Content pages
|
|
total_pages = (
|
|
db.query(ContentPage).filter(ContentPage.store_id == store_id).count()
|
|
)
|
|
|
|
published_pages = (
|
|
db.query(ContentPage)
|
|
.filter(
|
|
ContentPage.store_id == store_id,
|
|
ContentPage.is_published == True,
|
|
)
|
|
.count()
|
|
)
|
|
|
|
# Media files
|
|
media_count = (
|
|
db.query(MediaFile).filter(MediaFile.store_id == store_id).count()
|
|
)
|
|
|
|
# Total media size (in MB)
|
|
total_media_size = (
|
|
db.query(func.sum(MediaFile.file_size))
|
|
.filter(MediaFile.store_id == store_id)
|
|
.scalar()
|
|
or 0
|
|
)
|
|
total_media_size_mb = round(total_media_size / (1024 * 1024), 2)
|
|
|
|
# Theme configured
|
|
has_theme = (
|
|
db.query(StoreTheme).filter(StoreTheme.store_id == store_id).first()
|
|
is not None
|
|
)
|
|
|
|
return [
|
|
MetricValue(
|
|
key="cms.total_pages",
|
|
value=total_pages,
|
|
label="Total Pages",
|
|
category="cms",
|
|
icon="file-text",
|
|
description="Total content pages created",
|
|
),
|
|
MetricValue(
|
|
key="cms.published_pages",
|
|
value=published_pages,
|
|
label="Published Pages",
|
|
category="cms",
|
|
icon="globe",
|
|
description="Content pages that are published",
|
|
),
|
|
MetricValue(
|
|
key="cms.media_count",
|
|
value=media_count,
|
|
label="Media Files",
|
|
category="cms",
|
|
icon="image",
|
|
description="Files in media library",
|
|
),
|
|
MetricValue(
|
|
key="cms.media_size",
|
|
value=total_media_size_mb,
|
|
label="Media Size",
|
|
category="cms",
|
|
icon="hard-drive",
|
|
unit="MB",
|
|
description="Total storage used by media",
|
|
),
|
|
MetricValue(
|
|
key="cms.has_theme",
|
|
value=1 if has_theme else 0,
|
|
label="Theme Configured",
|
|
category="cms",
|
|
icon="palette",
|
|
description="Whether a custom theme is configured",
|
|
),
|
|
]
|
|
except Exception as e:
|
|
logger.warning(f"Failed to get CMS store metrics: {e}")
|
|
return []
|
|
|
|
def get_platform_metrics(
|
|
self,
|
|
db: Session,
|
|
platform_id: int,
|
|
context: MetricsContext | None = None,
|
|
) -> list[MetricValue]:
|
|
"""
|
|
Get CMS metrics aggregated for a platform.
|
|
|
|
Aggregates content management data across all stores.
|
|
"""
|
|
from app.modules.cms.models import ContentPage, MediaFile, StoreTheme
|
|
from app.modules.tenancy.models import StorePlatform
|
|
|
|
try:
|
|
# Get all store IDs for this platform using StorePlatform junction table
|
|
store_ids = (
|
|
db.query(StorePlatform.store_id)
|
|
.filter(
|
|
StorePlatform.platform_id == platform_id,
|
|
StorePlatform.is_active == True,
|
|
)
|
|
.subquery()
|
|
)
|
|
|
|
# Content pages
|
|
total_pages = (
|
|
db.query(ContentPage)
|
|
.filter(ContentPage.store_id.in_(store_ids))
|
|
.count()
|
|
)
|
|
|
|
published_pages = (
|
|
db.query(ContentPage)
|
|
.filter(
|
|
ContentPage.store_id.in_(store_ids),
|
|
ContentPage.is_published == True,
|
|
)
|
|
.count()
|
|
)
|
|
|
|
# Media files
|
|
media_count = (
|
|
db.query(MediaFile).filter(MediaFile.store_id.in_(store_ids)).count()
|
|
)
|
|
|
|
# Total media size (in GB for platform-level)
|
|
total_media_size = (
|
|
db.query(func.sum(MediaFile.file_size))
|
|
.filter(MediaFile.store_id.in_(store_ids))
|
|
.scalar()
|
|
or 0
|
|
)
|
|
total_media_size_gb = round(total_media_size / (1024 * 1024 * 1024), 2)
|
|
|
|
# Stores with themes
|
|
stores_with_themes = (
|
|
db.query(func.count(func.distinct(StoreTheme.store_id)))
|
|
.filter(StoreTheme.store_id.in_(store_ids))
|
|
.scalar()
|
|
or 0
|
|
)
|
|
|
|
return [
|
|
MetricValue(
|
|
key="cms.total_pages",
|
|
value=total_pages,
|
|
label="Total Pages",
|
|
category="cms",
|
|
icon="file-text",
|
|
description="Total content pages across all stores",
|
|
),
|
|
MetricValue(
|
|
key="cms.published_pages",
|
|
value=published_pages,
|
|
label="Published Pages",
|
|
category="cms",
|
|
icon="globe",
|
|
description="Published content pages across all stores",
|
|
),
|
|
MetricValue(
|
|
key="cms.media_count",
|
|
value=media_count,
|
|
label="Media Files",
|
|
category="cms",
|
|
icon="image",
|
|
description="Total media files across all stores",
|
|
),
|
|
MetricValue(
|
|
key="cms.media_size",
|
|
value=total_media_size_gb,
|
|
label="Total Media Size",
|
|
category="cms",
|
|
icon="hard-drive",
|
|
unit="GB",
|
|
description="Total storage used by media",
|
|
),
|
|
MetricValue(
|
|
key="cms.stores_with_themes",
|
|
value=stores_with_themes,
|
|
label="Themed Stores",
|
|
category="cms",
|
|
icon="palette",
|
|
description="Stores with custom themes",
|
|
),
|
|
]
|
|
except Exception as e:
|
|
logger.warning(f"Failed to get CMS platform metrics: {e}")
|
|
return []
|
|
|
|
|
|
# Singleton instance
|
|
cms_metrics_provider = CMSMetricsProvider()
|
|
|
|
__all__ = ["CMSMetricsProvider", "cms_metrics_provider"]
|