This commit is contained in:
Шурупов Илья Викторович 2026-02-04 21:21:19 +03:00
parent 5d38a146a1
commit 35905ea942
11 changed files with 11 additions and 9 deletions

View file

@ -0,0 +1,101 @@
import xmltodict
import json
def flatten_dict(d):
out = []
for song in d:
newSong = {}
strCount = 0
intCount = 0
dateCount = 0
def getStr(id):
nonlocal strCount
if id not in song["key"]:
return "undef"
if len(song["string"]) <= strCount:
return "error"
strCount += 1
return song["string"][strCount - 1]
def getInt(id):
nonlocal intCount
if id not in song["key"]:
return -1
if len(song["integer"]) <= intCount:
return -1
intCount += 1
return song["integer"][intCount - 1]
def getDate(id):
nonlocal dateCount
if id not in song["key"]:
return "undef"
if len(song["date"]) <= dateCount:
return "error"
dateCount += 1
return song["date"][dateCount - 1]
newSong["Track ID"] = getInt("Track ID")
newSong["Name"] = getStr("Name")
newSong["Artist"] = getStr("Artist")
newSong["Album Artist"] = getStr("Album Artist")
newSong["Composer"] = getStr("Composer")
newSong["Album"] = getStr("Album")
newSong["Genre"] = getStr("Genre")
newSong["Kind"] = getStr("Kind")
newSong["Size"] = getInt("Size")
newSong["Total Time"] = getInt("Total Time")
newSong["Disc Number"] = getInt("Disc Number")
newSong["Disc Count"] = getInt("Disc Count")
newSong["Track Number"] = getInt("Track Number")
newSong["Track Count"] = getInt("Track Count")
newSong["Year"] = getInt("Year")
newSong["Date Modified"] = getDate("Date Modified")
newSong["Date Added"] = getDate("Date Added")
newSong["Bit Rate"] = getInt("Bit Rate")
newSong["Sample Rate"] = getInt("Sample Rate")
newSong["Play Count"] = getInt("Play Count")
newSong["Play Date"] = getInt("Play Date")
newSong["Play Date UTC"] = getDate("Play Date UTC")
newSong["Skip Count"] = getInt("Skip Count")
newSong["Skip Date"] = getDate("Skip Date")
newSong["Release Date"] = getDate("Release Date")
newSong["Album Rating"] = getInt("Album Rating")
newSong["Album Rating Computed"] = "Album Rating Computed" in song["key"]
newSong["Loved"] = "Loved" in song["key"]
newSong["Album Loved"] = "Album Loved" in song["key"]
newSong["Explicit"] = "Explicit" in song["key"]
newSong["Compilation"] = "Compilation" in song["key"]
newSong["Artwork Count"] = getInt("Artwork Count")
newSong["Sort Album"] = getStr("Sort Album")
newSong["Sort Artist"] = getStr("Sort Artist")
newSong["Sort Name"] = getStr("Sort Name")
newSong["Persistent ID"] = getStr("Persistent ID")
newSong["Track Type"] = getStr("Track Type")
out.append(newSong)
return out
def convert(filename, out_path):
with open(filename, 'r', encoding='utf-8') as xml_file:
data_dict = xmltodict.parse(xml_file.read())
# Extract the "Tracks" dictionary to be flattened
tracks_dict = data_dict['plist']['dict']['dict']
flat_tracks_dict = flatten_dict(tracks_dict["dict"])
json_data = json.dumps(flat_tracks_dict, indent=2)
with open(out_path, 'w', encoding='utf-8') as json_file:
json_file.write(json_data)
if __name__ == "__main__":
convert('./prefetched/ItunesLibrary.xml', './prefetched/ItunesLibrary.json')

View file

@ -0,0 +1,151 @@
import fnmatch
import mutagen
import eyed3
from mutagen.flac import FLAC
from mutagen import File as MutagenFile
from src.helpers import *
music_extensions = ('*.mp3', '*.flac', '*.wav', '*.aac', '*.ogg', '*.m4a')
exclude_directories = ("*mary--*", "*example-word*")
from mutagen.id3 import ID3
from mutagen.id3 import ID3NoHeaderError
def detect_lyrics(abs_path, ext):
ext = ext.lower()
# ---------- MP3 ----------
if ext == ".mp3":
try:
tags = ID3(abs_path)
except ID3NoHeaderError:
return False
return bool(
tags.getall("SYLT") or
tags.getall("USLT")
)
# ---------- FLAC / OGG / OPUS ----------
if ext in {".flac", ".ogg", ".opus"}:
try:
audio = MutagenFile(abs_path)
if not audio or not audio.tags:
return False
for key in audio.tags.keys():
k = key.upper()
if k in {"LYRICS", "LRC", "SYNCEDLYRICS"}:
if audio.tags[key]:
return True
except Exception:
return False
# ---------- Fallback ----------
try:
audio = MutagenFile(abs_path)
if audio and audio.tags:
for key in audio.tags.keys():
if "LYRIC" in key.upper():
return True
except Exception:
pass
return False
def update_local_library(root_dir, output_dir):
set_workdir(output_dir / "local")
old_library = get_data("local_library")
known_paths = {track['path']: track_id for track_id, track in old_library.items()}
new_library = {}
song_id = '0'
def get_new_song_id(prev_id, song_path):
if song_path in known_paths:
return int(known_paths[song_path])
while (prev_id in old_library) or (prev_id in new_library):
prev_id = str(int(prev_id) + 1)
return prev_id
for dir_path, dir_names, filenames in os.walk(root_dir):
dir_names[:] = [
d for d in dir_names
if not any(fnmatch.fnmatch(d, exclude) for exclude in exclude_directories)
]
filtered_filenames = [
filename for filename in filenames
if any(fnmatch.fnmatch(filename, ext) for ext in music_extensions)
]
for filename in filtered_filenames:
song_name, track_type = os.path.splitext(filename)
relative_path = os.path.relpath(os.path.join(dir_path, filename), root_dir)
song_id = get_new_song_id(song_id, relative_path)
new_library[song_id] = {
"name": song_name,
"path": relative_path,
"type": track_type,
"artist": "",
"album": "",
"duration": None
}
for track_id, track in new_library.items():
abs_path = os.path.join(root_dir, track['path'])
track_abs_path = os.path.join(root_dir, track["path"])
track["has_lyrics"] = detect_lyrics(track_abs_path, track['type'].lower())
match track['type'].lower():
case ".mp3":
mp3track = eyed3.load(abs_path)
if not mp3track or not mp3track.info:
print(f"Failed to load mp3 {abs_path}")
track['name'] = track['path']
else:
track['duration'] = mp3track.info.time_secs
tag = mp3track.tag
if tag:
track['artist'] = tag.artist
track['name'] = tag.title
track['album'] = tag.album
case ".flac":
try:
audio = FLAC(abs_path)
if audio.info:
track['duration'] = audio.info.length
metadata = audio.tags
if metadata:
if 'artist' in metadata:
track['artist'] = " ".join(metadata['artist'])
if 'TITLE' in metadata:
track['name'] = " ".join(metadata['TITLE'])
if 'album' in metadata:
track['album'] = " ".join(metadata['album'])
except mutagen.MutagenError:
print(f"Invalid flac header {abs_path}")
track['name'] = track['path']
case _:
try:
audio = MutagenFile(abs_path)
if audio and audio.info:
track['duration'] = audio.info.length
except Exception:
pass
save_data(new_library, "local_library")

View file

@ -0,0 +1,50 @@
from pathlib import Path
def ():
output_dir = Path(output_dir)
output_playlists = []
remote_to_local_map = {}
for local_id, links in link_pattern.items():
if len(links['items']):
remote_to_local_map[links['items'][0]] = local_id
for name, tracks in remote_playlists.items():
local_tracks = []
for track in tracks:
track = track["track"]
path = "error"
error = ""
if track is None:
error = f"error: track is none"
elif track["id"] in remote_to_local_map:
path = local_library[remote_to_local_map[track["id"]]]["path"]
else:
error = f"error: not_found - {track["name"]}"
if error != "":
local_tracks.append("# " + error)
else:
path = (output_dir / ".." ).resolve() / Path(path)
local_tracks.append(path)
pl = {
"name": name,
"tracks": local_tracks,
}
output_playlists.append(pl)
for pl in output_playlists:
output_dir.mkdir(parents=True, exist_ok=True)
playlist_path = Path(output_dir) / f"{pl['name']}.m3u"
with (playlist_path.open("w", encoding="utf-8") as f):
for track in pl["tracks"]:
f.write(str(track))
f.write("\n")

View file

@ -0,0 +1,137 @@
import json
import re
from src.Library import *
import re
PROTECTED_ARTISTS = {
"Tyler, The Creator",
"The Good, The Bad & The Queen",
}
def split_artists(artist_tag: str) -> list[str]:
if not artist_tag:
return []
placeholder_comma = "§COMMA§"
placeholder_amp = "§AMP§"
protected_map = {}
# Protect known artist names
for artist in PROTECTED_ARTISTS:
protected = (
artist
.replace(",", placeholder_comma)
.replace("&", placeholder_amp)
)
protected_map[protected] = artist
artist_tag = artist_tag.replace(artist, protected)
# Split remaining separators
parts = re.split(r"[,&]", artist_tag)
# Restore protected names and trim spaces
result = []
for part in parts:
part = part.strip()
if not part:
continue
part = (
part
.replace(placeholder_comma, ",")
.replace(placeholder_amp, "&")
)
result.append(part)
return result
class LocalParser:
def __init__(self):
self.track_map: dict[str, Track] = {}
self.artist_map: dict[str, Artist] = {}
self.album_map: dict[str, Album] = {}
self.library = Library()
def parse_artists(self, data: str) -> List[Artist]:
if not data:
data = "unknown"
data = split_artists(data)
artists = []
for artist_data in data:
name = artist_data
artist_id = name
if artist_id in self.artist_map:
artists.append(self.artist_map[artist_id])
continue
artist = Artist()
artist.id = artist_id
artist.title = name
self.artist_map[artist_id] = artist
self.library.artists.append(artist)
artists.append(artist)
return artists
def parse_album(self, data: str) -> Album:
if not data:
data = "unknown"
name = data
album_id = name
if album_id in self.album_map:
return self.album_map[album_id]
album = Album()
album.id = album_id
album.title = name
self.album_map[album_id] = album
self.library.albums.append(album)
return album
def parse(self, tracks_file: Path) -> Library:
tracks_content = json.loads(tracks_file.read_text())
for track_id, data in tracks_content.items():
track = Track()
track.id = track_id
track.title = data.get("name")
track.artists = self.parse_artists(data.get("artist"))
track.album = self.parse_album(data.get("album"))
track.local_path = data.get("path")
track.duration_ms = data.get("duration")
track.has_lyrics = data.get("has_lyrics")
self.track_map[track.id] = track
self.library.tracks.append(track)
for track_id, track in self.track_map.items():
for artist in track.artists:
if artist not in track.album.artists:
track.album.artists.append(artist)
self.library.update_cache()
return self.library
def parse_track_item(self, id, track: dict[str, Any]) -> Track:
return track

View file

@ -0,0 +1,119 @@
from src.Library import *
from typing import Any, Iterable
import json
from datetime import datetime
from pathlib import Path
def get_artists_str(artists_pack):
track_artists = [artist['name'] for artist in artists_pack]
artists = " ".join(track_artists)
return artists
class Parser:
def __init__(self):
self.track_map: dict[str, Track] = {}
self.artist_map: dict[str, Artist] = {}
self.album_map: dict[str, Album] = {}
self.library = Library()
def parse(self, playlists: Path, tracks: Path, top_tracks: Path, top_artists: Path) -> Library:
self.library = Library()
self.parse_tracks(json.loads(tracks.read_text()))
self.parse_playlists(json.loads(playlists.read_text()))
self.parse_top_artists(json.loads(top_artists.read_text()))
self.parse_top_tracks(json.loads(top_tracks.read_text()))
self.library.update_cache()
return self.library
def parse_playlists(self, data):
for name, tracks in data.items():
playlist = Playlist()
playlist.title = name
for track in tracks:
track_data = track.get("track")
if track_data:
track_id = track_data.get("id")
playlist.tracks.append(self.parse_track(track_data, track.get("added_at")))
self.library.playlists.append(playlist)
def parse_top_tracks(self, data):
for track in data:
if track.get("id") in self.track_map:
self.library.top_tracks.append(self.track_map.get(track.get("id")))
def parse_top_artists(self, data):
for artist in data:
if artist.get("id") in self.artist_map:
self.library.top_artists.append(self.artist_map.get(artist.get("id")))
def parse_tracks(self, data):
for track in data:
self.parse_track(track.get("track"), track.get("added_at"))
for track in data:
self.library.liked_tracks.append(self.track_map.get(track.get("track").get("id")))
def parse_track(self, data: dict[str, Any], added_time : datetime) -> Track:
track_id = data.get("id")
if track_id in self.track_map:
return self.track_map[track_id]
track = Track()
self.track_map[track_id] = track
track.id = track_id
track.title = data.get("name")
track.artists = [self.parse_artist(artist) for artist in data.get("artists")]
track.album = self.parse_album(data.get("album"))
track.duration_ms = data.get("duration_ms")
self.library.tracks.append(track)
return track
def parse_album(self, data):
album_id = data.get("id")
if album_id in self.album_map:
return self.album_map[album_id]
album = Album()
self.album_map[album_id] = album
album.id = data.get("id")
album.title = data.get("name")
album.artists = [self.parse_artist(artist) for artist in data.get("artists")]
self.library.albums.append(album)
return album
def parse_artist(self, data: dict[str, Any]) -> Artist:
artist_id = data.get("id")
if artist_id in self.artist_map:
return self.artist_map[artist_id]
artist = Artist()
self.artist_map[artist_id] = artist
artist.id = data.get("id")
artist.title = data.get("name")
self.library.artists.append(artist)
return artist

View file

@ -0,0 +1,116 @@
import base64
import requests
import webbrowser
from http.server import BaseHTTPRequestHandler, HTTPServer
import urllib.parse
import threading
import time
import os.path
client_id = None
client_secret = None
token = None
def retrieve_web_api_token(authorization_code):
global token
token_url = 'https://accounts.spotify.com/api/token'
token_data = {
"grant_type": "authorization_code",
"client_secret": client_secret,
"client_id": client_id,
"code": authorization_code,
'redirect_uri': 'http://127.0.0.1:8888',
}
print("\nAuthenticating WEB API\n")
auth_string = f'{client_id}:{client_secret}'
b64_auth_string = base64.b64encode(auth_string.encode()).decode()
# Define the headers
headers = {
'Content-Type': 'application/x-www-form-urlencoded',
'Authorization': f'Basic {b64_auth_string}'
}
token_response = requests.post(token_url, data=token_data, headers=headers)
if token_response.status_code not in range(200, 299):
print(f"Failed to retrieve access token: {token_response.status_code}")
print(token_response.json())
raise ValueError
token = token_response.json()['access_token']
class RequestHandler(BaseHTTPRequestHandler):
def do_GET(self):
parsed_url = urllib.parse.urlparse(self.path)
query_params = urllib.parse.parse_qs(parsed_url.query)
if 'code' in query_params:
authorization_code = query_params['code'][0]
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
self.wfile.write(b'<html><body>Authorization code received. You can close this window.</body></html>')
retrieve_web_api_token(authorization_code)
threading.Thread(target=self.server.shutdown).start()
else:
self.send_response(400)
self.send_header('Content-type', 'text/html')
self.end_headers()
self.wfile.write(b'<html><body>Authorization code not found.</body></html>')
def run_redirect_server():
httpd = HTTPServer(('localhost', 8888), RequestHandler)
print('Starting HTTP redirection server')
httpd.serve_forever()
def open_authenticate_page():
authorize_url = 'https://accounts.spotify.com/authorize'
params = {
'client_id': client_id,
'response_type': 'code',
'redirect_uri': 'http://127.0.0.1:8888',
"scope": "user-top-read playlist-read-collaborative user-read-email user-library-read user-read-private playlist-read-private",
}
auth_url = f"{authorize_url}?client_id={params['client_id']}&response_type={params['response_type']}&redirect_uri={params['redirect_uri']}&scope={params['scope']}"
webbrowser.open(auth_url)
def authenticate(id, secret):
global client_secret
global client_id
client_secret=secret
client_id=id
redirect_server = threading.Thread(target=run_redirect_server)
redirect_server.start()
time.sleep(1)
open_authenticate_page()
redirect_server.join()
return token
if __name__ == "__main__":
authenticate()

View file

@ -0,0 +1,196 @@
import requests
from PIL import Image
from io import BytesIO
from src.backends.spotify.SpotifyAuthenticator import authenticate
from src.helpers import *
FETCH_STEP = 50
token = ''
def fetch(endpoint, method='GET', body=None):
url = f'https://api.spotify.com/{endpoint}'
headers = {
'Authorization': f'Bearer {token}',
}
response = requests.request(method, url, headers=headers, json=body)
response.raise_for_status()
return response.json()
def find_song(song_pattern):
return fetch(f"v1/search/?type=track&track=1&q={song_pattern}")['tracks']['items']
def fetch_playlists(user_id):
print("Fetching playlists")
endpoint = f"v1/me/playlists"
pl_total = fetch(f'{endpoint}?limit=1')['total']
pl_count = 0
playlists = []
while pl_count < pl_total:
res = fetch(f'{endpoint}?limit={FETCH_STEP}&offset={pl_count}')
playlists += res['items']
pl_count += FETCH_STEP
return playlists
def fetch_playlist_items(playlist_id):
endpoint = f"v1/playlists/{playlist_id}/tracks"
items = fetch(f'{endpoint}?fields=total')['total']
loaded = 0
tracks = []
while loaded < items:
res = fetch(f'{endpoint}?limit={FETCH_STEP}&offset={loaded}')
tracks += res['items']
loaded += FETCH_STEP
return tracks
def fetch_tracks(user_id):
print("Fetching tracks")
endpoint = f"v1/me/tracks"
total = fetch(f"{endpoint}?limit=1&market=ES")['total']
current = 0
tracks = []
while total > current:
res = fetch(f"{endpoint}?limit={FETCH_STEP}&offset={current}&market=ES")
tracks += res['items']
current += FETCH_STEP
save_data(tracks, "tracks")
return tracks
def fetch_tracks_top():
print("Fetching top tracks")
total = fetch(f'v1/me/top/tracks?fields=total')['total']
current = 0
tracks = []
while current < total:
res = fetch(f'v1/me/top/tracks?limit={FETCH_STEP}&offset={current}&time_range=long_term')
tracks += res['items']
current += FETCH_STEP
save_data(tracks, "top_tracks")
return tracks
def fetch_artists_top():
print("Fetching top artists")
total = fetch(f'v1/me/top/artists?fields=total')['total']
current = 0
artists = []
while current < total:
res = fetch(f'v1/me/top/artists?limit={FETCH_STEP}&offset={current}&time_range=long_term')
artists += res['items']
current += FETCH_STEP
save_data(artists, "top_artists")
return artists
def fetch_all_playlist_data(user_id):
playlists = fetch_playlists(user_id)
save_data(playlists, 'playlists')
print(len(playlists))
playlist_items = {}
print("Fetching playlists items")
index = 0
for pl in playlists:
pl_id = pl['id']
name = pl['name']
playlist_items[name] = fetch_playlist_items(pl_id)
print(f"{index} - {len(playlist_items[name])}")
index += 1
save_data(playlist_items, "playlistTracks")
def get_user_data():
print("Fetching user data")
user_profile = fetch(f'v1/me')
user_id = user_profile['id']
user_data = fetch(f'v1/users/{user_id}')
save_data(user_profile, "user_profile")
save_data(user_data, "user_data")
return user_id
def save_image_from_url(url, name, image_dir="playlist_covers"):
response = requests.get(url)
if response.status_code != 200:
raise "Cannot fetch the playlist cover"
image = Image.open(BytesIO(response.content))
directory = os.path.join(get_workdir(), image_dir)
os.makedirs(directory, exist_ok=True)
file_path = os.path.join(directory, f"{name}.jpg")
print(f"Image saved {file_path}")
image.save(file_path)
def load_playlist_covers():
playlists = get_data('playlists')
for pl in playlists:
if len(pl['images']):
url = pl['images'][0]['url']
save_image_from_url(url, pl['name'])
def load_user_cover():
user_data = get_data("user_data")
url = user_data['images'][1]['url']
save_image_from_url(url, "user", ".")
def load_artworks():
load_user_cover()
load_playlist_covers()
def update_access_token(id, secret):
global token
token = authenticate(id, secret)
def fetch_data():
user_id = get_user_data()
fetch_tracks(user_id)
fetch_all_playlist_data(user_id)
fetch_tracks_top()
fetch_artists_top()
load_artworks()
def fetch_all(id, secret, workdir):
set_workdir(workdir / "spotify")
update_access_token(id, secret)
fetch_data()
if __name__ == "__main__":
update_access_token("", "")
fetch_data()