Land CSOC-WP-0006 PageOps: member space/page CRUD, copy, transfer
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.
This commit is contained in:
parent
00dbb86c93
commit
affed64839
24 changed files with 1812 additions and 151 deletions
331
coulomb_social/apps/spaces/pageops.py
Normal file
331
coulomb_social/apps/spaces/pageops.py
Normal file
|
|
@ -0,0 +1,331 @@
|
|||
"""PageOps — thin filesystem content plane (ADR-0003 / ADR-0004).
|
||||
|
||||
Layout under CONTENT_ROOT:
|
||||
|
||||
spaces/<space-slug>/
|
||||
pages/<page-slug>.md
|
||||
assets/…
|
||||
|
||||
Markdown files use YAML-like frontmatter (title, abstractor, visual, …).
|
||||
The app DB holds Space rows as an index; bodies live on disk.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import shutil
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
from django.conf import settings
|
||||
from django.utils.text import slugify as django_slugify
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_SLUG_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
|
||||
|
||||
|
||||
class PageOpsError(Exception):
|
||||
def __init__(self, message: str):
|
||||
self.message = message
|
||||
super().__init__(message)
|
||||
|
||||
|
||||
@dataclass
|
||||
class PageDoc:
|
||||
slug: str
|
||||
title: str = ""
|
||||
abstractor: str = ""
|
||||
visual: str = ""
|
||||
body: str = ""
|
||||
meta: dict = field(default_factory=dict)
|
||||
path: str = ""
|
||||
|
||||
@property
|
||||
def exists(self) -> bool:
|
||||
return bool(self.path)
|
||||
|
||||
|
||||
def content_root() -> Path:
|
||||
raw = (getattr(settings, "CONTENT_ROOT", None) or "").strip()
|
||||
if raw:
|
||||
return Path(raw).expanduser().resolve()
|
||||
# Default under project var/ (gitignored)
|
||||
base = Path(settings.BASE_DIR)
|
||||
return (base / "var" / "content").resolve()
|
||||
|
||||
|
||||
def space_dir(space_slug: str) -> Path:
|
||||
return content_root() / "spaces" / space_slug
|
||||
|
||||
|
||||
def pages_dir(space_slug: str) -> Path:
|
||||
return space_dir(space_slug) / "pages"
|
||||
|
||||
|
||||
def assets_dir(space_slug: str) -> Path:
|
||||
return space_dir(space_slug) / "assets"
|
||||
|
||||
|
||||
def page_file(space_slug: str, page_slug: str) -> Path:
|
||||
name = "index.md" if page_slug == "index" else f"{page_slug}.md"
|
||||
return pages_dir(space_slug) / name
|
||||
|
||||
|
||||
def normalize_slug(value: str, *, allow_index: bool = True) -> str:
|
||||
s = django_slugify(value or "")[:80]
|
||||
if not s:
|
||||
raise PageOpsError("Slug cannot be empty")
|
||||
if s == "index" and not allow_index:
|
||||
raise PageOpsError("Slug 'index' is reserved")
|
||||
if not _SLUG_RE.match(s):
|
||||
raise PageOpsError("Invalid slug")
|
||||
return s
|
||||
|
||||
|
||||
def parse_frontmatter(text: str) -> tuple[dict, str]:
|
||||
if not text.startswith("---\n"):
|
||||
return {}, text
|
||||
end = text.find("\n---\n", 4)
|
||||
if end < 0:
|
||||
# allow trailing --- at EOF
|
||||
end = text.find("\n---", 4)
|
||||
if end < 0 or end + 4 != len(text.rstrip()):
|
||||
return {}, text
|
||||
body = ""
|
||||
fm_raw = text[4:end]
|
||||
else:
|
||||
fm_raw = text[4:end]
|
||||
body = text[end + 5 :]
|
||||
data: dict = {}
|
||||
for line in fm_raw.splitlines():
|
||||
if not line.strip() or line.strip().startswith("#"):
|
||||
continue
|
||||
if ":" not in line:
|
||||
continue
|
||||
key, raw = line.split(":", 1)
|
||||
key = key.strip()
|
||||
raw = raw.strip()
|
||||
if not key:
|
||||
continue
|
||||
if raw in ("true", "false"):
|
||||
data[key] = raw == "true"
|
||||
elif raw.startswith('"') or raw.startswith("'"):
|
||||
try:
|
||||
data[key] = json.loads(raw.replace("'", '"') if raw.startswith("'") else raw)
|
||||
except json.JSONDecodeError:
|
||||
data[key] = raw.strip("\"'")
|
||||
else:
|
||||
data[key] = raw
|
||||
return data, body.lstrip("\n")
|
||||
|
||||
|
||||
def dump_frontmatter(meta: dict, body: str) -> str:
|
||||
lines = ["---"]
|
||||
for key, value in meta.items():
|
||||
if value is None or value == "":
|
||||
lines.append(f'{key}: ""')
|
||||
elif isinstance(value, bool):
|
||||
lines.append(f"{key}: {'true' if value else 'false'}")
|
||||
elif isinstance(value, (int, float)):
|
||||
lines.append(f"{key}: {value}")
|
||||
else:
|
||||
lines.append(f"{key}: {json.dumps(str(value))}")
|
||||
lines.append("---")
|
||||
lines.append("")
|
||||
lines.append(body.rstrip() + ("\n" if body.strip() else ""))
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def ensure_space_tree(space_slug: str) -> Path:
|
||||
root = space_dir(space_slug)
|
||||
pages_dir(space_slug).mkdir(parents=True, exist_ok=True)
|
||||
assets_dir(space_slug).mkdir(parents=True, exist_ok=True)
|
||||
return root
|
||||
|
||||
|
||||
def write_page(
|
||||
space_slug: str,
|
||||
page_slug: str,
|
||||
*,
|
||||
title: str = "",
|
||||
abstractor: str = "",
|
||||
visual: str = "",
|
||||
body: str = "",
|
||||
extra_meta: dict | None = None,
|
||||
) -> PageDoc:
|
||||
page_slug = normalize_slug(page_slug) if page_slug != "index" else "index"
|
||||
ensure_space_tree(space_slug)
|
||||
meta = {
|
||||
"title": title or page_slug,
|
||||
"abstractor": abstractor or "",
|
||||
"visual": visual or "",
|
||||
"space": space_slug,
|
||||
}
|
||||
if extra_meta:
|
||||
meta.update(extra_meta)
|
||||
path = page_file(space_slug, page_slug)
|
||||
path.write_text(dump_frontmatter(meta, body or ""), encoding="utf-8")
|
||||
return read_page(space_slug, page_slug)
|
||||
|
||||
|
||||
def read_page(space_slug: str, page_slug: str) -> PageDoc:
|
||||
page_slug = "index" if page_slug in ("", "index") else normalize_slug(page_slug)
|
||||
path = page_file(space_slug, page_slug)
|
||||
if not path.is_file():
|
||||
return PageDoc(slug=page_slug, path="")
|
||||
text = path.read_text(encoding="utf-8")
|
||||
meta, body = parse_frontmatter(text)
|
||||
return PageDoc(
|
||||
slug=page_slug,
|
||||
title=str(meta.get("title") or ""),
|
||||
abstractor=str(meta.get("abstractor") or meta.get("description") or ""),
|
||||
visual=str(meta.get("visual") or ""),
|
||||
body=body,
|
||||
meta=meta,
|
||||
path=str(path),
|
||||
)
|
||||
|
||||
|
||||
def list_pages(space_slug: str) -> list[PageDoc]:
|
||||
pdir = pages_dir(space_slug)
|
||||
if not pdir.is_dir():
|
||||
return []
|
||||
docs: list[PageDoc] = []
|
||||
for path in sorted(pdir.glob("*.md")):
|
||||
slug = path.stem
|
||||
docs.append(read_page(space_slug, slug))
|
||||
# index first
|
||||
docs.sort(key=lambda d: (0 if d.slug == "index" else 1, d.title.lower() or d.slug))
|
||||
return docs
|
||||
|
||||
|
||||
def delete_page(space_slug: str, page_slug: str) -> None:
|
||||
if page_slug == "index":
|
||||
raise PageOpsError("Cannot delete the space index page")
|
||||
path = page_file(space_slug, normalize_slug(page_slug))
|
||||
if path.is_file():
|
||||
path.unlink()
|
||||
|
||||
|
||||
def copy_page(
|
||||
source_space: str,
|
||||
source_page: str,
|
||||
dest_space: str,
|
||||
dest_page: str,
|
||||
*,
|
||||
title: str | None = None,
|
||||
) -> PageDoc:
|
||||
src = read_page(source_space, source_page)
|
||||
if not src.exists:
|
||||
raise PageOpsError("Source page not found")
|
||||
dest_page = normalize_slug(dest_page)
|
||||
if page_file(dest_space, dest_page).is_file():
|
||||
raise PageOpsError("Destination page already exists")
|
||||
ensure_space_tree(dest_space)
|
||||
# copy visual asset if relative
|
||||
visual = src.visual
|
||||
if visual and not visual.startswith("http") and not visual.startswith("/"):
|
||||
src_asset = space_dir(source_space) / visual
|
||||
if src_asset.is_file():
|
||||
dest_assets = assets_dir(dest_space)
|
||||
dest_assets.mkdir(parents=True, exist_ok=True)
|
||||
dest_name = f"{dest_page}-{src_asset.name}"
|
||||
dest_path = dest_assets / dest_name
|
||||
shutil.copy2(src_asset, dest_path)
|
||||
visual = f"assets/{dest_name}"
|
||||
new_title = title if title is not None else (src.title or dest_page)
|
||||
if new_title == src.title and source_space == dest_space:
|
||||
new_title = f"{src.title} (copy)" if src.title else f"{dest_page}"
|
||||
return write_page(
|
||||
dest_space,
|
||||
dest_page,
|
||||
title=new_title,
|
||||
abstractor=src.abstractor,
|
||||
visual=visual,
|
||||
body=src.body,
|
||||
extra_meta={k: v for k, v in src.meta.items() if k not in {"title", "abstractor", "visual", "space"}},
|
||||
)
|
||||
|
||||
|
||||
def transfer_page(
|
||||
source_space: str,
|
||||
source_page: str,
|
||||
dest_space: str,
|
||||
dest_page: str | None = None,
|
||||
) -> PageDoc:
|
||||
dest_page = dest_page or source_page
|
||||
if source_page == "index":
|
||||
raise PageOpsError("Cannot transfer the space index page")
|
||||
doc = copy_page(source_space, source_page, dest_space, dest_page, title=None)
|
||||
# restore original title (copy_page may suffix)
|
||||
src = read_page(source_space, source_page)
|
||||
write_page(
|
||||
dest_space,
|
||||
dest_page,
|
||||
title=src.title,
|
||||
abstractor=src.abstractor,
|
||||
visual=doc.visual,
|
||||
body=src.body,
|
||||
extra_meta={k: v for k, v in src.meta.items() if k not in {"title", "abstractor", "visual", "space"}},
|
||||
)
|
||||
delete_page(source_space, source_page)
|
||||
return read_page(dest_space, dest_page)
|
||||
|
||||
|
||||
def init_space_content(
|
||||
space_slug: str,
|
||||
*,
|
||||
title: str,
|
||||
abstractor: str = "",
|
||||
visual: str = "",
|
||||
) -> PageDoc:
|
||||
"""Create tree + index.md for a new space."""
|
||||
ensure_space_tree(space_slug)
|
||||
body = f"# {title}\n\n{abstractor}\n" if abstractor else f"# {title}\n"
|
||||
return write_page(
|
||||
space_slug,
|
||||
"index",
|
||||
title=title,
|
||||
abstractor=abstractor,
|
||||
visual=visual,
|
||||
body=body,
|
||||
extra_meta={"bubble_type": "space"},
|
||||
)
|
||||
|
||||
|
||||
def rename_space_tree(old_slug: str, new_slug: str) -> None:
|
||||
old = space_dir(old_slug)
|
||||
new = space_dir(new_slug)
|
||||
if not old.is_dir():
|
||||
return
|
||||
if new.exists():
|
||||
raise PageOpsError("Target space directory already exists")
|
||||
new.parent.mkdir(parents=True, exist_ok=True)
|
||||
old.rename(new)
|
||||
|
||||
|
||||
def archive_space_tree(space_slug: str) -> None:
|
||||
"""Soft-archive: rename tree to .archived-<slug> if present."""
|
||||
root = space_dir(space_slug)
|
||||
if not root.is_dir():
|
||||
return
|
||||
dest = content_root() / "spaces" / f".archived-{space_slug}"
|
||||
if dest.exists():
|
||||
shutil.rmtree(dest)
|
||||
root.rename(dest)
|
||||
|
||||
|
||||
def local_page_text(space_slug: str, page_slug: str) -> str | None:
|
||||
"""Raw markdown if present on the content plane."""
|
||||
path = page_file(space_slug, page_slug if page_slug != "index" else "index")
|
||||
# handle normalize
|
||||
try:
|
||||
doc = read_page(space_slug, page_slug)
|
||||
except PageOpsError:
|
||||
return None
|
||||
if not doc.exists:
|
||||
return None
|
||||
return Path(doc.path).read_text(encoding="utf-8")
|
||||
Loading…
Add table
Add a link
Reference in a new issue