414 lines
12 KiB
Python
414 lines
12 KiB
Python
import os
|
|
import shlex
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, HTTPException, Query
|
|
from pydantic import BaseModel, Field
|
|
|
|
import uframe
|
|
|
|
router = APIRouter()
|
|
files_router = APIRouter()
|
|
|
|
PAGES_DIR = Path(os.environ.get("PAGES_DIR", "/data/pages"))
|
|
SOURCES_DIR = Path(os.environ.get("SOURCES_DIR", "/data/sources"))
|
|
|
|
DEFAULT_INDEX_SOURCE = '''\
|
|
page "Welcome" 60
|
|
bigtitle "uFrame" thin
|
|
|
|
box rounded "Micronomicon"
|
|
align center
|
|
text "Decentralized Page Server"
|
|
text "Powered by @bold{Reticulum} and @bold{NomadNet}"
|
|
|
|
spacer
|
|
|
|
heading 2 "Pages"
|
|
text "This node is serving pages built with the uFrame DSL."
|
|
text "Use the web IDE to create and publish new pages."
|
|
|
|
spacer
|
|
|
|
divider light
|
|
|
|
label "Node" "Micronomicon"
|
|
label "Engine" "uFrame v1"
|
|
status "Node" online
|
|
'''
|
|
|
|
|
|
def ensure_default_pages():
|
|
"""Create default index page and .env file if they don't exist."""
|
|
PAGES_DIR.mkdir(parents=True, exist_ok=True)
|
|
SOURCES_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
index_mu = PAGES_DIR / "index.mu"
|
|
index_src = SOURCES_DIR / "index.uf"
|
|
|
|
if not index_mu.is_file():
|
|
# Compile and publish the default index
|
|
result = uframe.compile(DEFAULT_INDEX_SOURCE)
|
|
index_mu.write_text(result.micron, encoding="utf-8")
|
|
index_mu.chmod(0o644)
|
|
|
|
if not index_src.is_file():
|
|
index_src.write_text(DEFAULT_INDEX_SOURCE, encoding="utf-8")
|
|
|
|
env_path = SOURCES_DIR / ".env"
|
|
if not env_path.is_file():
|
|
env_path.write_text(
|
|
"# Environment variables for dynamic pages\n"
|
|
"# Access with: source name : env \"KEY\"\n"
|
|
"#\n"
|
|
"# Example:\n"
|
|
"# API_KEY=your-key-here\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
class PageMeta(BaseModel):
|
|
name: str
|
|
title: str | None = None
|
|
published: bool = False
|
|
has_source: bool = False
|
|
last_modified: float | None = None
|
|
size: int | None = None
|
|
|
|
|
|
class PageDetail(BaseModel):
|
|
name: str
|
|
source: str | None = None
|
|
|
|
|
|
class SaveRequest(BaseModel):
|
|
source: str
|
|
publish: bool = False
|
|
|
|
|
|
def _extract_title(source: str) -> str | None:
|
|
"""Extract title from µFrame source or legacy Micron."""
|
|
for line in source.splitlines():
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith("#"):
|
|
continue
|
|
# µFrame: page "Title" [width]
|
|
if stripped.lower().startswith("page "):
|
|
try:
|
|
parts = shlex.split(stripped)
|
|
if len(parts) >= 2:
|
|
return parts[1]
|
|
except ValueError:
|
|
pass
|
|
break
|
|
# µFrame: heading 1 "Title"
|
|
if stripped.lower().startswith("heading "):
|
|
try:
|
|
parts = shlex.split(stripped)
|
|
if len(parts) >= 3:
|
|
return parts[2]
|
|
except ValueError:
|
|
pass
|
|
break
|
|
# Legacy Micron: >Title
|
|
if stripped.startswith(">") and not stripped.startswith(">>"):
|
|
return stripped[1:].strip()
|
|
break
|
|
return None
|
|
|
|
|
|
def _list_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 _source_path(name: str) -> Path:
|
|
"""Get source file path, preferring .uf over legacy .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 uf # default to .uf for new files
|
|
|
|
|
|
def _page_meta(name: str) -> PageMeta:
|
|
src_path = _source_path(name)
|
|
mu_path = PAGES_DIR / f"{name}.mu"
|
|
|
|
title = None
|
|
if src_path.is_file():
|
|
title = _extract_title(src_path.read_text(encoding="utf-8"))
|
|
elif mu_path.is_file():
|
|
title = _extract_title(mu_path.read_text(encoding="utf-8"))
|
|
|
|
published = mu_path.is_file()
|
|
last_modified = mu_path.stat().st_mtime if published else None
|
|
size = mu_path.stat().st_size if published else None
|
|
|
|
return PageMeta(
|
|
name=name,
|
|
title=title,
|
|
published=published,
|
|
has_source=src_path.is_file(),
|
|
last_modified=last_modified,
|
|
size=size,
|
|
)
|
|
|
|
|
|
@router.get("/pages", response_model=list[PageMeta])
|
|
async def list_pages():
|
|
return [_page_meta(n) for n in sorted(_list_all_page_names())]
|
|
|
|
|
|
@router.get("/pages/{name}", response_model=PageDetail)
|
|
async def get_page(name: str):
|
|
src_path = _source_path(name)
|
|
mu_path = PAGES_DIR / f"{name}.mu"
|
|
|
|
if not src_path.is_file() and not mu_path.is_file():
|
|
raise HTTPException(status_code=404, detail="Page not found")
|
|
|
|
source = (
|
|
src_path.read_text(encoding="utf-8")
|
|
if src_path.is_file()
|
|
else mu_path.read_text(encoding="utf-8")
|
|
)
|
|
|
|
return PageDetail(name=name, source=source)
|
|
|
|
|
|
@router.post("/pages/{name}", response_model=PageMeta)
|
|
async def save_page(name: str, req: SaveRequest):
|
|
SOURCES_DIR.mkdir(parents=True, exist_ok=True)
|
|
PAGES_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Save source as .uf
|
|
src_path = SOURCES_DIR / f"{name}.uf"
|
|
src_path.write_text(req.source, encoding="utf-8")
|
|
|
|
# Remove legacy .mu source if it exists
|
|
legacy_mu = SOURCES_DIR / f"{name}.mu"
|
|
if legacy_mu.is_file():
|
|
legacy_mu.unlink()
|
|
|
|
# Publish: compile .uf → .mu and write to pages dir
|
|
if req.publish:
|
|
try:
|
|
result = uframe.compile(req.source)
|
|
mu_path = PAGES_DIR / f"{name}.mu"
|
|
|
|
if result.is_dynamic and result.script:
|
|
# Dynamic page: write executable Python script
|
|
mu_path.write_text(result.script, encoding="utf-8")
|
|
mu_path.chmod(0o755) # Set execute bit for NomadNet
|
|
else:
|
|
# Static page: write compiled Micron
|
|
mu_path.write_text(result.micron, encoding="utf-8")
|
|
# Remove execute bit if it was previously dynamic
|
|
mu_path.chmod(0o644)
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=f"Compile failed during publish: {e}",
|
|
)
|
|
|
|
return _page_meta(name)
|
|
|
|
|
|
@router.delete("/pages/{name}")
|
|
async def delete_page(name: str):
|
|
src_path = _source_path(name)
|
|
mu_path = PAGES_DIR / f"{name}.mu"
|
|
|
|
if not src_path.is_file() and not mu_path.is_file():
|
|
raise HTTPException(status_code=404, detail="Page not found")
|
|
|
|
if src_path.is_file():
|
|
src_path.unlink()
|
|
if mu_path.is_file():
|
|
mu_path.unlink()
|
|
|
|
return {"deleted": name}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# File browser endpoints
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class FileEntry(BaseModel):
|
|
name: str
|
|
path: str
|
|
type: str # "file" | "folder" | "env"
|
|
size: int | None = None
|
|
last_modified: float | None = None
|
|
title: str | None = None
|
|
published: bool = False
|
|
|
|
|
|
class MkdirRequest(BaseModel):
|
|
path: str
|
|
|
|
|
|
class MoveRequest(BaseModel):
|
|
model_config = {"populate_by_name": True}
|
|
from_path: str = Field(alias="from")
|
|
to: str
|
|
|
|
|
|
class EnvRequest(BaseModel):
|
|
content: str
|
|
|
|
|
|
def _validate_relative_path(rel: str) -> Path:
|
|
"""Validate that a relative path has no traversal components and resolves
|
|
inside the expected base directories. Returns the cleaned relative Path."""
|
|
p = Path(rel)
|
|
# Reject absolute paths and any ".." components
|
|
if p.is_absolute():
|
|
raise HTTPException(status_code=400, detail="Absolute paths not allowed")
|
|
for part in p.parts:
|
|
if part == "..":
|
|
raise HTTPException(status_code=400, detail="Directory traversal not allowed")
|
|
# Extra safety: resolve against SOURCES_DIR and verify containment
|
|
resolved = (SOURCES_DIR / p).resolve()
|
|
if not str(resolved).startswith(str(SOURCES_DIR.resolve())):
|
|
raise HTTPException(status_code=400, detail="Path escapes base directory")
|
|
return p
|
|
|
|
|
|
def _file_entry(base: Path, rel_path: Path) -> FileEntry:
|
|
"""Build a FileEntry for a file or directory at base/rel_path."""
|
|
full = base / rel_path
|
|
name = rel_path.name
|
|
|
|
if full.is_dir():
|
|
return FileEntry(
|
|
name=name,
|
|
path=str(rel_path),
|
|
type="folder",
|
|
)
|
|
|
|
# .env file
|
|
if name == ".env":
|
|
stat = full.stat()
|
|
return FileEntry(
|
|
name=name,
|
|
path=str(rel_path),
|
|
type="env",
|
|
size=stat.st_size,
|
|
last_modified=stat.st_mtime,
|
|
)
|
|
|
|
# Regular file
|
|
stat = full.stat()
|
|
title = None
|
|
published = False
|
|
|
|
if full.suffix == ".uf":
|
|
try:
|
|
title = _extract_title(full.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
pass
|
|
# Check published status: corresponding .mu in PAGES_DIR at same relative path
|
|
mu_rel = rel_path.with_suffix(".mu")
|
|
published = (PAGES_DIR / mu_rel).is_file()
|
|
|
|
return FileEntry(
|
|
name=name,
|
|
path=str(rel_path),
|
|
type="file",
|
|
size=stat.st_size,
|
|
last_modified=stat.st_mtime,
|
|
title=title,
|
|
published=published,
|
|
)
|
|
|
|
|
|
@files_router.get("/files", response_model=list[FileEntry])
|
|
async def list_files(path: str = Query(default="")):
|
|
"""List files and folders in SOURCES_DIR, optionally scoped to a subfolder."""
|
|
if path:
|
|
rel = _validate_relative_path(path)
|
|
else:
|
|
rel = Path(".")
|
|
|
|
target = (SOURCES_DIR / rel).resolve()
|
|
if not str(target).startswith(str(SOURCES_DIR.resolve())):
|
|
raise HTTPException(status_code=400, detail="Path escapes base directory")
|
|
if not target.is_dir():
|
|
raise HTTPException(status_code=404, detail="Directory not found")
|
|
|
|
entries: list[FileEntry] = []
|
|
for item in sorted(target.iterdir(), key=lambda p: (not p.is_dir(), p.name.lower())):
|
|
item_rel = item.relative_to(SOURCES_DIR)
|
|
entries.append(_file_entry(SOURCES_DIR, item_rel))
|
|
|
|
return entries
|
|
|
|
|
|
@files_router.post("/files/mkdir")
|
|
async def mkdir(req: MkdirRequest):
|
|
"""Create a folder in both SOURCES_DIR and PAGES_DIR."""
|
|
rel = _validate_relative_path(req.path)
|
|
|
|
(SOURCES_DIR / rel).mkdir(parents=True, exist_ok=True)
|
|
(PAGES_DIR / rel).mkdir(parents=True, exist_ok=True)
|
|
|
|
return {"created": str(rel)}
|
|
|
|
|
|
@files_router.post("/files/move")
|
|
async def move_file(req: MoveRequest):
|
|
"""Move/rename a file or folder in both SOURCES_DIR and PAGES_DIR."""
|
|
from_rel = _validate_relative_path(req.from_path)
|
|
to_rel = _validate_relative_path(req.to)
|
|
|
|
# Move in SOURCES_DIR
|
|
src_from = SOURCES_DIR / from_rel
|
|
src_to = SOURCES_DIR / to_rel
|
|
if src_from.exists():
|
|
src_to.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.move(str(src_from), str(src_to))
|
|
|
|
# Move in PAGES_DIR (for .uf files, look for .mu counterpart)
|
|
if src_from.suffix == ".uf" or (not src_from.exists() and from_rel.suffix == ".uf"):
|
|
pages_from = PAGES_DIR / from_rel.with_suffix(".mu")
|
|
pages_to = PAGES_DIR / to_rel.with_suffix(".mu")
|
|
else:
|
|
pages_from = PAGES_DIR / from_rel
|
|
pages_to = PAGES_DIR / to_rel
|
|
|
|
if pages_from.exists():
|
|
pages_to.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.move(str(pages_from), str(pages_to))
|
|
|
|
return {"moved": {"from": str(from_rel), "to": str(to_rel)}}
|
|
|
|
|
|
@files_router.get("/files/env")
|
|
async def read_env():
|
|
"""Read the .env file from SOURCES_DIR root."""
|
|
env_path = SOURCES_DIR / ".env"
|
|
if env_path.is_file():
|
|
return {"content": env_path.read_text(encoding="utf-8")}
|
|
return {"content": ""}
|
|
|
|
|
|
@files_router.post("/files/env")
|
|
async def save_env(req: EnvRequest):
|
|
"""Save the .env file to SOURCES_DIR root."""
|
|
SOURCES_DIR.mkdir(parents=True, exist_ok=True)
|
|
env_path = SOURCES_DIR / ".env"
|
|
env_path.write_text(req.content, encoding="utf-8")
|
|
return {"saved": True}
|