summaryrefslogtreecommitdiff
path: root/scripts
diff options
context:
space:
mode:
Diffstat (limited to 'scripts')
-rw-r--r--scripts/import_myst.py36
-rw-r--r--scripts/swap_sources.py29
2 files changed, 54 insertions, 11 deletions
diff --git a/scripts/import_myst.py b/scripts/import_myst.py
index 88ba7006..b9dd7898 100644
--- a/scripts/import_myst.py
+++ b/scripts/import_myst.py
@@ -57,7 +57,10 @@ def list_myst_files(source: str) -> list[str]:
cwd=REPO_ROOT,
)
if result.returncode != 0:
- return []
+ raise SystemExit(
+ f"import_myst: git ls-tree {source!r} failed "
+ f"(exit {result.returncode}): {result.stderr.strip()}"
+ )
stems = []
for line in result.stdout.splitlines():
@@ -73,8 +76,14 @@ def list_myst_files(source: str) -> list[str]:
def list_rst_files(docs_dir: Path) -> set[str]:
"""Return set of stems (relative to docs_dir, no extension) for all *.rst files."""
+ build_dir = (docs_dir / "_build").resolve()
stems = set()
for rst in docs_dir.rglob("*.rst"):
+ try:
+ rst.resolve().relative_to(build_dir)
+ continue # inside _build/, skip
+ except ValueError:
+ pass
rel = rst.relative_to(docs_dir)
stem = str(rel)[:-4] # strip ".rst"
stems.add(stem)
@@ -89,8 +98,12 @@ def import_page(stem: str, md_content: bytes, docs_dir: Path, force: bool) -> bo
- Without --force: warns and skips (returns False) if different content exists.
- With --force: overwrites (returns True).
- Returns True on successful write.
+
+ Refuses (raises ValueError) if *stem* would resolve outside *docs_dir*.
"""
p = Path(stem)
+ if p.is_absolute() or ".." in p.parts:
+ raise ValueError(f"import_myst: refusing unsafe stem {stem!r}")
name = p.name
parent = p.parent
if str(parent) == ".":
@@ -99,6 +112,9 @@ def import_page(stem: str, md_content: bytes, docs_dir: Path, force: bool) -> bo
target_dir = docs_dir / parent
dest = target_dir / f"md-{name}.md"
+ docs_root = docs_dir.resolve()
+ if not dest.resolve().parent.is_relative_to(docs_root):
+ raise ValueError(f"import_myst: refusing stem {stem!r} — escapes docs/")
if dest.exists():
existing = dest.read_bytes()
@@ -147,6 +163,7 @@ def do_import(
imported = 0
skipped = 0
+ would_import = 0
for stem in candidate_stems:
# Path inside the branch: docs/{stem}.md
@@ -162,7 +179,7 @@ def do_import(
if dry_run:
print(f" (dry-run) would import {stem}")
- imported += 1
+ would_import += 1
continue
wrote = import_page(stem, content, docs_dir, force)
@@ -171,10 +188,17 @@ def do_import(
else:
skipped += 1
- print(
- f"import_myst: imported {imported}, skipped {skipped}.",
- file=sys.stderr,
- )
+ if dry_run:
+ print(
+ f"import_myst: (dry-run) would import {would_import}, "
+ f"skipped {skipped}.",
+ file=sys.stderr,
+ )
+ else:
+ print(
+ f"import_myst: imported {imported}, skipped {skipped}.",
+ file=sys.stderr,
+ )
# ---------------------------------------------------------------------------
diff --git a/scripts/swap_sources.py b/scripts/swap_sources.py
index 7e82e876..683d5ef9 100644
--- a/scripts/swap_sources.py
+++ b/scripts/swap_sources.py
@@ -52,7 +52,7 @@ def parse_swap_list(path: Path) -> list:
if not path.exists():
return []
stems = []
- for line in path.read_text().splitlines():
+ for line in path.read_text(encoding="utf-8").splitlines():
stripped = line.strip()
if stripped and not stripped.startswith("#"):
stems.append(stripped)
@@ -179,8 +179,8 @@ def do_swap(docs_dir: Path) -> None:
exclude_lines.append(str(rel_rst))
state = {"version": STATE_VERSION, "swaps": swaps}
- state_path.write_text(json.dumps(state, indent=2))
- _exclude_path(docs_dir).write_text("\n".join(exclude_lines) + "\n")
+ state_path.write_text(json.dumps(state, indent=2), encoding="utf-8")
+ _exclude_path(docs_dir).write_text("\n".join(exclude_lines) + "\n", encoding="utf-8")
print(f"swap_sources: swapped {len(completed)} file(s).", file=sys.stderr)
@@ -197,10 +197,29 @@ def do_restore(docs_dir: Path) -> None:
if not state_path.exists():
return # no-op
- state = json.loads(state_path.read_text())
+ try:
+ state = json.loads(state_path.read_text(encoding="utf-8"))
+ except (json.JSONDecodeError, OSError) as exc:
+ raise RuntimeError(
+ f"swap_sources: cannot read state file {state_path}: {exc}. "
+ f"Inspect or delete it manually."
+ ) from exc
+
+ version = state.get("version")
+ if version != STATE_VERSION:
+ raise RuntimeError(
+ f"swap_sources: state file version {version!r} does not match "
+ f"expected {STATE_VERSION!r}. Inspect {state_path} manually."
+ )
+
swaps = state.get("swaps", [])
for entry in reversed(swaps):
+ if not isinstance(entry, dict) or "md_from" not in entry or "md_to" not in entry:
+ raise RuntimeError(
+ f"swap_sources: malformed entry in {state_path}: {entry!r}. "
+ f"Inspect manually."
+ )
md_from = docs_dir / entry["md_from"] # original source (md-prefixed)
md_to = docs_dir / entry["md_to"] # current location (plain .md)
if md_to.exists():
@@ -254,7 +273,7 @@ def do_status(docs_dir: Path) -> None:
print("swap_sources: no active swap state.")
return
- state = json.loads(state_path.read_text())
+ state = json.loads(state_path.read_text(encoding="utf-8"))
swaps = state.get("swaps", [])
print(f"swap_sources: {len(swaps)} file(s) currently swapped:")
for entry in swaps: