main.py

import json
import os
import random
import re
import shutil
import string
import sys
import threading
import time
import urllib.request
import webbrowser
from http.server import BaseHTTPRequestHandler, HTTPServer
from os import devnull
from subprocess import call
from urllib.parse import parse_qs, urlencode, urlparse

import questionary
import requests
import yt_dlp
from alive_progress import alive_bar
from appdirs import AppDirs
from loguru import logger as log
from mutagen.id3 import APIC, TXXX, ID3, TPE1, TIT2, TPOS, TRCK, TDRC, TALB, TPE2
from mutagen import File
from mutagen.mp3 import MP3, HeaderNotFoundError
from questionary import Style
from sanitize_filename import sanitize
from youtube_search import YoutubeSearch
from yt_dlp import utils


def disablePrint():
    sys.stdout = open(devnull, "w")


def enablePrint():
    sys.stdout = sys.__stdout__


def clearScreen():
    os.system("clear")
    print("""
   ▗   ▗ ▘  
 ▛▘▜▘▀▌▜▘▌▛▘
 ▄▌▐▖█▌▐▖▌▙▖  v1.1.4
""")


def getAppConfig():
    if not os.path.exists(appFolder):
        os.makedirs(appFolder)
    try:
        with open(os.path.join(appFolder, "data.json"), "x") as f:
            emptyConfig = {
                "syncFolder": None,
                "accessToken": None,
                "userName": None,
                "clientId": None,
                "clientSecret": None,
                "selectedPlaylists": [],
            }
            jsonEmptyConfig = json.dumps(emptyConfig)
            f.write(jsonEmptyConfig)
            return emptyConfig
    except FileExistsError:
        with open(os.path.join(appFolder, "data.json"), "r") as f:
            return json.loads(f.read())


def writeAppConfig(appConfig):
    with open(os.path.join(appFolder, "data.json"), "w") as f:
        f.write(json.dumps(appConfig))


def chooseSyncFolder():
    clearScreen()

    if appConfig["syncFolder"]:
        userAction = questionary.confirm(
            "are you sure you want to override the already selected folder?",
            style=Style([("question", "nobold")]),
            qmark="",
        ).ask()

        if not userAction:
            return

        clearScreen()

    userFolderChoice = questionary.path(
        "enter spotify sync folder path [<tab> to autocomplete]",
        style=Style([("question", "nobold")]),
        qmark="",
        only_directories=True,
    ).ask()

    clearScreen()

    userFolderChoice = os.path.expanduser(userFolderChoice)

    contentsOfFolder = os.listdir(userFolderChoice)

    if len(contentsOfFolder) != 0:
        userAction = questionary.confirm(
            "are you sure you want to overwrite the contents of the selected folder? (all existing data will be deleted)",
            style=Style([("question", "nobold")]),
            qmark="",
        ).ask()
        if not userAction:
            return

    appConfig["syncFolder"] = userFolderChoice
    writeAppConfig(appConfig)
    return


def randomString():
    return "".join(random.choices(string.ascii_lowercase + string.digits, k=10))


def getYoutubeID(url):
    parsed = urlparse(url)

    if parsed.query:
        qs = parse_qs(parsed.query)
        if "v" in qs:
            return qs["v"][0]

    pathParts = parsed.path.strip("/").split("/")
    if "shorts" in pathParts and len(pathParts) > pathParts.index("shorts"):
        return pathParts[pathParts.index("shorts") + 1]

    return "dQw4w9WgXcQ"


def createAccessToken():
    authCode = [None]

    class SimpleHandler(BaseHTTPRequestHandler):
        def do_GET(self):
            parsed_url = urlparse(self.path)
            query_params = parse_qs(parsed_url.query)

            log.info("GET query parameters received:", query_params)

            authCode[0] = query_params["code"][0]

            self.send_response(200)
            self.end_headers()
            self.wfile.write(b"you can close this window and head back to static!\n")
            threading.Thread(target=server.shutdown).start()  # shutdown the server

    state = randomString()
    scope = "user-read-private user-read-email playlist-read-private"

    authParams = {
        "response_type": "code",
        "client_id": appConfig["clientId"],
        "scope": scope,
        "redirect_uri": "http://127.0.0.1:9321",
        "state": state,
    }

    log.debug(urlencode(authParams))

    PORT = 9321
    global server
    server = HTTPServer(("127.0.0.1", PORT), SimpleHandler)

    thread = threading.Thread(target=server.serve_forever)
    thread.start()

    webbrowser.open("https://accounts.spotify.com/authorize?" + urlencode(authParams))

    thread.join()

    log.info("threads closed")
    log.info(authCode[0])

    response = requests.post(
        "https://accounts.spotify.com/api/token",
        data={
            "code": authCode[0],
            "redirect_uri": "http://127.0.0.1:9321",
            "grant_type": "authorization_code",
            "client_id": appConfig["clientId"],
            "client_secret": appConfig["clientSecret"],
        },
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )

    return response.json()["access_token"]


def selectPlaylists():
    clearScreen()

    def getData():
        response = requests.get(
            "https://api.spotify.com/v1/me/playlists",
            headers={"Authorization": f"Bearer {appConfig['accessToken']}"},
        )
        return response

    response = getData()

    if response.status_code == 401:
        print("looks like your spotify account got disconnected (token expired)")
        userAction = questionary.confirm(
            "do you want to reconnect?",
            style=Style([("question", "nobold")]),
            qmark="",
        ).ask()
        if not userAction:
            return

        setupUser(True)
        response = getData()
        clearScreen()

    data = response.json()

    playlistChoices = []

    selectedIds = {list(d.keys())[0] for d in appConfig["selectedPlaylists"]}

    log.debug(json.dumps(data, indent=4))
    for x in data["items"]:
        log.debug(x)

        alreadySelected = x["id"] in selectedIds

        currentSnapshotIsNull = False
        for y in appConfig["selectedPlaylists"]:
            if (
                list(y.keys())[0] == x["id"]
                and y[list(y.keys())[0]]["snapshotId"] is None
            ):
                currentSnapshotIsNull = True

        playlistChoices.append(
            questionary.Choice(
                f"{x['name']} [{x['items']['total']}]",
                value={
                    x["id"]: {
                        "name": x["name"],
                        "snapshotId": x["snapshot_id"]
                        if alreadySelected and not currentSnapshotIsNull
                        else None,
                    }
                },
                checked=alreadySelected,
            )
        )

    playlistsSelected = questionary.checkbox(
        "select the playlists you want to sync",
        choices=playlistChoices,
        qmark="",
        pointer="⏵",
        instruction="[<space> to pick, <enter> to confirm]",
        style=Style([("question", "nobold")]),
    ).ask()

    if playlistsSelected is None:
        return False

    appConfig["selectedPlaylists"] = playlistsSelected
    writeAppConfig(appConfig)


dirs = AppDirs("static", "adithya")
appFolder = dirs.user_data_dir
appConfig = getAppConfig()


def setupClient(preConfirm=False):
    if not preConfirm:
        if appConfig["clientId"] or appConfig["clientSecret"]:
            userAction = questionary.confirm(
                "do you want to overwrite the current client secrets?",
                style=Style([("question", "nobold")]),
                qmark="",
            ).ask()

            clearScreen()

            if not userAction:
                return

        print(
            "in order for static to access your spotify data, it needs a client id and secret associated to your account"
        )
        print()
        print("follow the given step to obtain them")
        print("- create a spotify developers account at https://developer.spotify.com")
        print("- go to the dashboard and create an app")
        print("- make sure to add the redirect uri as http://127.0.0.1:9321")
        print()
        input("press enter to continue")

        clearScreen()

        clientId = questionary.text(
            "enter your spotify client id",
            style=Style([("question", "nobold")]),
            qmark="",
        ).ask()

        clearScreen()

        clientSecret = questionary.text(
            "enter your spotify client secret",
            style=Style([("question", "nobold")]),
            qmark="",
        ).ask()

        clearScreen()

        appConfig["clientId"] = clientId
        appConfig["clientSecret"] = clientSecret

        writeAppConfig(appConfig)


def setupUser(preConfirm=False):
    clearScreen()

    setupClient(preConfirm)

    if not preConfirm and appConfig["userName"]:
        userAction = questionary.confirm(
            "are you sure you want to overwrite the already connected account?",
            style=Style([("question", "nobold")]),
            qmark="",
        ).ask()
        if not userAction:
            return

    accessToken = createAccessToken()

    appConfig["accessToken"] = accessToken

    response = requests.get(
        "https://api.spotify.com/v1/me",
        headers={"Authorization": f"Bearer {accessToken}"},
    )

    data = response.json()

    appConfig["userName"] = data.get("display_name")
    appConfig["userId"] = data.get("id")

    writeAppConfig(appConfig)


def downloadSong(songId, playlistFolder):
    os.makedirs(playlistFolder, exist_ok=True)

    response = requests.get(
        f"https://api.spotify.com/v1/tracks/{songId}",
        headers={"Authorization": f"Bearer {appConfig['accessToken']}"},
    )
    if response.status_code == 400:
        return

    trackData = response.json()

    try:
        songTitle = (
            trackData.get("name")
            + " - "
            + ", ".join([artist["name"] for artist in trackData.get("artists")])
        )

        log.info(f"Searching for: {songTitle}")
        searchResults = YoutubeSearch(songTitle, max_results=1).to_dict()
        if not searchResults:
            log.info("No results found.")
            return

        videoData = searchResults[0]
        videoUrl = f"https://www.youtube.com{videoData['url_suffix']}"
        videoTitle = videoData["title"]

        log.info(f"Downloading: {videoTitle}")
        mp3FileName = f"{songTitle}"
        mp3FileName = sanitize(mp3FileName)
        mp3FullPath = os.path.join(playlistFolder, mp3FileName)
        mp3FullPath = os.path.expanduser(mp3FullPath)

        ytdlOptions = {
            "format": "bestaudio/best",
            "outtmpl": mp3FullPath,
            "noplaylist": True,
            "cookiefile": os.path.expanduser(os.path.join(appFolder, "cookies.txt")),
            "postprocessors": [
                {
                    "key": "FFmpegExtractAudio",
                    "preferredcodec": "mp3",
                    "preferredquality": "192",
                },
                {
                    "key": "FFmpegMetadata",
                },
            ],
        }

        try:
            log.info(f"Downloading {videoTitle} from {videoUrl}")
            with yt_dlp.YoutubeDL(ytdlOptions) as ydl:
                ydl.download([videoUrl])
        except utils.ExtractorError:
            log.info(f"Video unavailable to download: {mp3FileName}")
            return  # stop processing this song
        except utils.DownloadError:
            log.info(f"Video unavailable to download: {mp3FileName}")
            return  # stop processing this song

        time.sleep(1)

        try:
            # downloading and embedding cover art
            coverArtUrl = trackData.get("album")["images"][0]["url"]
            coverArtPath = os.path.join(playlistFolder, "thumb.jpg")
            urllib.request.urlretrieve(coverArtUrl, coverArtPath)

            audio = MP3(mp3FullPath + ".mp3", ID3=ID3)

            audio.delete()

            audio.tags.add(
                APIC(
                    encoding=3,
                    mime="image/jpeg",
                    type=3,
                    desc="Cover",
                    data=open(coverArtPath, "rb").read(),
                )
            )

            audio.save()

            os.remove(coverArtPath)

            log.info(f"tagged image for: {mp3FullPath}")
        except Exception as e:
            log.info(f"tagging image failed for {mp3FileName}: {e}. {videoUrl}")

        try:
            audio = MP3(mp3FullPath + ".mp3", ID3=ID3)

            log.debug(videoUrl)
            youtubeVideoId = getYoutubeID(videoUrl)

            log.info("Embedding cover art...")

            audio.tags.add(
                TXXX(encoding=3, desc="STATIC_SPOTIFY_ID", text=[str(songId)])
            )

            audio.tags.add(
                TXXX(encoding=3, desc="STATIC_YOUTUBE_ID", text=[str(youtubeVideoId)])
            )

            audio.tags.add(
                TXXX(
                    encoding=3,
                    desc="ITUNESADVISORY",
                    text=[str(1 if bool(trackData.get("explicit")) else 0)],
                ),
            )

            audio.tags.add(
                TPE1(
                    encoding=3,
                    text=[artist["name"] for artist in trackData.get("artists")],
                )
            )

            audio.tags.add(
                TPE2(
                    encoding=3,
                    text=[
                        artist["name"] for artist in trackData.get("album")["artists"]
                    ],
                )
            )

            audio.tags.add(TIT2(encoding=3, text=[str(trackData.get("name"))]))
            audio.tags.add(TPOS(encoding=3, text=[str(trackData.get("disc_number"))]))
            audio.tags.add(TRCK(encoding=3, text=[str(trackData.get("track_number"))]))
            audio.tags.add(
                TDRC(encoding=3, text=[str(trackData.get("album")["release_date"])])
            )
            audio.tags.add(TALB(encoding=3, text=[str(trackData.get("album")["name"])]))

            audio.save()

            log.info(f"Downloaded and tagged: {mp3FullPath}")
        except Exception as e:
            log.info(
                f"Tagging or file operations failed for {mp3FileName}: {e}. {videoUrl}"
            )
    except Exception as e:
        log.info(f"failed to dowload current track; skipping; reason: {e}")


def downloadPlaylist(data):
    global appConfig
    clearScreen()

    playlistId = data.get("id")
    playlistName = data.get("name")
    spotifySnapshotId = data.get("snapshot_id")

    noOfSongs = data.get("items")["total"]

    playlistFolder = os.path.join(appConfig["syncFolder"], playlistName)
    playlistFolder = os.path.expanduser(playlistFolder)

    if not os.path.exists(playlistFolder):
        os.makedirs(playlistFolder)

    songsDownloadedFull = os.listdir(playlistFolder)

    songsDownloadedIds = set()

    for x in songsDownloadedFull:
        if x in [".DS_Store"] or x.endswith((".part", ".ytdl", ".jpg")):
            continue

        try:
            audio = File(os.path.join(playlistFolder, x))

            songsDownloadedIds.add(str(audio["TXXX:STATIC_SPOTIFY_ID"]))
        except HeaderNotFoundError:
            try:
                os.remove(os.path.join(playlistFolder, x))
            except PermissionError:
                log.info(f"could not delete {os.path.join(playlistFolder, x)}")
            continue
        except KeyError:
            # current song hasnt been tagged properly and we'll just redownload it

            try:
                os.remove(os.path.join(playlistFolder, x))
            except PermissionError:
                log.info(f"could not delete {os.path.join(playlistFolder, x)}")

    url = f"https://api.spotify.com/v1/playlists/{playlistId}/items"

    songsOrder = []

    print(f"spotify/{appConfig['userName']}/{playlistName} ->  {playlistFolder}")
    print()
    print("downloading and tagging files")
    print()

    with alive_bar(noOfSongs) as bar:
        disablePrint()
        while url:
            response = requests.get(
                url, headers={"Authorization": f"Bearer {appConfig['accessToken']}"}
            )
            data = response.json()

            for item in data["items"]:
                if item.get("item"):
                    songsOrder.append(item["item"]["id"])
                    if item["item"]["id"] not in songsDownloadedIds:
                        downloadSong(item["item"]["id"], playlistFolder)
                bar()

            url = data.get("next")
        enablePrint()

    songsDownloadedFull = os.listdir(playlistFolder)

    clearScreen()
    print(f"spotify/{appConfig['userName']}/{playlistName} ->  {playlistFolder}")
    print()
    print("organizing files")
    print()

    with alive_bar(noOfSongs + 1) as bar:
        time.sleep(2)

        bar()

        for x in songsDownloadedFull:
            if x in [".DS_Store"] or x.endswith((".part", ".ytdl", ".jpg")):
                continue

            audio = File(os.path.join(playlistFolder, x))
            songId = audio["TXXX:STATIC_SPOTIFY_ID"]
            if songId in songsOrder:
                songNumber = str(songsOrder.index(songId) + 1).zfill(3)

                newName = x
                if re.match(r"^\d{3}\. ", x):
                    newName = x[5:]

                os.rename(
                    os.path.join(playlistFolder, x),
                    os.path.join(playlistFolder, f"{songNumber}. {newName}"),
                )
            bar()

    # writing the new snapshot id only after everything has been confirmed downloaded
    appConfig = getAppConfig()
    for x in appConfig["selectedPlaylists"]:
        key = list(x.keys())[0]
        if key == playlistId:
            x[key]["snapshotId"] = spotifySnapshotId
            break
    writeAppConfig(appConfig)


def getUnsyncedPlaylists():
    def getData(playlistId):
        response = requests.get(
            f"https://api.spotify.com/v1/playlists/{playlistId}",
            headers={"Authorization": f"Bearer {appConfig['accessToken']}"},
        )
        return response

    # checking for auth
    playlistId = list(appConfig["selectedPlaylists"][0].keys())[0]
    response = getData(playlistId)

    if response.status_code == 401:
        print("looks like your spotify account got disconnected (token expired)")
        userAction = questionary.confirm(
            "do you want to reconnect?",
            style=Style([("question", "nobold")]),
            qmark="",
        ).ask()
        if not userAction:
            return

        setupUser(True)
        clearScreen()

    unsyncedPlaylistsData = []
    syncedPlaylistsData = []

    clearScreen()

    print("checking for unsynced playlists")
    print()

    with alive_bar(len(appConfig["selectedPlaylists"])) as bar:
        for x in appConfig["selectedPlaylists"]:
            playlistId = list(x.keys())[0]
            snapshotId = x[playlistId]["snapshotId"]

            response = getData(playlistId)
            data = response.json()

            spotifySnapshotId = data.get("snapshot_id")

            if snapshotId == spotifySnapshotId:
                syncedPlaylistsData.append(data)
            else:
                unsyncedPlaylistsData.append(data)

            bar()

        log.debug(unsyncedPlaylistsData)
    return unsyncedPlaylistsData, syncedPlaylistsData


def syncPlaylists():
    clearScreen()

    syncFolder = os.path.expanduser(appConfig["syncFolder"])

    finalPlaylists = set()

    unsyncedPlaylistsData, syncedPlaylistsData = getUnsyncedPlaylists()
    log.debug(unsyncedPlaylistsData)

    for data in unsyncedPlaylistsData:
        playlistName = data.get("name")

        log.info(f"downloading {playlistName}")
        downloadPlaylist(data)
        finalPlaylists.add(playlistName)

    for data in syncedPlaylistsData:
        finalPlaylists.add(data.get("name"))

    playlistsDownloadedFull = os.listdir(syncFolder)

    for x in playlistsDownloadedFull:
        if x not in finalPlaylists:
            unnecessary = os.path.join(syncFolder, x)
            if os.path.isdir(unnecessary):
                shutil.rmtree(unnecessary)
            else:
                os.remove(unnecessary)


def resetSnapshotIds():
    for playlist in appConfig["selectedPlaylists"]:
        for key in playlist:
            playlist[key]["snapshotId"] = ""
    writeAppConfig(appConfig)


def main():
    log.remove(0)

    while True:
        clearScreen()
        appConfig = getAppConfig()

        log.add(
            os.path.expanduser(os.path.join(appFolder, "static.log")), rotation="1 week"
        )

        log.info("starting")

        mainMenuChoices = []

        if appConfig["userName"]:
            if len(appConfig["selectedPlaylists"]) != 0:
                mainMenuChoices.extend(["sync", "reset snapshot ids (force sync)"])

            mainMenuChoices.append("choose playlists to sync")

        mainMenuChoices.extend(
            [
                f"choose sync folder [{appConfig['syncFolder']}]",
                f"connect spotify account [{appConfig['userName']}]",
            ]
        )

        if appConfig["syncFolder"]:
            mainMenuChoices.append("open sync folder")

        mainMenuChoices.extend(
            [
                "open data folder",
                "built by adithya.zip",
                "coffee?",
            ]
        )

        userAction = questionary.select(
            "hey there!",
            choices=mainMenuChoices,
            qmark="",
            pointer="⏵",
            instruction=" ",
            style=Style([("question", "nobold")]),
        ).ask()

        chooseSyncFolderAction = f"choose sync folder [{appConfig['syncFolder']}]"
        connectAcountAction = f"connect spotify account [{appConfig['userName']}]"

        match userAction:
            case "sync":
                syncPlaylists()

            case "reset snapshot ids (force sync)":
                resetSnapshotIds()

            case "choose playlists to sync":
                selectPlaylists()

            case _ if userAction == chooseSyncFolderAction:
                chooseSyncFolder()

            case _ if userAction == connectAcountAction:
                setupUser()

            case "open sync folder":
                call(["open", "-R", os.path.expanduser(appConfig["syncFolder"])])

            case "open data folder":
                call(["open", "-R", os.path.expanduser(appFolder)])

            case "built by adithya.zip":
                webbrowser.open("https://adithya.zip")

            case "coffee?":
                webbrowser.open("https://ko-fi.com/adithyasource")

            case None:
                return False


main()