gmusicplay.py
· 2.3 KiB · Python
Raw
#!/usr/bin/env python3
from gmusicapi import Mobileclient
import os
from json import load
import sys
def mkdir_p(path):
import errno
try:
os.makedirs(path)
except OSError as exc:
if exc.errno == errno.EEXIST and os.path.isdir(path):
pass
else:
raise
mkdir_p(os.path.expanduser("~/.config"))
with open(os.path.expanduser("~/.config/gmusic.json")) as f:
CONFIG = load(f)
client = Mobileclient()
client.login(*CONFIG)
def search(term, results=10):
results = client.search(term, max_results=results)
return {"album": [client.get_album_info(x["album"]["albumId"]) for x in results["album_hits"]],
"artist": [client.get_artist_info(x["artist"]["artistId"], max_rel_artist=0) for x in results["artist_hits"]],
"playlist": [client.get_shared_playlist_contents(x["playlist"]["shareToken"]) for x in results["playlist_hits"]],
"podcast": [client.get_podcast_series_info(x["series"]["seriesId"]) for x in results["podcast_hits"]],
"song": [x["track"] for x in results["song_hits"]],
"video": [x.get("youtube_video", None) for x in results["video_hits"]],
}
def get_stream(results, selection):
section, _, n = selection.partition("#")
if not section in results:
raise ValueError("Invalid section name")
n = int(n)
if n+1 > len(results[section]):
raise ValueError("Invalid number")
result = results[section][n]
if section == "song":
return client.get_stream_url(result.get("storeId") or result.get("id"))
else:
raise NotImplemented
get_stream.supported = ["song"]
def main():
from argparse import ArgumentParser
parser = ArgumentParser()
parser.add_argument("term", nargs="+")
parser.add_argument("-n", help="number of results returned", type=int, default=10)
args = parser.parse_args()
results = search(args.term, results=args.n)
for section in results:
if section in get_stream.supported:
print("\n".join("{}#{}: {}".format(section, n, i["title"]) for n,i in enumerate(results[section]) if type(i) is dict and i.get("title")), file=sys.stderr)
print("Selection (section#number): ", end="", file=sys.stderr)
print(get_stream(results, input()))
return 0
if __name__ == "__main__":
sys.exit(main())
| 1 | #!/usr/bin/env python3 |
| 2 | from gmusicapi import Mobileclient |
| 3 | import os |
| 4 | from json import load |
| 5 | import sys |
| 6 | |
| 7 | def mkdir_p(path): |
| 8 | import errno |
| 9 | try: |
| 10 | os.makedirs(path) |
| 11 | except OSError as exc: |
| 12 | if exc.errno == errno.EEXIST and os.path.isdir(path): |
| 13 | pass |
| 14 | else: |
| 15 | raise |
| 16 | |
| 17 | mkdir_p(os.path.expanduser("~/.config")) |
| 18 | with open(os.path.expanduser("~/.config/gmusic.json")) as f: |
| 19 | CONFIG = load(f) |
| 20 | |
| 21 | client = Mobileclient() |
| 22 | client.login(*CONFIG) |
| 23 | |
| 24 | def search(term, results=10): |
| 25 | results = client.search(term, max_results=results) |
| 26 | return {"album": [client.get_album_info(x["album"]["albumId"]) for x in results["album_hits"]], |
| 27 | "artist": [client.get_artist_info(x["artist"]["artistId"], max_rel_artist=0) for x in results["artist_hits"]], |
| 28 | "playlist": [client.get_shared_playlist_contents(x["playlist"]["shareToken"]) for x in results["playlist_hits"]], |
| 29 | "podcast": [client.get_podcast_series_info(x["series"]["seriesId"]) for x in results["podcast_hits"]], |
| 30 | "song": [x["track"] for x in results["song_hits"]], |
| 31 | "video": [x.get("youtube_video", None) for x in results["video_hits"]], |
| 32 | } |
| 33 | |
| 34 | def get_stream(results, selection): |
| 35 | section, _, n = selection.partition("#") |
| 36 | if not section in results: |
| 37 | raise ValueError("Invalid section name") |
| 38 | n = int(n) |
| 39 | if n+1 > len(results[section]): |
| 40 | raise ValueError("Invalid number") |
| 41 | result = results[section][n] |
| 42 | if section == "song": |
| 43 | return client.get_stream_url(result.get("storeId") or result.get("id")) |
| 44 | else: |
| 45 | raise NotImplemented |
| 46 | |
| 47 | get_stream.supported = ["song"] |
| 48 | |
| 49 | def main(): |
| 50 | from argparse import ArgumentParser |
| 51 | parser = ArgumentParser() |
| 52 | parser.add_argument("term", nargs="+") |
| 53 | parser.add_argument("-n", help="number of results returned", type=int, default=10) |
| 54 | args = parser.parse_args() |
| 55 | results = search(args.term, results=args.n) |
| 56 | for section in results: |
| 57 | if section in get_stream.supported: |
| 58 | print("\n".join("{}#{}: {}".format(section, n, i["title"]) for n,i in enumerate(results[section]) if type(i) is dict and i.get("title")), file=sys.stderr) |
| 59 | print("Selection (section#number): ", end="", file=sys.stderr) |
| 60 | print(get_stream(results, input())) |
| 61 | return 0 |
| 62 | |
| 63 | if __name__ == "__main__": |
| 64 | sys.exit(main()) |
| 65 |