processing: remove multiple input files support

Author: László Vaskó

tests/test_cli.py

@@ -86,7 +86,7 @@ def test_help(params):
     result = runner.invoke(unblob.cli.cli, params)
     assert result.exit_code == 0
     # NOTE: In practice, it writes "Usage: unblob ...", this is done in the `cli.main` with `click.make_context`
-    assert result.output.startswith("Usage: cli [OPTIONS] FILES...")
+    assert result.output.startswith("Usage: cli [OPTIONS] FILE")
 
 
 @pytest.mark.parametrize(
@@ -118,15 +118,15 @@ def test_without_file(params: List[str]):
     runner = CliRunner()
     result = runner.invoke(unblob.cli.cli, params)
     assert result.exit_code == 2
-    assert "Missing argument 'FILES...'" in result.output
+    assert "Missing argument 'FILE'" in result.output
 
 
 def test_non_existing_file(tmp_path: Path):
     runner = CliRunner()
     path = Path("non/existing/path/54")
     result = runner.invoke(unblob.cli.cli, ["--extract-dir", str(tmp_path), str(path)])
     assert result.exit_code == 2
-    assert "Invalid value for 'FILES...'" in result.output
+    assert "Invalid value for 'FILE'" in result.output
     assert f"Path '{str(path)}' does not exist" in result.output
 
 

unblob/cli.py

@@ -77,8 +77,7 @@ def __init__(
 
 @click.command(help=get_help_text())
 @click.argument(
-    "files",
-    nargs=-1,
+    "file",
     type=click.Path(path_type=Path, exists=True, resolve_path=True),
     required=True,
 )
@@ -160,7 +159,7 @@ def __init__(
     expose_value=False,
 )
 def cli(
-    files: List[Path],
+    file: Path,
     extract_root: Path,
     force: bool,
     depth: int,
@@ -191,8 +190,8 @@ def cli(
         keep_extracted_chunks=keep_extracted_chunks,
     )
 
-    logger.info("Start processing files", count=noformat(len(files)))
-    all_reports = process_files(config, *files)
+    logger.info("Start processing file", file=file)
+    all_reports = process_files(config, file)
     return all_reports
 
 

unblob/processing.py

@@ -20,8 +20,9 @@
 from .models import ExtractError, File, Task, TaskResult, UnknownChunk, ValidChunk
 from .pool import make_pool
 from .report import (
-    ExtractDirectoriesExistReport,
+    ExtractDirectoryExistsReport,
     FileMagicReport,
+    Report,
     StatReport,
     UnknownError,
 )
@@ -59,24 +60,34 @@ class ExtractionConfig:
 
 
 @terminate_gracefully
-def process_files(config: ExtractionConfig, *paths: Path) -> List[Report]:
-    existing_extract_dirs = get_existing_extract_dirs(config, paths)
+def process_files(config: ExtractionConfig, path: Path) -> List[Report]:
+    task = Task(
+        path=path,
+        depth=0,
+    )
 
-    if config.force_extract:
-        for d in existing_extract_dirs:
-            shutil.rmtree(d)
-    elif existing_extract_dirs:
-        report = ExtractDirectoriesExistReport(paths=existing_extract_dirs)
-        logger.error("Extraction directories already exist", **report.asdict())
-        return [report]
+    errors = check_extract_directory(task, config)
+    if errors:
+        return errors
 
-    all_reports = []
-    for path in paths:
-        report = _process_one_file(config, path)
-        all_reports.extend(report)
+    result = _process_one_file(config, task)
+
+    return result
 
-    return all_reports
 
+def check_extract_directory(task: Task, config: ExtractionConfig):
+    errors = []
+
+    extract_dir = get_extract_dir_for_input(config, task.path)
+    if extract_dir.exists():
+        if config.force_extract:
+            shutil.rmtree(extract_dir)
+        else:
+            report = ExtractDirectoryExistsReport(path=extract_dir)
+            logger.error("Extraction directory already exist", **report.asdict())
+            errors.append(report)
+
+    return errors
 
 def get_existing_extract_dirs(
     config: ExtractionConfig, paths: Iterable[Path]
@@ -95,12 +106,7 @@ def get_existing_extract_dirs(
     return extract_dirs
 
 
-def _process_one_file(config: ExtractionConfig, path: Path) -> List[Report]:
-    root_task = Task(
-        path=path,
-        depth=0,
-    )
-
+def _process_one_file(config: ExtractionConfig, root_task: Task) -> List[Report]:
     processor = Processor(config)
     all_reports = []
 

unblob/report.py

@@ -90,9 +90,9 @@ class ExtractCommandFailedReport(ErrorReport):
 
 
 @attr.define(kw_only=True)
-class ExtractDirectoriesExistReport(ErrorReport):
+class ExtractDirectoryExistsReport(ErrorReport):
     severity: Severity = Severity.ERROR
-    paths: List[Path]
+    path: Path
 
 
 @attr.define(kw_only=True)