From 4ff184fed28da4358a4961e45f3d12987157130e Mon Sep 17 00:00:00 2001 From: BukeLy Date: Wed, 27 May 2026 14:28:03 +0800 Subject: [PATCH] feat(filesystem): add pifs workspace default --- pageindex/filesystem/cli.py | 74 +++++++++++++++++++++++++++++++++---- pifs | 6 +++ tests/test_pifs_cli.py | 51 +++++++++++++++++++++++++ 3 files changed, 124 insertions(+), 7 deletions(-) create mode 100755 pifs diff --git a/pageindex/filesystem/cli.py b/pageindex/filesystem/cli.py index 8af12e6..e808d32 100644 --- a/pageindex/filesystem/cli.py +++ b/pageindex/filesystem/cli.py @@ -2,6 +2,7 @@ from __future__ import annotations import argparse import contextlib +import json import os import re import shlex @@ -23,6 +24,45 @@ AGENT_STREAM_MODE_CHOICES = ("off", "tools", "model", "all") DEFAULT_AGENT_MODEL = "gpt-5.4-mini" EXIT_COMMANDS = {"exit", "quit", ":q"} ANSI_ESCAPE_RE = re.compile(r"\x1b(?:\[[0-?]*[ -/]*[@-~]|.)") +PIFS_CONFIG_FILE_ENV = "PIFS_CONFIG_FILE" +PIFS_WORKSPACE_ENV = "PIFS_WORKSPACE" + + +def _config_path() -> Path: + override = os.environ.get(PIFS_CONFIG_FILE_ENV) + if override: + return Path(override).expanduser() + config_home = os.environ.get("XDG_CONFIG_HOME") + root = Path(config_home).expanduser() if config_home else Path.home() / ".config" + return root / "pageindex" / "pifs.json" + + +def _read_config() -> dict[str, str]: + path = _config_path() + if not path.exists(): + return {} + with path.open("r", encoding="utf-8") as handle: + payload = json.load(handle) + if not isinstance(payload, dict): + raise ValueError(f"invalid PIFS config file: {path}") + return {str(key): str(value) for key, value in payload.items() if value is not None} + + +def _write_config(config: dict[str, str]) -> Path: + path = _config_path() + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", encoding="utf-8") as handle: + json.dump(config, handle, indent=2, sort_keys=True) + handle.write("\n") + return path + + +def _configured_workspace() -> str | None: + return _read_config().get("workspace") + + +def _resolve_workspace(value: str | None) -> str | None: + return value or os.environ.get(PIFS_WORKSPACE_ENV) or _configured_workspace() def _load_env_file(path: str | None = None, *, workspace: str | None = None) -> Path | None: @@ -114,10 +154,9 @@ def _parse_agent_command( parser.add_argument("question", nargs=argparse.REMAINDER) args = parser.parse_args(argv) _load_env_file(args.env_file, workspace=args.workspace) + args.workspace = _resolve_workspace(args.workspace) if not args.workspace: - args.workspace = os.environ.get("PIFS_WORKSPACE") - if not args.workspace: - parser.error("--workspace is required unless PIFS_WORKSPACE is set") + parser.error("--workspace is required unless PIFS_WORKSPACE is set or `pifs set workspace ` has been run") return args @@ -241,18 +280,37 @@ def _run_passthrough( return 0 +def _run_set(argv: list[str]) -> int: + parser = argparse.ArgumentParser( + prog="pifs set", + description="Set PageIndex FileSystem CLI defaults", + ) + parser.add_argument("name", choices=["workspace"]) + parser.add_argument("value") + args = parser.parse_args(argv) + + config = _read_config() + if args.name == "workspace": + workspace = Path(args.value).expanduser().resolve(strict=False) + config["workspace"] = str(workspace) + path = _write_config(config) + print(f"workspace: {workspace}") + print(f"config: {path}") + return 0 + raise ValueError(f"unknown config key: {args.name}") + + def main(argv: list[str] | None = None) -> int: argv = list(sys.argv[1:] if argv is None else argv) _load_env_file() parser = argparse.ArgumentParser(description="PageIndex FileSystem CLI") - parser.add_argument("--workspace", default=os.environ.get("PIFS_WORKSPACE")) + parser.add_argument("--workspace", default=None) parser.add_argument("--env-file", default=None) parser.add_argument("--json", action="store_true", dest="json_output") parser.add_argument("command", nargs=argparse.REMAINDER) args = parser.parse_args(argv) _load_env_file(args.env_file, workspace=args.workspace) - if not args.workspace: - args.workspace = os.environ.get("PIFS_WORKSPACE") + args.workspace = _resolve_workspace(args.workspace) command_tokens = [token for token in args.command if token != "--"] json_output = args.json_output @@ -263,6 +321,8 @@ def main(argv: list[str] | None = None) -> int: try: command_name = command_tokens[0] command_args = command_tokens[1:] + if command_name == "set": + return _run_set(command_args) if command_name == "ask": return _run_ask(command_args, workspace_default=args.workspace) if command_name == "chat": @@ -272,7 +332,7 @@ def main(argv: list[str] | None = None) -> int: command_tokens = [token for token in command_tokens if token != "--json"] json_output = True if not args.workspace: - parser.error("--workspace is required unless PIFS_WORKSPACE is set") + parser.error("--workspace is required unless PIFS_WORKSPACE is set or `pifs set workspace ` has been run") return _run_passthrough( command_tokens, workspace=args.workspace, diff --git a/pifs b/pifs new file mode 100755 index 0000000..46b6e0e --- /dev/null +++ b/pifs @@ -0,0 +1,6 @@ +#!/usr/bin/env python3 +from pageindex.filesystem.cli import main + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_pifs_cli.py b/tests/test_pifs_cli.py index 74832b8..491cbb9 100644 --- a/tests/test_pifs_cli.py +++ b/tests/test_pifs_cli.py @@ -55,6 +55,57 @@ def test_cli_passthrough_invokes_pifs_command_executor(monkeypatch, capsys, tmp_ assert executor_instances[0].commands == ["ls /documents"] +def test_cli_set_workspace_persists_default(monkeypatch, capsys, tmp_path): + from pageindex.filesystem import cli + + config_path = tmp_path / "pifs.json" + workspace = tmp_path / "workspace" + monkeypatch.setenv("PIFS_CONFIG_FILE", str(config_path)) + + status = cli.main(["set", "workspace", str(workspace)]) + + assert status == 0 + output = capsys.readouterr().out + assert f"workspace: {workspace}" in output + assert f"config: {config_path}" in output + assert config_path.read_text(encoding="utf-8") == ( + '{\n "workspace": "' + str(workspace) + '"\n}\n' + ) + + +def test_cli_passthrough_uses_configured_workspace(monkeypatch, capsys, tmp_path): + from pageindex.filesystem import cli + + config_path = tmp_path / "pifs.json" + workspace = tmp_path / "workspace" + executor_instances = [] + monkeypatch.setenv("PIFS_CONFIG_FILE", str(config_path)) + monkeypatch.delenv("PIFS_WORKSPACE", raising=False) + + class FakeExecutor: + def __init__(self, filesystem, *, json_output=False): + self.filesystem = filesystem + self.json_output = json_output + self.commands = [] + executor_instances.append(self) + + def execute(self, command): + self.commands.append(command) + return f"executed:{command}" + + monkeypatch.setattr(cli, "PageIndexFileSystem", FakeFileSystem) + monkeypatch.setattr(cli, "PIFSCommandExecutor", FakeExecutor) + + assert cli.main(["set", "workspace", str(workspace)]) == 0 + capsys.readouterr() + + status = cli.main(["ls", "/documents"]) + + assert status == 0 + assert capsys.readouterr().out == "executed:ls /documents\n" + assert executor_instances[0].filesystem.workspace == workspace + + def test_cli_ask_invokes_agent_with_question(monkeypatch, capsys, tmp_path): from pageindex.filesystem import cli