Add Empty Folder Cleaner library-maintenance job (corruption's request)
A maintenance job to keep the music library tidy — finds empty folders left behind by imports/relocations/deletions (empty artist/album dirs, or dirs holding only OS junk like .DS_Store/Thumbs.db) and removes them. Safety is the focus (deleting directories is destructive): - only TRULY empty folders are flagged — a folder with a cover image or any audio is never touched; only OS-junk files count as "no real content" (a setting), - the library root + symlinked dirs are never removed, - walks bottom-up so a parent left empty by its removable children cascades, - the apply handler RE-CHECKS emptiness at delete time, so a folder that gained a file between scan and apply is left alone. dir_is_removable + remove_empty_folder are pure/injectable seams. Wired through the job registry, repair_worker apply handler (_fix_empty_folder), fixable-types, and the findings UI. Opt-in (default off), weekly interval. Tests (10): removable decision (empty / real-file / surviving-subdir / junk-only / strict mode) + apply re-check (removes empty + junk, refuses content/root/symlink). Repair + integrity suites green; ruff clean.
This commit is contained in:
parent
4dd09ff48a
commit
8118a2c6bd
5 changed files with 302 additions and 2 deletions
|
|
@ -35,6 +35,7 @@ _JOB_MODULES = [
|
|||
'core.repair_jobs.missing_cover_art',
|
||||
'core.repair_jobs.missing_lyrics',
|
||||
'core.repair_jobs.replaygain_filler',
|
||||
'core.repair_jobs.empty_folder_cleaner',
|
||||
'core.repair_jobs.expired_download_cleaner',
|
||||
'core.repair_jobs.metadata_gap_filler',
|
||||
'core.repair_jobs.album_completeness',
|
||||
|
|
|
|||
195
core/repair_jobs/empty_folder_cleaner.py
Normal file
195
core/repair_jobs/empty_folder_cleaner.py
Normal file
|
|
@ -0,0 +1,195 @@
|
|||
"""Empty Folder Cleaner maintenance job (corruption's request).
|
||||
|
||||
After imports, relocations, and deletions, the music library accumulates empty
|
||||
artist/album folders (and folders left holding only OS junk like .DS_Store). This
|
||||
scans the library root and flags folders that are safe to remove, so the library
|
||||
stays tidy.
|
||||
|
||||
Safety is the whole point — deleting directories is destructive:
|
||||
- only TRULY empty folders (no real files) are ever flagged; a folder with a
|
||||
cover.jpg or any audio is never touched,
|
||||
- optionally folders holding *only* OS-junk files (.DS_Store, Thumbs.db, …),
|
||||
- the library root itself is never removed, nor symlinked directories,
|
||||
- it walks bottom-up so a parent left empty by its (removable) children cascades,
|
||||
- and the apply handler RE-CHECKS emptiness at delete time, so anything that
|
||||
gained a file between scan and apply is left alone.
|
||||
|
||||
``dir_is_removable`` is the pure decision seam — unit-tested independent of the FS.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Iterable, List
|
||||
|
||||
from core.repair_jobs import register_job
|
||||
from core.repair_jobs.base import JobContext, JobResult, RepairJob
|
||||
from utils.logging_config import get_logger
|
||||
|
||||
logger = get_logger("repair_jobs.empty_folder_cleaner")
|
||||
|
||||
# Files that don't count as real content — safe to delete along with the folder.
|
||||
JUNK_FILES = {'.ds_store', 'thumbs.db', 'desktop.ini', '.directory', 'album.nfo~'}
|
||||
|
||||
|
||||
def is_junk(name: str) -> bool:
|
||||
return (name or '').lower() in JUNK_FILES
|
||||
|
||||
|
||||
def dir_is_removable(files: Iterable[str], surviving_subdirs: Iterable[str],
|
||||
*, ignore_junk: bool = True) -> bool:
|
||||
"""Pure: is a directory safe to remove?
|
||||
|
||||
Removable iff it has **no surviving subdirectories** and **no real files** —
|
||||
where "no real files" means literally empty, or (when ``ignore_junk``) only
|
||||
OS-junk files. ``surviving_subdirs`` is the list of child dirs that are NOT
|
||||
themselves being removed (i.e. still hold content).
|
||||
"""
|
||||
if list(surviving_subdirs):
|
||||
return False
|
||||
files = list(files)
|
||||
if not files:
|
||||
return True
|
||||
if not ignore_junk:
|
||||
return False
|
||||
return all(is_junk(f) for f in files)
|
||||
|
||||
|
||||
@register_job
|
||||
class EmptyFolderCleanerJob(RepairJob):
|
||||
job_id = 'empty_folder_cleaner'
|
||||
display_name = 'Empty Folder Cleaner'
|
||||
description = 'Finds empty (or junk-only) folders in the library and removes them'
|
||||
help_text = (
|
||||
'Scans your music library for empty folders left behind after imports, '
|
||||
'relocations, and deletions — empty artist/album folders, or folders that '
|
||||
'hold only OS junk like .DS_Store / Thumbs.db.\n\n'
|
||||
'A finding is created for each. Applying one deletes the folder (after '
|
||||
're-checking it is still empty). Folders that contain any real file — a '
|
||||
'cover image, an audio track, anything — are never touched, the library '
|
||||
'root is never removed, and it cascades: a folder left empty once its '
|
||||
'empty children are removed is cleaned too.'
|
||||
)
|
||||
icon = 'repair-icon-folder'
|
||||
default_enabled = False
|
||||
default_interval_hours = 168 # weekly — empties accrue slowly
|
||||
default_settings = {'remove_junk_files': True}
|
||||
auto_fix = False
|
||||
|
||||
def scan(self, context: JobContext) -> JobResult:
|
||||
result = JobResult()
|
||||
|
||||
root = context.transfer_folder
|
||||
if not root or not os.path.isdir(root):
|
||||
logger.info("[Empty Folder Cleaner] library root not available — skipping")
|
||||
return result
|
||||
root = os.path.realpath(root)
|
||||
|
||||
ignore_junk = True
|
||||
try:
|
||||
if context.config_manager:
|
||||
ignore_junk = bool(context.config_manager.get(
|
||||
'repair.jobs.empty_folder_cleaner.remove_junk_files', True))
|
||||
except Exception: # noqa: S110 — setting read is best-effort; defaults to True
|
||||
pass
|
||||
|
||||
flagged = set() # dir paths we'd remove → a parent sees them as "gone"
|
||||
|
||||
# topdown=False ⇒ deepest first, so children are decided before parents.
|
||||
for dirpath, dirnames, filenames in os.walk(root, topdown=False):
|
||||
if context.check_stop():
|
||||
return result
|
||||
real = os.path.realpath(dirpath)
|
||||
if real == root:
|
||||
continue # never the library root itself
|
||||
if os.path.islink(dirpath):
|
||||
continue # don't delete symlinked dirs
|
||||
result.scanned += 1
|
||||
|
||||
surviving = [d for d in dirnames
|
||||
if os.path.join(dirpath, d) not in flagged]
|
||||
if not dir_is_removable(filenames, surviving, ignore_junk=ignore_junk):
|
||||
result.skipped += 1
|
||||
continue
|
||||
|
||||
flagged.add(dirpath)
|
||||
junk = [f for f in filenames if is_junk(f)]
|
||||
rel = os.path.relpath(dirpath, root)
|
||||
if context.report_progress:
|
||||
context.report_progress(log_line=f'Empty folder: {rel}', log_type='info')
|
||||
if context.create_finding:
|
||||
try:
|
||||
inserted = context.create_finding(
|
||||
job_id=self.job_id,
|
||||
finding_type='empty_folder',
|
||||
severity='info',
|
||||
entity_type='folder',
|
||||
entity_id=dirpath,
|
||||
file_path=dirpath,
|
||||
title=f'Empty folder: {os.path.basename(dirpath) or rel}',
|
||||
description=(f'"{rel}" holds no music'
|
||||
+ (f' (only {len(junk)} junk file(s))' if junk else '')
|
||||
+ ' — safe to remove.'),
|
||||
details={
|
||||
'folder_path': dirpath,
|
||||
'junk_files': junk,
|
||||
'remove_junk': ignore_junk,
|
||||
})
|
||||
if inserted:
|
||||
result.findings_created += 1
|
||||
else:
|
||||
result.findings_skipped_dedup += 1
|
||||
except Exception as e:
|
||||
logger.debug("[Empty Folder Cleaner] create finding failed for %s: %s", dirpath, e)
|
||||
result.errors += 1
|
||||
|
||||
logger.info("[Empty Folder Cleaner] %d folders scanned, %d empty flagged",
|
||||
result.scanned, result.findings_created)
|
||||
return result
|
||||
|
||||
def estimate_scope(self, context: JobContext) -> int:
|
||||
root = context.transfer_folder
|
||||
if not root or not os.path.isdir(root):
|
||||
return 0
|
||||
total = 0
|
||||
for _dp, dirnames, _f in os.walk(root):
|
||||
total += len(dirnames)
|
||||
return total
|
||||
|
||||
|
||||
def remove_empty_folder(folder_path: str, *, junk_files: List[str], remove_junk: bool,
|
||||
root: str, listdir, isdir, islink, remove_file, rmdir) -> dict:
|
||||
"""Pure-ish orchestration for the apply handler — RE-CHECKS the folder is still
|
||||
removable, then deletes any junk + the folder. Effects injected for testing.
|
||||
|
||||
Returns ``{'removed': bool, 'error': str|None}``. Refuses to touch the root, a
|
||||
symlink, a non-dir, or a folder that gained real content since the scan.
|
||||
"""
|
||||
if not folder_path or not isdir(folder_path):
|
||||
return {'removed': False, 'error': 'Folder no longer exists'}
|
||||
if islink(folder_path):
|
||||
return {'removed': False, 'error': 'Refusing to remove a symlinked folder'}
|
||||
if root and os.path.realpath(folder_path) == os.path.realpath(root):
|
||||
return {'removed': False, 'error': 'Refusing to remove the library root'}
|
||||
|
||||
# Re-check at apply time: only junk/empty now? (Anything else = leave it.)
|
||||
entries = list(listdir(folder_path))
|
||||
real_entries = [e for e in entries if not (remove_junk and is_junk(e))]
|
||||
if real_entries:
|
||||
return {'removed': False, 'error': 'Folder is no longer empty — left untouched'}
|
||||
|
||||
if remove_junk:
|
||||
for j in entries:
|
||||
if is_junk(j):
|
||||
try:
|
||||
remove_file(os.path.join(folder_path, j))
|
||||
except Exception: # noqa: S110 — junk best-effort; rmdir below fails loudly if blocked
|
||||
pass
|
||||
try:
|
||||
rmdir(folder_path)
|
||||
except Exception as e:
|
||||
return {'removed': False, 'error': f'Could not remove folder: {e}'}
|
||||
return {'removed': True, 'error': None}
|
||||
|
||||
|
||||
__all__ = ['dir_is_removable', 'is_junk', 'remove_empty_folder', 'EmptyFolderCleanerJob']
|
||||
|
|
@ -965,6 +965,7 @@ class RepairWorker:
|
|||
'missing_cover_art': self._fix_missing_cover_art,
|
||||
'missing_lyrics': self._fix_missing_lyrics,
|
||||
'missing_replaygain': self._fix_missing_replaygain,
|
||||
'empty_folder': self._fix_empty_folder,
|
||||
'expired_download': self._fix_expired_download,
|
||||
'metadata_gap': self._fix_metadata_gap,
|
||||
'duplicate_tracks': self._fix_duplicates,
|
||||
|
|
@ -1507,6 +1508,29 @@ class RepairWorker:
|
|||
return {'success': True, 'action': 'applied_replaygain',
|
||||
'message': f'Wrote ReplayGain ({gain_db:+.2f} dB)'}
|
||||
|
||||
def _fix_empty_folder(self, entity_type, entity_id, file_path, details):
|
||||
"""Apply an empty-folder finding: re-check the folder is still empty/junk-
|
||||
only (anything that gained a real file since the scan is left alone), then
|
||||
remove it. The library root + symlinked dirs are refused."""
|
||||
from core.repair_jobs.empty_folder_cleaner import remove_empty_folder
|
||||
raw = details.get('folder_path') or file_path
|
||||
if not raw:
|
||||
return {'success': False, 'error': 'No folder path in finding'}
|
||||
resolved = self._resolve_path(raw) if hasattr(self, '_resolve_path') else raw
|
||||
res = remove_empty_folder(
|
||||
resolved,
|
||||
junk_files=details.get('junk_files') or [],
|
||||
remove_junk=bool(details.get('remove_junk', True)),
|
||||
root=self.transfer_folder,
|
||||
listdir=os.listdir, isdir=os.path.isdir, islink=os.path.islink,
|
||||
remove_file=os.remove, rmdir=os.rmdir,
|
||||
)
|
||||
if not res.get('removed'):
|
||||
return {'success': False, 'error': res.get('error') or 'Could not remove folder'}
|
||||
_name = os.path.basename(resolved.rstrip('/\\')) or resolved
|
||||
return {'success': True, 'action': 'removed_empty_folder',
|
||||
'message': f'Removed empty folder: {_name}'}
|
||||
|
||||
def _fix_expired_download(self, entity_type, entity_id, file_path, details):
|
||||
"""Apply an expired-download finding: delete the file + library row +
|
||||
history entry, via the same helper the cleaner's auto mode uses."""
|
||||
|
|
@ -3312,7 +3336,7 @@ class RepairWorker:
|
|||
'album_mbid_mismatch',
|
||||
'album_tag_inconsistency',
|
||||
'incomplete_album', 'path_mismatch',
|
||||
'missing_lossy_copy', 'missing_replaygain',
|
||||
'missing_lossy_copy', 'missing_replaygain', 'empty_folder',
|
||||
'missing_discography_track', 'acoustid_mismatch')
|
||||
placeholders = ','.join(['?'] * len(fixable_types))
|
||||
where_parts = [f"finding_type IN ({placeholders})", "status = 'pending'"]
|
||||
|
|
|
|||
74
tests/test_empty_folder_cleaner.py
Normal file
74
tests/test_empty_folder_cleaner.py
Normal file
|
|
@ -0,0 +1,74 @@
|
|||
"""Empty Folder Cleaner (corruption's request) — pure removable decision + the
|
||||
apply handler's re-check safety (never deletes a folder that gained content, the
|
||||
library root, or a symlink)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
from core.repair_jobs.empty_folder_cleaner import dir_is_removable, remove_empty_folder, is_junk
|
||||
|
||||
|
||||
# ── pure decision ───────────────────────────────────────────────────────────
|
||||
def test_empty_dir_is_removable():
|
||||
assert dir_is_removable([], []) is True
|
||||
|
||||
|
||||
def test_dir_with_real_file_is_not_removable():
|
||||
assert dir_is_removable(['cover.jpg'], []) is False
|
||||
assert dir_is_removable(['song.flac'], []) is False
|
||||
|
||||
|
||||
def test_dir_with_surviving_subdir_is_not_removable():
|
||||
assert dir_is_removable([], ['Album']) is False
|
||||
|
||||
|
||||
def test_junk_only_dir_removable_when_ignore_junk():
|
||||
assert dir_is_removable(['.DS_Store', 'Thumbs.db'], []) is True
|
||||
assert dir_is_removable(['.DS_Store'], [], ignore_junk=False) is False # strict mode keeps it
|
||||
|
||||
|
||||
def test_junk_plus_real_file_not_removable():
|
||||
assert dir_is_removable(['.DS_Store', 'cover.jpg'], []) is False
|
||||
|
||||
|
||||
def test_is_junk():
|
||||
assert is_junk('.DS_Store') and is_junk('thumbs.db') and not is_junk('cover.jpg')
|
||||
|
||||
|
||||
# ── apply re-check (real FS) ────────────────────────────────────────────────
|
||||
def _fx():
|
||||
return dict(listdir=os.listdir, isdir=os.path.isdir, islink=os.path.islink,
|
||||
remove_file=os.remove, rmdir=os.rmdir)
|
||||
|
||||
|
||||
def test_apply_removes_truly_empty_folder(tmp_path):
|
||||
root = tmp_path / 'lib'; root.mkdir()
|
||||
empty = root / 'Artist' / 'Album'; empty.mkdir(parents=True)
|
||||
res = remove_empty_folder(str(empty), junk_files=[], remove_junk=True, root=str(root), **_fx())
|
||||
assert res['removed'] is True
|
||||
assert not empty.exists()
|
||||
|
||||
|
||||
def test_apply_deletes_junk_then_folder(tmp_path):
|
||||
root = tmp_path / 'lib'; root.mkdir()
|
||||
d = root / 'Empty'; d.mkdir()
|
||||
(d / '.DS_Store').write_text('x')
|
||||
res = remove_empty_folder(str(d), junk_files=['.DS_Store'], remove_junk=True, root=str(root), **_fx())
|
||||
assert res['removed'] is True and not d.exists()
|
||||
|
||||
|
||||
def test_apply_refuses_folder_that_gained_a_file(tmp_path):
|
||||
root = tmp_path / 'lib'; root.mkdir()
|
||||
d = root / 'NowFull'; d.mkdir()
|
||||
(d / 'new.flac').write_text('audio') # appeared between scan and apply
|
||||
res = remove_empty_folder(str(d), junk_files=[], remove_junk=True, root=str(root), **_fx())
|
||||
assert res['removed'] is False and 'no longer empty' in res['error'].lower()
|
||||
assert d.exists() # left untouched
|
||||
|
||||
|
||||
def test_apply_refuses_library_root(tmp_path):
|
||||
root = tmp_path / 'lib'; root.mkdir()
|
||||
res = remove_empty_folder(str(root), junk_files=[], remove_junk=True, root=str(root), **_fx())
|
||||
assert res['removed'] is False and 'root' in res['error'].lower()
|
||||
assert root.exists()
|
||||
|
|
@ -2735,7 +2735,7 @@ async function loadRepairFindings() {
|
|||
path_mismatch: 'Path Mismatch', metadata_gap: 'Missing Metadata',
|
||||
missing_cover_art: 'Missing Art', track_number_mismatch: 'Track Number',
|
||||
missing_lyrics: 'Missing Lyrics', expired_download: 'Expired',
|
||||
missing_replaygain: 'No ReplayGain',
|
||||
missing_replaygain: 'No ReplayGain', empty_folder: 'Empty Folder',
|
||||
missing_lossy_copy: 'No Lossy Copy', library_retag: 'Re-tag'
|
||||
};
|
||||
|
||||
|
|
@ -2747,6 +2747,7 @@ async function loadRepairFindings() {
|
|||
missing_cover_art: 'Apply Art',
|
||||
missing_lyrics: 'Apply Lyrics',
|
||||
missing_replaygain: 'Apply RG',
|
||||
empty_folder: 'Delete Folder',
|
||||
expired_download: 'Delete',
|
||||
metadata_gap: 'Apply',
|
||||
duplicate_tracks: 'Keep Best',
|
||||
|
|
@ -3145,6 +3146,11 @@ function _renderFindingDetail(f) {
|
|||
if (d.artist) rows.push(['Artist', d.artist]);
|
||||
return _gridRows(rows);
|
||||
|
||||
case 'empty_folder':
|
||||
if (d.folder_path) rows.push(['Folder', d.folder_path, 'path']);
|
||||
if (d.junk_files && d.junk_files.length) rows.push(['Junk files', d.junk_files.join(', ')]);
|
||||
return _gridRows(rows);
|
||||
|
||||
case 'expired_download':
|
||||
if (d.title) rows.push(['Track', d.title]);
|
||||
if (d.artist) rows.push(['Artist', d.artist]);
|
||||
|
|
|
|||
Loading…
Reference in a new issue