Add examples

examples/player/main.py

@@ -0,0 +1,153 @@
+import os
+import platform
+import re
+import subprocess
+import time
+
+import requests
+
+from librespot.audio.decoders import AudioQuality, VorbisOnlyAudioQuality
+from librespot.core import Session
+from librespot.metadata import TrackId
+
+quality: AudioQuality = AudioQuality.VERY_HIGH
+session: Session = None
+
+
+def clear():
+    if platform.system() == "Windows":
+        os.system("cls")
+    else:
+        os.system("clear")
+
+
+def client():
+    global quality, session
+    while True:
+        clear()
+        splash()
+        cmd = input("Player >>> ")
+        args = cmd.split(" ")
+        if args[0] == "exit" or args[0] == "quit":
+            return
+        if (args[0] == "p" or args[0] == "play") and len(args) == 2:
+            track_uri_search = re.search(
+                r"^spotify:track:(?P<TrackID>[0-9a-zA-Z]{22})$", args[1])
+            track_url_search = re.search(
+                r"^(https?://)?open\.spotify\.com/track/(?P<TrackID>[0-9a-zA-Z]{22})(\?si=.+?)?$",
+                args[1],
+            )
+            if track_uri_search is not None or track_url_search is not None:
+                track_id_str = (track_uri_search
+                                if track_uri_search is not None else
+                                track_url_search).group("TrackID")
+                play(track_id_str)
+                wait()
+        if args[0] == "q" or args[0] == "quality":
+            if len(args) == 1:
+                print("Current Quality: " + quality.name)
+                wait()
+            elif len(args) == 2:
+                if args[1] == "normal" or args[1] == "96":
+                    quality = AudioQuality.NORMAL
+                elif args[1] == "high" or args[1] == "160":
+                    quality = AudioQuality.HIGH
+                elif args[1] == "veryhigh" or args[1] == "320":
+                    quality = AudioQuality.VERY_HIGH
+                print("Set Quality to %s" % quality.name)
+                wait()
+        if (args[0] == "s" or args[0] == "search") and len(args) >= 2:
+            token = session.tokens().get("user-read-email")
+            resp = requests.get(
+                "https://api.spotify.com/v1/search",
+                {
+                    "limit": "5",
+                    "offset": "0",
+                    "q": cmd[2:],
+                    "type": "track"
+                },
+                headers={"Authorization": "Bearer %s" % token},
+            )
+            i = 1
+            tracks = resp.json()["tracks"]["items"]
+            for track in tracks:
+                print("%d, %s | %s" % (
+                    i,
+                    track["name"],
+                    ",".join([artist["name"] for artist in track["artists"]]),
+                ))
+                i += 1
+            position = -1
+            while True:
+                num_str = input("Select [1-5]: ")
+                if num_str == "exit" or num_str == "quit":
+                    return
+                try:
+                    num = int(num_str)
+                except ValueError:
+                    continue
+                if num in range(1, 5, 1):
+                    position = num - 1
+                    break
+            play(tracks[position]["id"])
+            wait()
+
+
+def login():
+    global session
+
+    if os.path.isfile("credentials.json"):
+        try:
+            session = Session.Builder().stored_file().create()
+            return
+        except RuntimeError:
+            pass
+    while True:
+        user_name = input("UserName: ")
+        password = input("Password: ")
+        try:
+            session = Session.Builder().user_pass(user_name, password).create()
+            return
+        except RuntimeError:
+            pass
+
+
+def play(track_id_str: str):
+    track_id = TrackId.from_base62(track_id_str)
+    stream = session.content_feeder().load(track_id,
+                                           VorbisOnlyAudioQuality(quality),
+                                           False, None)
+    ffplay = subprocess.Popen(
+        ["ffplay", "-"],
+        stdin=subprocess.PIPE,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    while True:
+        byte = stream.input_stream.stream().read()
+        if byte == -1:
+            return
+        ffplay.stdin.write(bytes([byte]))
+
+
+def splash():
+    print("=================================\n"
+          "| Librespot-Python Player       |\n"
+          "|                               |\n"
+          "| by kokarare1212               |\n"
+          "=================================\n\n\n")
+
+
+def main():
+    login()
+    client()
+
+
+def wait(seconds: int = 3):
+    for i in range(seconds)[::-1]:
+        print("\rWait for %d second(s)..." % (i + 1), end="")
+        time.sleep(1)
+
+
+if __name__ == "__main__":
+    main()

examples/server/main.py

@@ -0,0 +1,121 @@
+import os
+import re
+import socket
+import threading
+
+from librespot.audio.decoders import AudioQuality, VorbisOnlyAudioQuality
+from librespot.core import Session
+from librespot.metadata import TrackId
+
+session: Session
+sock: socket
+
+
+def handler(client: socket.socket, address: str):
+    req_raw = client.recv(1024 * 1024)
+    if len(req_raw) == 0:
+        return
+    req_arr = req_raw.split(b"\r\n")
+    req_http_raw = req_arr[0]
+    req_header_str = req_raw.split(b"\r\n\r\n")[0]
+    req_body_str = req_raw.split(b"\r\n\r\n")[1]
+    req_http_arr = req_http_raw.split(b" ")
+    req_method = req_http_arr[0]
+    req_uri = req_http_arr[1]
+    req_http_version = req_http_arr[2]
+    req_header = {}
+    for header in req_header_str.split(b"\r\n"):
+        try:
+            key, value = header.split(b": ")
+        except ValueError:
+            continue
+        else:
+            req_header[key.decode().lower()] = value.decode()
+    status, headers, content, manually = response(client, req_uri.decode(),
+                                                  req_header, req_body_str)
+    if not manually:
+        client.send(req_http_version + b" " + status.encode() + b"\r\n")
+        client.send(b"Access-Control-Allow-Origin: *\r\n")
+        for header in headers:
+            client.send(header.encode() + "\r\n")
+        client.send(b"\r\n")
+        client.send(content)
+    client.close()
+
+
+class HttpCode:
+    http_200 = "200 OK"
+    http_204 = "204 No Content"
+    http_400 = "400 Bad Request"
+    http_403 = "403 Forbidden"
+    http_404 = "404 Not Found"
+    http_500 = "500 Internal Server Error"
+
+
+def main():
+    global session, sock
+    session = None
+    if os.path.isfile("credentials.json"):
+        try:
+            session = Session.Builder().stored_file().create()
+        except RuntimeError:
+            pass
+    if session is None or not session.is_valid():
+        username = input("Username: ")
+        password = input("Password: ")
+        session = Session.Builder().user_pass(username, password).create()
+    if not session.is_valid():
+        return
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.bind(("127.0.0.1", 8080))
+    sock.listen(5)
+    while True:
+        threading.Thread(target=handler, args=sock.accept()).start()
+
+
+def response(client: socket.socket, uri: str, header: dict,
+             body: bytes) -> tuple[str, list, bytes, bool]:
+    if re.search(r"^/audio/track/([0-9a-zA-Z]{22})$", uri) is not None:
+        track_id_search = re.search(
+            r"^/audio/track/(?P<TrackID>[0-9a-zA-Z]{22})$", uri)
+        track_id_str = track_id_search.group("TrackID")
+        track_id = TrackId.from_base62(track_id_str)
+        stream = session.content_feeder().load(
+            track_id, VorbisOnlyAudioQuality(AudioQuality.VERY_HIGH), False,
+            None)
+        start = 0
+        end = stream.input_stream.stream().size()
+        if header.get("range") is not None:
+            range_search = re.search(
+                "^bytes=(?P<start>[0-9]+?)-(?P<end>[0-9]+?)$",
+                header.get("range"))
+            if range_search is not None:
+                start = int(range_search.group("start"))
+                end = (int(range_search.group("end"))
+                       if int(range_search.group("end")) <=
+                       stream.input_stream.stream().size() else
+                       stream.input_stream.stream().size())
+                stream.input_stream.stream().skip(start)
+        client.send(b"HTTP/1.0 200 OK\r\n")
+        client.send(b"Access-Control-Allow-Origin: *\r\n")
+        client.send(b"Content-Length: " +
+                    (str(stream.input_stream.stream().size()).encode() if
+                     stream.input_stream.stream().size() == end else "{}-{}/{}"
+                     .format(start, end,
+                             stream.input_stream.stream().size()).encode()) +
+                    b"\r\n")
+        client.send(b"Content-Type: audio/ogg\r\n")
+        client.send(b"\r\n")
+        while True:
+            if (stream.input_stream.stream().pos() >=
+                    stream.input_stream.stream().size()):
+                break
+            byte = stream.input_stream.stream().read()
+            client.send(bytes([byte]))
+        return "", [], b"", True
+    else:
+        return HttpCode.http_404, [], HttpCode.http_404.encode(), False
+
+
+if __name__ == "__main__":
+    main()