mirror of
https://github.com/tcsenpai/swingmusic.git
synced 2025-06-10 13:07:35 +00:00
79 lines
2.0 KiB
Python
79 lines
2.0 KiB
Python
from dataclasses import dataclass
|
|
from os import scandir
|
|
import time
|
|
from typing import Tuple
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
from app.models import Folder
|
|
from app.models import Track
|
|
|
|
from app import instances
|
|
from app.logger import Log
|
|
|
|
|
|
@dataclass
|
|
class Dir:
|
|
path: str
|
|
is_sym: bool
|
|
|
|
|
|
def get_folder_track_count(path: str) -> int:
|
|
"""
|
|
Returns the number of files associated with a folder.
|
|
"""
|
|
tracks = instances.tracks_instance.find_tracks_inside_path_regex(path)
|
|
return len(tracks)
|
|
|
|
|
|
def create_folder(dir: Dir) -> Folder:
|
|
"""Create a single Folder object"""
|
|
folder = {
|
|
"name": dir.path.split("/")[-1],
|
|
"path": dir.path,
|
|
"is_sym": dir.is_sym,
|
|
"trackcount": instances.tracks_instance.find_tracks_inside_path_regex(dir.path),
|
|
}
|
|
|
|
return Folder(folder)
|
|
|
|
|
|
class getFnF:
|
|
"""
|
|
Get files and folders from a directory.
|
|
"""
|
|
|
|
def __init__(self, path: str) -> None:
|
|
self.path = path
|
|
|
|
def __call__(self) -> Tuple[Track, Folder]:
|
|
try:
|
|
all = scandir(self.path)
|
|
except FileNotFoundError:
|
|
return ([], [])
|
|
|
|
dirs, files = [], []
|
|
|
|
for entry in all:
|
|
if entry.is_dir() and not entry.name.startswith("."):
|
|
dir = {
|
|
"path": entry.path,
|
|
"is_sym": entry.is_symlink(),
|
|
}
|
|
dirs.append(Dir(**dir))
|
|
elif entry.is_file() and entry.name.endswith((".mp3", ".flac")):
|
|
files.append(entry.path)
|
|
|
|
tracks = instances.tracks_instance.find_songs_by_filenames(files)
|
|
tracks = [Track(track) for track in tracks]
|
|
s = time.time()
|
|
|
|
# folders = [create_folder(dir) for dir in dirs]
|
|
with ThreadPoolExecutor() as pool:
|
|
iter = pool.map(create_folder, dirs)
|
|
folders = [i for i in iter if i is not None]
|
|
d = time.time() - s
|
|
Log(f"Did that in {d} seconds")
|
|
folders = filter(lambda f: f.trackcount > 0, folders)
|
|
|
|
return tracks, folders
|