"""Load and render space markdown pages (Forgejo-backed).""" from __future__ import annotations import logging import re from dataclasses import dataclass from pathlib import Path import bleach import markdown as md_lib from django.conf import settings from .forgejo import ContentFetchError, FetchedFile, fetch_raw_file from .models import Space logger = logging.getLogger(__name__) _PAGE_SLUG_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$") # Conservative allowlist for rendered markdown HTML _ALLOWED_TAGS = list(bleach.sanitizer.ALLOWED_TAGS) + [ "p", "pre", "code", "h1", "h2", "h3", "h4", "h5", "h6", "table", "thead", "tbody", "tr", "th", "td", "hr", "br", "img", "blockquote", "ul", "ol", "li", "strong", "em", "a", ] _ALLOWED_ATTRS = { **bleach.sanitizer.ALLOWED_ATTRIBUTES, "img": ["src", "alt", "title"], "a": ["href", "title", "rel"], "code": ["class"], "th": ["align"], "td": ["align"], } _ALLOWED_PROTOCOLS = ["http", "https", "mailto"] @dataclass(frozen=True) class RenderedPage: slug: str path: str title: str html: str source: str error: str | None = None def normalize_page_slug(page: str | None) -> str: slug = (page or "index").strip().lower() or "index" if slug.endswith(".md"): slug = slug[: -len(".md")] if slug != "index" and not _PAGE_SLUG_RE.match(slug): raise ContentFetchError("Invalid page slug") return slug def page_path_for(space: Space, page_slug: str) -> str: root = (space.content_root or "pages").strip().strip("/") filename = "index.md" if page_slug == "index" else f"{page_slug}.md" return f"{root}/{filename}" if root else filename def render_markdown(text: str) -> str: raw_html = md_lib.markdown( text, extensions=["fenced_code", "tables", "nl2br", "sane_lists"], output_format="html", ) return bleach.clean( raw_html, tags=_ALLOWED_TAGS, attributes=_ALLOWED_ATTRS, protocols=_ALLOWED_PROTOCOLS, strip=True, ) def _title_from_markdown(text: str, fallback: str) -> str: for line in text.splitlines(): line = line.strip() if line.startswith("# "): return line[2:].strip() or fallback return fallback def load_space_page(space: Space, page: str | None = None) -> RenderedPage: """Fetch and render a page for a space. Fail closed with error field set.""" try: page_slug = normalize_page_slug(page) path = page_path_for(space, page_slug) except ContentFetchError as exc: return RenderedPage( slug=page or "index", path="", title=space.title, html="", source="", error=exc.message, ) if not space.has_content_binding: return RenderedPage( slug=page_slug, path=path, title=space.title, html="", source="", error="This space is not bound to a Forgejo repository yet.", ) try: fetched = _fetch(space, path) html = render_markdown(fetched.text) title = _title_from_markdown(fetched.text, space.title) return RenderedPage( slug=page_slug, path=path, title=title, html=html, source=fetched.source, error=None, ) except ContentFetchError as exc: return RenderedPage( slug=page_slug, path=path, title=space.title, html="", source="", error=exc.message, ) def _fetch(space: Space, path: str) -> FetchedFile: # Optional local fixture root for offline tests / air-gapped demos fixture_root = (getattr(settings, "SPACE_CONTENT_FIXTURE_ROOT", None) or "").strip() if fixture_root: local = Path(fixture_root) / space.slug / path if local.is_file(): return FetchedFile(path=path, text=local.read_text(encoding="utf-8"), source="fixture") cache_key = ( space.forgejo_owner, space.forgejo_repo, space.default_branch or "main", path, getattr(settings, "FORGEJO_BASE_URL", ""), ) hit = _FETCH_CACHE.get(cache_key) if hit is not None: return hit fetched = fetch_raw_file( owner=space.forgejo_owner, repo=space.forgejo_repo, ref=space.default_branch or "main", path=path, ) # Cache successes only (avoid sticky 404s while authoring) if len(_FETCH_CACHE) > 128: _FETCH_CACHE.clear() _FETCH_CACHE[cache_key] = fetched return fetched _FETCH_CACHE: dict[tuple[str, str, str, str, str], FetchedFile] = {} def clear_content_cache() -> None: _FETCH_CACHE.clear() def clear_content_cache_for_repo(owner: str, repo: str) -> int: """Drop cached files for one Forgejo repo. Returns number of entries removed.""" owner = (owner or "").strip() repo = (repo or "").strip() if not owner or not repo: return 0 victims = [k for k in _FETCH_CACHE if k[0] == owner and k[1] == repo] for key in victims: del _FETCH_CACHE[key] return len(victims) def clear_content_cache_for_space(space: Space) -> int: if not space.has_content_binding: return 0 return clear_content_cache_for_repo(space.forgejo_owner, space.forgejo_repo) def forgejo_repo_url(space: Space) -> str: base = (getattr(settings, "FORGEJO_BASE_URL", None) or "").rstrip("/") if not space.has_content_binding or not base: return "" return f"{base}/{space.forgejo_owner}/{space.forgejo_repo}" def forgejo_edit_url(space: Space, page: str | None = None) -> str: """Deep-link to Forgejo's file editor for the page markdown (git remains SoR).""" base = forgejo_repo_url(space) if not base: return "" try: page_slug = normalize_page_slug(page) path = page_path_for(space, page_slug) except ContentFetchError: return base branch = space.default_branch or "main" # Gitea/Forgejo: /{owner}/{repo}/_edit/{branch}/{path} return f"{base}/_edit/{branch}/{path}" def forgejo_blob_url(space: Space, page: str | None = None) -> str: base = forgejo_repo_url(space) if not base: return "" try: page_slug = normalize_page_slug(page) path = page_path_for(space, page_slug) except ContentFetchError: return base branch = space.default_branch or "main" return f"{base}/src/branch/{branch}/{path}"