feat(cli): bundle offline workflow guides and expose dpdata skills

MANIFEST.in

@@ -1,3 +1,4 @@
 prune docs
 prune tests
 prune plugin_example
+graft skills

docs/cli.rst

@@ -5,3 +5,22 @@ Command line interface
    :module: dpdata.cli
    :func: dpdata_parser
    :prog: dpdata
+
+Offline workflow guides
+-----------------------
+
+The installed ``dpdata`` package includes the public workflow guides from
+its ``skills/`` directory. Read the instructions matching your installed
+version without installing a separate agent integration:
+
+.. code-block:: console
+
+   dpdata skills list
+   dpdata skills read dpdata-cli
+   dpdata skills read dpdata-cli README.md
+
+The optional file argument must be a relative path inside the selected guide;
+without one, ``skills read`` displays ``SKILL.md``. These commands work
+offline and only print documentation: they do not execute workflow examples
+or convert data. To convert a file literally named ``skills``, pass
+``./skills`` or use ``--`` to separate the filename from CLI options.

dpdata/cli.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import argparse
+import sys
 
 from . import __version__
 from .system import LabeledSystem, MultiSystems, System
@@ -16,9 +17,14 @@ def dpdata_parser() -> argparse.ArgumentParser:
     argparse.ArgumentParser
         dpdata cli parser
     """
+    from .skills import workflow_help
+
     parser = argparse.ArgumentParser(
         description="dpdata: Manipulating multiple atomic simulation data formats",
         formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+        epilog=workflow_help("dpdata-cli")
+        + " Discover other guides: dpdata skills list. "
+        "To convert a file named skills, use ./skills or -- skills.",
     )
 
     parser.add_argument("from_file", type=str, help="read data from a file")
@@ -51,6 +57,10 @@ def dpdata_cli():
 
         $ dpdata -iposcar POSCAR -odeepmd/npy -O data -n
     """
+    if sys.argv[1:2] == ["skills"]:
+        from .skills import skills_cli
+
+        return skills_cli(sys.argv[2:])
     parser = dpdata_parser()
     parsed_args = parser.parse_args()
     convert(**vars(parsed_args))

dpdata/skills.py

@@ -0,0 +1,89 @@
+"""Read the public workflow skills shipped with this dpdata version."""
+
+from __future__ import annotations
+
+import argparse
+import re
+import sys
+from importlib.resources import files
+from pathlib import Path
+
+
+def _root():
+    bundled = files("dpdata").joinpath("_skills")
+    if bundled.is_dir():
+        return bundled
+    # Editable/source checkouts use the same canonical directory as the build.
+    source = Path(__file__).resolve().parent.parent
+    if (source / "pyproject.toml").is_file() and (source / "skills").is_dir():
+        return source / "skills"
+    raise ValueError("Bundled skills are missing; reinstall dpdata.")
+
+
+def _parts(value):
+    if "\\" in value or "\x00" in value or ":" in value:
+        raise ValueError("Use a relative skill path without backslashes or colons.")
+    parts = value.split("/")
+    if any(part in ("", ".", "..") for part in parts):
+        raise ValueError("Skill paths cannot be absolute or contain '.' or '..'.")
+    return parts
+
+
+def _read(root, name, path):
+    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]*", name):
+        raise ValueError(f"Invalid skill name: {name!r}")
+    skill = root.joinpath(name)
+    if not skill.joinpath("SKILL.md").is_file():
+        raise ValueError(f"Unknown skill: {name}")
+    target = skill.joinpath(*_parts(path))
+    if isinstance(target, Path) and (
+        not skill.resolve().is_relative_to(root.resolve())
+        or not target.resolve().is_relative_to(skill.resolve())
+    ):
+        raise ValueError("Skill paths must stay within the named bundled skill.")
+    if not target.is_file():
+        raise ValueError(f"Unknown skill file: {name}/{path}")
+    try:
+        return target.read_bytes().decode("utf-8")
+    except UnicodeError as exc:
+        raise ValueError("Only UTF-8 text skill files can be read.") from exc
+
+
+def skills_cli(argv):
+    """List and read bundled instructions without running a workflow."""
+    parser = argparse.ArgumentParser(
+        prog="dpdata skills", description="Read offline, bundled workflow instructions."
+    )
+    commands = parser.add_subparsers(dest="command", required=True)
+    commands.add_parser("list", help="list available workflows")
+    reading = commands.add_parser("read", help="read a workflow or its reference file")
+    reading.add_argument("name", help="skill name from 'dpdata skills list'")
+    reading.add_argument(
+        "path", nargs="?", default="SKILL.md", help="relative text file"
+    )
+    args = parser.parse_args(argv)
+    try:
+        root = _root()
+        if args.command == "list":
+            names = sorted(
+                item.name
+                for item in root.iterdir()
+                if item.joinpath("SKILL.md").is_file()
+            )
+            print("\n".join(names))
+        else:
+            content = _read(root, args.name, args.path)
+            sys.stdout.write(content)
+    except (ValueError, OSError) as exc:
+        parser.error(str(exc))
+
+
+def workflow_help(name: str) -> str:
+    """Point help readers to an available, version-matched workflow guide."""
+    try:
+        available = _root().joinpath(name, "SKILL.md").is_file()
+    except (ValueError, OSError):
+        return ""
+    if not available:
+        return ""
+    return f"Workflow guide (read first): dpdata skills read {name}"

setup.py

@@ -0,0 +1,24 @@
+"""Bundle the canonical public skills with binary distributions."""
+
+from __future__ import annotations
+
+import shutil
+from pathlib import Path
+
+from setuptools import setup
+from setuptools.command.build_py import build_py
+
+
+class BuildPy(build_py):
+    """Copy public workflow documentation without a second maintained copy."""
+
+    def run(self):
+        super().run()
+        source = Path(__file__).parent / "skills"
+        destination = Path(self.build_lib) / "dpdata" / "_skills"
+        if destination.exists():
+            shutil.rmtree(destination)
+        shutil.copytree(source, destination)
+
+
+setup(cmdclass={"build_py": BuildPy})

tests/test_skill_help.py

@@ -0,0 +1,32 @@
+"""Discoverable workflow guidance belongs to help, not command output."""
+
+from __future__ import annotations
+
+import contextlib
+import io
+import unittest
+from unittest.mock import patch
+
+from dpdata.cli import dpdata_parser
+from dpdata.skills import workflow_help
+
+
+class TestSkillHelp(unittest.TestCase):
+    def test_contextual_help(self) -> None:
+        for command, skill in {"": "dpdata-cli"}.items():
+            with self.subTest(command=command):
+                output = io.StringIO()
+                with (
+                    contextlib.redirect_stdout(output),
+                    self.assertRaises(SystemExit) as result,
+                ):
+                    dpdata_parser().parse_args(
+                        ([command] if command else []) + ["--help"]
+                    )
+                self.assertEqual(result.exception.code, 0)
+                self.assertIn(f"dpdata skills read {skill}", output.getvalue())
+
+    def test_missing_guide_has_no_pointer(self) -> None:
+        self.assertEqual(workflow_help("not-a-bundled-skill"), "")
+        with patch("dpdata.skills._root", side_effect=ValueError("missing resources")):
+            self.assertEqual(workflow_help("dpdata-cli"), "")

tests/test_skills.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import contextlib
+import io
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from dpdata.cli import dpdata_cli, dpdata_parser
+from dpdata.skills import _read, _root, skills_cli
+
+
+class TestSkills(unittest.TestCase):
+    def test_list_and_read(self):
+        root = _root()
+        expected = sorted(
+            p.name for p in root.iterdir() if p.joinpath("SKILL.md").is_file()
+        )
+        output = io.StringIO()
+        with contextlib.redirect_stdout(output):
+            skills_cli(["list"])
+        self.assertEqual(output.getvalue().splitlines(), expected)
+        self.assertTrue(expected)
+        for name in expected:
+            output = io.StringIO()
+            with contextlib.redirect_stdout(output):
+                skills_cli(["read", name])
+            self.assertEqual(
+                output.getvalue(),
+                root.joinpath(name, "SKILL.md").read_bytes().decode("utf-8"),
+            )
+
+    def test_reference(self):
+        self.assertEqual(
+            _read(_root(), "dpdata-cli", "README.md"),
+            _root().joinpath("dpdata-cli", "README.md").read_bytes().decode("utf-8"),
+        )
+
+    def test_bad_paths(self):
+        for path in [
+            "/etc/passwd",
+            "../SKILL.md",
+            "references/../../SKILL.md",
+            "a\\b",
+            "C:/a",
+            "a\x00b",
+            "./SKILL.md",
+            "missing.md",
+        ]:
+            with self.subTest(path=path), self.assertRaises(ValueError):
+                _read(_root(), "dpdata-cli", path)
+        for name in ["../dpdata-cli", "/dpdata-cli", "missing"]:
+            with self.subTest(name=name), self.assertRaises(ValueError):
+                _read(_root(), name, "SKILL.md")
+
+    def test_symlink_escape(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory) / "skills"
+            skill = root / "example"
+            skill.mkdir(parents=True)
+            (skill / "SKILL.md").write_text("example", encoding="utf-8")
+            secret = Path(directory) / "secret"
+            secret.write_text("outside", encoding="utf-8")
+            (skill / "escape.md").symlink_to(secret)
+            with self.assertRaises(ValueError):
+                _read(root, "example", "escape.md")
+
+    def test_help_and_legacy_parser(self):
+        self.assertIn("skills list", dpdata_parser().format_help())
+        for argv in [["POSCAR", "-n"], ["./skills"], ["--", "skills"]]:
+            self.assertEqual(
+                dpdata_parser().parse_args(argv).from_file,
+                argv[0] if argv[0] != "--" else "skills",
+            )
+
+    def test_no_conversion(self):
+        with (
+            patch("sys.argv", ["dpdata", "skills", "list"]),
+            patch("dpdata.cli.convert") as convert,
+            contextlib.redirect_stdout(io.StringIO()),
+        ):
+            dpdata_cli()
+        convert.assert_not_called()
+
+    def test_unknown_is_usage_error(self):
+        with (
+            contextlib.redirect_stderr(io.StringIO()),
+            self.assertRaises(SystemExit) as error,
+        ):
+            skills_cli(["read", "missing"])
+        self.assertEqual(error.exception.code, 2)