import json from argparse import Namespace from typing import Any, Final from pydantic.dataclasses import dataclass from mdrsclient.api import FoldersApi from mdrsclient.client import MdrsClient from mdrsclient.commands.base import BaseCommand from mdrsclient.config import build_download_url from mdrsclient.exceptions import UnauthorizedException from mdrsclient.models import File, Folder, FolderSimple, Laboratory class Config: arbitrary_types_allowed = True frozen = True @dataclass(config=Config) class LsCommandContext: prefix: str client: MdrsClient laboratory: Laboratory password: str is_json: bool is_quiet: bool is_recursive: bool class LsCommand(BaseCommand): @classmethod def register(cls, parsers: Any) -> None: ls_parser = parsers.add_parser("ls", help="list the folder contents") ls_parser.add_argument("-p", "--password", help="password to use when open locked folder") ls_parser.add_argument("-J", "--json", help="turn on json output", action="store_true") ls_parser.add_argument( "-q", "--quiet", help="don't output header row. this option is forced if the -r option is specified", action="store_true", ) ls_parser.add_argument("-r", "--recursive", help="list the folder contents recursive", action="store_true") ls_parser.add_argument("remote_path", help="remote folder path (remote:/lab/path/)") ls_parser.set_defaults(func=cls.func) @classmethod def func(cls, args: Namespace) -> None: remote_path = str(args.remote_path) password = str(args.password) if args.password else None is_json = bool(args.json) is_recursive = bool(args.recursive) is_quiet = bool(args.quiet) if not is_recursive else True cls.ls(remote_path, password, is_json, is_recursive, is_quiet) @classmethod def ls(cls, remote_path: str, password: str | None, is_json: bool, is_recursive: bool, is_quiet: bool) -> None: remote = remote_path.split(":", 1)[0] if ":" in remote_path else "" from mdrsclient.client import MdrsClient client = MdrsClient.from_remote(remote) cls._ls_logic(client, remote_path, password, is_json, is_recursive, is_quiet) return @classmethod def _ls_logic( cls, client: MdrsClient, remote_path: str, password: str | None, is_json: bool, is_recursive: bool, is_quiet: bool, ) -> None: remote = remote_path.split(":", 1)[0] if ":" in remote_path else "" folder, laboratory = client.resolve_folder(remote_path, password) laboratory_name = laboratory.name files = client.find_files(folder.id) context = LsCommandContext( f"{remote}:/{laboratory_name}", client, laboratory, password if password is not None else "", is_json, is_quiet, is_recursive, ) if context.is_json: cls._ls_json(context, folder, files) else: cls._ls_plain(context, folder, files) @classmethod def _ls_json(cls, context: LsCommandContext, folder: Folder, files: list[File]) -> None: print(json.dumps(cls._folder2dict(context, folder, files), ensure_ascii=False)) LABELS: Final[dict[str, str]] = { "type": "Type", "acl": "Access", "laboratory": "Laboratory", "size": "Size", "date": "Date", "name": "Name", } @classmethod def _column_widths(cls, context: LsCommandContext, folder: Folder, files: list[File]) -> dict[str, int]: """Width of each column: the widest of its heading and everything printed under it.""" length = {key: len(label) if not context.is_quiet else 0 for key, label in cls.LABELS.items()} for sub_folder in folder.sub_folders: sub_laboratory_name = cls._laboratory_name(context, sub_folder.laboratory_id) length["acl"] = max(length["acl"], len(sub_folder.access_level_name)) length["laboratory"] = max(length["laboratory"], len(sub_laboratory_name)) length["size"] = max(length["size"], len(str(sub_folder.size))) length["date"] = max(length["date"], len(sub_folder.updated_at_name)) length["name"] = max(length["name"], len(sub_folder.name)) for file in files: length["size"] = max(length["size"], len(str(file.size))) length["date"] = max(length["date"], len(file.updated_at_name)) length["name"] = max(length["name"], len(file.name)) length["acl"] = max(length["acl"], len(folder.access_level_name)) length["laboratory"] = max(length["laboratory"], len(context.laboratory.name)) return length @classmethod def _ls_plain_children(cls, context: LsCommandContext, folder: Folder) -> None: """List each sub-folder in turn, passing over the ones the caller cannot open.""" print("") folder_api = FoldersApi(context.client.connection) for sub_folder in sorted(folder.sub_folders, key=lambda x: x.name): try: if sub_folder.lock: folder_api.auth(sub_folder.id, context.password) sub_detail = folder_api.retrieve(sub_folder.id) sub_files = context.client.find_files(sub_folder.id) except UnauthorizedException: continue cls._ls_plain(context, sub_detail, sub_files) @classmethod def _ls_plain(cls, context: LsCommandContext, folder: Folder, files: list[File]) -> None: label = cls.LABELS length = cls._column_widths(context, folder, files) header = ( f"{label['type']:{length['type']}}\t{label['acl']:{length['acl']}}\t" f"{label['laboratory']:{length['laboratory']}}\t{label['size']:{length['size']}}\t" f"{label['date']:{length['date']}}\t{label['name']:{length['name']}}" ) if context.is_recursive: print(f"{context.prefix}{folder.path}:") print(f"total {sum(f.size for f in files)}") if not context.is_quiet: print(header) print("-" * len(header.expandtabs())) for sub_folder in sorted(folder.sub_folders, key=lambda x: x.name): sub_laboratory_name = cls._laboratory_name(context, sub_folder.laboratory_id) sub_folder_type = "[d]" if sub_folder.lock is False else "[l]" print( f"{sub_folder_type:{length['type']}}\t{sub_folder.access_level_name:{length['acl']}}\t" f"{sub_laboratory_name:{length['laboratory']}}\t{sub_folder.size:{length['size']}}\t" f"{sub_folder.updated_at_name:{length['date']}}\t{sub_folder.name:{length['name']}}" ) for file in sorted(files, key=lambda x: x.name): print( f"{'[f]':{length['type']}}\t{folder.access_level_name:{length['acl']}}\t" f"{context.laboratory.name:{length['laboratory']}}\t{file.size:{length['size']}}\t" f"{file.updated_at_name:{length['date']}}\t{file.name:{length['name']}}" ) if context.is_recursive: cls._ls_plain_children(context, folder) @classmethod def _folder2dict( cls, context: LsCommandContext, folder: Folder | FolderSimple, files: list[File] ) -> dict[str, Any]: data: dict[str, Any] = { "id": folder.id, "pid": folder.pid, "name": folder.name, "size": folder.size, "access_level": folder.access_level_name, "lock": folder.lock, "laboratory": cls._laboratory_name(context, folder.laboratory_id), "description": folder.description, "created_at": folder.created_at, "updated_at": folder.updated_at, } if isinstance(folder, Folder): folder_api = FoldersApi(context.client.connection) data["metadata"] = folder_api.metadata(folder.id) if context.is_recursive: sub_folders: list[dict[str, Any]] = [] for sub_folder in sorted(folder.sub_folders, key=lambda x: x.name): try: if sub_folder.lock: folder_api.auth(sub_folder.id, context.password) folder2 = folder_api.retrieve(sub_folder.id) files2 = context.client.find_files(sub_folder.id) sub_folders.append(cls._folder2dict(context, folder2, files2)) except UnauthorizedException: pass data["sub_folders"] = sub_folders else: data["sub_folders"] = list( map(lambda x: cls._folder2dict(context, x, []), sorted(folder.sub_folders, key=lambda x: x.name)) ) data["files"] = list(map(lambda x: cls._file2dict(context, x), sorted(files, key=lambda x: x.name))) return data @classmethod def _file2dict(cls, context: LsCommandContext, file: File) -> dict[str, Any]: data: dict[str, Any] = { "id": file.id, "name": file.name, "type": file.type, "size": file.size, # "thumbnail": file.thumbnail, "description": file.description, "metadata": file.metadata, "download_url": build_download_url(context.client.connection.url, file.download_url), "created_at": file.created_at, "updated_at": file.updated_at, } return data @classmethod def _laboratory_name(cls, context: LsCommandContext, laboratory_id: int) -> str: laboratory = context.client.connection.laboratories.find_by_id(laboratory_id) return laboratory.name if laboratory is not None else "(invalid)"