Improve content loading, TOML utilities, runtime setup, theme caching, and login throttle pruning

apps/panel/services.py

@@ -110,6 +110,7 @@ def build_throttle_keys(request, username: str) -> list[str]:
 
 
 def get_login_lockout(request, username: str) -> LockoutState:
+    prune_stale_login_throttles()
     now = timezone.now()
     locked_until = None
 
@@ -126,6 +127,7 @@ def get_login_lockout(request, username: str) -> LockoutState:
 
 
 def register_failed_login_attempt(request, username: str) -> None:
+    prune_stale_login_throttles()
     now = timezone.now()
     window = timedelta(seconds=settings.PANEL_LOGIN_FAILURE_WINDOW_SECONDS)
     lockout = timedelta(seconds=settings.PANEL_LOGIN_LOCKOUT_SECONDS)
@@ -158,3 +160,11 @@ def clear_login_throttle(request, username: str) -> None:
     LoginThrottle.objects.filter(
         key__in=build_throttle_keys(request, username)
     ).delete()
+
+
+def prune_stale_login_throttles(*, now: datetime | None = None) -> int:
+    cutoff = (now or timezone.now()) - timedelta(
+        seconds=settings.PANEL_LOGIN_THROTTLE_RETENTION_SECONDS
+    )
+    deleted, _ = LoginThrottle.objects.filter(last_failure_at__lt=cutoff).delete()
+    return deleted

apps/web/apps.py

@@ -1,31 +1,6 @@
-import logging
-import os
-
 from django.apps import AppConfig
 
-from .content_loader import PortfolioContentLoader
-
-logger = logging.getLogger(__name__)
-_SYNC_HAS_RUN = False
-
-
-def _sync_examples_on_startup_enabled() -> bool:
-    """Return whether startup sync is enabled via environment configuration."""
-    value = os.getenv("MYLONITE_SYNC_CONTENT_EXAMPLES_ON_STARTUP", "1").strip().lower()
-    return value not in {"0", "false", "no", "off"}
-
 
 class WebConfig(AppConfig):
     default_auto_field = "django.db.models.BigAutoField"
     name = "apps.web"
-
-    def ready(self) -> None:
-        """Run one-time, best-effort local content example synchronization."""
-        global _SYNC_HAS_RUN
-        if _SYNC_HAS_RUN or not _sync_examples_on_startup_enabled():
-            return
-
-        _SYNC_HAS_RUN = True
-        synced = PortfolioContentLoader().sync_example_content()
-        if synced:
-            logger.info("Synchronized schema-driven content examples at startup.")

apps/web/content_loader.py

@@ -1,6 +1,6 @@
 import logging
 from datetime import date
-from typing import Callable, Protocol, TypeVar
+from typing import Protocol
 
 from .content_mappers import map_site_config
 from .content_registry import ContentEntityRegistry
@@ -25,10 +25,13 @@
     ValidationStatus,
 )
 
-EntityModel = TypeVar("EntityModel")
 logger = logging.getLogger(__name__)
 
 
+class ContentValidationError(ValueError):
+    """Raised when strict content-validation mode encounters schema errors."""
+
+
 class ContentRepository(Protocol):
     """Protocol for content source backends used by the loader."""
 
@@ -67,9 +70,12 @@ def __init__(
         self,
         repository: ContentRepository | None = None,
         entity_registry: ContentEntityRegistry | None = None,
+        *,
+        strict_validation: bool = False,
     ):
         self.repository = repository or FileSystemContentRepository()
         self.entity_registry = entity_registry or ContentEntityRegistry()
+        self.strict_validation = strict_validation
         self._sources: list[SourceInfo] = []
         self._validation_errors: list[str] = []
 
@@ -104,7 +110,10 @@ def _track_sources(self, sources: list[SourceInfo]) -> None:
 
     def _validate(self, scope: str, payload: dict, schema: SchemaDefinition) -> dict:
         normalized, errors = validate_record(schema, payload)
-        self._validation_errors.extend([f"{scope}: {error}" for error in errors])
+        scoped_errors = [f"{scope}: {error}" for error in errors]
+        self._validation_errors.extend(scoped_errors)
+        if self.strict_validation and scoped_errors:
+            raise ContentValidationError("; ".join(scoped_errors))
         return {**payload, **normalized}
 
     def load_site(self) -> SiteConfig:
@@ -113,29 +122,6 @@ def load_site(self) -> SiteConfig:
         validated_site_data = self._validate("site", site_data, SITE_CONFIG_SCHEMA)
         return map_site_config(validated_site_data)
 
-    def load_entity(
-        self,
-        object_id: str,
-        mapper: Callable[[str, dict, str], EntityModel],
-        *,
-        text_filename: str | None = None,
-        schema: SchemaDefinition | None = None,
-    ) -> EntityModel:
-        entry, body, sources = self.repository.load_entity_record(
-            object_id,
-            text_filename=text_filename,
-        )
-        self._track_sources(sources)
-
-        payload = dict(entry)
-
-        validated_entry = (
-            self._validate(object_id, payload, schema)
-            if schema is not None
-            else payload
-        )
-        return mapper(object_id, validated_entry, body)
-
     def load_registered_entity(self, entity_type: str, object_id: str):
         """Load an entity via registry metadata and body-source strategy."""
         definition = self.entity_registry.get(entity_type)

apps/web/content_mappers.py

@@ -18,8 +18,10 @@
 def map_site_config(site_data: dict) -> SiteConfig:
     defaults = schema_defaults(SITE_CONFIG_SCHEMA)
     merged_site_data = {**defaults, **site_data}
-    theme_data = merged_site_data.get("theme", {})
-    install_data = merged_site_data.get("install", {})
+    raw_theme_data = merged_site_data.get("theme", {})
+    theme_data = raw_theme_data if isinstance(raw_theme_data, dict) else {}
+    raw_install_data = merged_site_data.get("install", {})
+    install_data = raw_install_data if isinstance(raw_install_data, dict) else {}
     hosting_mode_value = merged_site_data.get("hosting_mode", HostingMode.LOCAL.value)
 
     try:
@@ -57,7 +59,7 @@ def map_person_profile(object_id: str, entry: dict, _: str) -> PersonProfile:
         display_name=merged_entry.get("display_name", ""),
         headline=merged_entry.get("headline", ""),
         summary=merged_entry.get("summary", ""),
-        bio=merged_entry.get("summary", ""),
+        bio=merged_entry.get("bio") or merged_entry.get("summary", ""),
         profile_image_path=merged_entry.get("profile_image_path", ""),
     )
 

apps/web/content_repository.py

@@ -1,8 +1,9 @@
-import tomllib
 from pathlib import Path
 
 from django.conf import settings
 
+from mylonite.core.content_conventions import is_valid_entity_id
+from mylonite.core.toml_utils import load_toml_file as load_toml_document
 from mylonite.core.content_types import SourceInfo
 
 CONTENT_ROOT = Path(settings.MYLONITE_CONTENT_ROOT)
@@ -70,8 +71,7 @@ def load_toml_file(
     if resolved_path is None:
         return {}, source
 
-    with resolved_path.open("rb") as handle:
-        return tomllib.load(handle), source
+    return load_toml_document(resolved_path), source
 
 
 def load_text_file(
@@ -106,7 +106,12 @@ def load_entity_record(
         *,
         text_filename: str | None = None,
     ) -> tuple[dict, str, list[SourceInfo]]:
+        if not is_valid_entity_id(object_id):
+            raise ValueError(f"invalid entity object_id: {object_id!r}")
+
         root = self.content_root / "entities" / object_id
+        if self.content_root not in root.resolve().parents:
+            raise ValueError(f"invalid entity object_id: {object_id!r}")
 
         entry, entry_source = load_toml_file(
             root / "entry.toml",

apps/web/content_scaffold.py

@@ -5,7 +5,9 @@
 from typing import Any
 
 from .content_registry import ContentEntityRegistry, EntityDefinition
+from mylonite.core.content_conventions import is_valid_entity_id
 from mylonite.core.content_schema import SITE_CONFIG_SCHEMA, schema_defaults
+from mylonite.core.toml_utils import render_toml
 
 
 @dataclass(frozen=True)
@@ -18,75 +20,6 @@ class ScaffoldEntityDefinition:
     text_content: str | None
 
 
-@dataclass(frozen=True)
-class _TomlSection:
-    key_path: tuple[str, ...]
-    values: dict[str, Any]
-
-
-def _format_toml_value(value: Any) -> str:
-    if isinstance(value, bool):
-        return "true" if value else "false"
-    if isinstance(value, (int, float)):
-        return str(value)
-    if isinstance(value, str):
-        escaped = value.replace("\\", "\\\\").replace('"', '\\"')
-        return f'"{escaped}"'
-    if isinstance(value, list):
-        rendered = ", ".join(_format_toml_value(item) for item in value)
-        return f"[{rendered}]"
-    raise TypeError(f"Unsupported TOML value type: {type(value)!r}")
-
-
-def _walk_toml_sections(
-    payload: dict[str, Any],
-    *,
-    key_path: tuple[str, ...] = (),
-) -> tuple[list[tuple[str, Any]], list[_TomlSection]]:
-    """Split payload into scalar fields and nested table sections recursively."""
-    scalars: list[tuple[str, Any]] = []
-    sections: list[_TomlSection] = []
-
-    for key, value in payload.items():
-        if isinstance(value, dict):
-            section_scalars, nested_sections = _walk_toml_sections(
-                value,
-                key_path=(*key_path, key),
-            )
-            sections.append(
-                _TomlSection(
-                    key_path=(*key_path, key),
-                    values={name: item for name, item in section_scalars},
-                )
-            )
-            sections.extend(nested_sections)
-            continue
-
-        scalars.append((key, value))
-
-    return scalars, sections
-
-
-def _render_toml(data: dict[str, Any]) -> str:
-    """Render deterministic TOML for scaffold defaults (supports nested tables)."""
-    scalar_fields, sections = _walk_toml_sections(data)
-
-    lines: list[str] = [
-        f"{key} = {_format_toml_value(value)}" for key, value in scalar_fields
-    ]
-
-    for section in sections:
-        if lines:
-            lines.append("")
-        lines.append("[" + ".".join(section.key_path) + "]")
-        lines.extend(
-            f"{key} = {_format_toml_value(value)}"
-            for key, value in section.values.items()
-        )
-
-    return "\n".join(lines).rstrip() + "\n"
-
-
 def _write_if_changed(path: Path, content: str) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
     current = path.read_text(encoding="utf-8") if path.exists() else None
@@ -98,16 +31,6 @@ def _write_if_changed(path: Path, content: str) -> None:
     tmp_path.replace(path)
 
 
-def _is_valid_entity_id(object_id: str) -> bool:
-    """Allow dotted IDs only; block path traversal or empty segments."""
-    if not object_id:
-        return False
-    if "/" in object_id or "\\" in object_id:
-        return False
-    parts = object_id.split(".")
-    return all(part.strip() for part in parts)
-
-
 def _build_entity_scaffold(
     definition: EntityDefinition,
 ) -> list[ScaffoldEntityDefinition]:
@@ -124,7 +47,7 @@ def _build_entity_scaffold(
 
     scaffolds: list[ScaffoldEntityDefinition] = []
     for object_id in definition.example_object_ids:
-        if not _is_valid_entity_id(object_id):
+        if not is_valid_entity_id(object_id):
             raise ValueError(f"invalid entity object_id: {object_id!r}")
         scaffolds.append(
             ScaffoldEntityDefinition(
@@ -161,13 +84,13 @@ def _prune_stale_text_examples(
 def sync_content_examples(content_root: Path, registry: ContentEntityRegistry) -> None:
     """Generate or update local `*.example` files from schema defaults."""
     site_example = content_root / "config" / "site.toml.example"
-    _write_if_changed(site_example, _render_toml(schema_defaults(SITE_CONFIG_SCHEMA)))
+    _write_if_changed(site_example, render_toml(schema_defaults(SITE_CONFIG_SCHEMA)))
 
     for definition in registry.definitions():
         for scaffold in _build_entity_scaffold(definition):
             entity_root = content_root / "entities" / scaffold.object_id
             _write_if_changed(
-                entity_root / "entry.toml.example", _render_toml(scaffold.entry)
+                entity_root / "entry.toml.example", render_toml(scaffold.entry)
             )
 
             _prune_stale_text_examples(entity_root, scaffold.text_filename)
@@ -182,5 +105,4 @@ def sync_content_examples(content_root: Path, registry: ContentEntityRegistry) -
     "ScaffoldEntityDefinition",
     "sync_content_examples",
     "_build_entity_scaffold",
-    "_render_toml",
 ]

apps/web/page_contexts.py

@@ -1,6 +1,8 @@
 from dataclasses import dataclass
 from typing import Protocol
 
+from django.conf import settings
+
 from .content_loader import PortfolioContentLoader
 from mylonite.core.content_types import HomePageContent, PersonProfile, SiteConfig
 from .view_models import HomeHeroViewModel, HomePageViewModel, LayoutViewModel
@@ -29,7 +31,8 @@ def build(self, shared: SharedPageContext, loader: PortfolioContentLoader) -> di
         hero = HomeHeroViewModel(
             display_name=shared.owner.display_name or shared.owner.full_name,
             headline=shared.owner.headline,
-            bio=shared.homepage.markdown,
+            bio=shared.owner.bio,
+            intro_markdown=shared.homepage.markdown,
             summary=shared.owner.summary,
         )
         page_model = HomePageViewModel(page_title="Home", hero=hero)
@@ -47,7 +50,9 @@ def __init__(
         loader: PortfolioContentLoader | None = None,
         builders: dict[str, PageContextBuilder] | None = None,
     ):
-        self.loader = loader or PortfolioContentLoader()
+        self.loader = loader or PortfolioContentLoader(
+            strict_validation=settings.MYLONITE_STRICT_CONTENT_VALIDATION
+        )
         self.builders = builders or {
             HomePageContextBuilder.page_name: HomePageContextBuilder(),
         }

apps/web/view_models.py

@@ -32,13 +32,15 @@ class HomeHeroViewModel:
     display_name: str
     headline: str
     bio: str
+    intro_markdown: str
     summary: str
 
     def to_context(self) -> dict:
         return {
             "display_name": self.display_name,
             "headline": self.headline,
             "bio": self.bio,
+            "intro_markdown": self.intro_markdown,
             "summary": self.summary,
         }
 

manage.py

@@ -5,6 +5,24 @@
 
 def main() -> None:
     os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mylonite.settings")
+
+    from pathlib import Path
+
+    from mylonite.runtime import ensure_runtime_directories, resolve_runtime_paths
+
+    base_dir = Path(__file__).resolve().parent
+    runtime_paths = resolve_runtime_paths(base_dir)
+
+    ensure_runtime_directories(
+        [
+            runtime_paths.config_root,
+            runtime_paths.data_root,
+            runtime_paths.db_path.parent,
+            runtime_paths.static_root,
+            runtime_paths.media_root,
+        ]
+    )
+
     from django.core.management import execute_from_command_line
 
     execute_from_command_line(sys.argv)