-
Notifications
You must be signed in to change notification settings - Fork 1
/
difm.py
135 lines (114 loc) · 4.33 KB
/
difm.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
import requests
import json
import time
from urllib.parse import urlparse
from urllib.parse import parse_qs
import datetime
from datetime import timezone
import os
networks = []
channels = []
user = []
audio_token = None
session_key = None
network_url = "https://www.di.fm"
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:66.0) Gecko/20100101 Firefox/66.0",
"Accept-Encoding": "*",
"Connection": "keep-alive"
}
def get_page():
response = requests.get(f"{network_url}/login",headers=headers)
result = json.loads(response.text.split("di.app.start(")[1].split(");")[0])
return result
def update_audio_token():
global audio_token
global channels
global session_key
global user
global networks
# There is probbaly a better way to get this that also allows for using credentials.
result = get_page()
user = result["user"]
networks = []
for network in result["networks"]:
# figure out what's wrong with parsing ClassicalRadio.com thml
if network["name"] != "ClassicalRadio.com":
networks.append(network)
audio_token = result["user"]["audio_token"]
session_key = result["user"]["session_key"]
channels = result["channels"]
def get_all_channels():
channel_list = []
result = get_page()
for network in result["networks"]:
if network["name"] != "ClassicalRadio.com":
result = get_page()
for channel in result["channels"]:
if channel not in channel_list:
channel_list.append(channel)
return channel_list
def get_url_expiration(url):
parsed_url = urlparse(url)
exp = parse_qs(parsed_url.query)['exp'][0]
return exp
def is_url_expired(url):
# This puts the burden on difm endpoint to validate
# make request and see if "URL expired" is returned
# Invalid signature if token is invalid
#response = requests.options(url)
# if response.text == "URL expired":
# return True
# else:
# return False
# This is less overhead since we are doing it before playing each track.
# Expired track is reall only an issue if the user keeps the app open for 24+ hrs.
# return true if expiration date is less than current date
current_time = datetime.datetime.now(timezone.utc).replace(tzinfo=timezone.utc)
expiration_time = datetime.datetime.strptime(
get_url_expiration(url), '%Y-%m-%dT%H:%M:%S%z').replace(tzinfo=timezone.utc)
return expiration_time < current_time
def get_channels():
return channels
def get_networks():
return networks
def set_network_url(url):
global network_url
network_url = url
update_audio_token()
def download_track(track,channel,url,directory="tracks"):
path = os.path.join(directory,channel)
if not os.path.exists(path):
os.makedirs(path)
r = requests.get(url, allow_redirects=True,headers=headers)
open(os.path.join(path,f"{track}.mp4"), 'wb').write(r.content)
def get_tracks_by_channel_id(id):
tracks = []
epoch = time.time()
channel_url = f'https://api.audioaddict.com/v1/di/routines/channel/{id}?tune_in=false&audio_token={audio_token}&_={epoch}'
channel_repsonse = requests.get(channel_url,headers=headers)
for track in json.loads(channel_repsonse.text)["tracks"]:
tracks.append(track)
return tracks
def generate_playlist(channel_id,channel_name,playlist_directory="playlists"):
print(f"Generating pls file for {channel_name} in {playlist_directory} ...")
index = 0
pls_tracks = {}
for n in range(4):
tracks = get_tracks_by_channel_id(channel_id)
for track in tracks:
pls_tracks[track["track"]] = f'https:{track["content"]["assets"][0]["url"]}'
expiration = get_url_expiration(list(pls_tracks.items())[0][1]).replace(":","-")
if not os.path.exists(playlist_directory):
os.makedirs(playlist_directory)
with open(os.path.join(playlist_directory,f"{channel_name} - Expires {expiration}.pls"), 'a', encoding="utf-8") as f:
f.write("[playlist]\n")
f.write(f"NumberOfEntries={len(pls_tracks)}\n")
for track in pls_tracks:
index += 1
f.write(f"File{index}={pls_tracks[track]}\n")
f.write(f"Title{index}={track} Expires: {get_url_expiration(pls_tracks[track])}\n")
print("Done. See playlists directory for file")
update_audio_token()
if __name__ == "__main__":
print(get_tracks_by_channel_id("124"))