diff --git a/src/usdb_syncer/constants.py b/src/usdb_syncer/constants.py index 15ae1c39..6bd7dc8e 100644 --- a/src/usdb_syncer/constants.py +++ b/src/usdb_syncer/constants.py @@ -9,6 +9,8 @@ MINIMUM_BPM = 200.0 +CLEANUP_DELETE_IMMEDIATELY_ENV_VAR = "SYNCER_CLEANUP_DELETE_IMMEDIATELY" + class UsdbStrings: """Relevant strings from USDB""" diff --git a/src/usdb_syncer/gui/song_table/song_table.py b/src/usdb_syncer/gui/song_table/song_table.py index b570f3d1..21019d3b 100644 --- a/src/usdb_syncer/gui/song_table/song_table.py +++ b/src/usdb_syncer/gui/song_table/song_table.py @@ -5,7 +5,6 @@ from functools import partial from typing import TYPE_CHECKING, Any, Callable, Iterable, Iterator -import send2trash from PySide6 import QtCore, QtGui, QtMultimedia, QtWidgets from PySide6.QtCore import QItemSelectionModel, Qt @@ -18,6 +17,7 @@ from usdb_syncer.logger import song_logger from usdb_syncer.song_loader import DownloadManager from usdb_syncer.usdb_song import DownloadStatus, UsdbSong +from usdb_syncer.utils import trash_or_delete_file_paths if TYPE_CHECKING: from usdb_syncer.gui.mw import MainWindow @@ -278,10 +278,10 @@ def delete_selected_songs(self) -> None: logger.info("Not trashing song folder as it is pinned.") continue if song.sync_meta.path.exists(): - send2trash.send2trash(song.sync_meta.path.parent) + trash_or_delete_file_paths(song.sync_meta.path.parent) song.remove_sync_meta() events.SongChanged(song.song_id) - logger.debug("Trashed song folder.") + logger.debug("Trashed/deleted song folder.") def set_pin_selected_songs(self, pin: bool) -> None: for song in self.selected_songs(): diff --git a/src/usdb_syncer/song_loader.py b/src/usdb_syncer/song_loader.py index c1067dc9..c91d5600 100644 --- a/src/usdb_syncer/song_loader.py +++ b/src/usdb_syncer/song_loader.py @@ -17,7 +17,6 @@ import mutagen.ogg import mutagen.oggopus import mutagen.oggvorbis -import send2trash import shiboken6 from mutagen import id3 from mutagen.flac import Picture @@ -44,7 +43,7 @@ from usdb_syncer.sync_meta import ResourceFile, SyncMeta from usdb_syncer.usdb_scraper import SongDetails from usdb_syncer.usdb_song import DownloadStatus, UsdbSong -from usdb_syncer.utils import video_url_from_resource +from usdb_syncer.utils import trash_or_delete_file_paths, video_url_from_resource class DownloadManager: @@ -754,11 +753,11 @@ def _cleanup_existing_resources(ctx: _Context) -> None: if not out.old_fname: # out of sync if old_path.exists(): - send2trash.send2trash(old_path) - ctx.logger.debug(f"Trashed untracked file: '{old_path}'.") + trash_or_delete_file_paths(old_path) + ctx.logger.debug(f"Trashed/deleted untracked file: '{old_path}'.") elif out.new_fname: - send2trash.send2trash(old_path) - ctx.logger.debug(f"Trashed existing file: '{old_path}'.") + trash_or_delete_file_paths(old_path) + ctx.logger.debug(f"Trashed/deleted existing file: '{old_path}'.") else: target = ctx.locations.filename(ext=utils.resource_file_ending(old.fname)) if out.old_fname != target: @@ -775,8 +774,8 @@ def _persist_tempfiles(ctx: _Context) -> None: ): target = ctx.locations.target_path(temp_path.name) if target.exists(): - send2trash.send2trash(target) - ctx.logger.debug(f"Trashed existing file: '{target}'.") + trash_or_delete_file_paths(target) + ctx.logger.debug(f"Trashed/deleted existing file: '{target}'.") shutil.move(temp_path, target) diff --git a/src/usdb_syncer/song_routines.py b/src/usdb_syncer/song_routines.py index a0950b47..57273fa7 100644 --- a/src/usdb_syncer/song_routines.py +++ b/src/usdb_syncer/song_routines.py @@ -5,7 +5,6 @@ from pathlib import Path from typing import Generator -import send2trash from requests import Session from usdb_syncer import ( @@ -23,7 +22,7 @@ from usdb_syncer.sync_meta import SyncMeta from usdb_syncer.usdb_scraper import get_usdb_available_songs from usdb_syncer.usdb_song import UsdbSong, UsdbSongEncoder -from usdb_syncer.utils import AppPaths +from usdb_syncer.utils import AppPaths, trash_or_delete_file_paths def load_available_songs(force_reload: bool, session: Session | None = None) -> None: @@ -96,8 +95,8 @@ def synchronize_sync_meta_folder(folder: Path) -> None: for path in _iterate_usdb_files_in_folder_recursively(folder=folder): if meta_id := SyncMetaId.from_path(path): if meta_id in found_metas: - send2trash.send2trash(path) - logger.warning(f"Trashed duplicated meta file: '{path}'") + trash_or_delete_file_paths(path) + logger.warning(f"Trashed/deleted duplicated meta file: '{path}'") continue found_metas.add(meta_id) diff --git a/src/usdb_syncer/utils.py b/src/usdb_syncer/utils.py index c735d669..cbe07e59 100644 --- a/src/usdb_syncer/utils.py +++ b/src/usdb_syncer/utils.py @@ -10,9 +10,14 @@ import time import unicodedata from pathlib import Path +from shutil import rmtree +from typing import Any from appdirs import AppDirs +from send2trash import send2trash +from send2trash.util import preprocess_paths +from usdb_syncer.constants import CLEANUP_DELETE_IMMEDIATELY_ENV_VAR from usdb_syncer.logger import logger CACHE_LIFETIME = 60 * 60 @@ -241,3 +246,19 @@ def format_timestamp(micros: int) -> str: return datetime.datetime.fromtimestamp(micros / 1_000_000).strftime( "%Y-%m-%d %H:%M:%S" ) + + +def trash_or_delete_file_paths(paths: str | bytes | os.PathLike | list[Any]) -> None: + """Either sends the paths to trash or deletes them immediately, depending on the environment variable. + + Args: + paths (str | bytes | os.PathLike | list[Any]): The paths to trash/delete. + """ + if os.environ.get(CLEANUP_DELETE_IMMEDIATELY_ENV_VAR) is not None: + for _path in preprocess_paths(paths): + if os.path.isfile(_path): + os.remove(_path) + elif os.path.isdir(_path): + rmtree(_path) + else: + send2trash(paths) diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index b7119b5f..304e3978 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -1,10 +1,17 @@ """Tests for utils.""" +import os from pathlib import Path +from unittest.mock import MagicMock, patch import pytest -from usdb_syncer.utils import extract_youtube_id, resource_file_ending +from usdb_syncer.constants import CLEANUP_DELETE_IMMEDIATELY_ENV_VAR +from usdb_syncer.utils import ( + extract_youtube_id, + resource_file_ending, + trash_or_delete_file_paths, +) FAKE_YOUTUBE_ID = "fake_YT-id0" @@ -26,3 +33,29 @@ def test_extract_youtube_id(resource_dir: Path) -> None: ) def test_resource_file_ending(name: str, expected: str) -> None: assert resource_file_ending(name) == expected + + +@patch.dict(os.environ, {CLEANUP_DELETE_IMMEDIATELY_ENV_VAR: "True"}) +@patch("usdb_syncer.utils.send2trash") +def test_trash_or_delete_file_paths_cleanup_immediately( + send2trash_mock: MagicMock, tmp_path: Path +) -> None: + path_1 = tmp_path / "foo.txt" + path_2 = tmp_path / "foo" + path_1.write_text("I am foo") + path_2.mkdir() + assert path_1.is_file() + assert path_2.is_dir() + trash_or_delete_file_paths(paths=[path_1, path_2]) + send2trash_mock.assert_not_called() + assert not path_1.is_file() + assert not path_2.is_dir() + + +@patch.dict(os.environ, clear=True) +@patch("usdb_syncer.utils.send2trash") +def test_trash_or_delete_file_paths_trash(send2trash_mock: MagicMock) -> None: + path_1 = MagicMock() + path_2 = MagicMock() + trash_or_delete_file_paths(paths=[path_1, path_2]) + send2trash_mock.assert_called_once_with([path_1, path_2])