forked from superolmo/pyxtream
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathfunctional_test.py
executable file
·159 lines (131 loc) · 4.56 KB
/
functional_test.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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
#!/usr/bin/python3
"""
Test application to validate provider account access
"""
import sys
from time import sleep
from pyxtream import XTream, __version__
PROVIDER_NAME = ""
PROVIDER_URL = ""
PROVIDER_USERNAME = ""
PROVIDER_PASSWORD = ""
if PROVIDER_URL == "" or PROVIDER_USERNAME == "" or PROVIDER_PASSWORD == "":
print("Please edit this file with the provider credentials")
sys.exit()
def str2list(input_string: str) -> list:
"""Convert a string with comma delimited numbers into a python list of integers
Args:
input_string (str): A list of commands comma delimited
Returns:
list: list of integers containing the commands
"""
# conver to the list
output_list = input_string.split(",")
# convert each element as integers
li = []
for list_index in output_list:
try:
li.append(int(list_index))
except ValueError:
pass
return li
print(f"pyxtream version {__version__}")
xt = XTream(
PROVIDER_NAME,
PROVIDER_USERNAME,
PROVIDER_PASSWORD,
PROVIDER_URL,
reload_time_sec=60*60*8,
debug_flask=False
)
sleep(0.5)
# If we could not connect, exit
if xt.auth_data == {}:
print("Authentication failed")
sleep(0.5)
sys.exit(0)
while True:
print(
"""
** Menu **
----------
(1) Load IPTV
(2) Search Streams Regex
(3) Search Streams Text
(4) Download Video (stream_id)
(5) Download Video Impl (URL, filename)
(6) Show how many movies added in past 30 days
(7) Show how many movies added in past 7 days
(8) Get Series Info By ID (series_id)
(9) Get Live EPG by Stream ID
----------
(0) Quit
"""
)
sleep(0.1)
# Get user input
commands = input(">>> ").lower().rstrip()
# Convert string of commands to list of integers
command_list = str2list(commands)
for choice in command_list:
print(f"\t[{choice}]: ")
if choice == 0:
#xt.flaskapp.shutdown()
sys.exit(0)
elif choice == 1:
xt.load_iptv()
elif choice == 2:
search_string = input("Search for REGEX (ex. '^Destiny.*$'): ")
search_result_obj = xt.search_stream(search_string, stream_type=("movies","series"))
result_number = len(search_result_obj)
print(f"\tFound {result_number} results")
if result_number < 10:
for stream in search_result_obj:
if 'url' in stream:
print(f"Found `{stream['name']}` at URL: {stream['url']}")
else:
print(stream)
elif choice == 3:
search_string = input("Search for text: ")
search_result_obj = xt.search_stream(
rf"^.*{search_string}.*$", stream_type=("movies","series")
)
result_number = len(search_result_obj)
print(f"\tFound {result_number} results")
if result_number < 10:
for stream in search_result_obj:
if 'url' in stream:
print(f"Found {stream['name']} at URL: {stream['url']}")
else:
print(stream)
elif choice == 4:
stream_id = input("Stream ID: ")
stream_id_number = int(stream_id)
if stream_id_number > 0:
print(f"\n\tFile saved at `{xt.download_video(int(stream_id))}`")
else:
print("\n\tInvalid number")
elif choice == 5:
url = input("Enter URL to download: ")
filename = input("Enter Fullpath Filename: ")
xt._download_video_impl(url,filename)
elif choice == 6:
NUM_MOVIES = len(xt.movies_30days)
print(f"Found {NUM_MOVIES} new movies in the past 30 days")
if NUM_MOVIES < 20:
for i in range(0,NUM_MOVIES):
print(xt.movies_30days[i].title)
elif choice == 7:
NUM_MOVIES = len(xt.movies_7days)
print(f"Found {NUM_MOVIES} new movies in the past 7 days")
if NUM_MOVIES < 20:
for i in range(0,NUM_MOVIES):
print(xt.movies_7days[i].title)
elif choice == 8:
series_id = input("Series ID: ")
data = xt._load_series_info_by_id_from_provider(series_id)
print(data)
elif choice == 9:
stream_id = input("Stream ID: ")
data = xt.liveEpgByStream(stream_id)
print(data)