# "-ar 44100" for 44.1 kHz audio output.
# "-codec:v copy" to copy the integrated image from source
# instead of converting it to png (default encoder).
- commonFlags: list[str] = ["-loglevel", "error", "-n",
+ commonFlags: list[str] = ["-loglevel", "warning", "-n",
"-ar", "44100", "-codec:v", "copy"]
# "-sample_fmt s16" for 16 bits audio output.
class Stats:
def __init__(self) -> None:
self.errors: int = 0
+ self.warnings: int = 0
self.skipped: int = 0
self.duplicates: int = 0
self.removed: int = 0
def add(self, stats) -> None:
self.errors += stats.errors
+ self.warnings += stats.warnings
self.skipped += stats.skipped
self.duplicates += stats.duplicates
self.removed += stats.removed
def _check(self, playlistDir: str, libraryDir: str) -> None:
if not os.path.isdir(playlistDir):
- self._log("Error: Invalid playlist directory '"
- + playlistDir + "'.")
+ self._log("Error: Invalid playlist directory \""
+ + playlistDir + "\".")
return
if not os.path.isdir(libraryDir):
- self._log("Error: Invalid library directory '"
- + libraryDir + "'.")
+ self._log("Error: Invalid library directory \""
+ + libraryDir + "\".")
return
# List of albums/directories from library directory.
track.trackFile)
if not os.path.isfile(trackFilePath):
- self._log("Warning: '" + trackFilePath
- + "' in playlist '" + libPlaylist.name + "' not found.")
+ self._log("Warning: \"" + trackFilePath
+ + "\" in playlist \"" + libPlaylist.name + "\" not found.")
# If an album contains a track from any playlist remove it.
if track.trackDir in albums:
def _convert(self, convertParams: ConvertParams) -> None:
if not os.path.isdir(convertParams.playlistDir):
- self._log("Error: Invalid playlist directory '"
- + convertParams.playlistDir + "'.")
+ self._log("Error: Invalid playlist directory \""
+ + convertParams.playlistDir + "\".")
return
if not os.path.isdir(convertParams.targetDir):
- self._log("Error: Invalid target directory '"
- + convertParams.targetDir + "'.")
+ self._log("Error: Invalid target directory \""
+ + convertParams.targetDir + "\".")
return
convertProfile: ConvertProfile = ConvertProfile()
if not convertProfile.init(convertParams.convertProfile):
- self._log("Error: Invalid convert profile '"
- + convertParams.convertProfile + "'.")
+ self._log("Error: Invalid convert profile \""
+ + convertParams.convertProfile + "\".")
return
if not self.checkConverter(convertParams.converterPath,
self._CONVERTER_CHECK):
- self._log("Error: Could not start converter '"
- + convertParams.converterPath + "'.")
+ self._log("Error: Could not start converter \""
+ + convertParams.converterPath + "\".")
return
if convertParams.jobs < 1:
track.trackFile)
if not os.path.isfile(trackFilePath):
- self._log("Error: '" + trackFilePath + "' in playlist '"
- + playlist.name + "' is not a file.")
+ self._log("Error: \"" + trackFilePath + "\" in playlist \""
+ + playlist.name + "\" is not a file.")
return
# Default task type: copy.
taskType = "convert"
# Do nothing if extension not found in profile.
elif not convertProfile.isCopyExtension(targetExt):
- self._log("Warning: Unsupported track file '"
- + trackFilePath + "' in playlist '"
- + playlist.name + "' for profile '"
- + convertProfile.name + "'.")
+ self._log("Warning: Unsupported track file \""
+ + trackFilePath + "\" in playlist \""
+ + playlist.name + "\" for profile \""
+ + convertProfile.name + "\".")
+ stats.warnings += 1
continue
targetFileName: str = targetName
unmatchedFiles.remove(targetFile)
if not os.path.isfile(targetFilePath):
- self._log("Error: Existing target element '"
- + targetFilePath + "' is not a valid file.")
+ self._log("Error: Existing target element \""
+ + targetFilePath + "\" is not a valid file.")
return
elif self.isFileChanged(trackFilePath, targetFilePath):
processedFiles.add(targetFile)
# Delete non matched playlists.
for f in unmatchedPlaylistFiles:
- self._log("Removing non matched playlist '" + f + "'.")
+ self._log("Removing non matched playlist \"" + f + "\".")
os.remove(os.path.join(convertParams.targetDir, f))
# Delete non matched tracks.
for f in unmatchedFiles:
- self._log("Removing non matched file '" + f + "'.")
+ self._log("Removing non matched file \"" + f + "\".")
os.remove(os.path.join(convertParams.targetDir, f))
stats.removed += 1
self._log("From " + str(trackCount) + " tracks in "
+ str(len(libPlaylists)) + " playlists:"
+ " Errors " + str(stats.errors) + ","
+ + " Warnings " + str(stats.warnings) + ","
+ " Skipped " + str(stats.skipped) + ","
+ " Duplicates " + str(stats.duplicates) + ","
+ " Removed " + str(stats.removed) + ","
+ " Copied " + str(stats.copied) + ".")
# Main function started by threads. Process queue elements
- # until 'None' is found or stop signal is received.
+ # until "None" is found or stop signal is received.
def _processQueue(self, q: queue.SimpleQueue,
converterPath: str, stats: Stats) -> None:
self._statsLock.acquire()
localStats: Stats = Stats()
- # Process elements until getting 'None'.
+ # Process elements until getting "None".
while True:
task: Task = q.get()
def _processTask(self, task: Task, converterPath: str, totalTasks: int,
stats: Stats) -> None:
if not os.path.isfile(task.inputFilePath):
- self._log("Error: Playlist element '" + task.inputFilePath
- + "' is not a file.")
+ self._log("Error: Playlist element \"" + task.inputFilePath
+ + "\" is not a file.")
stats.errors += 1
return
for f in [task.outputFilePath, outputFilePathWip]:
if os.path.isfile(f):
- self._log("Removing outdated file '" + f + "'.")
+ self._log("Removing outdated file \"" + f + "\".")
os.remove(f)
elif os.path.exists(f):
- self._log("Error: Existing non file element'" + f + "'.")
+ self._log("Error: Existing non file element \"" + f + "\".")
stats.errors += 1
return
if task.taskType == "copy":
self._log(str(task.taskId) + "/" + str(totalTasks)
- + ": Copying '" + task.inputFilePath
- + "' into '" + task.outputFilePath + "'.")
+ + ": Copying \"" + task.inputFilePath
+ + "\" into \"" + task.outputFilePath + "\".")
self.copyFile(task.inputFilePath, outputFilePathWip)
os.rename(outputFilePathWip, task.outputFilePath)
stats.copied += 1
+ [task.inputFilePath] + task.converterArgs
+ [outputFilePathWip])
self._log(str(task.taskId) + "/" + str(totalTasks)
- + ": Converting '" + task.inputFilePath
- + "' to '" + task.outputFilePath + "'.")
+ + ": Converting \"" + task.inputFilePath
+ + "\" to \"" + task.outputFilePath + "\".")
result: subprocess.CompletedProcess = subprocess.run(
command, capture_output=True, text=True)
- if result.returncode == 0:
+ if result.returncode != 0:
+ self._log("Error from converter on file \"" + task.inputFilePath + "\": " + result.stderr)
+ stats.errors += 1
+ else:
+ if len(result.stderr) != 0:
+ self._log("Warning from converter on file \""
+ + task.inputFilePath + "\": " + result.stderr)
+ stats.warnings += 1
+
os.rename(outputFilePathWip, task.outputFilePath)
stats.converted += 1
if update:
stats.updated +=1
- else:
- self._log("Error from converter: " + result.stderr)
- stats.errors += 1
+
else: # Unknown task
- self._log("Error: Skipping non valid task type '" + task.taskType
- + "' file '" + task.inputFilePath + "'.")
+ self._log("Error: Skipping non valid task type \"" + task.taskType
+ + "\" file \"" + task.inputFilePath + "\".")
stats.errors += 1
# Create library playlists from a base directory.
libPlaylists: dict[str, LibraryPlaylist] = {}
if not os.path.isdir(directory):
- self._log("Warning: '" + directory + "' is not a valid directory.")
+ self._log("Warning: \"" + directory + "\" is not a valid directory.")
else:
for element in os.listdir(directory):
path: str = os.path.join(directory, element)
playlistName: str = self.trimFileExtension(element)
if playlistName in libPlaylists:
- self._log("Warning: Duplicate playlist name '"
- + playlistName + "'.")
+ self._log("Warning: Duplicate playlist name \""
+ + playlistName + "\".")
else:
libPlaylist: LibraryPlaylist = self._createLibraryPlaylistFromFile(path)
if len(libPlaylist.tracks) == 0:
- self._log("Warning: Empty library playlist '"
- + libPlaylist.name + "'.")
+ self._log("Warning: Empty library playlist \""
+ + libPlaylist.name + "\".")
else:
libPlaylists[playlistName] = libPlaylist
elif self.getFileExtension(filePath).lower() == ".xspf":
self._loadTracksFromXspf(filePath, playlist)
else:
- self._log("Warning: Unsupported playlist extension for file '"
- + filePath + "'.")
+ self._log("Warning: Unsupported playlist extension for file \""
+ + filePath + "\".")
return playlist
string: str = line.rstrip()
# Starting with '#' means it's a comment.
- if len(string) == 0 or string[0] == "#":
+ if len(string) == 0 or string[0] == '#':
continue
playlist.add(os.path.dirname(string), os.path.basename(string))
os.remove(repPlFilePath)
return
- self._log("Updating outdated playlist file '" + oriPlFilePath + "'.")
+ self._log("Updating outdated playlist file \"" + oriPlFilePath + "\".")
os.remove(oriPlFilePath)
else:
- self._log("Adding playlist file '" + oriPlFilePath + "'.")
+ self._log("Adding playlist file \"" + oriPlFilePath + "\".")
os.rename(repPlFilePath, oriPlFilePath)