Render space pages from Forgejo markdown (CSOC-WP-0004-T04)

Fetch raw files from Forgejo, sanitize markdown to HTML, and show them on
space detail. Ship a public demo fixture path and seed_demo_space command.
This commit is contained in:
tegwick 2026-08-12 01:56:30 +02:00
parent 86b54243b8
commit 1cedd8f219
13 changed files with 606 additions and 26 deletions

View file

@ -0,0 +1,191 @@
"""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()

View file

@ -0,0 +1,113 @@
"""Fetch raw files from Forgejo (Gitea-compatible) HTTP API / raw URLs."""
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Any
from urllib.parse import quote
import httpx
from django.conf import settings
logger = logging.getLogger(__name__)
class ContentFetchError(Exception):
"""Fail-closed content load error (safe message for UI)."""
def __init__(self, message: str, *, status_code: int | None = None):
super().__init__(message)
self.message = message
self.status_code = status_code
@dataclass(frozen=True)
class FetchedFile:
path: str
text: str
source: str # forgejo-raw | forgejo-api | fixture
def _auth_headers() -> dict[str, str]:
token = (getattr(settings, "FORGEJO_TOKEN", None) or "").strip()
if not token:
return {}
return {"Authorization": f"token {token}"}
def fetch_raw_file(
*,
owner: str,
repo: str,
ref: str,
path: str,
timeout: float | None = None,
) -> FetchedFile:
"""GET raw file content from Forgejo.
Prefer public/raw URL (works without token for public repos). Fall back to
contents API when raw fails and a token is configured.
"""
base = (getattr(settings, "FORGEJO_BASE_URL", None) or "").rstrip("/")
if not base:
raise ContentFetchError("FORGEJO_BASE_URL is not configured")
timeout = timeout if timeout is not None else float(
getattr(settings, "FORGEJO_TIMEOUT_SECONDS", 10.0)
)
# Raw URL path segments
path_enc = "/".join(quote(p, safe="") for p in path.strip("/").split("/") if p)
raw_url = f"{base}/{quote(owner)}/{quote(repo)}/raw/branch/{quote(ref, safe='')}/{path_enc}"
try:
with httpx.Client(timeout=timeout, follow_redirects=True) as client:
resp = client.get(raw_url, headers=_auth_headers())
if resp.status_code == 200:
return FetchedFile(path=path, text=resp.text, source="forgejo-raw")
if resp.status_code in (401, 403) and _auth_headers():
return _fetch_via_contents_api(
client, base=base, owner=owner, repo=repo, ref=ref, path=path
)
if resp.status_code == 404:
raise ContentFetchError(f"File not found: {path}", status_code=404)
raise ContentFetchError(
f"Forgejo returned HTTP {resp.status_code} for {path}",
status_code=resp.status_code,
)
except ContentFetchError:
raise
except httpx.HTTPError as exc:
logger.exception("Forgejo fetch failed for %s/%s %s", owner, repo, path)
raise ContentFetchError("Could not reach Forgejo content store") from exc
def _fetch_via_contents_api(
client: httpx.Client,
*,
base: str,
owner: str,
repo: str,
ref: str,
path: str,
) -> FetchedFile:
"""Gitea/Forgejo contents API returns base64 content for a file."""
import base64
import json
path_enc = "/".join(quote(p, safe="") for p in path.strip("/").split("/") if p)
api = f"{base}/api/v1/repos/{quote(owner)}/{quote(repo)}/contents/{path_enc}"
resp = client.get(api, params={"ref": ref}, headers=_auth_headers())
if resp.status_code != 200:
raise ContentFetchError(
f"Forgejo API returned HTTP {resp.status_code} for {path}",
status_code=resp.status_code,
)
payload: dict[str, Any] = resp.json()
if payload.get("type") != "file":
raise ContentFetchError(f"Path is not a file: {path}")
encoded = payload.get("content") or ""
# API may wrap base64 with newlines
raw = base64.b64decode(encoded)
text = raw.decode("utf-8")
return FetchedFile(path=path, text=text, source="forgejo-api")

View file

@ -0,0 +1,64 @@
"""Seed a demo Space bound to the in-repo Forgejo fixture path (public raw)."""
from __future__ import annotations
from django.conf import settings
from django.core.management.base import BaseCommand
from coulomb_social.apps.spaces.models import Space
class Command(BaseCommand):
help = (
"Create or update the demo space (tenant from DEFAULT_TENANT_ID) bound to "
"coulomb/coulomb-social docs/space-fixtures/demo/pages for T04 smoke."
)
def add_arguments(self, parser):
parser.add_argument("--slug", default="demo")
parser.add_argument("--title", default="Demo space")
parser.add_argument(
"--tenant",
default="",
help="Defaults to DEFAULT_TENANT_ID",
)
parser.add_argument(
"--owner",
default="coulomb",
help="Forgejo owner",
)
parser.add_argument(
"--repo",
default="coulomb-social",
help="Forgejo repo containing fixture markdown",
)
parser.add_argument(
"--content-root",
default="docs/space-fixtures/demo/pages",
)
parser.add_argument("--branch", default="main")
def handle(self, *args, **options):
tenant = options["tenant"] or settings.DEFAULT_TENANT_ID
slug = options["slug"]
space, created = Space.objects.update_or_create(
tenant_id=tenant,
slug=slug,
defaults={
"title": options["title"],
"description": "Seeded demo space with markdown from Forgejo (CSOC-WP-0004-T04).",
"forgejo_owner": options["owner"],
"forgejo_repo": options["repo"],
"default_branch": options["branch"],
"content_root": options["content_root"],
"is_active": True,
},
)
action = "Created" if created else "Updated"
self.stdout.write(
self.style.SUCCESS(
f"{action} space {space.slug} tenant={space.tenant_id} "
f"→ {space.forgejo_owner}/{space.forgejo_repo} "
f"({space.content_root}/index.md @ {space.default_branch})"
)
)

View file

@ -4,6 +4,7 @@ from django.shortcuts import render
from coulomb_social.apps.core.principal import build_principal
from .content import load_space_page
from .services import get_space_for_member, spaces_for_member
@ -18,6 +19,9 @@ def space_detail(request: HttpRequest, slug: str) -> HttpResponse:
space = get_space_for_member(member, slug)
if space is None:
return HttpResponseNotFound("Space not found.")
page = request.GET.get("page") or "index"
rendered = load_space_page(space, page)
return render(
request,
"spaces/detail.html",
@ -25,10 +29,10 @@ def space_detail(request: HttpRequest, slug: str) -> HttpResponse:
"principal": principal,
"display_name": principal["display_name"],
"space": space,
"page": rendered,
},
)
# used by core.app_home — re-export list helper
def list_spaces_context(member) -> dict:
return {"spaces": list(spaces_for_member(member))}