140 lines
4.0 KiB
Python
140 lines
4.0 KiB
Python
import os
|
|
import re
|
|
import shlex
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter
|
|
from pydantic import BaseModel
|
|
|
|
router = APIRouter()
|
|
|
|
SOURCES_DIR = Path(os.environ.get("SOURCES_DIR", "/data/sources"))
|
|
PAGES_DIR = Path(os.environ.get("PAGES_DIR", "/data/pages"))
|
|
|
|
# Match µFrame link nodes: link "display" "/page/slug.mu" or link "display" "slug"
|
|
_UF_LINK = re.compile(r'^\s*link\s+', re.IGNORECASE)
|
|
# Fallback: Micron links [label`slug] or [label`slug.mu]
|
|
_MICRON_LINK = re.compile(r'\[([^`\]]+)`([a-zA-Z0-9_-]+)(?:\.mu)?\]')
|
|
|
|
|
|
class GraphNode(BaseModel):
|
|
id: str
|
|
published: bool
|
|
title: str | None = None
|
|
|
|
|
|
class GraphEdge(BaseModel):
|
|
source: str
|
|
target: str
|
|
|
|
|
|
class GraphData(BaseModel):
|
|
nodes: list[GraphNode]
|
|
edges: list[GraphEdge]
|
|
|
|
|
|
def _all_page_names() -> set[str]:
|
|
names: set[str] = set()
|
|
if PAGES_DIR.is_dir():
|
|
for f in PAGES_DIR.iterdir():
|
|
if f.suffix == ".mu" and f.is_file():
|
|
names.add(f.stem)
|
|
if SOURCES_DIR.is_dir():
|
|
for f in SOURCES_DIR.iterdir():
|
|
if f.suffix in (".uf", ".mu") and f.is_file():
|
|
names.add(f.stem)
|
|
return names
|
|
|
|
|
|
def _extract_title(source: str) -> str | None:
|
|
"""Extract title from µFrame page or heading, or legacy Micron >Title."""
|
|
for line in source.splitlines():
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith("#"):
|
|
continue
|
|
if stripped.lower().startswith("page "):
|
|
try:
|
|
parts = shlex.split(stripped)
|
|
if len(parts) >= 2:
|
|
return parts[1]
|
|
except ValueError:
|
|
pass
|
|
break
|
|
if stripped.lower().startswith("heading "):
|
|
try:
|
|
parts = shlex.split(stripped)
|
|
if len(parts) >= 3:
|
|
return parts[2]
|
|
except ValueError:
|
|
pass
|
|
break
|
|
if stripped.startswith(">") and not stripped.startswith(">>"):
|
|
return stripped[1:].strip()
|
|
break
|
|
return None
|
|
|
|
|
|
def _extract_links(source: str, all_names: set[str]) -> list[str]:
|
|
"""Extract internal link targets from µFrame or Micron source."""
|
|
targets: list[str] = []
|
|
|
|
for line in source.splitlines():
|
|
stripped = line.strip()
|
|
|
|
# µFrame: link "display" "/page/slug.mu" or link "display" "slug"
|
|
if _UF_LINK.match(stripped):
|
|
try:
|
|
parts = shlex.split(stripped)
|
|
if len(parts) >= 3:
|
|
dest = parts[2]
|
|
# Normalize: /page/slug.mu → slug
|
|
slug = dest.rsplit("/", 1)[-1].removesuffix(".mu")
|
|
if slug in all_names:
|
|
targets.append(slug)
|
|
except ValueError:
|
|
pass
|
|
continue
|
|
|
|
# Fallback: Micron link syntax [label`slug]
|
|
for m in _MICRON_LINK.finditer(stripped):
|
|
slug = m.group(2)
|
|
if slug in all_names:
|
|
targets.append(slug)
|
|
|
|
return targets
|
|
|
|
|
|
def _source_path(name: str) -> Path | None:
|
|
"""Get source file path, preferring .uf over .mu."""
|
|
uf = SOURCES_DIR / f"{name}.uf"
|
|
if uf.is_file():
|
|
return uf
|
|
mu = SOURCES_DIR / f"{name}.mu"
|
|
return mu if mu.is_file() else None
|
|
|
|
|
|
@router.get("/graph", response_model=GraphData)
|
|
async def get_graph():
|
|
all_names = _all_page_names()
|
|
nodes: list[GraphNode] = []
|
|
edges: list[GraphEdge] = []
|
|
|
|
for name in sorted(all_names):
|
|
src_path = _source_path(name)
|
|
mu_path = PAGES_DIR / f"{name}.mu"
|
|
|
|
title = None
|
|
if src_path:
|
|
content = src_path.read_text(encoding="utf-8")
|
|
title = _extract_title(content)
|
|
for target in _extract_links(content, all_names):
|
|
edges.append(GraphEdge(source=name, target=target))
|
|
|
|
nodes.append(GraphNode(
|
|
id=name,
|
|
published=mu_path.is_file(),
|
|
title=title,
|
|
))
|
|
|
|
return GraphData(nodes=nodes, edges=edges)
|