fix/refactor: use generic ModuleStatus class for getting status of optional modules

src/tagstudio/core/utils/ffmpeg_status.py

@@ -0,0 +1,69 @@
+# SPDX-FileCopyrightText: (c) TagStudio Contributors
+# SPDX-License-Identifier: MIT
+
+
+import contextlib
+from typing import override
+
+import structlog
+
+from tagstudio.core.utils.module_status import ModuleStatus
+from tagstudio.core.utils.silent_subprocess import (
+    silent_run,  # pyright: ignore[reportUnknownVariableType]
+)
+
+logger = structlog.get_logger(__name__)
+
+
+class FfmpegStatus(ModuleStatus):
+    """Class for getting the location and version of FFmpeg, if it exists."""
+
+    @override
+    @classmethod
+    def which(cls):
+        if cls._cached_location:
+            return cls._cached_location
+
+        cls._cached_location = cls._which("ffmpeg")
+        return cls._cached_location
+
+    @override
+    @classmethod
+    def version(cls):
+        if cls._cached_version:
+            return cls._cached_version
+
+        ffmpeg_cmd = cls._which("ffmpeg")
+        if ffmpeg_cmd:
+            out = silent_run([ffmpeg_cmd, "-version"], shell=False, capture_output=True, text=True)
+            if out.returncode == 0:
+                with contextlib.suppress(Exception):
+                    cls._cached_version = str(out.stdout).split(" ")[2]
+                    return cls._cached_version
+
+
+class FfprobeStatus(ModuleStatus):
+    """Class for getting the location and version of FFprobe, if it exists."""
+
+    @override
+    @classmethod
+    def which(cls):
+        if cls._cached_location:
+            return cls._cached_location
+
+        cls._cached_location = cls._which("ffprobe")
+        return cls._cached_location
+
+    @override
+    @classmethod
+    def version(cls):
+        if cls._cached_version:
+            return cls._cached_version
+
+        ffmpeg_cmd = cls._which("ffprobe")
+        if ffmpeg_cmd:
+            out = silent_run([ffmpeg_cmd, "-version"], shell=False, capture_output=True, text=True)
+            if out.returncode == 0:
+                with contextlib.suppress(Exception):
+                    cls._cached_version = str(out.stdout).split(" ")[2]
+                    return cls._cached_version

src/tagstudio/core/utils/module_status.py

@@ -0,0 +1,50 @@
+# SPDX-FileCopyrightText: (c) TagStudio Contributors
+# SPDX-License-Identifier: MIT
+
+import os
+import platform
+from shutil import which
+
+import structlog
+
+logger = structlog.get_logger(__name__)
+user = os.environ.get("USER", None)
+
+# NOTE: macOS does not make its PATH variable available to processes started outside the terminal.
+# The following is a list of common directories to search for binaries in.
+MACOS_BIN_LOCATIONS: list[str] = [
+    "",
+    "/opt/homebrew/bin/",
+    "/usr/local/bin/",
+    f"/etc/profiles/per-user/{user}/bin/",
+    "~/.nix-profile/bin/",
+]
+
+
+class ModuleStatus:
+    """An abstract base class for module status logic including the binary location and version."""
+
+    _cached_location: str | None = None
+    _cached_version: str | None = None
+
+    @classmethod
+    def which(cls) -> str | None:
+        raise NotImplementedError()
+
+    @classmethod
+    def version(cls) -> str | None:
+        raise NotImplementedError()
+
+    @staticmethod
+    def _which(cmd: str) -> str | None:
+        """Internal method for determining the correct location for which().
+
+        Args:
+            cmd (str): The process command to search for.
+        """
+        if platform.system() == "Darwin":
+            for loc in MACOS_BIN_LOCATIONS:
+                if which(loc + cmd):
+                    cmd = loc + cmd
+                    break
+        return cmd

src/tagstudio/core/utils/ripgrep_status.py

@@ -0,0 +1,42 @@
+# SPDX-FileCopyrightText: (c) TagStudio Contributors
+# SPDX-License-Identifier: MIT
+
+
+import contextlib
+from typing import override
+
+import structlog
+
+from tagstudio.core.utils.module_status import ModuleStatus
+from tagstudio.core.utils.silent_subprocess import (
+    silent_run,  # pyright: ignore[reportUnknownVariableType]
+)
+
+logger = structlog.get_logger(__name__)
+
+
+class RipgrepStatus(ModuleStatus):
+    """Class for getting the location and version of ripgrep, if it exists."""
+
+    @override
+    @classmethod
+    def which(cls):
+        if cls._cached_location:
+            return cls._cached_location
+
+        cls._cached_location = cls._which("rg")
+        return cls._cached_location
+
+    @override
+    @classmethod
+    def version(cls):
+        if cls._cached_version:
+            return cls._cached_version
+
+        ripgrep_cmd = cls._which("rg")
+        if ripgrep_cmd:
+            out = silent_run([ripgrep_cmd, "-V"], shell=False, capture_output=True, text=True)
+            if out.returncode == 0:
+                with contextlib.suppress(Exception):
+                    cls._cached_version = str(out.stdout).split(" ")[1].rstrip("\n")
+                    return cls._cached_version

src/tagstudio/core/utils/silent_subprocess.py

@@ -1,5 +1,5 @@
 # SPDX-FileCopyrightText: (c) TagStudio Contributors
-# SPDX-License-Identifier: GPL-3.0-only
+# SPDX-License-Identifier: MIT
 
 
 import os

src/tagstudio/qt/controllers/preview_panel_controller.py

@@ -3,14 +3,13 @@
 
 
 import typing
-from shutil import which
 from warnings import catch_warnings
 
 from tagstudio.core.library.alchemy.fields import BaseFieldTemplate
 from tagstudio.core.library.alchemy.library import Library
+from tagstudio.core.utils.ffmpeg_status import FfmpegStatus, FfprobeStatus
 from tagstudio.qt.controllers.field_template_search_panel_controller import FieldTemplateSearchModal
 from tagstudio.qt.controllers.tag_search_panel_controller import TagSearchModal
-from tagstudio.qt.previews.vendored.ffmpeg import FFMPEG_CMD, FFPROBE_CMD
 from tagstudio.qt.translations import Translations
 from tagstudio.qt.views.preview_panel_view import PreviewPanelView
 
@@ -59,7 +58,7 @@ class PreviewPanel(PreviewPanelView):
             self._containers.update_from_entry(self._selected[0])
 
     def _toggle_ffmpeg_warning(self, enable_warning: bool = True) -> None:
-        if enable_warning and (not which(FFMPEG_CMD) or not which(FFPROBE_CMD)):
+        if enable_warning and (not FfmpegStatus.which() or not FfprobeStatus.which()):
             self._ffmpeg_warning_widget.show()
             return
 

src/tagstudio/qt/helpers/file_tester.py

@@ -6,9 +6,7 @@ from pathlib import Path
 
 import ffmpeg
 
-from tagstudio.qt.previews.vendored.ffmpeg import (
-    probe,  # pyright: ignore[reportUnknownVariableType]
-)
+from tagstudio.qt.previews.vendored.probe import probe
 
 
 def is_readable_video(filepath: Path | str):
@@ -21,6 +19,8 @@ def is_readable_video(filepath: Path | str):
     """
     try:
         result = probe(Path(filepath))
+        if not result:
+            return False
         for stream in result["streams"]:
             # DRM check
             if stream.get("codec_tag_string") in [

src/tagstudio/qt/mixed/about_modal.py

@@ -4,7 +4,6 @@
 
 import math
 from pathlib import Path
-from shutil import which
 
 from PIL import ImageQt
 from PySide6.QtCore import QSize, Qt
@@ -28,11 +27,12 @@ from tagstudio.core.constants import (
     VERSION_BRANCH,
 )
 from tagstudio.core.ts_core import TagStudioCore
+from tagstudio.core.utils.ffmpeg_status import FfmpegStatus, FfprobeStatus
+from tagstudio.core.utils.ripgrep_status import RipgrepStatus
 from tagstudio.core.utils.str_formatting import is_version_outdated
 from tagstudio.core.utils.types import unwrap
 from tagstudio.qt.controllers.clickable_label import ClickableLabel
 from tagstudio.qt.models.palette import ColorType, UiColor, get_ui_color
-from tagstudio.qt.previews.vendored import ffmpeg
 from tagstudio.qt.resource_manager import ResourceManager
 from tagstudio.qt.translations import Translations
 from tagstudio.qt.utils.file_opener import open_file
@@ -94,7 +94,9 @@ class AboutModal(QWidget):
         self.desc_label.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding)
 
         # System Info ----------------------------------------------------------
-        ff_version = ffmpeg.version()
+        ffmpeg_ver = FfmpegStatus.version()
+        ffprobe_ver = FfprobeStatus.version()
+        ripgrep_ver = RipgrepStatus.version()
         red = get_ui_color(ColorType.PRIMARY, UiColor.RED)
         green = get_ui_color(ColorType.PRIMARY, UiColor.GREEN)
         amber = get_ui_color(ColorType.PRIMARY, UiColor.AMBER)
@@ -102,20 +104,16 @@ class AboutModal(QWidget):
         found = Translations["about.module.found"]
 
         ffmpeg_status = f'<span style="color:{red}">{missing}</span>'
-        if ff_version["ffmpeg"] is not None:
-            ffmpeg_status = (
-                f'<span style="color:{green}">{found}</span> (' + ff_version["ffmpeg"] + ")"
-            )
+        if ffmpeg_ver is not None:
+            ffmpeg_status = f'<span style="color:{green}">{found}</span> (' + ffmpeg_ver + ")"
 
         ffprobe_status = f'<span style="color:{red}">{missing}</span>'
-        if ff_version["ffprobe"] is not None:
-            ffprobe_status = (
-                f'<span style="color:{green}">{found}</span> (' + ff_version["ffprobe"] + ")"
-            )
+        if ffprobe_ver is not None:
+            ffprobe_status = f'<span style="color:{green}">{found}</span> (' + ffprobe_ver + ")"
 
         ripgrep_status = f'<span style="color:{amber}">{missing}</span>'
-        if which("rg") is not None:
-            ripgrep_status = f'<span style="color:{green}">{found}</span>'
+        if ripgrep_ver is not None:
+            ripgrep_status = f'<span style="color:{green}">{found}</span> (' + ripgrep_ver + ")"
 
         self.system_info_widget = QWidget()
         self.system_info_layout = QFormLayout(self.system_info_widget)
@@ -152,7 +150,7 @@ class AboutModal(QWidget):
         # FFmpeg Status
         ffmpeg_path_title = QLabel("FFmpeg")
         ffmpeg_path_content = ClickableLabel(f"{ffmpeg_status}")
-        ffmpeg_location = which(ffmpeg._get_ffmpeg_location())  # pyright: ignore[reportPrivateUsage]
+        ffmpeg_location = FfmpegStatus.which()
         if ffmpeg_location:
             ffmpeg_path_content.clicked.connect(
                 lambda: open_file(ffmpeg_location, file_manager=True)
@@ -165,7 +163,7 @@ class AboutModal(QWidget):
         # FFprobe Status
         ffprobe_path_title = QLabel("FFprobe")
         ffprobe_path_content = ClickableLabel(f"{ffprobe_status}")
-        ffprobe_location = which(ffmpeg._get_ffprobe_location())  # pyright: ignore[reportPrivateUsage]
+        ffprobe_location = FfprobeStatus.which()
         if ffprobe_location:
             ffprobe_path_content.clicked.connect(
                 lambda: open_file(ffprobe_location, file_manager=True)
@@ -180,7 +178,7 @@ class AboutModal(QWidget):
         ripgrep_path_title = QLabel("ripgrep")  # NOTE: Don't localize
         ripgrep_path_content = ClickableLabel()
         ripgrep_path_content.setText(f"{ripgrep_status}")  # TODO: Pass in constructor after #1386
-        ripgrep_location = which("rg")
+        ripgrep_location = RipgrepStatus.which()
         if ripgrep_location:
             ripgrep_path_content.clicked.connect(
                 lambda: open_file(ripgrep_location, file_manager=True)

src/tagstudio/qt/previews/vendored/ffmpeg.py

@@ -1,102 +0,0 @@
-# SPDX-FileCopyrightText: (c) 2022  Karl Kroening (kkroening)
-# SPDX-FileCopyrightText: (c) TagStudio Contributors
-# SPDX-License-Identifier: GPL-3.0-only
-# Vendored from ffmpeg-python and ffmpeg-python PR#790 by amamic1803
-
-
-import contextlib
-import json
-import os
-import platform
-import subprocess
-from shutil import which
-
-import ffmpeg
-import structlog
-
-from tagstudio.core.utils.silent_subprocess import silent_popen, silent_run
-
-logger = structlog.get_logger(__name__)
-
-user = os.environ.get("USER", None)
-FFMPEG_MACOS_LOCATIONS: list[str] = [
-    "",
-    "/opt/homebrew/bin/",
-    "/usr/local/bin/",
-    f"/etc/profiles/per-user/{user}/bin",
-]
-
-
-# TODO: Make this more intuitive to use in other classes
-def _get_ffprobe_location() -> str:
-    cmd: str = "ffprobe"
-    if platform.system() == "Darwin":
-        for loc in FFMPEG_MACOS_LOCATIONS:
-            if which(loc + cmd):
-                cmd = loc + cmd
-                break
-    logger.info(
-        f"[FFmpeg] Using FFprobe location: {cmd}{' (Found)' if which(cmd) else ' (Not Found)'}"
-    )
-    return cmd
-
-
-# TODO: Make this more intuitive to use in other classes
-def _get_ffmpeg_location() -> str:
-    cmd: str = "ffmpeg"
-    if platform.system() == "Darwin":
-        for loc in FFMPEG_MACOS_LOCATIONS:
-            if which(loc + cmd):
-                cmd = loc + cmd
-                break
-    logger.info(
-        f"[FFmpeg] Using FFmpeg location: {cmd}{' (Found)' if which(cmd) else ' (Not Found)'}"
-    )
-    return cmd
-
-
-FFPROBE_CMD = _get_ffprobe_location()
-FFMPEG_CMD = _get_ffmpeg_location()
-
-
-def probe(filename, cmd=FFPROBE_CMD, timeout=None, **kwargs):
-    """Run ffprobe on the specified file and return a JSON representation of the output.
-
-    Raises:
-        :class:`ffmpeg.Error`: if ffprobe returns a non-zero exit code,
-            an :class:`Error` is returned with a generic error message.
-            The stderr output can be retrieved by accessing the
-            ``stderr`` property of the exception.
-    """
-    args = [cmd, "-show_format", "-show_streams", "-of", "json"]
-    args += ffmpeg._utils.convert_kwargs_to_cmd_line_args(kwargs)  # pyright: ignore[reportAttributeAccessIssue]
-    args += [filename]
-
-    # PATCHED
-    p = silent_popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-    communicate_kwargs = {}
-    if timeout is not None:
-        communicate_kwargs["timeout"] = timeout
-    out, err = p.communicate(**communicate_kwargs)
-    if p.returncode != 0:
-        raise ffmpeg.Error("ffprobe", out, err)
-    return json.loads(out.decode("utf-8"))
-
-
-def version():
-    """Checks the version of FFmpeg and FFprobe and returns None if they dont exist."""
-    version: dict[str, str | None] = {"ffmpeg": None, "ffprobe": None}
-
-    if which(FFMPEG_CMD):
-        ret = silent_run([FFMPEG_CMD, "-version"], shell=False, capture_output=True, text=True)
-        if ret.returncode == 0:
-            with contextlib.suppress(Exception):
-                version["ffmpeg"] = str(ret.stdout).split(" ")[2]
-
-    if which(FFPROBE_CMD):
-        ret = silent_run([FFPROBE_CMD, "-version"], shell=False, capture_output=True, text=True)
-        if ret.returncode == 0:
-            with contextlib.suppress(Exception):
-                version["ffprobe"] = str(ret.stdout).split(" ")[2]
-
-    return version

src/tagstudio/qt/previews/vendored/probe.py

@@ -0,0 +1,45 @@
+# SPDX-FileCopyrightText: (c) 2022  Karl Kroening (kkroening)
+# SPDX-FileCopyrightText: (c) TagStudio Contributors
+# SPDX-License-Identifier: GPL-3.0-only
+# Vendored from ffmpeg-python and ffmpeg-python PR#790 by amamic1803
+
+
+import json
+import subprocess
+from pathlib import Path
+
+import ffmpeg
+import structlog
+
+from tagstudio.core.utils.ffmpeg_status import FfprobeStatus
+from tagstudio.core.utils.silent_subprocess import (
+    silent_popen,  # pyright: ignore[reportUnknownVariableType]
+)
+
+logger = structlog.get_logger(__name__)
+
+
+def probe(filename: Path | str, timeout: int | None = None, **kwargs: ...):
+    """Run ffprobe on the specified file and return a JSON representation of the output.
+
+    Raises:
+        Error: If ffprobe returns a non-zero exit code, an Error is raised
+            with a generic error message. The stderr output can be retrieved
+            by accessing the `stderr` property of the exception.
+    """
+    ffprobe_cmd: str | None = FfprobeStatus.which()
+    if not ffprobe_cmd:
+        return
+    args: list[str] = [ffprobe_cmd, "-show_format", "-show_streams", "-of", "json"]
+    args += ffmpeg._utils.convert_kwargs_to_cmd_line_args(kwargs)  # pyright: ignore
+    args += [filename]  # pyright: ignore
+
+    # PATCHED
+    p = silent_popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    communicate_kwargs = {}
+    if timeout is not None:
+        communicate_kwargs["timeout"] = timeout
+    out, err = p.communicate(**communicate_kwargs)
+    if p.returncode != 0:
+        raise ffmpeg.Error("ffprobe", out, err)
+    return json.loads(out.decode("utf-8"))

src/tagstudio/qt/previews/vendored/pydub/audio_segment.py

@@ -18,7 +18,7 @@ from tempfile import NamedTemporaryFile
 from pydub.logging_utils import log_conversion, log_subprocess_output
 from pydub.utils import fsdecode
 
-from tagstudio.qt.previews.vendored.ffmpeg import FFMPEG_CMD
+from tagstudio.core.utils.ffmpeg_status import FfmpegStatus
 
 try:
     from itertools import izip
@@ -161,7 +161,7 @@ class _AudioSegment:
         slice = a[5000:10000] # get a slice from 5 to 10 seconds of an mp3
     """
 
-    converter = FFMPEG_CMD
+    converter = FfmpegStatus.which()
 
     # TODO: remove in 1.0 release
     # maintain backwards compatibility for ffmpeg attr (now called converter)
@@ -727,7 +727,7 @@ class _AudioSegment:
             stdin_parameter = None
             stdin_data = None
         else:
-            if cls.converter == FFMPEG_CMD:
+            if cls.converter == FfmpegStatus.which():
                 conversion_command += [
                     "-read_ahead_limit",
                     str(read_ahead_limit),

src/tagstudio/qt/previews/vendored/pydub/utils.py

@@ -14,13 +14,13 @@ from pydub.utils import (
     get_extra_info,
 )
 
+from tagstudio.core.utils.ffmpeg_status import FfprobeStatus
 from tagstudio.core.utils.silent_subprocess import silent_popen
-from tagstudio.qt.previews.vendored.ffmpeg import FFPROBE_CMD
 
 
 def _mediainfo_json(filepath, read_ahead_limit=-1):
     """Return json dictionary with media info(codec, duration, size, bitrate...) from filepath."""
-    prober = FFPROBE_CMD
+    prober = FfprobeStatus.which()
     command_args = [
         "-v",
         "info",

src/tagstudio/qt/ts_qt.py

@@ -39,7 +39,6 @@ from PySide6.QtGui import (
 )
 from PySide6.QtWidgets import QApplication, QFileDialog, QMessageBox, QPushButton, QScrollArea
 
-# This import has side-effect of importing PySide resources
 import tagstudio.qt.resources_rc  # noqa: F401  # pyright: ignore[reportUnusedImport]
 from tagstudio.core.constants import TAG_ARCHIVED, TAG_FAVORITE, VERSION, VERSION_BRANCH
 from tagstudio.core.driver import DriverMixin
@@ -52,6 +51,10 @@ from tagstudio.core.library.refresh import RefreshTracker
 from tagstudio.core.media_types import MediaCategories
 from tagstudio.core.query_lang.util import ParsingError
 from tagstudio.core.ts_core import TagStudioCore
+
+# This import has side-effect of importing PySide resources
+from tagstudio.core.utils.ffmpeg_status import FfmpegStatus, FfprobeStatus
+from tagstudio.core.utils.ripgrep_status import RipgrepStatus
 from tagstudio.core.utils.str_formatting import is_version_outdated
 from tagstudio.core.utils.types import unwrap
 from tagstudio.qt.cache_manager import CacheManager
@@ -320,6 +323,9 @@ class QtDriver(DriverMixin, QObject):
         timer.start(500)
         timer.timeout.connect(lambda: None)
 
+        # Detect optional modules and versions for logging
+        self.log_optional_modules()
+
         # self.main_window = loader.load(home_path)
         self.main_window = MainWindow(self)
         self.main_window.setWindowTitle(self.base_title)
@@ -1755,3 +1761,32 @@ class QtDriver(DriverMixin, QObject):
     def clear_selected(self):
         self._selected.clear()
         self.main_window.thumb_layout.update_selected()
+
+    def log_optional_modules(self) -> None:
+        """Logs the status of optional modules."""
+        if FfmpegStatus.which():
+            logger.info(
+                "[QtDriver] FFmpeg found",
+                bin_location=FfmpegStatus.which(),
+                version=FfmpegStatus.version(),
+            )
+        else:
+            logger.warning("[QtDriver] FFmpeg not found")
+
+        if FfprobeStatus.which():
+            logger.info(
+                "[QtDriver] FFprobe found",
+                bin_location=FfprobeStatus.which(),
+                version=FfprobeStatus.version(),
+            )
+        else:
+            logger.warning("[QtDriver] FFprobe not found")
+
+        if RipgrepStatus.which():
+            logger.info(
+                "[QtDriver] ripgrep found",
+                bin_location=RipgrepStatus.which(),
+                version=RipgrepStatus.version(),
+            )
+        else:
+            logger.warning("[QtDriver] ripgrep not found")