Merge pull request #14 from MAK-Relic-Tool/bugfix-and-cli-help

Bugfix and new `info` cli

Author: ModernMAK

setup.cfg

@@ -39,6 +39,7 @@ relic.cli.sga =
     unpack = relic.sga.core.cli:RelicSgaUnpackCli
     pack = relic.sga.core.cli:RelicSgaPackCli
     repack = relic.sga.core.cli:RelicSgaRepackCli
+    info = relic.sga.core.cli:RelicSgaInfoCli
 
 [options.packages.find]
 where = src

src/relic/sga/core/__init__.py

@@ -3,4 +3,4 @@
 """
 from relic.sga.core.definitions import Version, MagicWord, StorageType, VerificationType
 
-__version__ = "1.1.3"
+__version__ = "1.1.4"

src/relic/sga/core/cli.py

@@ -1,14 +1,19 @@
 from __future__ import annotations
 
 import argparse
+import datetime
 import os.path
 from argparse import ArgumentParser, Namespace
-from typing import Optional, Callable
+from typing import Optional, Callable, Dict, List, Any, Tuple, Set, TextIO
 
 import fs.copy
 from fs.base import FS
+from fs.multifs import MultiFS
 from relic.core.cli import CliPluginGroup, _SubParsersAction, CliPlugin
 
+from relic.sga.core.definitions import StorageType
+from relic.sga.core.filesystem import EssenceFS, _EssenceDriveFS
+
 
 class RelicSgaCli(CliPluginGroup):
     GROUP = "relic.cli.sga"
@@ -130,3 +135,144 @@ def _create_parser(
         # pack further delegates to version plugins
 
         return parser
+
+
+class RelicSgaInfoCli(CliPlugin):
+    def _create_parser(
+        self, command_group: Optional[_SubParsersAction] = None
+    ) -> ArgumentParser:
+        parser: ArgumentParser
+        description = "Dumps metadata packed into an SGA file."
+        if command_group is None:
+            parser = ArgumentParser("info", description=description)
+        else:
+            parser = command_group.add_parser("info", description=description)
+
+        parser.add_argument(
+            "sga",
+            type=_get_file_type_validator(exists=True),
+            help="SGA File to inspect",
+        )
+        parser.add_argument(
+            "log_file",
+            nargs="?",
+            type=_get_file_type_validator(exists=False),
+            help="Optional file to write messages to, required if `-q/--quiet` is used",
+            default=None,
+        )
+        parser.add_argument(
+            "-q",
+            "--quiet",
+            action="store_true",
+            default=False,
+            help="When specified, SGA info is not printed to the console",
+        )
+        return parser
+
+    def command(self, ns: Namespace) -> Optional[int]:
+        sga: str = ns.sga
+        log_file: str = ns.log_file
+        quiet: bool = ns.quiet
+
+        logger: Optional[TextIO] = None
+        try:
+            if log_file is not None:
+                logger = open(log_file, "w")
+
+            outputs: List[Optional[TextIO]] = []
+            if quiet is False:
+                outputs.append(None)  # None is a sentinel for stdout
+            if logger is not None:
+                outputs.append(logger)
+
+            if len(outputs) == 0:
+                print(
+                    "Please specify a `log_file` if using the `-q` or `--quiet` command"
+                )
+                return 1
+
+            def _print(
+                *msg: str, sep: Optional[str] = None, end: Optional[str] = None
+            ) -> None:
+                for output in outputs:
+                    print(*msg, sep=sep, end=end, file=output)
+
+            def _is_container(d: Any) -> bool:
+                return isinstance(d, (Dict, List, Tuple, Set))  # type: ignore
+
+            def _stringify(d: Any, indent: int = 0) -> None:
+                _TAB = "\t"
+                if isinstance(d, Dict):
+                    for k, v in d.items():
+                        if _is_container(v):
+                            _print(f"{_TAB * indent}{k}:")
+                            _stringify(v, indent + 1)
+                        else:
+                            _print(f"{_TAB * indent}{k}: {v}")
+                elif isinstance(d, (List, Tuple, Set)):  # type: ignore
+                    _print(f"{_TAB * indent}{', '.join(*d)}")
+                else:
+                    _print(f"{_TAB * indent}{d}")
+
+            def _getessence(fs: FS, path: str = "/") -> Dict[str, Any]:
+                return fs.getinfo(path, "essence").raw.get("essence", {})  # type: ignore
+
+            _print(f"File: `{sga}`")
+            sgafs: EssenceFS
+            with fs.open_fs(f"sga://{sga}") as sgafs:  # type: ignore
+                _print("Archive Metadata:")
+                _stringify(sgafs.getmeta("essence"), indent=1)
+
+                drive: _EssenceDriveFS
+                for alias, drive in sgafs.iterate_fs():  # type: ignore
+                    _print(f"Drive: `{drive.name}` (`{drive.alias}`)")
+                    _print("\tDrive Metadata:")
+                    info = _getessence(drive)
+                    if len(info) > 0:
+                        _stringify(info, indent=2)
+                    else:
+                        _print(f"\t\tNo Metadata")
+
+                    _print("\tDrive Files Metadata:")
+                    for f in drive.walk.files():
+                        _print(f"\t\t`{f}`:")
+                        finfo: Dict[str, Any] = _getessence(drive, f)
+                        finfo = finfo.copy()
+                        # We alter storage_type cause it *should* always be present, if its not, we dont do anything
+                        key = "storage_type"
+                        if key in finfo:
+                            stv: int = finfo[key]
+                            st: StorageType = StorageType(stv)
+                            finfo[key] = f"{stv} ({st.name})"
+
+                        # We alter modified too, cause when it is present, its garbage
+                        key = "modified"
+                        if key in finfo:
+                            mtv: int = finfo[key]
+                            mt = datetime.datetime.fromtimestamp(
+                                mtv, datetime.timezone.utc
+                            )
+                            finfo[key] = str(mt)
+
+                        # And CRC32 if it's in bytes; this should be removed ASAP tho # I only put this in because its such a minor patch to V2
+                        key = "crc32"
+                        if key in finfo:
+                            crcv: bytes = finfo[key]
+                            if isinstance(crcv, bytes):
+                                crc32 = int.from_bytes(crcv, "little", signed=False)
+                                finfo[key] = crc32
+
+                        if len(finfo) > 0:
+                            _stringify(finfo, indent=3)
+                        else:
+                            _print(f"\t\t\tNo Metadata")
+
+        finally:
+            if logger is not None:
+                logger.close()
+
+        if log_file is not None:
+            print(
+                f"Saved to `{os.path.join(os.getcwd(), log_file)}`"
+            )  # DO NOT USE _PRINT
+        return None

src/relic/sga/core/filesystem.py

@@ -228,11 +228,7 @@ def __init__(self, resource_type: ResourceType, name: Text):
     def to_info(self, namespaces=None):
         # type: (Optional[Collection[Text]]) -> Info
         info = super().to_info(namespaces)
-        if (
-            namespaces is not None
-            and not self.is_dir
-            and ESSENCE_NAMESPACE in namespaces
-        ):
+        if namespaces is not None and ESSENCE_NAMESPACE in namespaces:
             info_dict = dict(info.raw)
             info_dict[ESSENCE_NAMESPACE] = self.essence.copy()
             info = Info(info_dict)
@@ -302,8 +298,8 @@ def getinfo(
         if _path == "/" and (
             namespaces is not None and ESSENCE_NAMESPACE in namespaces
         ):
-            raw_info = info.raw
-            essence_ns = dict(raw_info[ESSENCE_NAMESPACE])
+            raw_info = dict(info.raw)
+            essence_ns = raw_info[ESSENCE_NAMESPACE] = {}
             essence_ns["alias"] = self.alias
             essence_ns["name"] = self.name
             info = Info(raw_info)

src/relic/sga/core/serialization.py

@@ -489,6 +489,9 @@ def flatten_file_collection(self, container_fs: FS) -> Tuple[int, int]:
         ]
         self.flat_files.extend(subfile_defs)
         subfile_end = len(self.flat_files)
+
+        if subfile_start == subfile_end:
+            subfile_start = subfile_end = 0  #
         return subfile_start, subfile_end
 
     def flatten_folder_collection(self, container_fs: FS, path: str) -> Tuple[int, int]:

tests/regressions/test_version_comparisons.py

@@ -7,8 +7,8 @@
 # Chunky versions start at 1
 # Max i've seen is probably 16ish?
 # Minor has a lot mor variety than Chunky; so we test a bit more
-_VERSION_MAJORS = range(1,11)  # So far we only go up to V10
-_VERSION_MINORS = [0,1]  # Allegedly CoHO was v4.1 so... we do 0,1
+_VERSION_MAJORS = range(1, 11)  # So far we only go up to V10
+_VERSION_MINORS = [0, 1]  # Allegedly CoHO was v4.1 so... we do 0,1
 
 
 _VERSION_ARGS = list(itertools.product(_VERSION_MAJORS, _VERSION_MINORS))

tests/test_cli.py

@@ -31,7 +31,7 @@ def test_run_with(self, args: Sequence[str], output: str, exit_code: int):
             assert status == exit_code
 
 
-_SGA_HELP = ["sga", "-h"], """usage: relic sga [-h] {pack,repack,unpack} ...""", 0
+_SGA_HELP = ["sga", "-h"], """usage: relic sga [-h] {info,pack,repack,unpack} ...""", 0
 _SGA_PACK_HELP = ["sga", "pack", "-h"], """usage: relic sga pack [-h] {} ...""", 0
 _SGA_UNPACK_HELP = ["sga", "unpack", "-h"], """usage: relic sga unpack [-h]""", 0