Update openssl check.

This commit is contained in:
Lovi 2024-06-17 22:33:10 +02:00
parent e05ffbef25
commit 4766a1017d
7 changed files with 6 additions and 508 deletions

View File

@ -71,11 +71,14 @@ if crypto_installed:
else:
# Check if openssl command is available
try:
openssl_available = subprocess.run(["openssl", "version"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode == 0
logging.info("Decrypy use: OPENSSL")
except:
openssl_available = False
if not openssl_available:
console.log("[red]Neither Crypto nor openssl is installed. Please install either one of them.")
console.log("[red]Neither python library: pycryptodome nor openssl software is installed. Please install either one of them. Read readme.md [Requirement].")
sys.exit(0)
class M3U8_Decryption:

View File

@ -1,84 +0,0 @@
# 15.05.24
# Fix import
import sys
import os
src_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
sys.path.append(src_path)
# Import
from Src.Api.Streamingcommunity.Core.Util.get_domain import grab_sc_top_level_domain
from Src.Api.Animeunity.Core.Util.get_domain import grab_au_top_level_domain
import unittest
import time
import logging
class URLFilter(logging.Filter):
def __init__(self, url):
super().__init__()
self.url = url
def filter(self, record):
return self.url not in record.getMessage()
# Configure logging
logging.basicConfig()
logger = logging.getLogger()
logger.setLevel(logging.ERROR) # Set logger level to ERROR
# Add custom filter to suppress URLs
url_filters = ['https://streamingcommunity', 'https://www.animeunity']
for url in url_filters:
logger.addFilter(URLFilter(url))
# Variable
real_stream_domain = "foo"
real_anime_domain = "to"
class TestGrabStreamingDomain(unittest.TestCase):
def test_light_streaming(self):
start = time.time()
result = grab_sc_top_level_domain(method='light')
end = time.time()
print(f"Light streaming: {result}, in: {end - start}")
# Assert that result is as expected
self.assertEqual(result, real_stream_domain)
def test_strong_streaming(self):
start = time.time()
result = grab_sc_top_level_domain(method='strong')
end = time.time()
print(f"Strong streaming: {result}, in: {end - start}")
# Assert that result is as expected
self.assertEqual(result, real_stream_domain)
def test_light_anime(self):
start = time.time()
result = grab_au_top_level_domain(method='light')
end = time.time()
print(f"Light anime: {result}, in: {end - start}")
# Assert that result is as expected
self.assertEqual(result, real_anime_domain)
def test_strong_anime(self):
start = time.time()
result = grab_au_top_level_domain(method='strong')
end = time.time()
print(f"Strong anime: {result}, in: {end - start}")
# Assert that result is as expected
self.assertEqual(result, real_anime_domain)
if __name__ == '__main__':
unittest.main()

View File

@ -1,96 +0,0 @@
# 15.05.24
# Fix import
import sys
import os
src_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
sys.path.append(src_path)
# Import
from Src.Api.Streamingcommunity.Core.Class.EpisodeType import EpisodeManager
import unittest
class TestEpisodeManager(unittest.TestCase):
def test_get_length(self):
# Test data
json_ep_1 = {
"id":59517,
"number":1,
"name":"La Fine",
"plot":"Okey dokey...",
"duration":75,
"scws_id":220399,
"season_id":3883,
"created_by":"None",
"created_at":"2024-04-11T01:31:02.000000Z",
"updated_at":"2024-04-11T01:31:02.000000Z",
"images":[
{
"id":103989,
"filename":"b7bd00f0-c420-471a-8cc0-1877b3c6182b.webp",
"type":"cover",
"imageable_type":"episode",
"imageable_id":59517,
"created_at":"2024-04-11T01:31:02.000000Z",
"updated_at":"2024-04-11T01:31:02.000000Z",
"original_url_field":"None"
}
]
}
json_ep_2 = {'id': 59525, 'number': 2, 'name': "L'obiettivo", 'plot': 'So che quassù non è stato facile per voi', 'duration': 66, 'scws_id': 220404, 'season_id': 3883, 'created_by': None, 'created_at': '2024-04-11T01:38:02.000000Z', 'updated_at': '2024-04-11T01:38:03.000000Z', 'images': [{'id': 103997, 'filename': 'fe838c36-50d0-4096-b4e3-68228c7c9e40.webp', 'type': 'cover', 'imageable_type': 'episode', 'imageable_id': 59525, 'created_at': '2024-04-11T01:38:02.000000Z', 'updated_at': '2024-04-11T01:38:02.000000Z', 'original_url_field': None}]}
json_ep_3 = {'id': 59531, 'number': 3, 'name': 'La Testa', 'plot': "La Zona Contaminata ha la sua Regola d'Oro...", 'duration': 57, 'scws_id': 220409, 'season_id': 3883, 'created_by': None, 'created_at': '2024-04-11T01:50:02.000000Z', 'updated_at': '2024-04-11T01:50:03.000000Z', 'images': [{'id': 104003, 'filename': 'e0d105aa-01a9-400d-b257-bcd3cddd164c.webp', 'type': 'cover', 'imageable_type': 'episode', 'imageable_id': 59531, 'created_at': '2024-04-11T01:50:02.000000Z', 'updated_at': '2024-04-11T01:50:02.000000Z', 'original_url_field': None}]}
# Initialize the episode manager
eps_manager = EpisodeManager()
# Add episodes to the episode manager
eps_manager.add_episode(json_ep_1)
eps_manager.add_episode(json_ep_2)
eps_manager.add_episode(json_ep_3)
# Verify if the number of episodes in the episode manager is correct
self.assertEqual(eps_manager.get_length(), 3)
def test_add_episode(self):
# Test data
json_ep_1 = {
"id":59517,
"number":1,
"name":"La Fine",
"plot":"Okey dokey...",
"duration":75,
"scws_id":220399,
"season_id":3883,
"created_by":"None",
"created_at":"2024-04-11T01:31:02.000000Z",
"updated_at":"2024-04-11T01:31:02.000000Z",
"images":[
{
"id":103989,
"filename":"b7bd00f0-c420-471a-8cc0-1877b3c6182b.webp",
"type":"cover",
"imageable_type":"episode",
"imageable_id":59517,
"created_at":"2024-04-11T01:31:02.000000Z",
"updated_at":"2024-04-11T01:31:02.000000Z",
"original_url_field":"None"
}
]
}
# Initialize the episode manager
eps_manager = EpisodeManager()
# Add the episode to the episode manager
eps_manager.add_episode(json_ep_1)
# Check the ID of the first episode added
first_episode_id = eps_manager.episodes[0].id
self.assertEqual(first_episode_id, json_ep_1['id'])
if __name__ == '__main__':
unittest.main()

View File

@ -1,31 +0,0 @@
# 15.05.24
# Fix import
import sys
import os
src_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
sys.path.append(src_path)
# Import
from Src.Api.Streamingcommunity.Core.Class.SearchType import MediaManager
# Test data
json_media_1 = {'id': 4006, 'slug': 'person-of-interest', 'name': 'Person of Interest', 'type': 'tv', 'score': '8.6', 'sub_ita': 0, 'last_air_date': None, 'seasons_count': 5, 'images': [{'imageable_id': 4006, 'imageable_type': 'title', 'filename': '967d17c0-2a64-43b9-860c-a14b650cfbd3.webp', 'type': 'poster', 'original_url_field': None}, {'imageable_id': 4006, 'imageable_type': 'title', 'filename': '544889c3-67bf-4c4c-999a-5b9480b42c0f.webp', 'type': 'background', 'original_url_field': None}, {'imageable_id': 4006, 'imageable_type': 'title', 'filename': '3c6cb9ec-4cac-4fcc-ae47-eb2bf9f9a397.webp', 'type': 'cover', 'original_url_field': None}, {'imageable_id': 4006, 'imageable_type': 'title', 'filename': '3ce34b59-ff49-41af-9886-b169043e3f4d.webp', 'type': 'cover_mobile', 'original_url_field': None}, {'imageable_id': 4006, 'imageable_type': 'title', 'filename': '9db4b4d2-2148-44c9-bede-c40faa582f33.webp', 'type': 'logo', 'original_url_field': None}]}
json_media_2 = {'id': 2794, 'slug': 'mucca-e-pollo', 'name': 'Mucca e Pollo', 'type': 'tv', 'score': '7.5', 'sub_ita': 0, 'last_air_date': None, 'seasons_count': 4, 'images': [{'imageable_id': 2794, 'imageable_type': 'title', 'filename': '704cc5d9-6e42-4659-a81b-e41c1c626719.webp', 'type': 'poster', 'original_url_field': None}, {'imageable_id': 2794, 'imageable_type': 'title', 'filename': '4608481d-e97c-44c2-8426-0eb86d435b5f.webp', 'type': 'background', 'original_url_field': None}, {'imageable_id': 2794, 'imageable_type': 'title', 'filename': '57240efa-e295-4905-aee5-3240578e3dfe.webp', 'type': 'cover', 'original_url_field': None}, {'imageable_id': 2794, 'imageable_type': 'title', 'filename': '05827a5f-30a1-4c89-ad14-3eac4b4931c9.webp', 'type': 'cover_mobile', 'original_url_field': None}, {'imageable_id': 2794, 'imageable_type': 'title', 'filename': 'e98ab29b-9ea8-4ce0-8dcb-3d28ebaa571c.webp', 'type': 'logo', 'original_url_field': None}]}
json_media_3 = {'id': 5694, 'slug': 'vatican-girl-la-scomparsa-di-emanuela-orlandi', 'name': 'Vatican Girl: la scomparsa di Emanuela Orlandi', 'type': 'tv', 'score': '8.2', 'sub_ita': 0, 'last_air_date': None, 'seasons_count': 1, 'images': [{'imageable_id': 5694, 'imageable_type': 'title', 'filename': '19e7d5c2-d9f0-4467-b1ca-b9d043de57d5.webp', 'type': 'poster', 'original_url_field': None}, {'imageable_id': 5694, 'imageable_type': 'title', 'filename': '6ae75507-4018-4bec-83e1-85718bf8663d.webp', 'type': 'background', 'original_url_field': None}, {'imageable_id': 5694, 'imageable_type': 'title', 'filename': '05c29b8e-e490-49d2-bc36-68ccd1e9183e.webp', 'type': 'cover', 'original_url_field': None}, {'imageable_id': 5694, 'imageable_type': 'title', 'filename': '331c1787-5a64-49f7-baed-ff47df894f2b.webp', 'type': 'cover_mobile', 'original_url_field': None}, {'imageable_id': 5694, 'imageable_type': 'title', 'filename': '70f487dc-08b9-4c95-9da3-53095e9649e5.webp', 'type': 'logo', 'original_url_field': None}]}
json_media_4 = {'id': 861, 'slug': 'pokemon-detective-pikachu', 'name': 'Pokémon Detective Pikachu', 'type': 'movie', 'score': '7.4', 'sub_ita': 0, 'last_air_date': '2019-05-03', 'seasons_count': 0, 'images': [{'imageable_id': 861, 'imageable_type': 'title', 'filename': '2097306c-6a83-438b-911f-7d600b01dbaf.webp', 'type': 'cover_mobile', 'original_url_field': None}, {'imageable_id': 861, 'imageable_type': 'title', 'filename': 'dc443093-e4cc-465b-8283-acf2d9005127.webp', 'type': 'cover_mobile', 'original_url_field': None}, {'imageable_id': 861, 'imageable_type': 'title', 'filename': '057c99f1-b2fb-455c-8ed1-9f33e19c7914.webp', 'type': 'poster', 'original_url_field': None}, {'imageable_id': 861, 'imageable_type': 'title', 'filename': '6598d047-1a82-4f73-adf5-49729a6f1f45.webp', 'type': 'cover', 'original_url_field': None}, {'imageable_id': 861, 'imageable_type': 'title', 'filename': '70368946-6199-41c2-af77-c5c0b698748d.webp', 'type': 'logo', 'original_url_field': None}, {'imageable_id': 861, 'imageable_type': 'title', 'filename': '55707415-c917-404f-a50a-fba1f0fe2df0.webp', 'type': 'background', 'original_url_field': None}]}
# Init class media search manager
search_ep = MediaManager()
search_ep.add_media(json_media_1)
search_ep.add_media(json_media_2)
search_ep.add_media(json_media_3)
print(search_ep)
print(search_ep.media_list[0])
print(search_ep.media_list[0].images[0])

View File

@ -1,33 +0,0 @@
# 15.05.24
# Fix import
import sys
import os
src_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
sys.path.append(src_path)
# Import
from Src.Api.Streamingcommunity.Core.Class.SeriesType import TitleManager
# Test data
title_data_1 = {'id': 1809, 'number': 1, 'name': None, 'plot': None, 'release_date': None, 'title_id': 4006, 'created_at': '2023-05-18T14:58:39.000000Z', 'updated_at': '2023-05-18T14:58:39.000000Z', 'episodes_count': 23}
title_data_2 = {'id': 1810, 'number': 2, 'name': None, 'plot': None, 'release_date': None, 'title_id': 4006, 'created_at': '2023-05-18T14:58:40.000000Z', 'updated_at': '2023-05-18T14:58:40.000000Z', 'episodes_count': 22}
title_data_3 = {'id': 1811, 'number': 3, 'name': None, 'plot': None, 'release_date': None, 'title_id': 4006, 'created_at': '2023-05-18T14:58:40.000000Z', 'updated_at': '2023-05-18T14:58:40.000000Z', 'episodes_count': 23}
title_data_4 = {'id': 1812, 'number': 4, 'name': None, 'plot': None, 'release_date': None, 'title_id': 4006, 'created_at': '2023-05-18T14:58:41.000000Z', 'updated_at': '2023-05-18T14:58:41.000000Z', 'episodes_count': 22}
title_data_5 = {'id': 1813, 'number': 5, 'name': None, 'plot': None, 'release_date': None, 'title_id': 4006, 'created_at': '2023-05-18T14:58:41.000000Z', 'updated_at': '2023-05-18T14:58:41.000000Z', 'episodes_count': 13}
# Init class series manager
title_man = TitleManager()
title_man.add_title(title_data_1)
title_man.add_title(title_data_2)
title_man.add_title(title_data_3)
title_man.add_title(title_data_4)
title_man.add_title(title_data_5)
print(title_man)
print(title_man.titles[0])

View File

@ -1,220 +0,0 @@
# 15.05.24
# Fix import
import sys
import os
import logging
from urllib.parse import urlencode
src_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
sys.path.append(src_path)
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# Library import
from Src.Lib.Request import requests
import time
import json
import re
import html
from datetime import datetime
def get_current_expiration_timestamp():
"""Get the current Unix timestamp."""
return int(time.time())
def get_current_year_month():
"""Get the current year and month."""
now = datetime.now()
return now.year, now.month
def create_json_file():
"""Create a JSON file if it does not exist."""
if not os.path.exists(JSON_FILE_PATH):
with open(JSON_FILE_PATH, 'w') as file:
json.dump({}, file, indent=4)
def save_playlist_info(year, month, playlist_info):
"""
Save playlist information for a specific year and month.
Args:
year (int): The year.
month (int): The month.
playlist_info (dict): Information about the playlist.
"""
logging.info(f"Saving playlist info for {year}-{month}")
with open(JSON_FILE_PATH, 'r') as file:
data = json.load(file)
year_month_key = f"{year}-{month}"
data[year_month_key] = playlist_info
with open(JSON_FILE_PATH, 'w') as file:
json.dump(data, file, indent=4)
# Variable
year, month = get_current_year_month()
JSON_FILE_PATH = 'playlist_info.json'
def load_playlist_info(year, month):
"""
Load playlist information for a specific year and month.
Args:
year (int): The year.
month (int): The month.
Returns:
dict: Information about the playlist for the specified year and month, or None if not found.
"""
logging.info(f"Loading playlist info for {year}-{month}")
if not os.path.exists(JSON_FILE_PATH):
return None
with open(JSON_FILE_PATH, 'r') as file:
data = json.load(file)
year_month_key = f"{year}-{month}"
return data.get(year_month_key)
def _fetch_webpage_text(url):
"""
Fetch webpage text from a given URL.
Args:
url (str): The URL of the webpage.
Returns:
str: The text content of the webpage.
"""
logging.info(f"Fetching webpage text from URL: {url}")
response = requests.get(url, timeout=3)
response.raise_for_status()
return html.unescape(response.text)
def _extract_playlist_info(webpage_text):
"""
Extract playlist information from webpage text.
Args:
webpage_text (str): The text content of the webpage.
Returns:
dict: Information about the playlist extracted from the webpage.
"""
logging.info("Extracting playlist info from webpage text")
info_match = re.search(r'data-page="([\s\S]+})"', webpage_text)
info_data = info_match.group(1)
print("1. =>", info_data)
info = json.loads(info_data)
logging.debug(f"Playlist info extracted: {info}")
return info
def _extract_iframe_info(playlist_info):
"""
Extract iframe info from playlist information.
Args:
playlist_info (dict): Information about the playlist.
Returns:
str: The text content of the iframe.
"""
embed_url = playlist_info['props']['embedUrl']
logging.info(f"Fetching iframe info from embed URL: {embed_url}")
iframe_text = _fetch_webpage_text(embed_url)
iframe_info_url_match = re.search(r'<iframe[^>]+src="([^"]+)', iframe_text)
iframe_info_url = iframe_info_url_match.group(1)
print("2. =>", iframe_info_url)
return _fetch_webpage_text(iframe_info_url)
def _extract_playlist_url(iframe_info_text):
"""
Extract playlist URL from iframe info text.
Args:
iframe_info_text (str): The text content of the iframe.
Returns:
tuple: A tuple containing playlist information and the playlist URL.
"""
logging.info("Extracting playlist URL from iframe info text")
playlist_match = re.search(r'window\.masterPlaylist[^:]+params:[^{]+({[^<]+?})', iframe_info_text)
playlist_info = json.loads(re.sub(r',[^"]+}', '}', playlist_match.group(1).replace('\'', '"')))
playlist_url_match = re.search(r'window\.masterPlaylist[^<]+url:[^<]+\'([^<]+?)\'', iframe_info_text)
playlist_url = playlist_url_match.group(1)
logging.debug(f"Playlist info: {playlist_info}, Playlist URL: {playlist_url}")
return playlist_info, playlist_url
def _construct_download_url(playlist_info, playlist_url):
"""
Construct download URL from playlist information and URL.
Args:
playlist_info (dict): Information about the playlist.
playlist_url (str): The URL of the playlist.
Returns:
str: The constructed download URL.
"""
logging.info("Constructing download URL")
year, month = get_current_year_month()
save_playlist_info(year, month, playlist_info)
query = urlencode(list(playlist_info.items()))
dl_url = playlist_url + '?' + query
logging.debug(f"Download URL constructed: {dl_url}")
return dl_url
def get_download_url(video_url):
"""
Get the download URL for a video from its URL.
Args:
video_url (str): The URL of the video.
Returns:
str: The download URL for the video.
"""
logging.info(f"Getting download URL for video: {video_url}")
webpage_text = _fetch_webpage_text(video_url)
playlist_info = _extract_playlist_info(webpage_text)
scws_id = playlist_info['props']['title']['scws_id']
# Get episode scws_id
if scws_id is None:
logging.info("Getting id from series")
scws_id = playlist_info['props']['episode']['scws_id']
else:
logging.info("Getting id from film")
loaded_parameter = load_playlist_info(year, month)
if loaded_parameter is None:
iframe_info_text = _extract_iframe_info(playlist_info)
playlist_info, playlist_url = _extract_playlist_url(iframe_info_text)
return _construct_download_url(playlist_info, playlist_url)
else:
playlist_url = f"https://vixcloud.co/playlist/{scws_id}?{urlencode(list(loaded_parameter.items()))}"
return playlist_url
if __name__ == "__main__":
# Create json file
create_json_file()
# Get m3u8 playlist from watch url
video_url = "https://streamingcommunity.foo/watch/8427?e=61248"
download_url = get_download_url(video_url)
print(download_url)

View File

@ -1,41 +0,0 @@
# 15.05.24
# Fix import
import sys
import os
src_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
sys.path.append(src_path)
# Import
from Src.Api.Streamingcommunity.Core.Class.WindowType import WindowVideo, WindowParameter
import unittest
class TestWindowVideo(unittest.TestCase):
def test_str(self):
# Test data
json_video = {'id': 220399, 'name': 'Fallout S:1 E:1', 'filename': 'Fallout.S01E01.La.Fine.1080p.AMZN.WEB-DL.ITA.ENG.DDP5.1.H.264.mkv', 'size': 5250906, 'quality': 1080, 'duration': 4498, 'views': 0, 'is_viewable': 1, 'status': 'public', 'fps': 24, 'legacy': 0, 'folder_id': '81aff513-a0c4-4723-bacc-fca2d1143eb5', 'created_at_diff': '3 settimane fa'}
expected_output = "WindowVideo(id=220399, name='Fallout S:1 E:1', filename='Fallout.S01E01.La.Fine.1080p.AMZN.WEB-DL.ITA.ENG.DDP5.1.H.264.mkv', size='5250906', quality='1080', duration='4498', views=0, is_viewable=1, status='public', fps=24, legacy=0, folder_id=81aff513-a0c4-4723-bacc-fca2d1143eb5, created_at_diff='3 settimane fa')"
# Initialize WindowVideo object
win_video = WindowVideo(json_video)
# Assert
self.assertEqual(str(win_video), expected_output)
class TestWindowParameter(unittest.TestCase):
def test_str(self):
# Test data
json_parameter = {'token': 'cWDn4XKemoNr7PnPghud2Q', 'token360p': '', 'token480p': 'arPfGVKQ7Dk7wh9xJ8QJDA', 'token720p': 'fuEf67Z1WuD-OgWzS7OxxA', 'token1080p': '3O-DrtT7I3ZmecYQgpC45A', 'expires': '1720094851'}
expected_output = "WindowParameter(token='cWDn4XKemoNr7PnPghud2Q', token360p='', token480p='arPfGVKQ7Dk7wh9xJ8QJDA', token720p='fuEf67Z1WuD-OgWzS7OxxA', token1080p='3O-DrtT7I3ZmecYQgpC45A', expires='1720094851')"
# Initialize WindowParameter object
win_par = WindowParameter(json_parameter)
# Assert
self.assertEqual(str(win_par), expected_output)
if __name__ == '__main__':
unittest.main()