Backup_Repos/beets
1
0
Fork 0
mirror of https://github.com/beetbox/beets.git synced 2025-12-24 01:25:47 +01:00
Code Issues Projects Releases Wiki Activity

Merge pull request #1127 from geigerzaehler/empty-dir-warning

Browse source 
Warn user when no files are imported from a directory
This commit is contained in:
Adrian Sampson 2014-12-01 11:30:26 -08:00
commit 4670a8485f
2 changed files with 72 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

103
beets/importer.py
View file

@ -949,20 +949,52 @@ class ArchiveImportTask(SentinelImportTask):
class ImportTaskFactory(object):
"""Create album and singleton import tasks from paths for toppaths
in session.
"""Create album and singleton import tasks for all media files in a
directory or path.
The `singleton()` and `album()` methods accept paths and return
instances of `SingletonImportTask` and `ImportTask`, respectively.
`None` is returned if either no media file items could be created
from the paths or if the paths have already been imported. In both
cases it logs messages.
Depending on the session's 'flat' and 'singleton' configuration, it
groups all media files contained in `toppath` into singleton or
album import tasks.
"""
def __init__(self, toppath, session):
self.toppath = toppath
self.session = session
self.skipped = 0
def tasks(self):
"""Yield all import tasks for `self.toppath`.
The behavior is configured by the session's 'flat', and
'singleton' flags.
"""
for dirs, paths in self.paths():
if self.session.config['singletons']:
for path in paths:
task = self.singleton(path)
if task:
yield task
yield self.sentinel(dirs)
else:
task = self.album(paths, dirs)
if task:
yield task
def paths(self):
"""Walk `self.toppath` and yield pairs of directory lists and
path lists.
"""
if not os.path.isdir(syspath(self.toppath)):
yield ([self.toppath], [self.toppath])
elif self.session.config['flat']:
paths = []
for dirs, paths_in_dir in albums_in_dir(self.toppath):
paths += paths_in_dir
yield ([self.toppath], paths)
else:
for dirs, paths in albums_in_dir(self.toppath):
yield (dirs, paths)
def singleton(self, path):
if self.session.already_imported(self.toppath, [path]):
log.debug(u'Skipping previously-imported path: {0}'
@ -976,17 +1008,16 @@ class ImportTaskFactory(object):
else:
return None
def album(self, paths, dir=None):
def album(self, paths, dirs=None):
"""Return `ImportTask` with all media files from paths.
`dir` is a common parent directory of all paths.
`dirs` is a list of parent directories used to record already
imported albums.
"""
if not paths:
return None
if dir:
dirs = [dir]
else:
if dirs is None:
dirs = list(set(os.path.dirname(p) for p in paths))
if self.session.already_imported(self.toppath, dirs):
@ -1038,10 +1069,9 @@ def read_tasks(session):
"""
skipped = 0
for toppath in session.paths:
task_factory = ImportTaskFactory(toppath, session)
# Determine if we want to resume import of the toppath
session.ask_resume(toppath)
user_toppath = toppath
# Extract archives.
archive_task = None
@ -1063,42 +1093,11 @@ def read_tasks(session):
# Continue reading albums from the extracted directory.
toppath = archive_task.toppath
# Check whether the path is to a file.
if not os.path.isdir(syspath(toppath)):
if session.config['singletons']:
task = task_factory.singleton(toppath)
else:
task = task_factory.album([toppath], dir=toppath)
if task:
yield task
yield task_factory.sentinel()
continue
# A flat album import merges all items into one album.
if session.config['flat'] and not session.config['singletons']:
paths = []
for _, item_paths in albums_in_dir(toppath):
paths += item_paths
task = task_factory.album(paths)
if task:
yield task
yield task_factory.sentinel()
continue
# Produce paths under this directory.
for dirs, paths in albums_in_dir(toppath):
if session.config['singletons']:
for path in paths:
task = task_factory.singleton(path)
if task:
yield task
yield task_factory.sentinel(dirs)
else:
task = task_factory.album(paths)
if task:
yield task
task_factory = ImportTaskFactory(toppath, session)
imported = False
for t in task_factory.tasks():
imported |= not t.skip
yield t
# Indicate the directory is finished.
# FIXME hack to delete extracted archives
@ -1107,6 +1106,10 @@ def read_tasks(session):
else:
yield archive_task
if not imported:
log.warn(u'No files imported from {0}'
.format(displayable_path(user_toppath)))
# Show skipped directories.
if skipped:
log.info(u'Skipped {0} directories.'.format(skipped))

20
test/test_importer.py
View file

@ -25,7 +25,7 @@ from mock import patch
import _common
from _common import unittest
from helper import TestImportSession, TestHelper, has_program
from helper import TestImportSession, TestHelper, has_program, capture_log
from beets import importer
from beets.importer import albums_in_dir
from beets.mediafile import MediaFile
@ -600,6 +600,24 @@ class ImportTest(_common.TestCase, ImportHelper):
self.importer.run()
self.assertEqual(len(self.lib.items()), 1)
def test_empty_directory_warning(self):
import_dir = os.path.join(self.temp_dir, 'empty')
self.touch('non-audio', dir=import_dir)
self._setup_import_session(import_dir=import_dir)
with capture_log() as logs:
self.importer.run()
self.assertIn('No files imported from {0}'.format(import_dir), logs)
def test_empty_directory_singleton_warning(self):
import_dir = os.path.join(self.temp_dir, 'empty')
self.touch('non-audio', dir=import_dir)
self._setup_import_session(import_dir=import_dir, singletons=True)
with capture_log() as logs:
self.importer.run()
self.assertIn('No files imported from {0}'.format(import_dir), logs)
class ImportTracksTest(_common.TestCase, ImportHelper):
"""Test TRACKS and APPLY choice.