feat(filesystem): add pifs workspace default

This commit is contained in:
BukeLy 2026-05-27 14:28:03 +08:00
parent 977f596666
commit 4ff184fed2
3 changed files with 124 additions and 7 deletions

View file

@ -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 <path>` 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 <path>` has been run")
return _run_passthrough(
command_tokens,
workspace=args.workspace,

6
pifs Executable file
View file

@ -0,0 +1,6 @@
#!/usr/bin/env python3
from pageindex.filesystem.cli import main
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -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