mirror of
https://github.com/eunomia-bpf/bpf-developer-tutorial.git
synced 2026-10-31 20:54:56 +08:00
Harden the tutorial writing workflow
This commit is contained in:
LinuxDev9002
219
.agents/skills/write-bpf-production-tutorial/scripts/sync-source-blocks.py
Executable file
219
.agents/skills/write-bpf-production-tutorial/scripts/sync-source-blocks.py
Executable file
@@ -0,0 +1,219 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Synchronize inventoried complete-source blocks in tutorial READMEs."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import pathlib
|
||||
import re
|
||||
import sys
|
||||
import tempfile
|
||||
|
||||
|
||||
BEGIN_RE = re.compile(r"^<!-- BEGIN FULL SOURCE: ([^ ]+) -->$")
|
||||
END = "<!-- END FULL SOURCE -->"
|
||||
LANGUAGES = {
|
||||
".c": "c",
|
||||
".cc": "cpp",
|
||||
".cpp": "cpp",
|
||||
".go": "go",
|
||||
".h": "c",
|
||||
".py": "python",
|
||||
".rs": "rust",
|
||||
".sh": "bash",
|
||||
}
|
||||
|
||||
|
||||
def contained_path(root: pathlib.Path, value: str, label: str) -> pathlib.Path:
|
||||
path = (root / value).resolve()
|
||||
try:
|
||||
path.relative_to(root)
|
||||
except ValueError as error:
|
||||
raise ValueError(f"{label} escapes repository root: {value}") from error
|
||||
return path
|
||||
|
||||
|
||||
def render(
|
||||
readme: pathlib.Path,
|
||||
root: pathlib.Path,
|
||||
forbidden_sources: set[pathlib.Path],
|
||||
) -> tuple[str, set[pathlib.Path]]:
|
||||
lines = readme.read_text(encoding="utf-8").splitlines()
|
||||
rendered: list[str] = []
|
||||
sources: set[pathlib.Path] = set()
|
||||
index = 0
|
||||
|
||||
while index < len(lines):
|
||||
match = BEGIN_RE.match(lines[index])
|
||||
if not match:
|
||||
rendered.append(lines[index])
|
||||
index += 1
|
||||
continue
|
||||
|
||||
source_name = match.group(1)
|
||||
source = contained_path(root, source_name, "source path")
|
||||
if not source.is_file():
|
||||
raise ValueError(f"source file does not exist: {source_name}")
|
||||
if source in forbidden_sources:
|
||||
raise ValueError(
|
||||
f"complete-source marker cannot name a tutorial README: {source_name}"
|
||||
)
|
||||
if source.suffix not in LANGUAGES:
|
||||
raise ValueError(f"unsupported complete-source file type: {source_name}")
|
||||
if source in sources:
|
||||
raise ValueError(f"duplicate complete-source marker: {source_name}")
|
||||
|
||||
end = index + 1
|
||||
while end < len(lines) and lines[end] != END:
|
||||
if BEGIN_RE.match(lines[end]):
|
||||
raise ValueError(
|
||||
f"nested complete-source marker before {END!r} after {source_name}"
|
||||
)
|
||||
end += 1
|
||||
if end == len(lines):
|
||||
raise ValueError(f"missing {END!r} after {source_name}")
|
||||
|
||||
language = LANGUAGES[source.suffix]
|
||||
source_text = source.read_text(encoding="utf-8").rstrip("\n")
|
||||
rendered.extend(
|
||||
[
|
||||
lines[index],
|
||||
f"```{language}",
|
||||
source_text,
|
||||
"```",
|
||||
END,
|
||||
]
|
||||
)
|
||||
sources.add(source)
|
||||
index = end + 1
|
||||
|
||||
return "\n".join(rendered) + "\n", sources
|
||||
|
||||
|
||||
def display_paths(paths: set[pathlib.Path], root: pathlib.Path) -> str:
|
||||
return ", ".join(sorted(str(path.relative_to(root)) for path in paths))
|
||||
|
||||
|
||||
def atomic_write(path: pathlib.Path, content: str) -> None:
|
||||
temporary_name: str | None = None
|
||||
try:
|
||||
with tempfile.NamedTemporaryFile(
|
||||
"w",
|
||||
encoding="utf-8",
|
||||
dir=path.parent,
|
||||
prefix=f".{path.name}.",
|
||||
delete=False,
|
||||
) as temporary:
|
||||
temporary_name = temporary.name
|
||||
temporary.write(content)
|
||||
temporary.flush()
|
||||
os.fsync(temporary.fileno())
|
||||
os.chmod(temporary_name, path.stat().st_mode)
|
||||
os.replace(temporary_name, path)
|
||||
temporary_name = None
|
||||
finally:
|
||||
if temporary_name is not None:
|
||||
pathlib.Path(temporary_name).unlink(missing_ok=True)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--repo", required=True, help="absolute repository root")
|
||||
parser.add_argument(
|
||||
"--readme", action="append", required=True, help="repository-relative README"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--expected-source",
|
||||
action="append",
|
||||
required=True,
|
||||
help="repository-relative complete core source; repeat for the full inventory",
|
||||
)
|
||||
mode = parser.add_mutually_exclusive_group(required=True)
|
||||
mode.add_argument("--write", action="store_true", help="replace marked blocks")
|
||||
mode.add_argument("--check", action="store_true", help="fail when blocks differ")
|
||||
args = parser.parse_args()
|
||||
|
||||
root = pathlib.Path(args.repo).resolve()
|
||||
if not (root / ".git").exists() and not (root / ".git").is_file():
|
||||
print(f"not a Git worktree: {root}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
failed = False
|
||||
readmes: list[tuple[str, pathlib.Path]] = []
|
||||
for value in args.readme:
|
||||
try:
|
||||
readme = contained_path(root, value, "README path")
|
||||
if not readme.is_file():
|
||||
raise ValueError(f"README does not exist: {value}")
|
||||
if readme in {path for _, path in readmes}:
|
||||
raise ValueError(f"duplicate README path: {value}")
|
||||
readmes.append((value, readme))
|
||||
except (OSError, ValueError) as error:
|
||||
print(error, file=sys.stderr)
|
||||
failed = True
|
||||
|
||||
if failed:
|
||||
return 1
|
||||
|
||||
readme_paths = {path for _, path in readmes}
|
||||
expected_sources: set[pathlib.Path] = set()
|
||||
for value in args.expected_source:
|
||||
try:
|
||||
source = contained_path(root, value, "expected source path")
|
||||
if not source.is_file():
|
||||
raise ValueError(f"expected source file does not exist: {value}")
|
||||
if source in readme_paths:
|
||||
raise ValueError(f"expected source cannot be a tutorial README: {value}")
|
||||
if source.suffix not in LANGUAGES:
|
||||
raise ValueError(f"unsupported expected-source file type: {value}")
|
||||
if source in expected_sources:
|
||||
raise ValueError(f"duplicate expected source: {value}")
|
||||
expected_sources.add(source)
|
||||
except (OSError, ValueError) as error:
|
||||
print(error, file=sys.stderr)
|
||||
failed = True
|
||||
|
||||
if failed:
|
||||
return 1
|
||||
|
||||
planned: list[tuple[str, pathlib.Path, str, str, int]] = []
|
||||
for value, readme in readmes:
|
||||
try:
|
||||
before = readme.read_text(encoding="utf-8")
|
||||
after, sources = render(readme, root, readme_paths)
|
||||
missing = expected_sources - sources
|
||||
extra = sources - expected_sources
|
||||
if missing or extra:
|
||||
details = []
|
||||
if missing:
|
||||
details.append(f"missing: {display_paths(missing, root)}")
|
||||
if extra:
|
||||
details.append(f"unexpected: {display_paths(extra, root)}")
|
||||
raise ValueError(
|
||||
f"complete-source inventory mismatch in {value} ({'; '.join(details)})"
|
||||
)
|
||||
planned.append((value, readme, before, after, len(sources)))
|
||||
except (OSError, ValueError) as error:
|
||||
print(error, file=sys.stderr)
|
||||
failed = True
|
||||
|
||||
if failed:
|
||||
return 1
|
||||
|
||||
for value, readme, before, after, blocks in planned:
|
||||
if args.write:
|
||||
if before != after:
|
||||
atomic_write(readme, after)
|
||||
print(f"synced {blocks} source blocks in {value}")
|
||||
elif before != after:
|
||||
print(f"stale complete-source block in {value}", file=sys.stderr)
|
||||
failed = True
|
||||
else:
|
||||
print(f"checked {blocks} source blocks in {value}")
|
||||
|
||||
return 1 if failed else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user