19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
|
#
#
#
import re
import os
import http
from config import conf, __print__, dbg
import platform
main = None
|
|
|
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
|
#
#
#
import re
import os
import ahttp as http
from config import conf, __print__, dbg
import platform
main = None
|
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
|
# media formats
mf = {"mp3":"audio/mp3", "ogg":"audio/ogg", "aac":"audio/aac"}
# web
@staticmethod
def browser(url):
__print__( conf.browser )
action.run(conf.browser + " " + action.quote(url))
# os shell cmd escaping
@staticmethod
def quote(s):
if conf.windows:
return s # should actually be "\\\"%s\\\"" % s
else:
return "%r" % s
# calls player for stream url and format
@staticmethod
def play(url, audioformat="audio/mp3", listformat="text/x-href"):
if (url):
url = action.url(url, listformat)
if (audioformat):
if audioformat == "audio/mpeg":
audioformat = "audio/mp3" # internally we use the more user-friendly moniker
cmd = conf.play.get(audioformat, conf.play.get("*/*", "vlc %u"))
__print__( "play", url, cmd )
try:
action.run( action.interpol(cmd, url) )
except:
pass
# exec wrapper
@staticmethod
def run(cmd):
if conf.windows:
os.system("start \"%s\"")
else:
os.system(cmd + " &")
# streamripper
@staticmethod
def record(url, audioformat="audio/mp3", listformat="text/x-href", append="", row={}):
__print__( "record", url )
cmd = conf.record.get(audioformat, conf.record.get("*/*", None))
try: action.run( action.interpol(cmd, url, row) + append )
except: pass
# save as .m3u
@staticmethod
|
|
|
|
|
|
|
|
|
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
|
# media formats
mf = {"mp3":"audio/mp3", "ogg":"audio/ogg", "aac":"audio/aac"}
# web
@staticmethod
def browser(url):
__print__( dbg.CONF, conf.browser )
action.run(conf.browser + " " + action.quote(url))
# os shell cmd escaping
@staticmethod
def quote(s):
if conf.windows:
return str(s) # should actually be "\\\"%s\\\"" % s
else:
return "%r" % str(s)
# calls player for stream url and format
@staticmethod
def play(url, audioformat="audio/mp3", listformat="text/x-href"):
if (url):
url = action.url(url, listformat)
if (audioformat):
if audioformat == "audio/mpeg":
audioformat = "audio/mp3" # internally we use the more user-friendly moniker
cmd = conf.play.get(audioformat, conf.play.get("*/*", "vlc %u"))
__print__( dbg.PROC,"play", url, cmd )
try:
action.run( action.interpol(cmd, url) )
except:
pass
# exec wrapper
@staticmethod
def run(cmd):
if conf.windows:
os.system("start \"%s\"")
else:
os.system(cmd + " &")
# streamripper
@staticmethod
def record(url, audioformat="audio/mp3", listformat="text/x-href", append="", row={}):
__print__( dbg.PROC, "record", url )
cmd = conf.record.get(audioformat, conf.record.get("*/*", None))
try: action.run( action.interpol(cmd, url, row) + append )
except: pass
# save as .m3u
@staticmethod
|
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
|
return url
# download a .pls resource and extract urls
@staticmethod
def pls(url):
text = http.get(url)
__print__( "pls_text=", text )
return re.findall("\s*File\d*\s*=\s*(\w+://[^\s]+)", text, re.I)
# currently misses out on the titles
# get a single direct ICY stream url (extract either from PLS or M3U)
@staticmethod
def srv(url):
return action.extract_urls(url)[0]
|
|
|
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
|
return url
# download a .pls resource and extract urls
@staticmethod
def pls(url):
text = http.get(url)
__print__( dbg.DATA, "pls_text=", text )
return re.findall("\s*File\d*\s*=\s*(\w+://[^\s]+)", text, re.I)
# currently misses out on the titles
# get a single direct ICY stream url (extract either from PLS or M3U)
@staticmethod
def srv(url):
return action.extract_urls(url)[0]
|
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
|
# use shoutcast unique stream id if available
stream_id = re.search("http://.+?/.*?(\d+)", pls, re.M)
stream_id = stream_id and stream_id.group(1) or "XXXXXX"
try:
channelname = main.current_channel
except:
channelname = "unknown"
return (conf.tmp + os.sep + "streamtuner2."+channelname+"."+stream_id+".m3u", len(stream_id) > 3 and stream_id != "XXXXXX")
# check if there are any urls in a given file
@staticmethod
def has_urls(tmp_fn):
if os.path.exists(tmp_fn):
return open(tmp_fn, "r").read().find("http://") > 0
# create a local .m3u file from it
@staticmethod
def m3u(pls):
# temp filename
(tmp_fn, unique) = action.tmp_fn(pls)
# does it already exist?
if tmp_fn and unique and conf.reuse_m3u and action.has_urls(tmp_fn):
return tmp_fn
# download PLS
__print__( "pls=",pls )
url_list = action.extract_urls(pls)
__print__( "urls=", url_list )
# output URL list to temporary .m3u file
if (len(url_list)):
#tmp_fn =
f = open(tmp_fn, "w")
f.write("#M3U\n")
f.write("\n".join(url_list) + "\n")
f.close()
# return path/name of temporary file
return tmp_fn
else:
__print__( "error, there were no URLs in ", pls )
raise "Empty PLS"
# open help browser
@staticmethod
def help(*args):
action.run("yelp /usr/share/doc/streamtuner2/help/")
#or action.browser("/usr/share/doc/streamtuner2/")
#class action
|
|
|
|
|
|
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
|
# use shoutcast unique stream id if available
stream_id = re.search("http://.+?/.*?(\d+)", pls, re.M)
stream_id = stream_id and stream_id.group(1) or "XXXXXX"
try:
channelname = main.current_channel
except:
channelname = "unknown"
return (str(conf.tmp) + os.sep + "streamtuner2."+channelname+"."+stream_id+".m3u", len(stream_id) > 3 and stream_id != "XXXXXX")
# check if there are any urls in a given file
@staticmethod
def has_urls(tmp_fn):
if os.path.exists(tmp_fn):
return open(tmp_fn, "r").read().find("http://") > 0
# create a local .m3u file from it
@staticmethod
def m3u(pls):
# temp filename
(tmp_fn, unique) = action.tmp_fn(pls)
# does it already exist?
if tmp_fn and unique and conf.reuse_m3u and action.has_urls(tmp_fn):
return tmp_fn
# download PLS
__print__( dbg.DATA, "pls=",pls )
url_list = action.extract_urls(pls)
__print__( dbg.DATA, "urls=", url_list )
# output URL list to temporary .m3u file
if (len(url_list)):
#tmp_fn =
f = open(tmp_fn, "w")
f.write("#M3U\n")
f.write("\n".join(url_list) + "\n")
f.close()
# return path/name of temporary file
return tmp_fn
else:
__print__( dbg.ERR, "error, there were no URLs in ", pls )
raise "Empty PLS"
# open help browser
@staticmethod
def help(*args):
action.run("yelp /usr/share/doc/streamtuner2/help/")
#or action.browser("/usr/share/doc/streamtuner2/")
#class action
|