feat(cli): improve file discovery and completion
- Render remote files as an aligned table with JSON and TSV alternatives. - Add Bash and Zsh completion for commands, options, paths, and remote files. - Allow downloads by exact filename while rejecting ambiguous duplicate names. - Cover table formatting, completion scripts, and selector resolution with tests.
This commit is contained in:
@@ -9,8 +9,10 @@ from pathlib import Path
|
||||
from .app import make_client
|
||||
from .auth import login as auth_login
|
||||
from .client import download, list_remote, upload
|
||||
from .completion import script as completion_script
|
||||
from .config import DEFAULT_CONFIG
|
||||
from .sync import sync_once
|
||||
from .table import format_table
|
||||
|
||||
DEFAULT_NODE = 'picoshare'
|
||||
|
||||
@@ -42,21 +44,44 @@ def _cmd_list(args) -> int:
|
||||
return 1
|
||||
finally:
|
||||
client.close()
|
||||
if args.format == 'json':
|
||||
print(json.dumps(files, indent=2, ensure_ascii=False))
|
||||
elif args.format == 'tsv':
|
||||
for item in files:
|
||||
print(f"{item.get('file_id')}\t{item.get('file_name')}\t{item.get('file_size', '')}")
|
||||
print(f"{item.get('file_id', '')}\t{item.get('file_name', '')}\t{item.get('file_size', '')}")
|
||||
else:
|
||||
rows = [
|
||||
[item.get('file_id'), item.get('file_name'), item.get('file_size', '')]
|
||||
for item in files
|
||||
]
|
||||
print(format_table(['ID', 'Name', 'Size'], rows))
|
||||
return 0
|
||||
|
||||
def _resolve_remote_file(files: list[dict], selector: str) -> dict:
|
||||
id_matches = [item for item in files if item.get('file_id') == selector]
|
||||
if id_matches:
|
||||
return id_matches[0]
|
||||
name_matches = [item for item in files if item.get('file_name') == selector]
|
||||
if not name_matches:
|
||||
raise FileNotFoundError(f'no remote file matches: {selector}')
|
||||
if len(name_matches) > 1:
|
||||
raise RuntimeError(f'multiple remote files named {selector!r}; use the file ID')
|
||||
return name_matches[0]
|
||||
|
||||
def _cmd_download(args) -> int:
|
||||
dest = Path(args.output).expanduser() if args.output else Path(args.file_id)
|
||||
client = make_client(args)
|
||||
try:
|
||||
download(client, args.node, args.file_id, dest)
|
||||
remote = _resolve_remote_file(list_remote(client, args.node), args.file)
|
||||
file_id = str(remote['file_id'])
|
||||
file_name = str(remote.get('file_name') or file_id)
|
||||
dest = Path(args.output).expanduser() if args.output else Path(file_name)
|
||||
download(client, args.node, file_id, dest)
|
||||
except Exception as exc:
|
||||
print(f'error: {exc}', file=sys.stderr)
|
||||
return 1
|
||||
finally:
|
||||
client.close()
|
||||
print(f'downloaded {args.file_id} -> {dest}')
|
||||
print(f'downloaded {file_name} ({file_id}) -> {dest}')
|
||||
return 0
|
||||
|
||||
def _cmd_sync(args) -> int:
|
||||
@@ -74,6 +99,14 @@ def _cmd_watch(args) -> int:
|
||||
return code
|
||||
time.sleep(args.interval)
|
||||
|
||||
def _cmd_completion(args) -> int:
|
||||
try:
|
||||
print(completion_script(args.shell))
|
||||
except ValueError as exc:
|
||||
print(f'error: {exc}', file=sys.stderr)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
def _add_common(parser: argparse.ArgumentParser) -> None:
|
||||
parser.add_argument('--config', default=str(DEFAULT_CONFIG), help='Path to config.toml.')
|
||||
parser.add_argument('--url', action='append', metavar='NODE=URL', help='Override/add a node URL.')
|
||||
@@ -111,11 +144,12 @@ def main(argv: list[str] | None = None) -> int:
|
||||
|
||||
list_cmd = sub.add_parser('list', help='List remote files.')
|
||||
_add_common(list_cmd)
|
||||
list_cmd.add_argument('--format', choices=['table', 'json', 'tsv'], default='table')
|
||||
list_cmd.set_defaults(func=_cmd_list)
|
||||
|
||||
download_cmd = sub.add_parser('download', help='Download one file via private API endpoint.')
|
||||
_add_common(download_cmd)
|
||||
download_cmd.add_argument('file_id')
|
||||
download_cmd.add_argument('file', help='Remote file ID or exact file name.')
|
||||
download_cmd.add_argument('-o', '--output')
|
||||
download_cmd.set_defaults(func=_cmd_download)
|
||||
|
||||
@@ -136,6 +170,10 @@ def main(argv: list[str] | None = None) -> int:
|
||||
watch_cmd.add_argument('--interval', type=float, default=10.0)
|
||||
watch_cmd.set_defaults(func=_cmd_watch)
|
||||
|
||||
completion_cmd = sub.add_parser('completion', help='Print shell completion script.')
|
||||
completion_cmd.add_argument('shell', choices=['zsh', 'bash'])
|
||||
completion_cmd.set_defaults(func=_cmd_completion)
|
||||
|
||||
args = parser.parse_args(argv)
|
||||
return args.func(args)
|
||||
|
||||
|
||||
@@ -0,0 +1,107 @@
|
||||
from __future__ import annotations
|
||||
|
||||
_ZSH = r'''# nanoshare zsh completion -- add to ~/.zshrc: eval "$(nanoshare completion zsh)"
|
||||
_nanoshare_files() {
|
||||
local id name size
|
||||
while IFS=$'\t' read -r id name size; do
|
||||
[[ -n $id ]] && printf '%s:%s\n' "$id" "$name"
|
||||
done < <(nanoshare list --format tsv 2>/dev/null)
|
||||
}
|
||||
|
||||
_nanoshare() {
|
||||
local -a commands common_opts login_opts sync_opts watch_opts download_opts
|
||||
commands=(login upload list download sync watch completion)
|
||||
common_opts=(--config --url --token --refresh-token --token-url --source --timeout --node)
|
||||
login_opts=(--config --node --source --scope --callback-host --callback-port --login-timeout --timeout --no-browser)
|
||||
sync_opts=($common_opts --note --expires --delete)
|
||||
watch_opts=($sync_opts --interval)
|
||||
download_opts=($common_opts --output -o)
|
||||
|
||||
if (( CURRENT == 2 )); then
|
||||
compadd -- $commands
|
||||
return
|
||||
fi
|
||||
|
||||
case ${words[2]} in
|
||||
login)
|
||||
compadd -- $login_opts
|
||||
;;
|
||||
upload)
|
||||
compadd -- $common_opts --name --note --expires
|
||||
_files
|
||||
;;
|
||||
list)
|
||||
compadd -- $common_opts --format
|
||||
;;
|
||||
download)
|
||||
if (( CURRENT == 3 )); then
|
||||
local -a remote_files
|
||||
remote_files=("${(@f)$(_nanoshare_files)}")
|
||||
_describe 'remote file' remote_files
|
||||
else
|
||||
compadd -- $download_opts
|
||||
_files
|
||||
fi
|
||||
;;
|
||||
sync|watch)
|
||||
compadd -- ${(P)${:-${words[2]}_opts}}
|
||||
_files -/
|
||||
;;
|
||||
completion)
|
||||
(( CURRENT == 3 )) && compadd -- zsh bash
|
||||
;;
|
||||
esac
|
||||
}
|
||||
compdef _nanoshare nanoshare
|
||||
'''
|
||||
|
||||
_BASH = r'''# nanoshare bash completion -- add to ~/.bashrc: eval "$(nanoshare completion bash)"
|
||||
_nanoshare_files() {
|
||||
nanoshare list --format tsv 2>/dev/null
|
||||
}
|
||||
|
||||
_nanoshare() {
|
||||
local cur="${COMP_WORDS[COMP_CWORD]}" cmd="${COMP_WORDS[1]}"
|
||||
local commands="login upload list download sync watch completion"
|
||||
local common_opts="--config --url --token --refresh-token --token-url --source --timeout --node"
|
||||
local login_opts="--config --node --source --scope --callback-host --callback-port --login-timeout --timeout --no-browser"
|
||||
local sync_opts="$common_opts --note --expires --delete"
|
||||
local watch_opts="$sync_opts --interval"
|
||||
local download_opts="$common_opts --output -o"
|
||||
|
||||
if [ "$COMP_CWORD" -eq 1 ]; then
|
||||
COMPREPLY=( $(compgen -W "$commands" -- "$cur") )
|
||||
return
|
||||
fi
|
||||
|
||||
case "$cmd" in
|
||||
login)
|
||||
COMPREPLY=( $(compgen -W "$login_opts" -- "$cur") ) ;;
|
||||
upload)
|
||||
COMPREPLY=( $(compgen -W "$common_opts --name --note --expires" -- "$cur") ) ;;
|
||||
list)
|
||||
COMPREPLY=( $(compgen -W "$common_opts --format" -- "$cur") ) ;;
|
||||
download)
|
||||
if [ "$COMP_CWORD" -eq 2 ]; then
|
||||
COMPREPLY=( $(compgen -W "$(_nanoshare_files | cut -f1)" -- "$cur") )
|
||||
else
|
||||
COMPREPLY=( $(compgen -W "$download_opts" -- "$cur") )
|
||||
fi ;;
|
||||
sync)
|
||||
COMPREPLY=( $(compgen -W "$sync_opts" -- "$cur") ) ;;
|
||||
watch)
|
||||
COMPREPLY=( $(compgen -W "$watch_opts" -- "$cur") ) ;;
|
||||
completion)
|
||||
[ "$COMP_CWORD" -eq 2 ] && COMPREPLY=( $(compgen -W "zsh bash" -- "$cur") ) ;;
|
||||
esac
|
||||
}
|
||||
complete -F _nanoshare nanoshare
|
||||
'''
|
||||
|
||||
|
||||
def script(shell: str) -> str:
|
||||
if shell == 'zsh':
|
||||
return _ZSH
|
||||
if shell == 'bash':
|
||||
return _BASH
|
||||
raise ValueError(f'unsupported shell: {shell}')
|
||||
@@ -0,0 +1,26 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Iterable
|
||||
|
||||
def _display_width(value: str) -> int:
|
||||
return len(value)
|
||||
|
||||
def _cell(value: object) -> str:
|
||||
return '' if value is None else str(value)
|
||||
|
||||
def format_table(headers: list[str], rows: Iterable[Iterable[object]]) -> str:
|
||||
string_rows = [[_cell(value) for value in row] for row in rows]
|
||||
widths = [_display_width(header) for header in headers]
|
||||
for row in string_rows:
|
||||
for index, value in enumerate(row):
|
||||
if index < len(widths):
|
||||
widths[index] = max(widths[index], _display_width(value))
|
||||
|
||||
def render_row(values: list[str]) -> str:
|
||||
padded = [value.ljust(widths[index]) for index, value in enumerate(values)]
|
||||
return ' '.join(padded).rstrip()
|
||||
|
||||
separator = ' '.join('-' * width for width in widths).rstrip()
|
||||
lines = [render_row(headers), separator]
|
||||
lines.extend(render_row(row) for row in string_rows)
|
||||
return '\n'.join(lines)
|
||||
@@ -8,6 +8,8 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'cli'))
|
||||
|
||||
from nanoshare_client import cli
|
||||
from nanoshare_client.auth import _CallbackServer
|
||||
from nanoshare_client.completion import script as completion_script
|
||||
from nanoshare_client.table import format_table
|
||||
|
||||
class FakeClient:
|
||||
def __init__(self):
|
||||
@@ -55,6 +57,53 @@ def test_sync_uploads_new_local_file_and_writes_state(tmp_path, monkeypatch):
|
||||
state = json.loads((tmp_path / '.nanoshare-sync' / 'state.json').read_text())
|
||||
assert state['files']['hello.txt']['file_id'] == 'file_1'
|
||||
|
||||
def test_completion_scripts_include_commands():
|
||||
zsh = completion_script('zsh')
|
||||
bash = completion_script('bash')
|
||||
|
||||
assert 'compdef _nanoshare nanoshare' in zsh
|
||||
assert 'complete -F _nanoshare nanoshare' in bash
|
||||
assert 'login upload list download sync watch completion' in zsh
|
||||
assert 'login upload list download sync watch completion' in bash
|
||||
|
||||
def test_resolve_remote_file_accepts_id_or_exact_name():
|
||||
files = [
|
||||
{'file_id': 'file_1', 'file_name': 'report.pdf'},
|
||||
{'file_id': 'file_2', 'file_name': 'photo.png'},
|
||||
]
|
||||
|
||||
assert cli._resolve_remote_file(files, 'file_1')['file_name'] == 'report.pdf'
|
||||
assert cli._resolve_remote_file(files, 'photo.png')['file_id'] == 'file_2'
|
||||
|
||||
def test_resolve_remote_file_rejects_duplicate_names():
|
||||
files = [
|
||||
{'file_id': 'file_1', 'file_name': 'report.pdf'},
|
||||
{'file_id': 'file_2', 'file_name': 'report.pdf'},
|
||||
]
|
||||
|
||||
try:
|
||||
cli._resolve_remote_file(files, 'report.pdf')
|
||||
except RuntimeError as exc:
|
||||
assert 'multiple remote files' in str(exc)
|
||||
else:
|
||||
raise AssertionError('expected duplicate names to raise')
|
||||
|
||||
def test_format_table_aligns_columns():
|
||||
output = format_table(
|
||||
['ID', 'Name', 'Size'],
|
||||
[
|
||||
['abc', 'short.txt', '1 KB'],
|
||||
['longer-id', 'a much longer file name.png', '22 MB'],
|
||||
],
|
||||
)
|
||||
|
||||
assert output.splitlines() == [
|
||||
'ID Name Size',
|
||||
'--------- --------------------------- -----',
|
||||
'abc short.txt 1 KB',
|
||||
'longer-id a much longer file name.png 22 MB',
|
||||
]
|
||||
|
||||
def test_login_callback_returns_without_waiting_for_timeout():
|
||||
server = _CallbackServer('127.0.0.1', 0, 'expected-state')
|
||||
result = {}
|
||||
|
||||
Reference in New Issue
Block a user