daniel156161
2b0215cd94
Build and Push Docker Container / build-and-push (push) Successful in 1m52s
- Leave sync and watch notes empty by default instead of writing placeholder text. - Preserve existing remote notes when moving or adopting files without --note. - Detect tracked local moves by SHA-256 and update remote metadata safely. - Skip files that vanish during watch scans instead of aborting the sync cycle. - Bump NanoShare to 1.24.0 and the standalone CLI to 0.3.0.
191 lines
7.4 KiB
Python
191 lines
7.4 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import time
|
|
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 .remote_path import join_remote_path
|
|
from .sync import sync_once
|
|
from .table import format_table
|
|
|
|
DEFAULT_NODE = 'picoshare'
|
|
|
|
def _cmd_login(args) -> int:
|
|
return auth_login(args)
|
|
|
|
def _cmd_upload(args) -> int:
|
|
path = Path(args.file).expanduser().resolve()
|
|
if not path.is_file():
|
|
print(f'not a file: {path}', file=sys.stderr)
|
|
return 1
|
|
client = make_client(args)
|
|
try:
|
|
result = upload(client, args.node, path, args.name or path.name, args.note or '', args.expires or '')
|
|
except Exception as exc:
|
|
print(f'error: {exc}', file=sys.stderr)
|
|
return 1
|
|
finally:
|
|
client.close()
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|
return 0
|
|
|
|
def _cmd_list(args) -> int:
|
|
client = make_client(args)
|
|
try:
|
|
files = list_remote(client, args.node)
|
|
except Exception as exc:
|
|
print(f'error: {exc}', file=sys.stderr)
|
|
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_path', '')}\t{item.get('file_name', '')}\t{item.get('file_size', '')}")
|
|
else:
|
|
rows = [
|
|
[item.get('file_id'), item.get('file_path', ''), item.get('file_name'), item.get('file_size', '')]
|
|
for item in files
|
|
]
|
|
print(format_table(['ID', 'Path', '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]
|
|
full_path_matches = [item for item in files if join_remote_path(str(item.get('file_name') or ''), str(item.get('file_path') or ''), str(item.get('file_id') or '')) == selector]
|
|
if full_path_matches:
|
|
if len(full_path_matches) > 1:
|
|
raise RuntimeError(f'multiple remote files at {selector!r}; use the file ID')
|
|
return full_path_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 or full path')
|
|
return name_matches[0]
|
|
|
|
def _cmd_download(args) -> int:
|
|
client = make_client(args)
|
|
try:
|
|
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)
|
|
remote_name = join_remote_path(file_name, str(remote.get('file_path') 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 {remote_name} ({file_id}) -> {dest}')
|
|
return 0
|
|
|
|
def _cmd_sync(args) -> int:
|
|
client = make_client(args)
|
|
try:
|
|
return sync_once(args, client)
|
|
finally:
|
|
client.close()
|
|
|
|
def _cmd_watch(args) -> int:
|
|
print(f'watching {args.folder} every {args.interval}s')
|
|
while True:
|
|
code = _cmd_sync(args)
|
|
if code:
|
|
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.')
|
|
parser.add_argument('--token', help='Static bearer token.')
|
|
parser.add_argument('--refresh-token', dest='refresh_token', help='Refresh token to auto-exchange for access tokens.')
|
|
parser.add_argument('--token-url', dest='token_url', help='Token-refresh endpoint.')
|
|
parser.add_argument('--source', default='nanoshare-cli', help='Source node name.')
|
|
parser.add_argument('--timeout', type=float, default=30.0)
|
|
parser.add_argument('--node', default=DEFAULT_NODE, help='NanoShare node name.')
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(prog='nanoshare', description='NanoShare CLI and folder sync client.')
|
|
sub = parser.add_subparsers(dest='command', required=True)
|
|
|
|
login = sub.add_parser('login', help='Login via browser and write CLI config.')
|
|
login.add_argument('base_url', help='NanoShare base URL, e.g. https://nanoshare.example')
|
|
login.add_argument('--config', type=Path, default=DEFAULT_CONFIG)
|
|
login.add_argument('--node', default=DEFAULT_NODE)
|
|
login.add_argument('--source', default='nanoshare-cli')
|
|
login.add_argument('--scope', default='files')
|
|
login.add_argument('--callback-host', default='127.0.0.1')
|
|
login.add_argument('--callback-port', type=int, default=0)
|
|
login.add_argument('--login-timeout', type=float, default=300.0)
|
|
login.add_argument('--timeout', type=float, default=30.0)
|
|
login.add_argument('--no-browser', action='store_true')
|
|
login.set_defaults(func=_cmd_login)
|
|
|
|
upload_cmd = sub.add_parser('upload', help='Upload one file.')
|
|
_add_common(upload_cmd)
|
|
upload_cmd.add_argument('file')
|
|
upload_cmd.add_argument('--name', help='Remote file name.')
|
|
upload_cmd.add_argument('--note', default='uploaded from nanoshare cli')
|
|
upload_cmd.add_argument('--expires', default='')
|
|
upload_cmd.set_defaults(func=_cmd_upload)
|
|
|
|
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', help='Remote file ID or exact file name.')
|
|
download_cmd.add_argument('-o', '--output')
|
|
download_cmd.set_defaults(func=_cmd_download)
|
|
|
|
sync_cmd = sub.add_parser('sync', help='Two-way folder sync MVP.')
|
|
_add_common(sync_cmd)
|
|
sync_cmd.add_argument('folder')
|
|
sync_cmd.add_argument('--note', default='')
|
|
sync_cmd.add_argument('--expires', default='')
|
|
sync_cmd.add_argument('--delete', action='store_true', help='Delete remote files that were deleted locally.')
|
|
sync_cmd.add_argument('--ignore', action='append', default=[], help='Ignore glob pattern for sync; repeatable.')
|
|
sync_cmd.set_defaults(func=_cmd_sync)
|
|
|
|
watch_cmd = sub.add_parser('watch', help='Run sync repeatedly.')
|
|
_add_common(watch_cmd)
|
|
watch_cmd.add_argument('folder')
|
|
watch_cmd.add_argument('--note', default='')
|
|
watch_cmd.add_argument('--expires', default='')
|
|
watch_cmd.add_argument('--delete', action='store_true')
|
|
watch_cmd.add_argument('--ignore', action='append', default=[], help='Ignore glob pattern for sync; repeatable.')
|
|
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)
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|