process-media: filter exts if supporting programs not installed

xklb/mediafiles/process_media.py

@@ -1,5 +1,6 @@
 import argparse, math, os, sqlite3
 from contextlib import suppress
+from shutil import which
 
 from xklb import usage
 from xklb.mediadb import db_history
@@ -88,21 +89,32 @@ def parse_args() -> argparse.Namespace:
 
 
 def collect_media(args) -> list[dict]:
+    FFMPEG_INSTALLED = which("ffmpeg") or which("ffmpeg.exe")
+    IM7_INSTALLED = which("magick")
+    CALIBRE_INSTALLED = which("ebook-convert")
+    UNAR_INSTALLED = which("lsar")
+
+    default_exts = (
+        (consts.AUDIO_ONLY_EXTENSIONS if FFMPEG_INSTALLED else set())
+        | (consts.VIDEO_EXTENSIONS if FFMPEG_INSTALLED else set())
+        | (consts.IMAGE_EXTENSIONS - set(("avif",)) if IM7_INSTALLED else set())
+        | (consts.CALIBRE_EXTENSIONS if CALIBRE_INSTALLED else set())
+        | (consts.ARCHIVE_EXTENSIONS if UNAR_INSTALLED else set())
+    )
+
     if args.database:
         db_history.create(args)
 
+        if not args.ext:
+            or_conditions = [f"path like '%.{ext}'" for ext in default_exts]
+            args.filter_sql.append(f" AND ({' OR '.join(or_conditions)})")
+
         try:
             media = list(args.db.query(*sqlgroups.media_sql(args)))
         except sqlite3.OperationalError:
             media = list(args.db.query(*sqlgroups.fs_sql(args, args.limit)))
     else:
-        media = arg_utils.gen_d(
-            args,
-            consts.AUDIO_ONLY_EXTENSIONS
-            | consts.VIDEO_EXTENSIONS
-            | consts.IMAGE_EXTENSIONS - set(("avif",))
-            | consts.CALIBRE_EXTENSIONS,
-        )
+        media = arg_utils.gen_d(args, default_exts)
         media = [d if "size" in d else file_utils.get_filesize(d) for d in media]
     return media
 
@@ -215,7 +227,7 @@ def check_shrink(args, m) -> list:
 
         m["media_type"] = "Text"
         m["future_size"] = future_size
-        m["savings"] = (m.get("compresseos.stat(new_path).st_sized_size") or m["size"]) - future_size
+        m["savings"] = (m.get("compressed_size") or m["size"]) - future_size
         m["processing_time"] = args.transcoding_image_time * 12
         if can_shrink:
             return [m]
@@ -228,7 +240,11 @@ def check_shrink(args, m) -> list:
         return [check_shrink(args, d) for d in contents]
     else:
         # TODO: csv, json => parquet
-        log.warning("[%s]: Skipping unknown filetype %s", m["path"], filetype)
+
+        if m.get("compressed_size"):
+            log.warning("[%s]: Skipping unknown filetype %s from archive %s", m["path"], m["ext"], m["archive_path"])
+        else:
+            log.warning("[%s]: Skipping unknown filetype %s", m["path"], filetype)
     return []
 
 

xklb/mediafiles/process_text.py

@@ -154,6 +154,10 @@ def process_path(args, path):
 def process_text():
     args = parse_args()
 
+    if not shutil.which("ebook-convert"):
+        print("Calibre is required for process-text")
+        raise SystemExit(1)
+
     for path in gen_paths(args, consts.CALIBRE_EXTENSIONS):
         if not path.startswith("http"):
             path = str(Path(path).resolve())

xklb/usage.py

@@ -1845,5 +1845,11 @@ def play(action) -> str:
 
         library process-media --invalid --no-valid --delete-unplayable video.db
 
+    If not installed, related file extensions will be skipped during scan:
+
+        - FFmpeg is required for shrinking video and audio
+        - ImageMagick is required for shrinking images
+        - Calibre is required for shrinking eBooks
+
 Inspired somewhat by https://nikkhokkho.sourceforge.io/?page=FileOptimizer
 """

xklb/utils/processes.py

@@ -393,6 +393,9 @@ def lsar(archive_path):
     unar_out = unar_out_path(archive_path)
     archive_info_list = []
     for entry in lsar_contents:
+        if entry.get("XADIsDirectory", 0):
+            continue
+
         archive_info_list.append(
             {
                 "archive_path": archive_path,

xklb/utils/web.py

@@ -888,6 +888,9 @@ def __new__(cls, *args):
     def __init__(self, path):
         self._path = str(path)
 
+    def __fspath__(self):
+        return str(self)
+
     @property
     def parent(self):
         scheme, netloc, path, params, query, fragment = urlparse(str(self))