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.
libPlaylist = self._createLibraryPlaylistFromFile(os.path.join(playlistDir,
playlistFile))
for track in libPlaylist.tracks:
- trackFilePath = os.path.join(track.trackDir,
- track.trackFile)
+ trackFilePath = os.path.join(track.trackDir, track.trackFile)
if not os.path.isfile(trackFilePath):
self._log("Warning: \"" + trackFilePath
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 + "\".")
+ if not self.checkConverter(convertParams.converterPath, self._CONVERTER_CHECK):
+ self._log("Error: Could not start converter \"" + convertParams.converterPath + "\".")
return
if convertParams.jobs < 1:
# Build existing file sets from target directory.
for tFile in os.listdir(convertParams.targetDir):
- if not os.path.isfile(os.path.join(convertParams.targetDir,
- tFile)):
+ if not os.path.isfile(os.path.join(convertParams.targetDir, tFile)):
continue
tExt: str = self.getFileExtension(tFile).lower()
- if (convertProfile.isTargetExtension(tExt)
- or convertProfile.isCopyExtension(tExt)):
+ if (convertProfile.isTargetExtension(tExt) or convertProfile.isCopyExtension(tExt)):
unmatchedFiles.add(tFile)
elif tExt == self._TARGET_PLAYLIST_EXT:
unmatchedPlaylistFiles.add(tFile)
# Build task queue and update playlist files.
for p in libPlaylists:
- if self._getStopSignal() == True:
+ if self._getStopSignal():
self._log("Stopped.")
return
playlist: LibraryPlaylist = libPlaylists[p]
- targetPlaylistFile: str = (playlist.name
- + self._TARGET_PLAYLIST_EXT)
+ targetPlaylistFile: str = (playlist.name + self._TARGET_PLAYLIST_EXT)
targetFileList: list[str] = []
if targetPlaylistFile in unmatchedPlaylistFiles:
unmatchedPlaylistFiles.remove(targetPlaylistFile)
for track in playlist.tracks:
- trackFilePath: str = os.path.join(track.trackDir,
- track.trackFile)
+ trackFilePath: str = os.path.join(track.trackDir, track.trackFile)
if not os.path.isfile(trackFilePath):
self._log("Error: \"" + trackFilePath + "\" in playlist \""
targetFileName = self.trimTrackFileName(targetFileName)
targetFile: str = targetFileName + targetExt
- targetFilePath: str = os.path.join(convertParams.targetDir,
- targetFile)
+ targetFilePath: str = os.path.join(convertParams.targetDir, targetFile)
targetFileList.append(targetFile)
# Main function started by threads. Process queue elements
# until "None" is found or stop signal is received.
- def _processQueue(self, q: queue.SimpleQueue,
- converterPath: str, stats: Stats) -> None:
+ def _processQueue(self, q: queue.SimpleQueue, converterPath: str, stats: Stats) -> None:
self._statsLock.acquire()
totalTasks: int = stats.tasks
self._statsLock.release()
self._statsLock.release()
# Copy or convert a music file.
- def _processTask(self, task: Task, converterPath: str, totalTasks: int,
- stats: Stats) -> None:
+ 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
self._log(str(task.taskId) + "/" + str(totalTasks)
+ ": Copying \"" + task.inputFilePath
+ "\" into \"" + task.outputFilePath + "\".")
- self.copyFile(task.inputFilePath, outputFilePathWip)
- os.rename(outputFilePathWip, task.outputFilePath)
+
+ try:
+ shutil.copy(task.inputFilePath, outputFilePathWip)
+ os.rename(outputFilePathWip, task.outputFilePath)
+ except:
+ self._log("Error copying file \"" + task.inputFilePath
+ + "\" into \"" + task.outputFilePath + "\".")
+ stats.errors += 1
+ return
+
stats.copied += 1
if update:
stats.updated +=1
elif task.taskType == "convert":
- command: list[str] = ([converterPath] + [self._CONVERTER_IN]
- + [task.inputFilePath] + task.converterArgs
- + [outputFilePathWip])
self._log(str(task.taskId) + "/" + str(totalTasks)
+ ": Converting \"" + task.inputFilePath
+ "\" to \"" + task.outputFilePath + "\".")
+
+ command: list[str] = ([converterPath] + [self._CONVERTER_IN]
+ + [task.inputFilePath] + task.converterArgs
+ + [outputFilePathWip])
result: subprocess.CompletedProcess = subprocess.run(
command, capture_output=True, text=True)
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
+ return
+ try:
os.rename(outputFilePathWip, task.outputFilePath)
- stats.converted += 1
+ except:
+ self._log("Error converting file \"" + task.inputFilePath
+ + "\" into \"" + task.outputFilePath + "\".")
+ stats.errors += 1
+ return
+
+ if len(result.stderr) != 0:
+ self._log("Warning from converter on file \""
+ + task.inputFilePath + "\": " + result.stderr)
+ stats.warnings += 1
- if update:
- stats.updated +=1
+ stats.converted += 1
+ if update:
+ stats.updated +=1
else: # Unknown task
self._log("Error: Skipping non valid task type \"" + task.taskType
+ "\" file \"" + task.inputFilePath + "\".")
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
# Read m3u playlist file to get all its tracks.
def _loadTracksFromM3u(self, filePath: str, playlist: LibraryPlaylist) -> None:
- m3uFile: typing.IO[str] = self.openFile(filePath, "rt")
+ m3uFile: typing.IO[str] = open(filePath, "rt", encoding="utf8")
# Each line is an string path.
for line in m3uFile:
if decodedString.startswith("file://"):
decodedString = decodedString[7:]
- playlist.add(os.path.dirname(decodedString),
- os.path.basename(decodedString))
+ playlist.add(os.path.dirname(decodedString), os.path.basename(decodedString))
# Create or update playlist file from a directory.
def _updatePlaylistFile(self, playlist: list[str], playlistPath: str) -> None:
# Build a m3u file from a playlist.
def _buildM3uPlaylistFile(self, playlist: list[str], filePath: str) -> None:
- plFile: typing.IO[str] = self.openFile(filePath, "wt")
+ plFile: typing.IO[str] = open(filePath, "wt", encoding="utf8")
# Build new playlist file.
for track in playlist:
def getFileExtension(self, filePath: str) -> str:
return os.path.splitext(filePath)[1]
- def openFile(self, path: str, args: str) -> typing.IO[str]:
- return open(path, args, encoding="utf8")
-
- def makeDir(self, dirPath: str) -> None:
- os.makedirs(dirPath, exist_ok=True)
+ # Compare last modification date.
+ def isFileChanged(self, sourceFilePath: str, targetFilePath: str) -> bool:
+ return os.path.getmtime(sourceFilePath) > os.path.getmtime(targetFilePath)
- def copyFile(self, oriFilePath: str, newFilePath: str) -> None:
- shutil.copy(oriFilePath, newFilePath)
+ def isFileIdentical(self, firstFilePath: str, secondFilePath: str) -> bool:
+ return filecmp.cmp(firstFilePath, secondFilePath, shallow=False)
- # Compare last modification date.
- def isFileChanged(self, sourceFilePath: str,
- targetFilePath: str) -> bool:
- return (os.path.getmtime(sourceFilePath)
- > os.path.getmtime(targetFilePath))
-
- def isFileIdentical(self, firstFilePath: str,
- secondFilePath: str) -> bool:
- return filecmp.cmp(firstFilePath, secondFilePath,
- shallow=False)
-
- def checkConverter(self, converterPath: str,
- converterArg: str) -> bool:
+ def checkConverter(self, converterPath: str, converterArg: str) -> bool:
try:
return subprocess.run([converterPath, converterArg],
stdout=subprocess.DEVNULL,
self.runButton: tk.Button = tk.Button(self.runFrame,
command=self.run, text="Run Configuration")
- self.cancelButton: tk.Button = tk.Button(self.runFrame,
- command=self.cancel, text="Cancel")
+ self.cancelButton: tk.Button = tk.Button(self.runFrame, command=self.cancel, text="Cancel")
self.cancelButton["state"] = "disabled"
- self.actionLabel: tk.Label = tk.Label(self.runFrame,
- text="Action")
+ self.actionLabel: tk.Label = tk.Label(self.runFrame, text="Action")
self.action: tk.StringVar = tk.StringVar(self.runFrame)
self.action.set("convert")
self.actionOption: tk.OptionMenu = tk.OptionMenu(self.runFrame,
self.action, "convert", "check")
self.actionOption.config(width=8)
- self.profileLabel: tk.Label = tk.Label(self.runFrame,
- text="Profile")
+ self.profileLabel: tk.Label = tk.Label(self.runFrame, text="Profile")
self.profile: tk.StringVar = tk.StringVar(self.runFrame)
self.profile.set("mp3")
self.profileOption: tk.OptionMenu = tk.OptionMenu(self.runFrame,
self.trimCheck: tk.Checkbutton = tk.Checkbutton(self.runFrame,
text="Trim track N", variable=self.trim)
- self.jobsLabel: tk.Label = tk.Label(self.runFrame,
- text="Jobs")
+ self.jobsLabel: tk.Label = tk.Label(self.runFrame, text="Jobs")
self.jobs: tk.StringVar = tk.StringVar(value="1")
self.jobsOption: tk.OptionMenu = tk.OptionMenu(self.runFrame,
self.jobs, "1", "2", "3", "4", "5", "6", "7", "8", "9", "10")
description="Music Library Converter",
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
- parser.add_argument("playlist-dir", help="Playlist directory",
- type=str)
- parser.add_argument("target-dir", help="Target directory",
- type=str)
- parser.add_argument("--profile", "-p", help="Conversion profile",
- type=str, default="mp3")
+ parser.add_argument("playlist-dir", help="Playlist directory", type=str)
+ parser.add_argument("target-dir", help="Target directory", type=str)
+ parser.add_argument("--profile", "-p", help="Conversion profile", type=str, default="mp3")
parser.add_argument("--converter", "-c", help="Converter path",
type=str, default=CONVERTER_NAME)
parser.add_argument("--trim", "-t", help="Trim track numbers in file names",
action="store_true")
- parser.add_argument("--jobs", "-j", help="Number of concurrent jobs",
- type=int, default=1)
+ parser.add_argument("--jobs", "-j", help="Number of concurrent jobs", type=int, default=1)
parser.add_argument("--check", help="Check mode", action="store_true")
args: dict = vars(parser.parse_args())
projects / converter.git / commitdiff