tegwick
affed64839
Thin CONTENT_ROOT content plane (ADR-0003/0004) with space visual field, product UI for Title/Abstract/Visual, page list sidebar, and tests. Update capability model and smoke checklist; mark WP-0006 finished.
257 lines
7 KiB
Python
257 lines
7 KiB
Python
"""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. Prefer CONTENT_ROOT (ADR-0004)."""
|
|
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,
|
|
)
|
|
|
|
try:
|
|
fetched = _fetch(space, path)
|
|
# Prefer frontmatter title when present
|
|
from .pageops import parse_frontmatter
|
|
|
|
meta, body = parse_frontmatter(fetched.text)
|
|
display_title = (
|
|
str(meta.get("title") or "").strip()
|
|
or _title_from_markdown(body or fetched.text, space.title)
|
|
)
|
|
html = render_markdown(body if meta else fetched.text)
|
|
return RenderedPage(
|
|
slug=page_slug,
|
|
path=path,
|
|
title=display_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:
|
|
# ADR-0004: CONTENT_ROOT / spaces/<slug>/pages/…
|
|
from . import pageops
|
|
|
|
try:
|
|
local_doc = pageops.read_page(space.slug, Path(path).stem)
|
|
if local_doc.exists and local_doc.path:
|
|
text = Path(local_doc.path).read_text(encoding="utf-8")
|
|
return FetchedFile(path=path, text=text, source="content-plane")
|
|
except pageops.PageOpsError:
|
|
pass
|
|
|
|
# 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")
|
|
|
|
if not space.has_content_binding:
|
|
raise ContentFetchError(
|
|
"No local page on the content plane and no Forgejo binding."
|
|
)
|
|
|
|
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}"
|