218 lines
8.9 KiB
Python
218 lines
8.9 KiB
Python
"""Admin routes: index, favicon, browse-dirs, robots."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, Body, HTTPException
|
|
from fastapi.responses import FileResponse, HTMLResponse, PlainTextResponse, Response
|
|
|
|
from core.constants import STATIC_DIR, __version__
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter()
|
|
|
|
_BROWSE_BLOCKED: frozenset[str] = frozenset({
|
|
"/proc", "/sys", "/dev", "/run", "/boot",
|
|
})
|
|
|
|
# Any folder a Browse dialog leads a user to pick only really "exists" for
|
|
# them if it survives the container being recreated and, ideally, is visible
|
|
# from their host machine too — i.e. it's on a bind/volume mount, not the
|
|
# container's own throwaway overlay filesystem. Read once from /proc/mounts
|
|
# (containers don't remount at runtime) rather than per-request: confirmed
|
|
# live as a real gap — a user picked a folder from the picker for the
|
|
# audiobook-export "Save Location" field with no way to tell it was actually
|
|
# just container-internal storage that would vanish and never show up on
|
|
# their host.
|
|
_PERSISTED_MOUNTS: frozenset[str] | None = None
|
|
_MOUNT_IGNORE_FSTYPES = frozenset({
|
|
"overlay", "proc", "sysfs", "cgroup", "cgroup2", "devpts", "mqueue",
|
|
"tmpfs", "devtmpfs", "securityfs", "debugfs", "tracefs", "pstore",
|
|
"bpf", "hugetlbfs", "autofs", "nsfs",
|
|
})
|
|
|
|
|
|
def _persisted_mounts() -> frozenset[str]:
|
|
global _PERSISTED_MOUNTS
|
|
if _PERSISTED_MOUNTS is not None:
|
|
return _PERSISTED_MOUNTS
|
|
mounts: set[str] = set()
|
|
try:
|
|
with open("/proc/mounts", "r", encoding="utf-8", errors="replace") as f:
|
|
for line in f:
|
|
parts = line.split()
|
|
if len(parts) < 3:
|
|
continue
|
|
mnt, fstype = parts[1], parts[2]
|
|
if fstype in _MOUNT_IGNORE_FSTYPES or mnt == "/":
|
|
continue
|
|
if any(mnt == b or mnt.startswith(b + "/") for b in _BROWSE_BLOCKED):
|
|
continue
|
|
# Individual files are sometimes bind-mounted too (this app's
|
|
# own dev setup does this for /etc/resolv.conf etc.) — those
|
|
# can never be "picked" as a folder and would otherwise mark
|
|
# their parent directory as a reachable waypoint that then
|
|
# dead-ends (confirmed live: /etc showed up pinned for this
|
|
# reason alone, with nothing browsable underneath it).
|
|
if not os.path.isdir(mnt):
|
|
continue
|
|
mounts.add(mnt)
|
|
except OSError as exc:
|
|
logger.warning("Could not read /proc/mounts for the folder browser: %s", exc)
|
|
_PERSISTED_MOUNTS = frozenset(mounts)
|
|
return _PERSISTED_MOUNTS
|
|
|
|
|
|
def _mount_relation(path_str: str, mounts: frozenset[str]) -> tuple[bool, bool]:
|
|
"""Returns (is_persisted, leads_to_persisted) for `path_str`.
|
|
is_persisted: the path itself is on (at or below) a real mount — safe to
|
|
actually save into. leads_to_persisted: an ancestor of some mount — not
|
|
itself persisted, but worth navigating into (e.g. /home on the way to
|
|
/home/app/.config/tts-voice-creator, which sits inside otherwise-ordinary
|
|
overlay directories)."""
|
|
is_persisted = any(path_str == m or path_str.startswith(m + "/") for m in mounts)
|
|
leads_to = is_persisted or any(m == path_str or m.startswith(path_str + "/") for m in mounts)
|
|
return is_persisted, leads_to
|
|
|
|
|
|
_ASSET_URL_RE = re.compile(r"(/static/[^\"'\s?]+)\?v=[^\"'\s&]+")
|
|
|
|
|
|
def asset_build_id() -> str:
|
|
"""A version string that changes whenever any script, stylesheet or module list changes.
|
|
|
|
Scripts and stylesheets are requested as ``file?v=<version>`` and cached for a year as
|
|
immutable, so a version that only moves at release time leaves a browser (or a proxy in front
|
|
of it) running old code after an edit -- a button added to a script simply does not appear.
|
|
The newest modification time under static/ js, css and the manifest is folded into the version.
|
|
"""
|
|
newest = 0
|
|
for pattern in ("js/*.js", "css/*.css", "*.css", "loader.js", "nav.js", "module-manifest.json"):
|
|
for path in STATIC_DIR.glob(pattern):
|
|
try:
|
|
newest = max(newest, int(path.stat().st_mtime))
|
|
except OSError:
|
|
pass
|
|
return f"{__version__}-{newest:x}"
|
|
|
|
|
|
@router.get("/")
|
|
async def index():
|
|
"""The app shell, with every /static asset URL carrying the current build id."""
|
|
html = (STATIC_DIR / "index.html").read_text("utf-8")
|
|
build = asset_build_id()
|
|
return HTMLResponse(_ASSET_URL_RE.sub(lambda m: f"{m.group(1)}?v={build}", html),
|
|
headers={"Cache-Control": "no-store, max-age=0"})
|
|
|
|
|
|
@router.get("/favicon.ico")
|
|
async def favicon():
|
|
return FileResponse(STATIC_DIR / "icon.svg", media_type="image/svg+xml")
|
|
|
|
|
|
# ── PWA: serve the service worker and manifest from root so the SW controls "/" ──
|
|
@router.get("/sw.js")
|
|
async def service_worker():
|
|
return FileResponse(
|
|
STATIC_DIR / "sw.js",
|
|
media_type="application/javascript",
|
|
headers={"Cache-Control": "no-cache", "Service-Worker-Allowed": "/"},
|
|
)
|
|
|
|
|
|
@router.get("/manifest.webmanifest")
|
|
async def manifest():
|
|
return FileResponse(STATIC_DIR / "manifest.webmanifest", media_type="application/manifest+json")
|
|
|
|
|
|
@router.get("/api/browse-dirs")
|
|
async def browse_dirs(path: str = "/"):
|
|
p = Path(path).resolve()
|
|
p_str = str(p)
|
|
if any(p_str == b or p_str.startswith(b + "/") for b in _BROWSE_BLOCKED):
|
|
raise HTTPException(403, "Access to this path is not permitted")
|
|
if not p.is_dir():
|
|
raise HTTPException(404, "Not a directory")
|
|
try:
|
|
all_dirs = sorted([d.name for d in p.iterdir() if d.is_dir()], key=str.lower)
|
|
entries = [n for n in all_dirs if not n.startswith(".")]
|
|
except PermissionError:
|
|
raise HTTPException(403, "Permission denied")
|
|
parent = str(p.parent) if p.parent != p else None
|
|
mounts = _persisted_mounts()
|
|
is_persisted, _ = _mount_relation(p_str, mounts)
|
|
# Additive fields only — existing callers (e.g. voice-library.js's own
|
|
# folder picker) read `dirs` as plain strings and are unaffected.
|
|
# Computed over ALL subdirectories, dotdirs included: the app's own
|
|
# default export location (~/.config/tts-voice-creator) sits behind a
|
|
# dotdir, so excluding those here would make it unreachable through
|
|
# mounted-only navigation even though it's the single most useful target
|
|
# (confirmed live) — `dirs` above still hides dotfiles for every other
|
|
# (unrestricted) use of this endpoint, unaffected.
|
|
reachable_dirs = []
|
|
persisted_dirs = []
|
|
for name in all_dirs:
|
|
child = p_str.rstrip("/") + "/" + name if p_str != "/" else "/" + name
|
|
child_persisted, child_leads_to = _mount_relation(child, mounts)
|
|
if child_leads_to:
|
|
reachable_dirs.append(name)
|
|
if child_persisted:
|
|
persisted_dirs.append(name)
|
|
return {
|
|
"path": p_str, "parent": parent, "dirs": entries,
|
|
"is_persisted": is_persisted,
|
|
"reachable_dirs": reachable_dirs,
|
|
"persisted_dirs": persisted_dirs,
|
|
}
|
|
|
|
|
|
_MKDIR_NAME_RE = re.compile(r"^[^/\\\x00]{1,100}$")
|
|
|
|
|
|
@router.post("/api/browse-dirs/mkdir")
|
|
async def browse_dirs_mkdir(payload: dict = Body(...)):
|
|
"""Create a subfolder from inside the folder browser modal — without
|
|
this, picking a destination that doesn't exist yet (e.g. a new export
|
|
folder) meant leaving the app to create it first (confirmed live)."""
|
|
path = str(payload.get("path") or "")
|
|
name = str(payload.get("name") or "").strip()
|
|
if not name or name in (".", "..") or not _MKDIR_NAME_RE.match(name):
|
|
raise HTTPException(400, "Invalid folder name")
|
|
p = Path(path).resolve()
|
|
p_str = str(p)
|
|
if any(p_str == b or p_str.startswith(b + "/") for b in _BROWSE_BLOCKED):
|
|
raise HTTPException(403, "Access to this path is not permitted")
|
|
if not p.is_dir():
|
|
raise HTTPException(404, "Not a directory")
|
|
new_dir = p / name
|
|
if new_dir.exists() and not new_dir.is_dir():
|
|
raise HTTPException(409, "A file with that name already exists here")
|
|
try:
|
|
new_dir.mkdir(exist_ok=True)
|
|
except PermissionError:
|
|
raise HTTPException(403, "Permission denied")
|
|
return {"ok": True, "path": str(new_dir)}
|
|
|
|
|
|
@router.get("/api/version")
|
|
async def get_version():
|
|
from core.vocal_isolation import available as _vocal_isolation
|
|
return {"version": __version__, "features": {"vocal_isolation": _vocal_isolation()}}
|
|
|
|
|
|
@router.get("/api/changelog", response_class=PlainTextResponse)
|
|
async def get_changelog():
|
|
root = Path(__file__).parent.parent
|
|
for cl in (root / "CHANGELOG.md", root / "static" / "CHANGELOG.md"):
|
|
if cl.is_file():
|
|
return cl.read_text(encoding="utf-8")
|
|
raise HTTPException(404, "CHANGELOG.md not found")
|
|
|
|
|
|
@router.get("/robots.txt", response_class=PlainTextResponse)
|
|
async def robots_txt():
|
|
return "User-agent: *\nDisallow: /"
|