BOF: local JSONID registry

.gitignore

@@ -142,3 +142,7 @@ jsonid-integration-files/
 # Secreta
 token.pypi
 jsonid_pronom.xml
+*.csv
+*.json
+*.diff
+token-pypi

README.md

@@ -405,9 +405,39 @@ within the `sops` object/value.
 
 ### Local rules
 
-The plan is to allow local rules to be run alongside the global ruleset. I
-expect this will be a bit further down the line when the ruleset and
-metaddata is more stabilised.
+You can define local rules in a local registry object. Local registries are
+defined in TOML, and look as follows:
+
+```toml
+[[entries]]
+
+name = "doctype1"
+identifier = "local0001"
+localref = "http://example.com/doctype/spec/ID"
+
+[[entries.markers]]
+
+key = "key1"
+is = "value1"
+
+[[entries]]
+
+name = "doctype2"
+identifier = "local0002"
+localref = "http://example.com/doctype/spec/ID
+
+[[entries.markers]]
+
+key = "key2"
+is = "value2"
+```
+
+Markers follow the same pattern as the standard registry.
+
+#### Local only
+
+Use the `--localonly` flag to use _just_ your custom markers in your format
+identification workflow.
 
 ## PRONOM
 

local/registry.toml

@@ -0,0 +1,25 @@
+[[entries]]
+
+name = "doctype1"
+identifier = "local0001"
+localref = "http://example.com/repository/ID"
+
+[[entries.markers]]
+
+key = "key1"
+is = "value1"
+
+[[entries.markers]]
+
+key = "key2"
+is = "value2"
+
+[[entries]]
+
+name = "doctype2"
+identifier = "local0002"
+
+[[entries.markers]]
+
+key = "key2"
+is = "value2"

src/jsonid/file_processing.py

@@ -164,6 +164,9 @@ async def analyse_json(paths: list[str], strategy: list):
     """Analyse a JSON object."""
     analysis_res = []
     for path in paths:
+        if os.path.islink(path):
+            logger.debug(f"'{path}' is a symlink")
+            continue
         if os.path.getsize(path) == 0:
             logger.debug("%s is an empty file", path)
             continue
@@ -200,8 +203,13 @@ async def process_result(
     base_obj: registry.BaseCharacteristics,
     padding: int,
     agentout: bool,
+    reg_data: list,
 ):
-    """Process something JSON/YAML/TOML"""
+    """Process something JSON/YAML/TOML.
+
+    TODO: ...
+
+    """
     results = []
     # NB. these switch-like ifs might not be needed in the fullness
     # of time. It depends if we need to do any custom processing of
@@ -225,21 +233,23 @@ async def process_result(
         return
     # If we don't exit early and we try and identify the file... we then
     # create a new class object with an identification...
-    if base_obj.doctype == registry.DOCTYPE_JSON:
-        results = registry.matcher(base_obj)
-    if base_obj.doctype == registry.DOCTYPE_JSONL:
-        results = registry.matcher(base_obj)
-    if base_obj.doctype == registry.DOCTYPE_YAML:
-        results = registry.matcher(base_obj)
-    if base_obj.doctype == registry.DOCTYPE_TOML:
-        results = registry.matcher(base_obj)
+    if base_obj.doctype not in (
+        registry.DOCTYPE_JSON,
+        registry.DOCTYPE_JSONL,
+        registry.DOCTYPE_YAML,
+        registry.DOCTYPE_TOML,
+    ):
+        return
+    results = registry.matcher(
+        base_obj=base_obj,
+        reg_data=reg_data,
+    )
     output.output_results(
         path=path,
         results=results,
         padding=padding,
         agentout=agentout,
     )
-    return
 
 
 def _get_padding(paths: list):
@@ -255,10 +265,18 @@ def _get_padding(paths: list):
     return padding
 
 
-async def identify_json(paths: list[str], strategy: list, binary: bool, agentout: bool):
-    """Identify objects."""
+async def identify_json(
+    paths: list[str], strategy: list, binary: bool, agentout: bool, reg_data: list
+):
+    """Identify objects.
+
+    TODO: ,,,
+    """
     padding = _get_padding(paths=paths)
     for _, path in enumerate(paths):
+        if os.path.islink(path):
+            logger.debug(f"'{path}' is a symlink")
+            continue
         if os.path.getsize(path) == 0:
             logger.debug("%s is an empty file", path)
             base_obj = registry.BaseCharacteristics(empty=True)
@@ -268,6 +286,7 @@ async def identify_json(paths: list[str], strategy: list, binary: bool, agentout
                     base_obj=base_obj,
                     padding=padding,
                     agentout=agentout,
+                    reg_data=reg_data,
                 )
             continue
         base_obj = await identify_plaintext_bytestream(
@@ -283,6 +302,7 @@ async def identify_json(paths: list[str], strategy: list, binary: bool, agentout
                     base_obj=base_obj,
                     padding=padding,
                     agentout=agentout,
+                    reg_data=reg_data,
                 )
             continue
         logger.debug("processing: %s (%s)", path, base_obj.doctype)
@@ -291,6 +311,7 @@ async def identify_json(paths: list[str], strategy: list, binary: bool, agentout
             base_obj=base_obj,
             padding=padding,
             agentout=agentout,
+            reg_data=reg_data,
         )
 
 
@@ -443,7 +464,9 @@ async def process_glob(glob_path: str):
     return paths
 
 
-async def process_data(path: str, strategy: list, binary: bool, agentout: bool):
+async def process_data(
+    path: str, strategy: list, binary: bool, agentout: bool, reg_data: list
+):
     """Process all objects at a given path."""
     logger.debug("processing: %s", path)
     if "*" in path:
@@ -453,6 +476,7 @@ async def process_data(path: str, strategy: list, binary: bool, agentout: bool):
             strategy=strategy,
             binary=binary,
             agentout=agentout,
+            reg_data=reg_data,
         )
         sys.exit(0)
     if not os.path.exists(path):
@@ -464,6 +488,7 @@ async def process_data(path: str, strategy: list, binary: bool, agentout: bool):
             strategy=strategy,
             binary=binary,
             agentout=agentout,
+            reg_data=reg_data,
         )
         sys.exit(0)
     paths = await create_manifest(path)
@@ -475,6 +500,7 @@ async def process_data(path: str, strategy: list, binary: bool, agentout: bool):
         strategy=strategy,
         binary=binary,
         agentout=agentout,
+        reg_data=reg_data,
     )
 
 

src/jsonid/jsonid.py

@@ -4,6 +4,7 @@
 
 import argparse
 import asyncio
+import copy
 import logging
 import signal
 import sys
@@ -16,11 +17,30 @@
     import helpers
     import lookup
     import registry
+    import registry_data
+
+    import local
 except ModuleNotFoundError:
     try:
-        from src.jsonid import export, file_processing, helpers, lookup, registry
+        from src.jsonid import (
+            export,
+            file_processing,
+            helpers,
+            local,
+            lookup,
+            registry,
+            registry_data,
+        )
     except ModuleNotFoundError:
-        from jsonid import export, file_processing, helpers, lookup, registry
+        from jsonid import (
+            export,
+            file_processing,
+            helpers,
+            local,
+            lookup,
+            registry,
+            registry_data,
+        )
 
 
 logger = None
@@ -216,9 +236,18 @@ def main() -> None:
     )
     parser.add_argument(
         "--registry",
+        "--local",
         help="path to a custom registry to lead into memory replacing the default",
         required=False,
     )
+    parser.add_argument(
+        "--localonly",
+        "--lonly",
+        "--lonely",
+        help="if a local registry is specified, use this and this only",
+        required=False,
+        action="store_true",
+    )
     # NB. consider output to stdout once the feature is more stable.
     parser.add_argument(
         "--pronom",
@@ -284,7 +313,12 @@ def main() -> None:
 
     # Primary application functions.
     if args.registry:
-        raise NotImplementedError("custom registry is not yet available")
+        if args.localonly:
+            reg_data = local.load_and_parse_local_registry(path=args.registry)
+        else:
+            reg_data = local.load_and_parse_local_registry(path=args.registry)
+    if not args.registry:
+        reg_data = copy.deepcopy(registry_data.registry())
     if args.pronom:
         export.export_pronom()
         sys.exit()
@@ -305,6 +339,8 @@ def main() -> None:
         logger.info("ok")
         sys.exit()
     if args.html:
+        if args.registry:
+            raise NotImplementedError("local registry output is not yet supported")
         helpers.html()
         sys.exit()
     if not strategy:
@@ -338,6 +374,8 @@ def signal_handler(*args):  # pylint: disable=W0613
             strategy=strategy,
             binary=args.binary,
             agentout=args.agentout,
+            # TODO: registry data here? or just feed the local config through?
+            reg_data=reg_data,
         )
     )