nganhkhoa/musipy
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge branch 'feature/refactor-code' into develop

Browse Source
This commit is contained in:
Nguyễn Anh Khoa 2018-09-27 02:01:24 +07:00
commit 735081db41
10 changed files with 350 additions and 273 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@ -123,3 +123,6 @@ tags
# End of https://www.gitignore.io/api/vim,python,visualstudiocode
test/test/*
*/*.log

211
CommandExecutioner.py Normal file
View File

@ -0,0 +1,211 @@
from tinytag import TinyTag
import os
# from functools import reduce
import pprint
from binascii import b2a_hex
class CommandExecutioner:
def __init__(self, kwargs):
self.options = kwargs
def work(self):
pass
def scan(self):
"""
set self.files to be a list of files
element in list are in full path
"""
self.files = []
for root, dirs, files in os.walk(self.source):
print("[+] Scan {}".format(root))
dirs[:] = [d for d in dirs if not d.startswith('.')]
for folder in dirs:
folder = os.path.join(root, folder)
for f in files:
self.files.append(os.path.join(root, f))
def _scan(self):
"""
deprecated
"""
# files to work on
# scan all files and store whole path on here
self.files = []
folder_queue = [self.source]
home_path_len = len(folder_queue[0])
while (len(folder_queue) > 0):
current_folder = folder_queue[0]
folder_queue = folder_queue[1:]
if len(current_folder[home_path_len:]) == 0:
print("[+] Scan /")
else:
print("[+] Scan {}".format(current_folder[home_path_len:]))
self.files += list(
filter(
lambda x: os.path.isfile(x),
os.listdir(current_folder))
)
for f in self.files:
f = current_folder + '/' + f
folders = list(
filter(
lambda x:
os.path.isdir(x) and x[0] != '.',
os.listdir(current_folder)))
for folder in folders:
folder_queue.append(current_folder + '/' + folder)
def makedict(self):
"""
from list of files
to dictionary of file by attribute specified
"""
self.data = {}
for f in self.files:
try:
tag = TinyTag.get(f)
except LookupError:
# not a music file
continue
except BaseException:
print("Cannot get tag from file --> Skip\n\t{}".format(f))
continue
tag = getattr(tag, self.attribute)
if tag in self.data:
self.data[tag].append(f)
else:
self.data[tag] = []
self.data[tag].append(f)
class CommandSort(CommandExecutioner):
def __init__(self, kwargs):
super().__init__(kwargs)
self.source = kwargs['source']
self.destination = kwargs['destination']
self.attribute = kwargs['attribute']
def work(self):
self.scan()
self.makedict()
pprint.pprint(self.data)
self.sort()
def sort(self):
# key is now the name of the folder
for folder_name, tracks in self.data.items():
# print(folder_name)
# continue
if folder_name is None:
# file attribute is None
folder = "Undefined"
elif folder_name == '':
# file attribute is empty string
folder = "Undefined"
else:
folder = folder_name
# because folder is taken from file tags
# some tags could have '/' in it
# which is not acceptable as a file name in linux
# folder = folder.replace('/', '-')
# print(self.destination)
folder = self.destination + '/' + folder
if not os.path.exists(folder):
os.makedirs(folder)
print("========================================")
print("Album: {}".format(folder_name))
print("Folder: {}".format(folder))
# continue
for track in tracks:
new_file = folder + '/' + os.path.basename(track)
if track == new_file:
# after sort, stay the same
continue
if os.path.exists(new_file):
pass
else:
print("[+] Replace\n\t{}\n\t{}".format(track, new_file))
# uncomment when you are ready
# os.rename(track, new_file)
pass
return
class CommandPlaylist(CommandExecutioner):
def __init__(self, kwargs):
super().__init__(kwargs)
self.source = kwargs['source']
self.destination = kwargs['destination']
self.attribute = kwargs['attribute']
self.playlist = kwargs['playlist']
def work(self):
self.scan()
self.makedict()
print("Create playlist {}".format(self.playlist))
playlist = open(self.playlist, 'w')
playlist.write('#EXTM3U\n')
for key, musics in self.data.items():
for music in musics:
tag = TinyTag.get(music)
if tag.artist is None:
tag.artist = ''
if tag.title is None:
tag.title = ''
tag.duration = int(tag.duration)
# write comment
playlist.write('#EXTINF:{},{} - {}\n'
.format(
tag.duration,
tag.artist,
tag.title))
# write file direction
music = encode_to_hex(music)
playlist.write('file://{}\n'.format(music))
print("{} created".format(self.playlist))
def encode_to_hex(string):
"""
change special char to hex
"""
chars = list(string)
for i in range(len(string)):
hex_c = ord(chars[i])
if hex_c >= ord('!') and hex_c <= ord('~'):
# skip ascii characters
# be aware of special ascii!!!
continue
elif hex_c == ord(' '):
chars[i] = '%20'
else:
# not ascii change to hex
u = b2a_hex(chars[i].encode('utf-8')).decode('utf-8')
u = list(u)
for j in range(len(u)):
u[j] = u[j].upper()
if j % 2 != 0:
continue
u[j] = '%' + u[j]
chars[i] = "".join(u)
string = "".join(chars)
return string

90
CommandHandler.py Normal file
View File

@ -0,0 +1,90 @@
import os
from CommandExecutioner import (
CommandSort,
CommandPlaylist
)
argument_requirement = {
'sort':
['source', 'destination', 'attribute'],
'playlist':
['source', 'destination', 'attribute', 'playlist_name'],
}
attributes = ('title', 'album', 'artist')
class CommandHandler:
"""
Pre validation on arugments
"""
def __init__(self, kwargs):
# print(kwargs)
self.run(kwargs)
def run(self, kwargs):
kwargs['source'] = os.path.abspath(kwargs['source'])
kwargs['destination'] = os.path.abspath(kwargs['destination'])
try:
self.validate(kwargs)
except Exception:
exit(-1)
mode = kwargs['mode']
worker = None
if mode == 'sort':
worker = CommandSort(kwargs)
elif mode == 'playlist':
worker = CommandPlaylist(kwargs)
else:
# not likely
return
worker.work()
def validate(self, kwargs):
requirements = argument_requirement[kwargs['mode']]
for requirement in requirements:
if requirement == 'source':
if not os.path.isdir(kwargs['source']):
print(
'source:',
kwargs['source'],
'is not a valid directory')
print('directory not exist or is not a directory')
raise Exception
if requirement == 'destination':
if not os.path.isdir(kwargs['destination']):
print(
'destination:',
kwargs['destination'],
'is not a valid directory')
print('directory not exist or is not a directory')
raise Exception
elif requirement == 'attribute':
if kwargs['attribute'] not in attributes:
print(
'attribute:',
kwargs['attribute'],
'is not a valid attribute')
print('valid attributes are:', attributes)
raise Exception
elif requirement == 'playlist_name':
if kwargs['playlist'] is None:
print("Playlist name must be given")
raise Exception
pl_file = kwargs['destination'] + \
'/' + kwargs['playlist'] + '.m3u'
while True:
if not os.path.exists(pl_file):
break
overwrite = input(
'File path {} is existed, overwrite?(y/n) '
.format(pl_file))
if overwrite in ('y', 'Y'):
break
newname = input("New file name: ")
pl_file = kwargs['destination'] + '/' + newname + '.m3u'
kwargs['playlist'] = pl_file

2
common/__init__.py
View File

@ -1,2 +0,0 @@
from .get_content import get_content
from .same_name_alert import same_name_alert

13
common/get_content.py
View File

@ -1,13 +0,0 @@
import os
def get_content(source):
files = []
folders = []
for f in os.listdir(source):
if os.path.isfile(os.path.join(source, f)):
files.append(f)
else:
folders.append(f)
return files, folders

24
common/same_name_alert.py
View File

@ -1,24 +0,0 @@
import os
def same_name_alert(oldfile, newfile):
print("Old: {}".format(oldfile))
print("New: {}".format(newfile))
print("File existed, overwrite?")
print("'y' for yes")
print("'m' to view more data")
print("'l' to listen to two song")
while True:
ans = input("Answer: ")
if ans == 'y':
os.rename(oldfile, newfile)
break
elif ans == 'm':
print("Old file data:")
print("New file data:")
elif ans == 'l':
print("Listen to song 1")
print("Listen to song 2")
else:
break
return

172
musipy.py
View File

@ -1,172 +0,0 @@
import os
from parser import Parser
from common import same_name_alert, get_content
from tinytag import TinyTag
from binascii import b2a_hex
class musipy:
def __init__(self):
# prepare data
self.data = {}
self.parser = Parser()
print(self.parser.source)
print(self.parser.output)
print(self.parser.mode)
print(self.parser.attr)
# run
self.run()
def run(self):
if self.parser.mode == 'sort':
self.collect()
self.move_files()
elif self.parser.mode == 'playlist':
self.collect()
self.playlist()
else:
pass
return
# sort files bases on attribute
def sort(self, f, tag):
# get attribute from tag
# using self.attr
tag = getattr(tag, self.parser.attr)
if tag in self.data:
self.data[tag].append(f)
else:
self.data[tag] = []
self.data[tag].append(f)
return
# move files to new destination based on attribute
def move_files(self):
for folder, tracks in self.data.items():
if folder is None:
folder = "Undefined"
if not folder:
folder = "Undefined"
new_folder = self.parser.output + '/' + folder
if not os.path.exists(new_folder):
os.makedirs(new_folder)
print("Folder: {}".format(folder))
moved_files = 0
total_files = len(tracks)
for track in tracks:
moved_files += 1
percent = int(moved_files / total_files * 100)
print("Processing ... {:3d}%".format(percent), end='\r')
new_file = new_folder + '/' + os.path.basename(track)
if track == new_file:
# after sort, stay the same
continue
if os.path.exists(new_file):
same_name_alert(track, new_file)
else:
os.rename(track, new_file)
print("")
return
def playlist(self):
print("Create playlist name {}".format(self.parser.playlistname))
pl_file = self.parser.output + '/' + self.parser.playlistname + '.m3u'
mode = 'w'
while os.path.exists(pl_file):
rewrite = input('Playlist existed, rewrite?(y/n) ')
if rewrite == 'y':
break
newname = input("New file name: ")
pl_file = self.parser.output + '/' + newname + '.m3u'
# return
playlist = open(pl_file, mode)
playlist.write('#EXTM3U\n')
for key, musics in self.data.items():
for music in musics:
tag = TinyTag.get(music)
if tag.artist is None:
tag.artist = ''
if tag.title is None:
tag.title = ''
tag.duration = int(tag.duration)
# write comment
playlist.write('#EXTINF:{},{} - {}\n'
.format(tag.duration, tag.artist, tag.title))
# change special char to hex
chars = list(music)
for i in range(len(music)):
hex_c = ord(chars[i])
if hex_c >= ord('!') and hex_c <= ord('~'):
# if in range of ascii characters
continue
elif hex_c == ord(' '):
chars[i] = '%20'
else:
u = b2a_hex(chars[i].encode('utf-8')).decode('utf-8')
u = list(u)
for j in range(len(u)):
u[j] = u[j].upper()
if j % 2 != 0:
continue
u[j] = '%' + u[j]
chars[i] = "".join(u)
# print("Cannot find this character: 0x{}"
# .format(hex(hex_c)))
# exit(-1)
music = "".join(chars)
# write file direction
playlist.write('file://{}\n'.format(music))
print("{} created".format(pl_file))
return
# collect all files and store in self.data
def collect(self):
folder_queue = [self.parser.source]
home_path_len = len(folder_queue[0])
while (len(folder_queue) > 0):
current_folder = folder_queue[0]
folder_queue = folder_queue[1:]
if len(current_folder[home_path_len:]) == 0:
print("[+] Scan /")
else:
print("[+] Scan {}".format(current_folder[home_path_len:]))
files, folders = get_content(current_folder)
# skip folder named '.folder'
# generate full path
for folder in folders:
if folder[0] != '.':
full_path = current_folder + '/' + folder
folder_queue.append(full_path)
# work with files
for f in files:
try:
fp = current_folder + '/' + f # full path to file
tag = TinyTag.get(fp)
except LookupError:
continue
except:
print("Cannot get tag from file --> Skip\n\t{}".format(fp))
continue
self.sort(fp, tag)
if __name__ == "__main__":
muse = musipy()

1
parser/__init__.py
View File

@ -1 +0,0 @@
from .parser import Parser

61
parser/parser.py
View File

@ -1,61 +0,0 @@
import os
import getopt
import sys
class Parser():
def __init__(self):
argv = sys.argv[1:]
self.source = None
self.output = None
self.attr = None
self.mode = None
self.playlistname = None
try:
opts, args = getopt.getopt(
argv, 'hs:o:a:m:pln:',
['source=', 'output=', 'attribute=', 'mode=', 'playlistname='])
except getopt.GetoptError:
print('')
exit(0)
for opt, arg in opts:
if opt == '-h':
print("Help")
exit(0)
elif opt in ('-s', '--source'):
self.source = arg
elif opt in ('-o', '--output'):
self.output = arg
elif opt in ('-a', '--attribute'):
self.attr = arg
elif opt in ('-m', '--mode'):
self.mode = arg
elif opt in ('-pl', '--playlistname'):
self.playlistname = arg
else:
print("Unknown flag {} {}".format(opt, arg))
if self.source is None:
self.source = os.getcwd()
if self.output is None:
self.output = self.source + '/output'
if self.attr is None:
self.attr = 'album'
if self.mode is None:
self.mode = 'sort'
if self.mode == 'playlist' and self.playlistname is None:
print("No play list name")
exit()
if __name__ == '__main__':
p = Parser()
print(p.source)
print(p.output)
print(p.attr)
print(p.mode)

46
run.py Normal file
View File

@ -0,0 +1,46 @@
# from .commandparser import commandparser
import click
import os
from CommandHandler import CommandHandler
@click.group()
# @click.option('--verbose', '-v', is_flag=True, default=False)
def main():
pass
@main.command()
@click.option('--source', '-src', default=os.getcwd())
@click.option('--destination', '-dst', default=os.getcwd() + '/dst/')
@click.option('--attribute', '-attr', default='album')
@click.option('--auto-overwrite', is_flag=True, default=False)
def sort(**kwargs):
commandparser('sort', **kwargs)
@main.command()
@click.option('--source', '-src', default=os.getcwd())
@click.option('--destination', '-dst', default=os.getcwd() + '/dst/')
@click.option('--attribute', '-attr', default='album')
@click.option('--auto-overwrite', is_flag=True, default=False)
@click.option('--playlist', '-name')
def playlist(**kwargs):
commandparser('playlist', **kwargs)
@main.command()
@click.option('--source', '-src', default=os.getcwd())
@click.option('--destination', '-dst', default=os.getcwd())
@click.option('--format', '-fmt')
def format(**kwargs):
commandparser('format', **kwargs)
def commandparser(mode, **kwargs):
kwargs['mode'] = mode
CommandHandler(kwargs)
if __name__ == "__main__":
main()