122 lines
4.2 KiB
Python
Executable file
122 lines
4.2 KiB
Python
Executable file
#!/usr/bin/env python3
|
||
|
||
import os
|
||
import subprocess
|
||
import progressbar
|
||
import logging
|
||
import coloredlogs
|
||
|
||
coloredlogs.install(level='DEBUG', fmt='%(levelname)s %(message)s')
|
||
log = logging.getLogger()
|
||
|
||
# Constants
|
||
SOURCE_FOLDER = os.path.join(os.path.expanduser("~"), "Musique")
|
||
OUTPUT_FOLDER = os.path.join(os.path.expanduser("~"), ".MusiqueCompressed")
|
||
CONVERSIONS = {"flac": "opus"}
|
||
FORBIDDEN_EXTENSIONS = ["jpg", "pdf", "ffs_db"]
|
||
FORGIVEN_FILENAMES = ["cover.jpg"]
|
||
IGNORED_EMPTY_FOLDER = [".stfolder"]
|
||
|
||
# TODO FEAT Make the directory structure the same as the base one and
|
||
# remove IGNORED_EMPTY_FOLDER variable
|
||
|
||
# Listing files
|
||
log.info("Listing files")
|
||
sourceFiles = dict()
|
||
for root, dirs, files in os.walk(SOURCE_FOLDER):
|
||
for f in files:
|
||
fullPath = os.path.join(root, f)
|
||
path = os.path.relpath(fullPath, SOURCE_FOLDER)
|
||
sourceFiles[path] = os.path.getctime(fullPath)
|
||
|
||
outputFiles = dict()
|
||
for root, dirs, files in os.walk(OUTPUT_FOLDER):
|
||
for f in files:
|
||
fullPath = os.path.join(root, f)
|
||
path = os.path.relpath(fullPath, OUTPUT_FOLDER)
|
||
outputFiles[path] = os.path.getctime(fullPath)
|
||
|
||
# Sorting files
|
||
remainingConversions = dict()
|
||
extraFiles = set(outputFiles.keys())
|
||
|
||
|
||
def convertPath(path):
|
||
filename, extension = os.path.splitext(path)
|
||
extension = extension[1:].lower()
|
||
# If the extension isn't allowed
|
||
if extension in FORBIDDEN_EXTENSIONS:
|
||
basename = os.path.basename(path)
|
||
# And the filename is not an exception
|
||
if basename not in FORGIVEN_FILENAMES:
|
||
# This file shouldn't be copied nor converted
|
||
return False
|
||
# If this needs a conversion
|
||
elif extension in CONVERSIONS:
|
||
extension = CONVERSIONS[extension]
|
||
return filename + "." + extension
|
||
# In all other case, this is a simple copy
|
||
return path
|
||
|
||
|
||
log.info("Determining action over {} files".format(len(sourceFiles)))
|
||
for sourceFile in sourceFiles:
|
||
outputFile = convertPath(sourceFile)
|
||
# If the file should not be converted, do nothing
|
||
if outputFile == False:
|
||
continue
|
||
# If the file already has something as an output
|
||
elif outputFile in outputFiles:
|
||
extraFiles.remove(outputFile)
|
||
# If the output file is newer than the source file, do not initiate a
|
||
# conversion
|
||
if outputFiles[outputFile] >= sourceFiles[sourceFile]:
|
||
continue
|
||
# If the file needs to be converted, do it
|
||
remainingConversions[sourceFile] = outputFile
|
||
|
||
log.debug("{} actions will need to be taken".format(len(remainingConversions)))
|
||
log.info("Copying files that do not require a conversion")
|
||
conversions = set()
|
||
for sourceFile in remainingConversions:
|
||
outputFile = remainingConversions[sourceFile]
|
||
|
||
# Creating folder if it doesn't exists
|
||
fullOutputFile = os.path.join(OUTPUT_FOLDER, outputFile)
|
||
fullOutputDir = os.path.dirname(fullOutputFile)
|
||
os.makedirs(fullOutputDir, exist_ok=True)
|
||
|
||
# Converting
|
||
fullSourceFile = os.path.join(SOURCE_FOLDER, sourceFile)
|
||
if sourceFile == outputFile:
|
||
log.debug('{} → {}'.format(fullSourceFile, fullOutputFile))
|
||
if os.path.isfile(fullOutputFile):
|
||
os.remove(fullOutputFile)
|
||
os.link(fullSourceFile, fullOutputFile)
|
||
else:
|
||
conversions.add((fullSourceFile, fullOutputFile))
|
||
|
||
log.info("Removing extra files")
|
||
for extraFile in extraFiles:
|
||
fullExtraFile = os.path.join(OUTPUT_FOLDER, extraFile)
|
||
log.debug('× {}'.format(fullExtraFile))
|
||
os.remove(fullExtraFile)
|
||
|
||
log.info("Listing files that will be converted")
|
||
for fullSourceFile, fullOutputFile in conversions:
|
||
log.debug('{} ⇒ {}'.format(fullSourceFile, fullOutputFile))
|
||
|
||
log.info("Converting files")
|
||
for fullSourceFile, fullOutputFile in progressbar.progressbar(conversions):
|
||
cmd = ["ffmpeg", "-y", "-i", fullSourceFile, "-c:a", "libopus",
|
||
"-movflags", "+faststart", "-b:a", "128k", "-vbr", "on",
|
||
"-compression_level", "10", fullOutputFile]
|
||
subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
|
||
# Removing empty dirs
|
||
for root, dirs, files in os.walk(OUTPUT_FOLDER):
|
||
if not dirs and not files:
|
||
dirBasename = os.path.basename(root)
|
||
if dirBasename not in IGNORED_EMPTY_FOLDER:
|
||
os.rmdir(root)
|