"""Blog app service layer - business logic for blog operations."""
import logging
import re
import time
from datetime import datetime
from typing import Dict, List, Optional, Any
import kumiho
logger = logging.getLogger("kumiho.fastapi.blog")
# Simple in-memory cache for settings
# Key: project_name, Value: (timestamp, settings_dict)
_settings_cache: Dict[str, tuple] = {}
CACHE_TTL_SECONDS = 300 # 5 minutes
[docs]
def load_blog_settings(
client: Any,
project_name: str,
cache: Optional[Dict[str, Dict[str, object]]] = None,
) -> Dict[str, object]:
"""Load blog settings for a project."""
if cache is not None and project_name in cache:
return cache[project_name]
settings: Dict[str, object] = {}
try:
# Use project traversal to get the settings item
project = kumiho.get_project(project_name)
if project:
space = project.get_space(f"/{project_name}")
item = space.get_item("blog.settings", "config")
revision = item.get_latest_revision()
if revision and revision.metadata:
settings = revision.metadata
except Exception as exc:
logger.debug("blog.settings lookup failed for %s: %s", project_name, exc)
if cache is not None:
cache[project_name] = settings
return settings
[docs]
def get_post_item_kind(
client: Any,
project_name: str,
cache: Optional[Dict[str, Dict[str, object]]] = None,
) -> str:
"""Get the configured item kind for blog posts from settings."""
settings = load_blog_settings(client, project_name, cache)
if "post_item_kind" in settings:
return str(settings.get("post_item_kind", "post"))
return str(settings.get("post_product_type", "post"))
[docs]
def allow_public_read(
client: Any,
project_name: str,
cache: Optional[Dict[str, Dict[str, object]]] = None,
) -> bool:
"""Check if public read is allowed for a project's blog."""
# 1. Check blog settings first (override)
settings = load_blog_settings(client, project_name, cache)
value = settings.get("allow_public")
if value is not None:
if isinstance(value, bool):
return value
if isinstance(value, str):
return value.strip().lower() in {"1", "true", "yes", "public", "on"}
if isinstance(value, (int, float)):
return bool(value)
# 2. Fallback to Project settings
try:
project = kumiho.get_project(project_name)
if project:
return project.allow_public
except Exception as e:
logger.warning(f"Failed to check project allow_public for {project_name}: {e}")
return False
[docs]
def resolve_project_name(space_path: Optional[str], project_filter: Optional[str]) -> Optional[str]:
"""Extract project name from space path or project filter."""
if project_filter:
return project_filter
if space_path:
path_parts = [p for p in space_path.strip('/').split('/') if p]
if path_parts:
return path_parts[0]
return None
[docs]
def get_or_create_space_hierarchy(
client: Any,
project: kumiho.Project,
space_names: List[str]
) -> kumiho.Space:
"""
Navigate or create the space hierarchy under a project.
Returns the leaf Space object.
"""
from fastapi import HTTPException
try:
current_space = project.get_space(f"/{project.name}")
except Exception:
raise HTTPException(
status_code=500,
detail=f"Project root space /{project.name} not found"
)
for name in space_names:
try:
current_space = current_space.get_space(name)
except Exception:
current_space = current_space.create_space(name)
return current_space
[docs]
def get_cached_settings(project_name: str) -> Optional[Dict[str, object]]:
"""Get cached settings if still valid."""
if project_name in _settings_cache:
timestamp, cached_settings = _settings_cache[project_name]
if time.time() - timestamp < CACHE_TTL_SECONDS:
return cached_settings
return None
[docs]
def set_cached_settings(project_name: str, settings: Dict[str, object]) -> None:
"""Update the settings cache."""
_settings_cache[project_name] = (time.time(), settings)
[docs]
def generate_slug(title: str) -> str:
"""Generate a URL-safe slug from a title."""
slug = title.lower().strip()
slug = re.sub(r'[^a-z0-9]+', '-', slug)
slug = slug.strip('-')
return slug or "post"
[docs]
def build_default_settings(project_name: str, allow_public: bool = False) -> Dict[str, object]:
"""Build default blog settings dict."""
return {
"project_name": project_name,
"post_item_kind": "post",
"pagination_count": 10,
"display_type": "title_only",
"display_category_filters": True,
"allow_public": allow_public
}
[docs]
def ensure_blog_project(project_name: str) -> kumiho.Project:
"""
Ensure the blog project and its settings exist.
Creates them if they don't exist.
"""
project = kumiho.get_project(project_name)
if not project:
logger.info(f"Creating missing blog project: {project_name}")
project = kumiho.create_project(name=project_name, description="Blog project")
project.set_public(True) # Default to public for blog projects
# Ensure root space exists
try:
project.get_space(f"/{project_name}")
except Exception:
logger.info(f"Creating missing root space for {project_name}")
try:
# Create root space using project.create_space with parent_path="/"
project.create_space(name=project_name, parent_path="/")
except Exception as e:
logger.error(f"Failed to create root space: {e}")
# Continue anyway, maybe it exists but get_space failed for other reasons?
# Ensure settings item exists
try:
space = project.get_space(f"/{project_name}")
try:
space.get_item("blog.settings", "config")
except Exception:
logger.info(f"Creating missing blog settings for {project_name}")
item = space.create_item("blog.settings", "config")
# Create initial revision with defaults
defaults = build_default_settings(project_name, allow_public=True)
metadata = {k: str(v) for k, v in defaults.items()}
metadata["updated_at"] = datetime.now().isoformat()
item.create_revision(metadata=metadata)
except Exception as e:
logger.warning(f"Failed to ensure blog settings: {e}")
return project