reloia/MammaMia
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

MammaMia New Version 1.5, fixed sources and Live TV, added support for MFP, added support for quick deployment with vercel and render, added proxy support

Browse Source
This commit is contained in:
UrloMythus
2025-01-01 10:13:57 +01:00 Unverified
parent 60b40eab6d
commit c073c50804
29 changed files with 1135 additions and 540 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Src/API/cb01.py
+80 -52
View File
@@ -5,12 +5,11 @@ from Src.Utilities.convert import get_TMDb_id_from_IMDb_id
from fake_headers import Headers
import Src.Utilities.config as config
from Src.Utilities.loadenv import load_env
HF = config.HF
env_vars = load_env()
ForwardProxy_list = config.ForwardProxy
ForwardProxy = ForwardProxy_list[0]
PROXY_CREDENTIALS = env_vars.get('PROXY_CREDENTIALS')
CB_DOMAIN = config.CB_DOMAIN
CB_PROXY = config.CB_PROXY
if CB_PROXY == "1":
env_vars = load_env()
PROXY_CREDENTIALS = env_vars.get('PROXY_CREDENTIALS')
fake_headers = Headers()
async def get_stayonline(link,client):
@@ -22,14 +21,13 @@ async def get_stayonline(link,client):
data = {'id': link.split("/")[-2], 'ref': ''}
response = await client.post('https://stayonline.pro/ajax/linkEmbedView.php', headers=headers, data=data)
real_url = response.json()['data']['value']
print(real_url)
return real_url
async def get_uprot(link,client):
if "msf" in link:
link = link.replace("msf","mse")
if PROXY_CREDENTIALS :
if CB_PROXY == "1":
import json
import random
proxy_list = json.loads(PROXY_CREDENTIALS)
@@ -45,18 +43,22 @@ async def get_uprot(link,client):
proxies = {}
headers = fake_headers.generate()
print(ForwardProxy + link)
response = await client.get(ForwardProxy + link, headers=headers, allow_redirects=True, timeout=10, proxies=proxies)
response = await client.get(link, headers=headers, allow_redirects=True, timeout=10, proxies=proxies, impersonate = "chrome124")
soup = BeautifulSoup(response.text, "lxml")
maxstream_url = soup.find("a")
maxstream_url = maxstream_url.get("href")
return maxstream_url
async def get_true_link_mixdrop(real_link,client):
async def get_true_link_mixdrop(real_link,client,MFP):
try:
import string
if "club" in real_link:
real_link = real_link.replace("club","ps").split('/2')[0]
if MFP == "1":
return real_link
headers = fake_headers.generate()
response = await client.get(real_link, headers=headers, allow_redirects=True,timeout = 30)
response = await client.get(real_link, headers=headers, allow_redirects=True,timeout = 30, impersonate = "chrome124")
[s1, s2] = re.search(r"\}\('(.+)',.+,'(.+)'\.split", response.text).group(1, 2)
schema = s1.split(";")[2][5:-1]
terms = s2.split("|")
@@ -70,11 +72,24 @@ async def get_true_link_mixdrop(real_link,client):
print(s)
return s
except Exception as e:
print(e)
return None
async def get_true_link_maxstream(maxstream_url,client):
headers = fake_headers.generate()
if CB_PROXY == "1":
import json
import random
proxy_list = json.loads(PROXY_CREDENTIALS)
proxy = random.choice(proxy_list)
if proxy == " ":
proxies = {
"http": proxy,
"https": proxy
}
else:
proxies = {}
# Send a GET request to the Maxstream URL
response = await client.get(ForwardProxy + maxstream_url, headers=headers, allow_redirects=True, timeout=10)
response = await client.get(maxstream_url, headers=headers, allow_redirects=True, timeout=10,proxies = proxies, impersonate = "chrome124")
[s1, s2] = re.search(r"\}\('(.+)',.+,'(.+)'\.split", response.text).group(1, 2)
terms = s2.split("|")
urlset_index = terms.index('urlset')
@@ -111,9 +126,11 @@ async def search_movie(showname,date,client):
try:
showname = showname.replace(" ","+")
headers = fake_headers.generate()
headers['Referer'] = 'https://cb01.technology/'
query = f'https://cb01.technology/?s={showname}'
response = await client.get(query,headers=headers)
headers['Referer'] = f'https://cb01new.{CB_DOMAIN}/'
query = f'https://cb01new.{CB_DOMAIN}/?s={showname}'
response = await client.get(query,headers=headers, impersonate = "chrome124")
if response.status_code != 200:
print(f"CB01 Failed to fetch search results: {response.status_code}")
soup = BeautifulSoup(response.text, 'lxml',parse_only=SoupStrainer('div', class_='card-content'))
cards = soup.find_all('div', class_='card-content')
year_pattern = re.compile(r'(19|20)\d{2}')
@@ -121,18 +138,15 @@ async def search_movie(showname,date,client):
# Find the link inside the current card
link_tag = card.find('h3', class_='card-title').find('a')
href = link_tag['href']
# Find the date span and extract possible years
date_span = card.find('span', style=re.compile('color'))
if date_span:
date_text = date_span.text
date_text = href.split("/")[-2]
# Search for the first occurrence of a year (starting with 19 or 20)
match = year_pattern.search(date_text)
if match:
year = match.group(0)
if year == date : # Check if the year is 2011
print(f"Link: {href}")
return href
match = year_pattern.search(date_text)
if match:
year = match.group(0)
if year == date : # Check if the year is 2011
return href
except Exception as e:
print(f'MammaMia: Error in search_series cb01: {e}')
@@ -141,9 +155,11 @@ async def search_series(showname,date,client):
try:
showname = showname.replace(" ","+")
headers = fake_headers.generate()
headers['Referer'] = 'https://cb01.technology/serietv/'
query = f'https://cb01.technology/serietv/?s={showname}'
response = await client.get(query,headers=headers)
headers['Referer'] = f'https://cb01new.{CB_DOMAIN}/serietv/'
query = f'https://cb01new.{CB_DOMAIN}/serietv/?s={showname}'
response = await client.get(query,headers=headers,impersonate = "chrome124")
if response.status_code != 200:
print(f"CB01 Failed to fetch search results: {response.status_code}")
soup = BeautifulSoup(response.text, 'lxml',parse_only=SoupStrainer('div', class_='card-content'))
cards = soup.find_all('div', class_='card-content')
year_pattern = re.compile(r'(19|20)\d{2}')
@@ -161,23 +177,35 @@ async def search_series(showname,date,client):
if match:
year = match.group(0)
if year == date : # Check if the year is 2011
print(f"Link: {href}")
return href
except Exception as e:
print(f'MammaMia: Error in search_series cb01: {e}')
async def movie_redirect_url(link,client):
async def movie_redirect_url(link,client,MFP):
headers = fake_headers.generate()
response = client.get(link, headers=headers, allow_redirects=True, timeout=10)
response = await client.get(link, headers=headers, allow_redirects=True, timeout=10, impersonate = "chrome124")
# Extract the redirect URL from the HTML
soup = BeautifulSoup(response.text, "lxml",parse_only=SoupStrainer('div', id='iframen1'))
redirect_url = soup.find("div", id="iframen1").get("data-src")
soup = BeautifulSoup(response.text, "lxml",parse_only=SoupStrainer('div'))
redirect_url = soup.find("div", id="iframen2").get("data-src")
try:
if "stayonline" in redirect_url:
mixdrop_real_link = await get_stayonline(redirect_url,client)
final_url = await get_true_link_mixdrop(mixdrop_real_link,client,MFP)
return final_url
except Exception as e:
redirect_url = soup.find("div", id="iframen1").get("data-src")
if "stayonline" in maxstream_link:
maxstream_link =await get_stayonline(maxstream_link,client)
maxstream_real_link = await get_uprot(maxstream_link,client)
final_url = await get_true_link_maxstream(maxstream_real_link,client)
return(final_url)
return redirect_url
async def series_redirect_url(link,season,episode,client):
async def series_redirect_url(link,season,episode,client,MFP):
if len(episode) == 1:
episode = f'0{episode}'
headers = fake_headers.generate()
response = await client.get(link, headers=headers, allow_redirects=True, timeout=10)
response = await client.get(link, headers=headers, allow_redirects=True, timeout=10,impersonate = "chrome124")
soup = BeautifulSoup(response.text, "lxml")
seasons_text = soup.find_all('div', class_='sp-head')
for season_text in seasons_text:
@@ -190,11 +218,11 @@ async def series_redirect_url(link,season,episode,client):
sp_body = season_text.find_next('div', class_='sp-body')
link_tag = sp_body.find('a')
uprot_long = link_tag.get('href')
response = await client.get(uprot_long, headers=headers, allow_redirects=True, timeout=10)
response = await client.get(uprot_long, headers=headers, allow_redirects=True, timeout=10, impersonate = "chrome124")
season = "01"
episode = "04"
pattern1 = rf"(?i)\S*?\.{season}x{episode}\S*?\.mkv.*?href=['\"](.*?)['\"]"
pattern2 = rf"\S*\.S{season}E{episode}\S*?\.mkv.*?href=['\"](.*?)['\"]"
pattern2 = rf"\S*\.S{season}E{episode}\S*?\.mkv|\.mp4|\.avi.*?href=['\"](.*?)['\"]"
match = re.search(pattern1, response.text)
if match:
maxstream_link = match.group(1)
@@ -202,13 +230,11 @@ async def series_redirect_url(link,season,episode,client):
match = re.search(pattern2, response.text)
if match:
maxstream_link = match.group(1)
print(maxstream_link)
if "stayonline" in maxstream_link:
maxstream_link =await get_stayonline(maxstream_link,client)
maxstream_real_link = await get_uprot(maxstream_link,client)
final_url = await get_true_link_maxstream(maxstream_real_link,client)
print(final_url)
return final_url
else:
pattern = re.compile(r'4&#215;03\s*&#8211;.*?<a href="(.*?)".*?>Maxstream</a>\s*&#8211;.*?<a href="(.*?)".*?>Mixdrop</a>', re.DOTALL)
match = pattern.search(response.text)
@@ -217,21 +243,19 @@ async def series_redirect_url(link,season,episode,client):
print(f"Maxstream: {maxstream_link}")
print(f"Mixdrop: {mixdrop_link}")
mixdrop_real_link = await get_stayonline(mixdrop_link,client)
final_url = await get_true_link_mixdrop(mixdrop_real_link,client)
final_url = await get_true_link_mixdrop(mixdrop_real_link,client,MFP)
if final_url == None:
if "stayonline" in maxstream_link:
maxstream_link =await get_stayonline(maxstream_link,client)
maxstream_real_link = await get_uprot(maxstream_link,client)
final_url = await get_true_link_maxstream(maxstream_real_link,client)
print(final_url)
final_url = await get_true_link_maxstream(maxstream_real_link,client)
return final_url
async def cb01(id,client):
async def cb01(id,client,MFP):
try:
general = is_movie(id)
general = await is_movie(id)
ismovie = general[0]
real_id = general[1]
type = "Cb01"
@@ -243,12 +267,14 @@ async def cb01(id,client):
season = general[2]
episode = general[3]
link = await search_series(showname,date,client)
redirect_url = await series_redirect_url(link,season,episode,client)
final_url = await series_redirect_url(link,season,episode,client,MFP)
return final_url
elif ismovie == 1:
season = None
episode = None
link = await search_movie(showname,date,client)
redirect_url = await movie_redirect_url(link,client)
redirect_url = await movie_redirect_url(link,client,MFP)
return redirect_url
except Exception as e:
print(f'MammaMia: Error in cb01: {e}')
return None
@@ -275,8 +301,10 @@ async def test_animeworld():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
# Replace with actual id, for example 'anime_id:episode' format
test_id = "tt0158552:1:1" # This is an example ID format
results = await cb01(test_id, client)
test_id = "tt14948432" # This is an example ID format
MFP = "1"
results = await cb01(test_id, client,MFP)
print(results)
if __name__ == "__main__":
import asyncio
Src/API/cool.py
+1 -7
View File
@@ -29,12 +29,9 @@ async def get_links(slug,season,episode,ismovie,client):
response = await client.get(f'https://altadefinizioneapp.com/api/post/urls/stream/{slug}',cookies=cookies,headers=headers)
elif ismovie == 0:
print("HERE SEASON",season)
print("HERE EPISODE",episode)
request_url =f'https://altadefinizioneapp.com/api/post/urls/stream/{slug}/{season}/{episode}'
print(request_url)
response = await client.get(request_url,cookies=cookies,headers=headers)
print(response.text)
try:
video_data = response.json() # Assuming this is the JSON response containing video streams
if 'streams' not in video_data:
@@ -92,7 +89,6 @@ async def search_imdb(showname,tmdba,client):
tmdb_id = ''.join(char for char in tmdb_id if char.isdigit())
if tmdb_id == tmdba:
slug = item.get('slug')
print(slug)
return slug
@@ -100,11 +96,9 @@ async def search_imdb(showname,tmdba,client):
def parse_links(resolution_links):
results = {}
if resolution_links:
print("Video links:")
for resolution, link in resolution_links.items():
if "cdn.altadefinizione-originale.com" in link or "cdn.altadefinizioneapp.com" in link:
link = link.replace("cdn.altadefinizione-originale.com","protectlinknt.b-cdn.net").replace("cdn.altadefinizioneapp.com","protectlinknt.b-cdn.net")
print(f"{resolution}: {link}")
results[resolution] = link
return results
else:
@@ -114,7 +108,7 @@ def parse_links(resolution_links):
async def cool(imdb,client):
try:
type = "Cool"
general = is_movie(imdb)
general = await is_movie(imdb)
ismovie = general[0]
imdb_id = general[1]
if ismovie == 0 :
Src/API/ddlstream.py
+19 -19
View File
@@ -3,16 +3,23 @@ import re
from bs4 import BeautifulSoup, SoupStrainer
from Src.Utilities.info import get_info_imdb, is_movie, get_info_tmdb
import urllib.parse
import Src.Utilities.config as config
import urllib.parse
DDL_DOMAIN = config.DDL_DOMAIN
ips4_device_key = config.ips4_device_key
ips4_login_key = config.ips4_login_key
ips4_member_id = config.ips4_member_id
ips4_IPSSessionFront = config.ips4_IPSSessionFront
cookies = {
'ips4_device_key': 'b6f084be79c28bb53ecfc556fd7a2a70',
'ips4_IPSSessionFront': 'f2b9db91793e084cf2003a07c563e6aa',
'ips4_member_id': '4029711',
'ips4_login_key': '41b6a70b16cce2fd39482f5b21a9c35f',
'ips4_device_key': ips4_device_key,
'ips4_IPSSessionFront': ips4_IPSSessionFront,
'ips4_member_id': ips4_member_id,
'ips4_login_key': ips4_login_key,
}
async def search_series(client,id,season,episode,showname):
showname = showname.replace(" ", "%20").replace("–", "+").replace("—","+")
response = await client.get(f"https://ddlstreamitaly.co/search/?&q={showname}%20{season}%20Streaming&type=videobox_video&quick=1&nodes=11&search_and_or=and&search_in=titles&sortby=relevancy")
response = await client.get(f"https://ddlstreamitaly.{DDL_DOMAIN}/search/?&q={showname}%20{season}%20Streaming&type=videobox_video&quick=1&nodes=11,36&search_and_or=and&search_in=titles&sortby=relevancy")
soup = BeautifulSoup(response.text, 'lxml',parse_only=SoupStrainer('a'))
a_tags = soup.find_all('a', {'data-linktype': 'link'})
for a in a_tags:
@@ -58,10 +65,7 @@ async def get_episode(client,link,episode):
}
episode = "6"
response = await client.get(link, cookies = cookies, headers = headers, params = params, impersonate="chrome120")
print(link)
pattern = rf'<a\s+href="([^"]+)"[^>]*>\s*Part {episode}\s*</a>'
match = re.search(pattern, response.text)
@@ -71,9 +75,9 @@ async def get_episode(client,link,episode):
return mp4_link
async def search_movie(client,showname,id):
showname = showname.replace(" ", "%20").replace("–", "+").replace("—","+")
showname = urllib.parse.quote_plus(showname)
link = f"https://ddlstreamitaly.co/search/?&q={showname}%20Streaming&quick=1&nodes=11&search_and_or=and&search_in=titles&sortby=relevancy"
showname = showname.replace("–", "+").replace("—","+")
showname = urllib.parse.quote(showname)
link = f"https://ddlstreamitaly.{DDL_DOMAIN}/search/?&q={showname}%20Streaming&quick=1&nodes=11&search_and_or=and&search_in=titles&sortby=relevancy"
response = await client.get(link,impersonate = "chrome120")
soup = BeautifulSoup(response.text, 'lxml',parse_only=SoupStrainer('a'))
a_tags = soup.find_all('a', {'data-linktype': 'link'})
@@ -116,13 +120,12 @@ async def get_mp4(client,link):
async def ddlstream(imdb,client):
try:
general = is_movie(imdb)
general = await is_movie(imdb)
ismovie = general[0]
id = general[1]
type = "DDLStream"
if "tt" in imdb:
showname = await get_info_imdb(id,ismovie,type,client)
print(showname)
else:
showname = get_info_tmdb(id,ismovie,type)
if ismovie == 0:
@@ -131,24 +134,22 @@ async def ddlstream(imdb,client):
page_link = await search_series(client,id,season,episode,showname)
mp4_link = await get_episode(client,page_link,episode)
final_url = await get_mp4(client,mp4_link)
print(final_url)
return final_url
else:
page_link = await search_movie(client,showname,id)
mp4_link = page_link + "?area=online"
final_url = await get_mp4(client,mp4_link)
print(final_url)
return final_url
except Exception as e:
print(f"MammaMia: DDLStream Failed {e}")
return None,None
return None
'''
async def test_animeworld():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
test_id = "tt6263850" # This is an example ID format
test_id = "tt14948432" # This is an example ID format
results = await ddlstream(test_id, client)
print(results)
@@ -158,4 +159,3 @@ if __name__ == "__main__":
#python3 -m Src.API.ddlstream
'''
Src/API/epg.py
+7 -6
View File
@@ -186,11 +186,11 @@ async def epg_guide(id,client):
elif id in convert_bho_2:
new_id = convert_bho_2[id]
new_id = new_id.replace(" ","%20")
response = await client.get(f"https://mammamia-urlo-boh2.hf.space/{new_id}/now")
response = await client.get(f"https://aimammam-boh2.hf.space/{new_id}/now")
elif id in convert_bho_3:
new_id = convert_bho_3[id]
new_id = new_id.replace(" ","%20")
response = await client.get(f"https://mammamia-urlo-boh3.hf.space/{new_id}/now")
response = await client.get(f"https://aimammam-boh3.hf.space/{new_id}/now")
data = response.json()
description = data['description'].replace("- EPG by epg-guide.com","").replace("No description","")
title = data['title']
@@ -204,10 +204,10 @@ async def epg_guide(id,client):
#async def test_animeworld():
'''
async def test_animeworld():
async with httpx.AsyncClient() as client:
test_id = "rai-news" # Replace with actual ID
test_id = "rai-1" # Replace with actual ID
try:
results = await epg_guide(test_id, client)
print(results)
@@ -216,7 +216,8 @@ async def epg_guide(id,client):
except Exception as e:
print(f"An error occurred: {e}")
#if __name__ == "__main__":
if __name__ == "__main__":
import httpx
import asyncio
asyncio.run(test_animeworld())
'''
Src/API/filmpertutti.py
+121 -95
View File
@@ -4,13 +4,12 @@ from bs4 import BeautifulSoup,SoupStrainer
import string
import re
import dateparser
from Src.Utilities.convert import get_TMDb_id_from_IMDb_id
from Src.Utilities.convert import get_TMDb_id_from_IMDb_id, get_IMDB_id_from_TMDb_id
from Src.Utilities.info import get_info_tmdb, is_movie, get_info_imdb
from Src.Utilities.convert_date import convert_US_date
import Src.Utilities.config as config
FT_DOMAIN = config.FT_DOMAIN
import urllib.parse
WOA = 0
#Some basic headers
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.10; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3',
@@ -22,8 +21,19 @@ month_mapping = {
'May': 'Maggio', 'Jun': 'Giugno', 'Jul': 'Luglio', 'Aug': 'Agosto',
'Sep': 'Settembre', 'Oct': 'Ottobre', 'Nov': 'Novembre', 'Dec': 'Dicembre'
}
async def get_streamtape(link,client):
headers = {
'user-agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36',
}
response = await client.get(link, headers=headers, impersonate = "chrome124")
regex = r"id=.*?(?=')"
matches = re.findall(regex, response.text)
i = 0
final_url = next((f"https://streamtape.com/get_video?{matches[i + 1]}" for i in range(len(matches) - 1) if matches[i] == matches[i + 1]), None)
return final_url
async def search(query,date,client,season,ismovie):
async def search(query,imdb_id,client,season,ismovie):
response = await client.get(query)
response = response.json()
#Get link tid of every item and then open the link to see if the date = date
@@ -32,36 +42,30 @@ async def search(query,date,client,season,ismovie):
tid = json['id']
series_response = await client.get(link, headers=headers, allow_redirects=True, timeout = 30)
series_soup = BeautifulSoup(series_response.text, 'lxml')
release_span = series_soup.find('span', class_='released')
if release_span:
if release_span.text != "Data di uscita: N/A":
date_string = release_span.text.split(': ')[-1] # Get the date part
release_date = date_string[-4:]
if WOA:
for eng, ita in month_mapping.items():
date_string = re.sub(rf'\b{eng}\b', ita, date_string)
# Swap to YY-MM-DD formatting using dateparser
release_date = dateparser.parse(date_string, languages=['it']).strftime("%Y-%m-%d")
if release_date == date:
url = link
tid = tid
if ismovie == 0:
Seasons = series_soup.find_all('span', class_="season-name")
i = 0
for item in Seasons:
season_text = item.text.strip()
if season in season_text and "SUB" not in season_text:
actual_season = i
return url, tid, actual_season
else:
i = i + 1
continue
else:
actual_season = None
return url, tid, actual_season
imdb_pattern = r"'imdb_id':\s*'([^']+)'"
imdb_match = re.search(imdb_pattern, series_response.text)
if imdb_match:
id = imdb_match.group(1)
if id == imdb_id:
url = link
tid = tid
if ismovie == 0:
Seasons = series_soup.find_all('span', class_="season-name")
i = 0
for item in Seasons:
season_text = item.text.strip()
if season in season_text and "SUB" not in season_text:
actual_season = i
return url, tid, actual_season
else:
i = i + 1
continue
else:
actual_season = None
return url, tid, actual_season
else:
print("")
print("There's no IMDB_ID")
def get_episode_link(actual_season,episode,tid,url):
#Get the link from where we have to obtain mixdrop link
@@ -75,18 +79,25 @@ def get_film(url):
return tlink
async def get_real_link(tlink,client):
#Some basic code to get the mixdrop link
page = await client.get(tlink, headers=headers, allow_redirects=True)
soup = BeautifulSoup(page.content, features="lxml",parse_only=SoupStrainer('iframe'))
iframe_src = soup.find('iframe')['src']
iframe_page = await client.get(iframe_src, headers=headers, allow_redirects=True, timeout = 30)
iframe_soup = BeautifulSoup(iframe_page.content, features="lxml")
try:
#Some basic code to get the mixdrop link
page = await client.get(tlink, headers=headers, allow_redirects=True)
soup = BeautifulSoup(page.content, features="lxml",parse_only=SoupStrainer('iframe'))
iframe_src = soup.find('iframe')['src']
iframe_page = await client.get(iframe_src, headers=headers, allow_redirects=True, timeout = 30)
iframe_soup = BeautifulSoup(iframe_page.content, features="lxml", parse_only=SoupStrainer('div',class_="megaButton"))
mega_button = iframe_soup.find('div', attrs={'class': 'megaButton', 'rel': 'nofollow'}, string='MIXDROP')
if mega_button:
real_link = mega_button.get('meta-link')
return real_link
mega_button = iframe_soup.find('div', attrs={'class': 'megaButton', 'rel': 'nofollow'}, string='MIXDROP')
if mega_button:
real_link = mega_button.get('meta-link')
return real_link
else:
mega_button = iframe_soup.find('div', attrs={'class': 'megaButton', 'rel': 'nofollow'}, string='STREAMTAPE')
real_link = mega_button.get('meta-link')
return real_link
except Exception as e:
print(f"Error: {e}")
return None
async def get_true_link(real_link,client):
response = await client.get(real_link, headers=headers, allow_redirects=True,timeout = 30)
[s1, s2] = re.search(r"\}\('(.+)',.+,'(.+)'\.split", response.text).group(1, 2)
@@ -101,70 +112,85 @@ async def get_true_link(real_link,client):
s += d[c] if c in d else c
return s
async def filmpertutti(imdb,client):
general = is_movie(imdb)
ismovie = general[0]
imdb_id = general[1]
type = "Filmpertutti"
if ismovie == 0 :
season = general[2]
episode = int(general[3])
if "tt" in imdb:
if ismovie == 0:
#Get showname and date
showname,date = await get_info_imdb(imdb_id,ismovie,type,client)
else:
#THIS IS needed cause the only way to get all releases dates is by giving a tmdb ID not a IMDB
showname,date = await get_info_imdb(imdb_id,ismovie,type, client)
season = None
elif "tmdb" in imdb:
#Get showname and date
tmdba = imdb_id.replace("tmdb:","")
showname,date = get_info_tmdb(tmdba,ismovie,type)
showname = showname.replace(" ", "+").replace("–", "+").replace("—","+")
showname = urllib.parse.quote_plus(showname)
#Build the query
query = f'https://filmpertutti.{FT_DOMAIN}/wp-json/wp/v2/posts?search={showname}&page=1&_fields=link,id'
async def filmpertutti(imdb,client,MFP):
try:
url,tid,actual_season = await search(query,date,client,season,ismovie)
except:
print("MammaMia: No results found for Filmpertutti")
return None
if ismovie == 0:
episode_link = get_episode_link(actual_season,episode,tid,url)
#Let's get mixdrop link
real_link = await get_real_link(episode_link,client)
#let's get delivery link, streaming link
streaming_link = await get_true_link(real_link,client)
print(streaming_link)
return streaming_link
elif ismovie == 1:
film_link = get_film(url)
#Let's get mixdrop link
real_link = await get_real_link(film_link,client)
#let's get delivery link, streaming link
streaming_link = await get_true_link(real_link,client)
return streaming_link
general = await is_movie(imdb)
ismovie = general[0]
imdb_id = general[1]
type = "Filmpertutti"
if ismovie == 0 :
season = general[2]
episode = int(general[3])
else:
season = None
episode = None
if "tt" in imdb:
showname = await get_info_imdb(imdb_id,ismovie,type,client)
elif "tmdb" in imdb:
#Get showname and date
tmdba = imdb_id.replace("tmdb:","")
imdb_id = await get_IMDB_id_from_TMDb_id(tmdba,client)
showname = get_info_tmdb(tmdba,ismovie,type)
showname = showname.replace(" ", "+").replace("–", "+").replace("—","+")
showname = urllib.parse.quote_plus(showname)
# Build the query
query = f'https://filmpertutti.{FT_DOMAIN}/wp-json/wp/v2/posts?search={showname}&page=1&_fields=link,id'
try:
url,tid,actual_season = await search(query,imdb_id,client,season,ismovie)
except:
print("MammaMia: No results found for Filmpertutti")
return None,None
if ismovie == 0:
episode_link = get_episode_link(actual_season,episode,tid,url)
#Let's get mixdrop link
real_link = await get_real_link(episode_link,client)
if MFP == "1":
if "mixdrop" in real_link:
Host = "Mixdrop"
elif "streamtape" in real_link:
Host = "Streamtape"
return real_link, Host
#let's get delivery link, streaming link
if "mixdrop" in real_link:
streaming_link = await get_true_link(real_link,client)
elif "streamtape" in real_link:
streaming_link = await get_streamtape(real_link,client)
Host = None
return streaming_link, Host
elif ismovie == 1:
film_link = get_film(url)
#Let's get mixdrop link
real_link = await get_real_link(film_link,client)
if MFP == "1":
if "mixdrop" in real_link:
Host = "Mixdrop"
elif "streamtape" in real_link:
Host = "Streamtape"
return real_link, Host
if "mixdrop" in real_link:
streaming_link = await get_true_link(real_link,client)
elif "streamtape" in real_link:
streaming_link = await get_streamtape(real_link,client)
#let's get delivery link, streaming link
Host = None
return streaming_link, Host
except Exception as e:
print(f"Filmpertutti failed: {e}")
return None,None
'''
async def test_animeworld():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
MFP = "1"
# Replace with actual id, for example 'anime_id:episode' format
test_id = "tt4655480:1:1" # This is an example ID format
results = await filmpertutti(test_id, client)
test_id = "tt0413573:19:1" # This is an example ID format
results = await filmpertutti(test_id, client,MFP)
if __name__ == "__main__":
import asyncio
asyncio.run(test_animeworld())
'''
Src/API/guardahd.py
+102
View File
@@ -0,0 +1,102 @@
from Src.Utilities.info import is_movie
from bs4 import BeautifulSoup,SoupStrainer
import re
import Src.Utilities.config as config
from fake_headers import Headers
from Src.Utilities.loadenv import load_env
import json, random
env_vars = load_env()
GH_PROXY = config.GH_PROXY
proxies = {}
if GH_PROXY == "1":
PROXY_CREDENTIALS = env_vars.get('PROXY_CREDENTIALS')
proxy_list = json.loads(PROXY_CREDENTIALS)
proxy = random.choice(proxy_list)
if proxy == "":
proxies = {}
else:
proxies = {
"http": proxy,
"https": proxy
}
GH_ForwardProxy = config.GH_ForwardProxy
if GH_ForwardProxy == "1":
ForwardProxy = env_vars.get('ForwardProxy')
else:
ForwardProxy = ""
#Get domain
GHD_DOMAIN = config.GHD_DOMAIN
random_headers = Headers()
async def get_supervideo_link(link,client):
headers = random_headers.generate()
response = await client.get(ForwardProxy + link, headers=headers, allow_redirects=True,timeout = 30)
s2 = re.search(r"\}\('(.+)',.+,'(.+)'\.split", response.text).group(2)
terms = s2.split("|")
file_index = terms.index('file')
#I start to loop from file cause hfs it's one or two position after it usually.
for i in range(file_index,len(terms)):
if "hfs" in terms[i]:
hfs = terms[i]
break
urlset_index = terms.index('urlset')
hls_index = terms.index('hls')
result = terms[urlset_index + 1 : hls_index]
#If the len is >1 then you have to merge the elements from the last to the first
reversed_elements = result[::-1]
base_url =f"https://{hfs}.serversicuro.cc/hls/"
if len(reversed_elements) == 1:
final_url = base_url + "," + reversed_elements[0] + ".urlset/master.m3u8"
lenght = len(reversed_elements)
i = 1
for element in reversed_elements:
base_url += element + ","
if lenght == i:
base_url += ".urlset/master.m3u8"
else:
i += 1
final_url = base_url
return final_url
async def search(clean_id,client):
headers = random_headers.generate()
response = await client.get(ForwardProxy + f"https://mostraguarda.{GHD_DOMAIN}/set-movie-a/{clean_id}", allow_redirects=True, impersonate = "chrome124", headers = headers, proxies = proxies)
if response.status_code != 200:
print(f"GuardaHD Failed to fetch search results: {response.status_code}")
soup = BeautifulSoup(response.text,'lxml',parse_only=SoupStrainer('li'))
li_tag = soup.find('li', class_='')
href = "https:" + li_tag['data-link']
return href
async def guardahd(id,client):
try:
general = await is_movie(id)
ismovie = general[0]
clean_id = general[1]
if ismovie == 0:
return None
supervideo_link = await search(clean_id,client)
final_url = await get_supervideo_link(supervideo_link,client)
return final_url
except Exception as e:
print("MammaMia: Guardaserie Failed",e)
return None
async def test_script():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
# Replace with actual id, for example 'anime_id:episode' format
test_id = "tt0816692" # This is an example ID format
results = await guardahd(test_id, client)
print(results)
if __name__ == "__main__":
import asyncio
asyncio.run(test_script())
Src/API/guardaserie.py
+129
View File
@@ -0,0 +1,129 @@
from Src.Utilities.info import is_movie
from bs4 import BeautifulSoup,SoupStrainer
import re
import Src.Utilities.config as config
from fake_headers import Headers
from Src.Utilities.loadenv import load_env
import json, random
env_vars = load_env()
GS_PROXY = config.GS_PROXY
proxies = {}
if GS_PROXY == "1":
PROXY_CREDENTIALS = env_vars.get('PROXY_CREDENTIALS')
proxy_list = json.loads(PROXY_CREDENTIALS)
proxy = random.choice(proxy_list)
if proxy == "":
proxies = {}
else:
proxies = {
"http": proxy,
"https": proxy
}
GS_ForwardProxy = config.GS_ForwardProxy
if GS_ForwardProxy == "1":
ForwardProxy = env_vars.get('ForwardProxy')
else:
ForwardProxy = ""
#Get domain
GS_DOMAIN = config.GS_DOMAIN
random_headers = Headers()
async def get_supervideo_link(link,client):
headers = random_headers.generate()
response = await client.get(ForwardProxy + link, headers=headers, allow_redirects=True,timeout = 30)
s2 = re.search(r"\}\('(.+)',.+,'(.+)'\.split", response.text).group(2)
terms = s2.split("|")
file_index = terms.index('file')
#I start to loop from file cause hfs it's one or two position after it usually.
for i in range(file_index,len(terms)):
if "hfs" in terms[i]:
hfs = terms[i]
break
urlset_index = terms.index('urlset')
hls_index = terms.index('hls')
result = terms[urlset_index + 1 : hls_index]
#If the len is >1 then you have to merge the elements from the last to the first
reversed_elements = result[::-1]
base_url =f"https://{hfs}.serversicuro.cc/hls/"
if len(reversed_elements) == 1:
final_url = base_url + "," + reversed_elements[0] + ".urlset/master.m3u8"
lenght = len(reversed_elements)
i = 1
for element in reversed_elements:
base_url += element + ","
if lenght == i:
base_url += ".urlset/master.m3u8"
else:
i += 1
final_url = base_url
return final_url
async def search(clean_id,client):
try:
headers = random_headers.generate()
response = await client.get(ForwardProxy + f'https://guardaserie.{GS_DOMAIN}/?story={clean_id}&do=search&subaction=search', allow_redirects=True, impersonate = "chrome124", headers = headers, proxies = proxies)
if response.status_code != 200:
print(f"Guardaserie Failed to fetch search results: {response.status_code}")
soup = BeautifulSoup(response.text,'lxml',parse_only=SoupStrainer('div',class_="mlnh-2"))
div_mlnh2 = soup.select_one('div.mlnh-2:nth-of-type(2)')
a_tag = div_mlnh2.find('h2').find('a')
href = a_tag['href']
return href
except Exception as e:
return None
async def player_url(page_url, season, episode,client):
try:
headers = random_headers.generate()
response = await client.get(ForwardProxy + page_url, allow_redirects=True, impersonate = "chrome124", headers = headers, proxies = proxies)
soup = BeautifulSoup(response.text,'lxml',parse_only=SoupStrainer('a'))
a_tag = soup.find('a', id = f"serie-{season}_{episode}")
href = a_tag['data-link']
return href
except Exception as e:
return None
async def guardaserie(id,client):
try:
general = await is_movie(id)
ismovie = general[0]
clean_id = general[1]
season = general[2]
episode = general[3]
if ismovie == 1:
return None
page_url = await search(clean_id,client)
supervideo_link =await player_url(page_url,season,episode,client)
final_url = await get_supervideo_link(supervideo_link,client)
return final_url
except Exception as e:
print("MammaMia: Guardaserie Failed",e)
return None
async def test_script():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
# Replace with actual id, for example 'anime_id:episode' format
test_id = "tt16288804:1:1" # This is an example ID format
results = await guardaserie(test_id, client)
print(results)
if __name__ == "__main__":
import asyncio
asyncio.run(test_script())
Src/API/lordchannel.py
+4 -5
View File
@@ -62,16 +62,15 @@ async def search(showname,date,season,episode,ismovie,client):
async def get_m3u8(video_url,client):
response = await client.get(video_url, allow_redirects=True, impersonate = "chrome120")
pattern = r'const\s+videoData\s*=\s*\[(.*?)\];'
pattern = r'https?://[^\s]+\.m3u8'
match = re.search(pattern, response.text)
if match:
video_data = match.group(1).strip().split(', ')
url = video_data[0]
url = match.group(0)
return url
async def lordchannel(imdb,client):
try:
general = is_movie(imdb)
general = await is_movie(imdb)
ismovie = general[0]
imdb_id = general[1]
type = "LordChannel"
@@ -99,7 +98,6 @@ async def lordchannel(imdb,client):
print("MammaMia: Lordchannel Failed",e)
return None,None
'''
async def test_animeworld():
from curl_cffi.requests import AsyncSession
@@ -110,4 +108,5 @@ async def test_animeworld():
if __name__ == "__main__":
import asyncio
asyncio.run(test_animeworld())
#python3 -m Src.API.lordchannel
'''
Src/API/streamingcommunity.py
+64 -24
View File
@@ -9,8 +9,27 @@ from urllib.parse import urlparse, parse_qs
from fake_headers import Headers
from Src.Utilities.loadenv import load_env
import urllib.parse
User_Agent= "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36 OPR/111.0.0.0"
env_vars = load_env()
SC_PROXY = config.SC_PROXY
proxies = {}
if SC_PROXY == "1":
PROXY_CREDENTIALS = env_vars.get('PROXY_CREDENTIALS')
proxy_list = json.loads(PROXY_CREDENTIALS)
proxy = random.choice(proxy_list)
if proxy == "":
proxies = {}
else:
proxies = {
"http": proxy,
"https": proxy
}
SC_ForwardProxy = config.SC_ForwardProxy
if SC_ForwardProxy == "1":
ForwardProxy = env_vars.get('ForwardProxy')
else:
ForwardProxy = ""
#Get domain
SC_DOMAIN = config.SC_DOMAIN
Public_Instance = config.Public_Instance
@@ -27,7 +46,7 @@ async def get_version(client):
random_headers['Referer'] = f"https://streamingcommunity.{SC_DOMAIN}/"
random_headers['Origin'] = f"https://streamingcommunity.{SC_DOMAIN}"
base_url = f'https://streamingcommunity.{SC_DOMAIN}/richiedi-un-titolo'
response = await client.get(base_url, headers=random_headers, allow_redirects = True, impersonate="chrome120")
response = await client.get(ForwardProxy + base_url, headers=random_headers, allow_redirects = True, impersonate="chrome124", proxies = proxies)
#Soup the response
soup = BeautifulSoup(response.text, "lxml")
@@ -46,7 +65,7 @@ async def search(query,date,ismovie, client,SC_FAST_SEARCH,movie_id):
random_headers['Accept'] = 'application/json' # Assuming the API returns JSON
random_headers['Content-Type'] = 'application/json'
#Do a request to get the ID of serie/move and it's slug in the URL
response = await client.get(query, headers = random_headers, allow_redirects=True)
response = await client.get(ForwardProxy + query, headers = random_headers, allow_redirects=True, impersonate = "chrome124", proxies = proxies)
print(response)
response = response.json()
@@ -64,7 +83,7 @@ async def search(query,date,ismovie, client,SC_FAST_SEARCH,movie_id):
random_headers = headers.generate()
random_headers['Referer'] = "https://streamingcommunity.buzz/"
random_headers['Origin'] = "https://streamingcommunity.buzz"
response = await client.get ( f'https://streamingcommunity.{SC_DOMAIN}/titles/{tid}-{slug}', headers = random_headers, allow_redirects=True,impersonate = "chrome120")
response = await client.get ( ForwardProxy + f'https://streamingcommunity.{SC_DOMAIN}/titles/{tid}-{slug}', headers = random_headers, allow_redirects=True,impersonate = "chrome124", proxies = proxies)
soup = BeautifulSoup(response.text, "lxml")
data = json.loads(soup.find("div", {"id": "app"}).get("data-page"))
version = data['version']
@@ -87,9 +106,11 @@ async def get_film(tid,version,client):
random_headers['Origin'] = "https://streamingcommunity.buzz"
random_headers['x-inertia'] = "true"
random_headers['x-inertia-version'] = version
random_headers['User-Agent'] = User_Agent
random_headers['user-agent'] = User_Agent
#Access the iframe
url = f'https://streamingcommunity.{SC_DOMAIN}/iframe/{tid}'
response = await client.get(url, headers=random_headers, allow_redirects=True,impersonate = "chrome120")
response = await client.get(ForwardProxy + url, headers=random_headers, allow_redirects=True,impersonate = "chrome124", proxies = proxies)
iframe = BeautifulSoup(response.text, 'lxml')
#Get the link of iframe
iframe = iframe.find('iframe').get("src")
@@ -98,25 +119,33 @@ async def get_film(tid,version,client):
parsed_url = urlparse(iframe)
query_params = parse_qs(parsed_url.query)
random_headers = headers.generate()
random_headers['Referer'] = "https://streamingcommunity.buzz/"
random_headers['Origin'] = "https://streamingcommunity.buzz"
random_headers['Referer'] = f"https://streamingcommunity.{SC_DOMAIN}/"
random_headers['Origin'] = f"https://streamingcommunity.{SC_DOMAIN}"
random_headers['x-inertia'] = "true"
random_headers['x-inertia-version'] = version
random_headers['User-Agent'] = User_Agent
random_headers['user-agent'] = User_Agent
#Get real token and expires by looking at the page in the iframe, vixcloud/embed
resp = await client.get(iframe, headers = random_headers, allow_redirects=True,impersonate= "chrome120")
resp = await client.get(ForwardProxy + iframe, headers = random_headers, allow_redirects=True,impersonate= "chrome124", proxies = proxies)
soup= BeautifulSoup(resp.text, "lxml")
script = soup.find("body").find("script").text
token = re.search(r"'token':\s*'(\w+)'", script).group(1)
expires = re.search(r"'expires':\s*'(\d+)'", script).group(1)
quality = re.search(r'"quality":(\d+)', script).group(1)
base_url = re.search(r"url:\s*'(https?://[^']+)'", script).group(1)
#Example url https://vixcloud.co/playlist/231315?b=1&token=bce060eec3dc9d1965a5d258dc78c964&expires=1728995040&rendition=1080p
url = f'https://vixcloud.co/playlist/{vixid}.m3u8?expires={expires}'
url = f'https://vixcloud.co/playlist/{vixid}.m3u8?token={token}&expires={expires}'
if 'canPlayFHD' in query_params:
canPlayFHD = 'h=1'
url += "&h=1"
if 'b=1' in base_url:
url += "&b=1"
'''
if 'b' in query_params:
b = 'b=1'
url += "&b=1"
'''
'''
if quality == "1080":
if "&h" in url:
url = url
@@ -125,7 +154,8 @@ async def get_film(tid,version,client):
elif quality == "1080" and "&b" and "&h" not in url:
url = url + "&h=1"
else:
url = url + f"&token={token}"
url = url + f"&token={token}"
'''
url720 = f'https://vixcloud.co/playlist/{vixid}.m3u8'
return url,url720,quality
@@ -135,9 +165,10 @@ async def get_season_episode_id(tid,slug,season,episode,version,client):
random_headers['Origin'] = "https://streamingcommunity.buzz"
random_headers['x-inertia'] = "true"
random_headers['x-inertia-version'] = version
#Set some basic headers for the request
#Get episode ID
response = await client.get(f'https://streamingcommunity.{SC_DOMAIN}/titles/{tid}-{slug}/stagione-{season}', headers=random_headers, allow_redirects=True, impersonate = "chrome120")
response = await client.get(ForwardProxy + f'https://streamingcommunity.{SC_DOMAIN}/titles/{tid}-{slug}/stagione-{season}', headers=random_headers, allow_redirects=True, impersonate = "chrome124", proxies = proxies)
# Print the json got
json_response = response.json().get('props', {}).get('loadedSeason', {}).get('episodes', [])
for dict_episode in json_response:
@@ -147,42 +178,49 @@ async def get_season_episode_id(tid,slug,season,episode,version,client):
async def get_episode_link(episode_id,tid,version,client):
#The parameters for the request
random_headers = headers.generate()
random_headers['Referer'] = "https://streamingcommunity.buzz/"
random_headers['Origin'] = "https://streamingcommunity.buzz"
random_headers['Referer'] = f"https://streamingcommunity.{SC_DOMAIN}/"
random_headers['Origin'] = f"https://streamingcommunity.{SC_DOMAIN}"
random_headers['User-Agent'] = User_Agent
random_headers['user-agent'] = User_Agent
params = {
'episode_id': episode_id,
'next_episode': '1'
}
#Let's try to get the link from iframe source
# Make a request to get iframe source
response = await client.get(f"https://streamingcommunity.{SC_DOMAIN}/iframe/{tid}", params=params, headers = random_headers, allow_redirects=True, impersonate = "chrome120")
response = await client.get(ForwardProxy + f"https://streamingcommunity.{SC_DOMAIN}/iframe/{tid}", params=params, headers = random_headers, allow_redirects=True, impersonate = "chrome124", proxies = proxies)
# Parse response with BeautifulSoup to get iframe source
soup = BeautifulSoup(response.text, "lxml")
iframe = soup.find("iframe").get("src")
vixid = iframe.split("/embed/")[1].split("?")[0]
random_headers = headers.generate()
random_headers['Referer'] = "https://streamingcommunity.buzz/"
random_headers['Origin'] = "https://streamingcommunity.buzz"
random_headers['Referer'] = f"https://streamingcommunity.{SC_DOMAIN}/"
random_headers['Origin'] = f"https://streamingcommunity.{SC_DOMAIN}"
random_headers['x-inertia'] = "true"
random_headers['x-inertia-version'] = version
random_headers['User-Agent'] = User_Agent
random_headers['user-agent'] = User_Agent
parsed_url = urlparse(iframe)
query_params = parse_qs(parsed_url.query)
#Get real token and expires by looking at the page in the iframe, vixcloud/embed
resp = await client.get(iframe, headers = random_headers, allow_redirects=True, impersonate = "chrome120")
resp = await client.get(ForwardProxy + iframe, headers = random_headers, allow_redirects=True, impersonate = "chrome124", proxies = proxies)
soup= BeautifulSoup(resp.text, "lxml")
script = soup.find("body").find("script").text
token = re.search(r"'token':\s*'(\w+)'", script).group(1)
expires = re.search(r"'expires':\s*'(\d+)'", script).group(1)
quality = re.search(r'"quality":(\d+)', script).group(1)
base_url = re.search(r"url:\s*'(https?://[^']+)'", script).group(1)
#Example url https://vixcloud.co/playlist/231315?b=1&token=bce060eec3dc9d1965a5d258dc78c964&expires=1728995040&rendition=1080p
url = f'https://vixcloud.co/playlist/{vixid}.m3u8?expires={expires}'
url = f'https://vixcloud.co/playlist/{vixid}.m3u8?token={token}&expires={expires}'
if 'canPlayFHD' in query_params:
canPlayFHD = 'h=1'
url += "&h=1"
if 'b' in query_params:
if 'b=1' in base_url:
b = 'b=1'
url += "&b=1"
'''
if quality == "1080":
if "&h" in url:
url = url
@@ -191,13 +229,15 @@ async def get_episode_link(episode_id,tid,version,client):
elif quality == "1080" and "&b" and "&h" not in url:
url = url + "&h=1"
else:
url = url + f"&token={token}"
url = url + f"&token={token}"
'''
url720 = f'https://vixcloud.co/playlist/{vixid}.m3u8'
return url,url720,quality
async def streaming_community(imdb,client,SC_FAST_SEARCH):
try:
'''
if Public_Instance == "1":
Weird_Link = json.loads(Alternative_Link)
link_post = random.choice(Weird_Link)
@@ -207,7 +247,8 @@ async def streaming_community(imdb,client,SC_FAST_SEARCH):
quality_sc = response.headers.get('x-quality-sc')
print(quality_sc,url_streaming_community)
return url_streaming_community,url_720_streaming_community,quality_sc
general = is_movie(imdb)
'''
general = await is_movie(imdb)
ismovie = general[0]
imdb_id = general[1]
@@ -271,16 +312,15 @@ async def streaming_community(imdb,client,SC_FAST_SEARCH):
except Exception as e:
print("MammaMia: StreamingCommunity failed",e)
return None,None,None,None
'''
async def test_animeworld():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
# Replace with actual id, for example 'anime_id:episode' format
test_id = "tt0095765" # This is an example ID format
test_id = "tt6468322:1:1" # This is an example ID format
results = await streaming_community(test_id, client,"0")
print(results)
if __name__ == "__main__":
import asyncio
asyncio.run(test_animeworld())
'''
asyncio.run(test_animeworld())
Src/API/streamingwatch.py
+22 -17
View File
@@ -4,7 +4,6 @@ from Src.Utilities.info import get_info_tmdb, is_movie, get_info_imdb
import Src.Utilities.config as config
import re
import json
import urllib.parse
SW_DOMAIN = config.SW_DOMAIN
async def wponce_get(client):
response = await client.get(f"https://www.streamingwatch.{SW_DOMAIN}/contatto/")
@@ -44,15 +43,19 @@ async def search(showname,season,episode,date,ismovie,client):
}
response = await client.post(query,cookies=cookies, headers=headers, data=data)
soup = BeautifulSoup(response.content,'lxml')
page_date = soup.find(id = 'search-cat-year').text.strip()
if page_date == date:
href = soup.find('a')['href']
response = await client.get(href, allow_redirects=True, impersonate = "chrome120")
soup = BeautifulSoup(response.text,'lxml',parse_only=SoupStrainer('iframe'))
iframe = soup.find('iframe')
hdplayer = iframe.get('data-lazy-src')
all_page_dates = soup.find_all(id = 'search-cat-year')
all_hrefs = soup.find_all('a')
for temp_data,temp_href in zip(all_page_dates,all_hrefs):
page_date = temp_data.text.strip()
if date == page_date:
href = temp_href['href']
href = soup.find('a')['href']
response = await client.get(href, allow_redirects=True, impersonate = "chrome120")
soup = BeautifulSoup(response.text,'lxml',parse_only=SoupStrainer('iframe'))
iframe = soup.find('iframe')
hdplayer = iframe.get('data-lazy-src')
return hdplayer
return hdplayer
elif ismovie == 0:
#Some series have the name in english so we first search with the categories option and then we use the obtained ID to get all the episodes
id_response = await client.get(f'https://streamingwatch.{SW_DOMAIN}/wp-json/wp/v2/categories?search={showname}&_fields=id', allow_redirects=True, impersonate = "chrome120")
@@ -75,10 +78,11 @@ async def hls_url(hdplayer,client):
response = await client.get(hdplayer, allow_redirects=True, impersonate = "chrome120")
match = re.search(r'sources:\s*\[\s*\{\s*file\s*:\s*"([^"]*)"', response.text)
url = match.group(1)
return url
async def streamingwatch(imdb,client):
try:
general = is_movie(imdb)
general = await is_movie(imdb)
ismovie = general[0]
imdb_id = general[1]
type = "StreamingWatch"
@@ -97,24 +101,25 @@ async def streamingwatch(imdb,client):
else:
tmdba = imdb_id
showname,date = get_info_tmdb(tmdba,ismovie,type)
showname = showname.replace(" ", "+").replace("–", "+").replace("—","+")
showname = showname.replace(" ", "+").replace("–", "+").replace("—","+").replace("&","")
hdplayer = await search(showname,season,episode,date,ismovie,client)
url = await hls_url(hdplayer,client)
print(f"MammaMia: StreamingWatch found results for {showname}")
return url
if url:
#Fix for Exoplayer which needs .m3u8 in the url to play the file
url = url + ".m3u8"
return url,hdplayer
except Exception as e:
print("MammaMia: StreamingWatch Failed",e)
return None
return None,None
'''
async def test_animeworld():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
# Replace with actual id, for example 'anime_id:episode' format
test_id = "tt16426418" # This is an example ID format
test_id = "tt1431045" # This is an example ID format
results = await streamingwatch(test_id, client)
if __name__ == "__main__":
import asyncio
asyncio.run(test_animeworld())
'''
Src/API/tantifilm.py
+24 -19
View File
@@ -4,23 +4,26 @@ import time
from Src.Utilities.info import is_movie,get_info_imdb,get_info_tmdb
import Src.Utilities.config as config
from Src.Utilities.loadenv import load_env
HF = config.HF
env_vars = load_env()
PROXY_CREDENTIALS = env_vars.get('PROXY_CREDENTIALS')
ForwardProxy_list = config.ForwardProxy
ForwardProxy = ForwardProxy_list[0]
TF_PROXY = config.TF_PROXY
if TF_PROXY == "1":
PROXY_CREDENTIALS = env_vars.get('PROXY_CREDENTIALS')
TF_FORWARDPROXY = config.TF_ForwardProxy
if TF_FORWARDPROXY == "1":
ForwardProxy = env_vars.get('ForwardProxy')
else:
ForwardProxy = ""
TF_DOMAIN = config.TF_DOMAIN
import urllib.parse
async def search(showname,ismovie,date,client):
showname = showname.replace(" ","%20")
showname = urllib.parse.quote_plus(showname)
showname = showname.replace(" ","+")
url = f'https://www.tanti.bond/ajax/posts?q={showname}'
response = await client.post(url, allow_redirects=True, impersonate = "chrome120")
response = await client.get(url, allow_redirects=True)
response = response.json()['data']
if ismovie == 1:
for link in response:
url = link['url']
response = client.get(url, allow_redirects=True, impersonate = "chrome120")
response = await client.get(url, allow_redirects=True, impersonate = "chrome120")
pattern = r'Data di rilascio\s*</div>\s*<div class="text">\s*(\d{4})\s*</div>'
found_date = re.search(pattern, response.text)
release_date = str(found_date.group(1))
@@ -107,7 +110,7 @@ async def get_protect_link(id,url,client):
soup = BeautifulSoup(response.text, "lxml", parse_only=SoupStrainer('iframe'))
protect_link = soup.iframe['src']
if "protect" in protect_link:
url = true_url(protect_link)
url = await true_url(protect_link,client)
links_dict[title] = url
return links_dict
# Get the value of the href attribute
@@ -159,7 +162,10 @@ async def true_url(protect_link,client):
"Range": "bytes=0-",
"Referer": "https://d000d.com/",
}
if HF == "1":
doodstream_url = protect_link
proxies = {}
if TF_PROXY == "1":
import random
import json
print(PROXY_CREDENTIALS)
@@ -172,14 +178,12 @@ async def true_url(protect_link,client):
"http": proxy,
"https": proxy
}
elif TF_FORWARDPROXY == "1":
response = await client.head(protect_link, allow_redirects=True, impersonate = "chrome120", proxies = proxies)
doodstream_url = response.url
else:
proxies = {}
doodstream_url = protect_link
response = await client.get(ForwardProxy + doodstream_url, allow_redirects=True, impersonate = "chrome120", proxies = proxies)
if response.status_code == 200:
# Get unique timestamp for the request
real_time = str(int(time.time()))
@@ -199,7 +203,7 @@ async def true_url(protect_link,client):
print("MammaMia: Found results for Tantifilm")
return real_url
else:
print("No match found in the text.")
print("Tantifilm: No match found in the text. Please be sure you are using a local instance")
return None
print("Error: Could not get the response.")
@@ -211,7 +215,7 @@ async def true_url(protect_link,client):
async def tantifilm(imdb,client,TF_FAST_SEARCH):
urls = None
try:
general = is_movie(imdb)
general = await is_movie(imdb)
ismovie = general[0]
imdb_id = general[1]
if ismovie == 0 :
@@ -278,17 +282,18 @@ async def tantifilm(imdb,client,TF_FAST_SEARCH):
return None
'''
async def test_animeworld():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
# Replace with actual id, for example 'anime_id:episode' format
test_id = "tt6468322:1:1" # This is an example ID format
test_id = "tt0816692" # This is an example ID format
results = await tantifilm(test_id, client,"0")
if __name__ == "__main__":
import asyncio
asyncio.run(test_animeworld())
'''
Src/API/webru.py
+52 -23
View File
@@ -3,6 +3,10 @@ from bs4 import BeautifulSoup,SoupStrainer
import Src.Utilities.config as config
from Src.Utilities.dictionaries import webru_vary,webru_dlhd,skystreaming
from Src.Utilities.loadenv import load_env
from urllib.parse import urlparse
import re
TF_DOMAIN = config.TF_DOMAIN
DLHD_DOMAIN = config.DLHD_DOMAIN
env_vars = load_env()
MEDIAFLOW_PASS = env_vars.get('MEDIAFLOW_PASS')
Referer = "https://ilovetoplay.xyz/"
@@ -25,23 +29,50 @@ headers = {
"Cache-Control": "no-cache",
}
medialink = config.MediaProxy
async def get_stream_link(id,site):
async def get_stream_link(id,site,MFP_CREDENTIALS,client):
try:
if site == "dlhd":
stream_url = "https://xyzdddd.mizhls.ru/lb/" + webru_dlhd[id] + "/index.m3u8"
elif site == "vary":
stream_url = "https://webuit.mizhls.ru/lb/"+ webru_vary[id] + "/index.m3u8"
mediaproxy = config.MediaProxy
medialink = random.choice(mediaproxy)
new_stream_url = f'{medialink}proxy/hls/manifest.m3u8?api_password={MEDIAFLOW_PASS}&d={stream_url}&h_Referer={Referer}&h_Origin={Origin}&h_User-Agent=Mozilla%2F5.0%20(Windows%20NT%2010.0%3B%20Win64%3B%20x64)%20AppleWebKit%2F537.36%20(KHTML%2C%20like%20Gecko)%20Chrome%2F58.0.3029.110%20Safari%2F537.3'
response = await client.get(f"https://thedaddy.{DLHD_DOMAIN}/embed/stream-853.php", impersonate = "chrome124", headers = headers)
soup = BeautifulSoup(response.text, 'lxml', parse_only=SoupStrainer('iframe'))
iframe = soup.find('iframe', id='thatframe')
real_link = iframe.get('src')
response = await client.get(real_link, allow_redirects = False)
pattern = r"source:\s*'([^']*\.m3u8)'"
match = re.search(pattern, response.text)
if match:
m3u8_url = match.group(1) # The URL is captured in the first capturing group
parsed_url = urlparse(m3u8_url)
domain = parsed_url.netloc
return stream_url,Referer,Origin
else:
print("No .m3u8 URL found.")
stream_url = f"https://{domain}/lb/" + webru_dlhd[id] + "/index.m3u8"
elif site == "vary":
response = await client.get(f"https://www.tanti.{TF_DOMAIN}/tv-channel/sky-cinema-action-2", impersonate = "chrome124", headers = headers)
soup = BeautifulSoup(response.text, 'lxml', parse_only=SoupStrainer('iframe'))
iframe = soup.find('iframe', class_='embed-responsive-item')
real_link = iframe.get('src')
response = await client.get(real_link, allow_redirects = False)
pattern = r"source:\s*'([^']*\.m3u8)'"
match = re.search(pattern, response.text)
if match:
m3u8_url = match.group(1) # The URL is captured in the first capturing group
parsed_url = urlparse(m3u8_url)
domain = parsed_url.netloc
else:
print("No .m3u8 URL found.")
stream_url = f"https://{domain}/lb/"+ webru_vary[id] + "/index.m3u8"
mfp_url = MFP_CREDENTIALS[0]
mfp_pass = MFP_CREDENTIALS[1]
new_stream_url = f'{mfp_url}/proxy/hls/manifest.m3u8?api_password={mfp_pass}&d={stream_url}&h_Referer={Referer}&h_Origin={Origin}&h_User-Agent=Mozilla%2F5.0%20(Windows%20NT%2010.0%3B%20Win64%3B%20x64)%20AppleWebKit%2F537.36%20(KHTML%2C%20like%20Gecko)%20Chrome%2F58.0.3029.110%20Safari%2F537.3'
return new_stream_url
except Exception as e:
return None
async def webru(id,site,client):
async def webru(id,site,client,MFP_CREDENTIALS):
try:
new_stream_url = await get_stream_link(id,site,client)
new_stream_url = await get_stream_link(id,site,MFP_CREDENTIALS,client)
return new_stream_url
except Exception as e:
@@ -94,22 +125,20 @@ async def get_skystreaming_url(skystreaming_link,client):
'''
async def webru(id,site,client):
'''
async def webru(id,site,client,MFP_CREDENTIALS):
try:
stream_url, Referer,Origin = await get_stream_link(id,site,client)
mediaproxy = config.MediaProxy
medialink = random.choice(mediaproxy)
new_stream_url = f'{medialink}proxy/hls/manifest.m3u8?api_password={MEDIAFLOW_PASS}&d={stream_url}&h_Referer={Referer}&h_Origin={Origin}&h_User-Agent=Mozilla%2F5.0%20(Windows%20NT%2010.0%3B%20Win64%3B%20x64)%20AppleWebKit%2F537.36%20(KHTML%2C%20like%20Gecko)%20Chrome%2F58.0.3029.110%20Safari%2F537.3'
mfp_url = MFP_CREDENTIALS[0]
mfp_pass = MFP_CREDENTIALS[1]
new_stream_url = f'{mfp_url}/proxy/hls/manifest.m3u8?api_password={mfp_pass}&d={stream_url}&h_Referer={Referer}&h_Origin={Origin}&h_User-Agent=Mozilla%2F5.0%20(Windows%20NT%2010.0%3B%20Win64%3B%20x64)%20AppleWebKit%2F537.36%20(KHTML%2C%20like%20Gecko)%20Chrome%2F58.0.3029.110%20Safari%2F537.3'
return new_stream_url
except Exception as e:
print("WebRu failed",e)
return None
'''
async def webru(id,site,client):
'''
'''
async def webru2(id,site,client):
try:
print(id)
if site == "vary":
@@ -129,8 +158,8 @@ async def webru(id,site,client):
print("WorldSport failed",e)
return None
'''
'''
'''
if id in skystreaming:
i = i+1
url,Host = await get_skystreaming(id,client)
Src/API/whvx.py
+1 -1
View File
@@ -1,7 +1,7 @@
from Src.Utilities.info import is_movie,get_info_imdb,get_info_tmdb
import json
import urllib.parse
#This doesn't work anymore.
headers = {
'User-Agent': 'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:127.0) Gecko/20100101 Firefox/127.0',
'Accept': '*/*',
Src/Utilities/config.py
+33 -3
View File
@@ -15,14 +15,44 @@ LC_DOMAIN = SITE["LordChannel"]['domain']
SW_DOMAIN = SITE["StreamingWatch"]['domain']
AW_DOMAIN = SITE['AnimeWorld']['domain']
SKY_DOMAIN = SITE['SkyStreaming']['domain']
CB_DOMAIN = SITE['CB01']['domain']
DDL_DOMAIN = SITE['DDLStream']['domain']
DLHD_DOMAIN = SITE['DaddyLiveHD']['domain']
GS_DOMAIN = SITE['Guardaserie']['domain']
GHD_DOMAIN = SITE['GuardaHD']['domain']
SC = SITE['StreamingCommunity']['enabled']
FT = SITE['Filmpertutti']['enabled']
TF = SITE['Tantifilm']['enabled']
LC = SITE['LordChannel']['enabled']
SW = SITE['StreamingWatch']['enabled']
AW = SITE['AnimeWorld']['enabled']
SKY = SITE['SkyStreaming']['enabled']
CB = SITE['CB01']['enabled']
DDL = SITE['DDLStream']['enabled']
MYSTERIUS = SITE['Mysterius']['enabled']
GS = SITE['Guardaserie']['enabled']
GHD = SITE['GuardaHD']['enabled']
DLHD = SITE['DaddyLiveHD']['enabled']
TF_ForwardProxy = SITE['Tantifilm']["TF_ForwardProxy"]
SC_ForwardProxy = SITE['StreamingCommunity']["SC_ForwardProxy"]
GS_ForwardProxy = SITE['Guardaserie']["GS_ForwardProxy"]
GH_ForwardProxy = SITE['GuardaHD']["GH_ForwardProxy"]
GS_PROXY = SITE['Guardaserie']["GS_PROXY"]
GH_PROXY = SITE['GuardaHD']["GH_PROXY"]
TF_PROXY = SITE['Tantifilm']["TF_PROXY"]
CB_PROXY = SITE['CB01']["CB_PROXY"]
SC_PROXY = SITE['StreamingCommunity']["SC_PROXY"]
ips4_device_key = SITE['DDLStream']['cookies']["ips4_device_key"]
ips4_IPSSessionFront = SITE['DDLStream']['cookies']["ips4_IPSSessionFront"]
ips4_member_id = SITE['DDLStream']['cookies']["ips4_member_id"]
ips4_login_key = SITE['DDLStream']['cookies']["ips4_login_key"]
#General
GENERAL = config['General']
dotenv = GENERAL["load_env"]
HOST = GENERAL["HOST"]
PORT = GENERAL["PORT"]
HF = GENERAL["HF"]
Icon = GENERAL["Icon"]
Name = GENERAL["Name"]
Public_Instance = GENERAL["Public_Instance"]
MediaProxy = GENERAL["MediaProxy"]
ForwardProxy = GENERAL["ForwardProxy"]
Remote_Instance = GENERAL["Remote_Instance"]
Global_Proxy = GENERAL["Global_Proxy"]
Src/Utilities/convert.py
+12 -1
View File
@@ -11,4 +11,15 @@ async def get_TMDb_id_from_IMDb_id(imdb_id,client):
elif tmbda['tv_results']:
return tmbda['tv_results'][0]['id']
else:
return None
return None
async def get_IMDB_id_from_TMDb_id(tmdb_id,client):
url = f"https://api.themoviedb.org/3/movie/{tmdb_id}?api_key={TMDB_KEY}"
response = await client.get(url)
if response.status_code == 200:
data = response.json()
imdb_id = data.get('imdb_id')
if imdb_id:
return imdb_id
Src/Utilities/dictionaries.py
+36 -30
View File
@@ -1,3 +1,5 @@
import Src.Utilities.config as config
SKY_DOMAIN = config.SKY_DOMAIN
okru = {
"rai-1": "https://ok.ru/videoembed/7868570672752?nochat=1",
"rai-2": "https://ok.ru/videoembed/7805618364016?nochat=1",
@@ -26,7 +28,7 @@ STREAM = {
"title": "Rai 1",
"name": "Full HD Http",
"genres": ["Rai"],
"poster": "https://i.imgur.com/OPV9jJD.png",
"poster": "https://i.imgur.com/bb72vqh.jpeg",
"url": "http://173.208.52.200/rai1/index.m3u8"
},
{
@@ -34,7 +36,7 @@ STREAM = {
"title": "Rai 2",
"name": "Full HD Http",
"genres": ["Rai"],
"poster": "https://i.imgur.com/uHtYy8w.png",
"poster": "https://i.imgur.com/jX95Qod.jpeg",
"url": "http://173.208.52.200/rai2/index.m3u8"
},
{
@@ -42,7 +44,7 @@ STREAM = {
"title": "Rai 3",
"name": "Full HD Http",
"genres": ["Rai"],
"poster": "https://i.imgur.com/iXst6me.png",
"poster": "https://i.imgur.com/fgAh9if.jpeg",
"url": "http://173.208.52.200/rai3/index.m3u8"
},
{
@@ -50,7 +52,7 @@ STREAM = {
"title": "Rai 4",
"name": "Full HD Http",
"genres": ["Rai"],
"poster": "https://i.imgur.com/KGg3hvr.png",
"poster": "https://i.imgur.com/NryLzhA.jpeg",
"url": "http://173.208.52.200/rai4/index.m3u8"
},
{
@@ -844,31 +846,31 @@ extra_sources = {
"doctor-who": ["https://amg00793-amg00793c17-samsung-it-4577.playouts.now.amagi.tv/playlist/amg00793-bbcstudios-drwhoitaly-samsungit/playlist.m3u8"]
}
skystreaming ={
"sky-sport-tennis": "https://skystreaming.guru/embed/MKCKP",
"sky-sport-24": "https://skystreaming.guru/embed/GB7zq",
"sky-sport-251": "https://skystreaming.guru/embed/drHbM",
"sky-sport-252": "https://skystreaming.guru/embed/m5kfg",
"sky-sport-253": "https://skystreaming.guru/embed/YprHf",
"sky-sport-254": "https://skystreaming.guru/embed/wroRV",
"sky-sport-255": "https://skystreaming.guru/embed/0TixN",
"sky-sport-256": "https://skystreaming.guru/embed/8g8bh",
"sky-sport-257": "https://skystreaming.guru/embed/ri859",
"sky-sport-258": "https://skystreaming.guru/embed/UoJuW",
"sky-sport-259": "https://skystreaming.guru/embed/squCy",
"sky-sport-260": "https://skystreaming.guru/embed/Z5h2e",
"sky-sport-261": "https://skystreaming.guru/embed/Hz9FW",
"sky-sport-uno": "https://skystreaming.guru/embed/tb1np",
"sky-sport-f1": "https://skystreaming.guru/embed/MLkym",
"sky-sport-motogp": "https://skystreaming.guru/embed/ARFFv",
"sky-sport-calcio": "https://skystreaming.guru/embed/STOZT",
"sky-sport-arena": "https://skystreaming.guru/embed/NkEda",
"sky-sport-max": "https://skystreaming.guru/embed/4QzWz",
"sky-sport-nba": "https://skystreaming.guru/embed/saR8j",
"sky-sport-golf": "https://skystreaming.guru/embed/s9L5f",
"eurosport-1": "https://skystreaming.guru/embed/Vyv99",
"eurosport-2": "https://skystreaming.guru/embed/um8E5",
"dazn-zona-a": ["https://hls.psz.pm/hls/serie/index.m3u8","https://hls.cls.autos/hls/serie/index.m3u8"],
"sportitalia": "https://hls.cls.autos/hls/sportitalia/index.m3u8",
"sky-sport-tennis": f"https://skystreaming.{SKY_DOMAIN}/embed/MKCKP",
"sky-sport-24": f"https://skystreaming.{SKY_DOMAIN}/embed/GB7zq",
"sky-sport-251": f"https://skystreaming.{SKY_DOMAIN}/embed/drHbM",
"sky-sport-252": f"https://skystreaming.{SKY_DOMAIN}/embed/m5kfg",
"sky-sport-253": f"https://skystreaming.{SKY_DOMAIN}/embed/YprHf",
"sky-sport-254": f"https://skystreaming.{SKY_DOMAIN}/embed/wroRV",
"sky-sport-255": f"https://skystreaming.{SKY_DOMAIN}/embed/0TixN",
"sky-sport-256": f"https://skystreaming.{SKY_DOMAIN}/embed/8g8bh",
"sky-sport-257": f"https://skystreaming.{SKY_DOMAIN}/embed/ri859",
"sky-sport-258": f"https://skystreaming.{SKY_DOMAIN}/embed/UoJuW",
"sky-sport-259": f"https://skystreaming.{SKY_DOMAIN}/embed/squCy",
"sky-sport-260": f"https://skystreaming.{SKY_DOMAIN}/embed/Z5h2e",
"sky-sport-261": f"https://skystreaming.{SKY_DOMAIN}/embed/Hz9FW",
"sky-sport-uno": f"https://skystreaming.{SKY_DOMAIN}/embed/tb1np",
"sky-sport-f1": f"https://skystreaming.{SKY_DOMAIN}/embed/MLkym",
"sky-sport-motogp": f"https://skystreaming.{SKY_DOMAIN}/embed/ARFFv",
"sky-sport-calcio": f"https://skystreaming.{SKY_DOMAIN}/embed/STOZT",
"sky-sport-arena": f"https://skystreaming.{SKY_DOMAIN}/embed/NkEda",
"sky-sport-max": f"https://skystreaming.{SKY_DOMAIN}/embed/4QzWz",
"sky-sport-nba": f"https://skystreaming.{SKY_DOMAIN}/embed/saR8j",
"sky-sport-golf": f"https://skystreaming.{SKY_DOMAIN}/embed/s9L5f",
"eurosport-1": f"https://skystreaming.{SKY_DOMAIN}/embed/Vyv99",
"eurosport-2": f"https://skystreaming.{SKY_DOMAIN}/embed/um8E5",
"dazn-zona-a": ["https://live.servis.hair/hls/serie/index.m3u8","https://live.servis.hair/hls/serie/index.m3u8"],
"sportitalia": "https://live.servis.hair/hls/sportitalia/index.m3u8",
}
huhu = {
"canale-5": "1101559666",
@@ -892,5 +894,9 @@ provider_map = {
'TF_FS': 'TF_FAST_SEARCH',
'TF': 'TANTIFILM', # Assuming 'TF' should also be handled
'FT': 'FILMPERTUTTI',
'AW': 'ANIMEWORLD'
'AW': 'ANIMEWORLD',
'DDL': 'DDLSTREAM',
'CB': 'CB01',
"GS": "GUARDASERIE",
"GHD": "GUARDAHD",
}
Src/Utilities/info.py
+5 -23
View File
@@ -16,21 +16,16 @@ def get_info_tmdb(tmbda,ismovie,type):
show= tv.details(tmbda)
showname = show.name
if type == "Filmpertutti":
date= show.first_air_date
date = date.split("-")[0]
print("Real date",date)
return showname,date
return showname
elif type == "StreamingCommunity":
full_date = show.first_air_date
date = full_date.split("-")[0]
print(date)
return showname,date
elif type == "StreamingCommunityFS":
return showname
elif type == "Tantifilm":
date = show.first_air_date
date = date.split("-")[0]
print("Real date",date)
return showname,date
elif type == "TantifilmFS":
return showname
@@ -39,12 +34,10 @@ def get_info_tmdb(tmbda,ismovie,type):
elif type == "LordChannel":
date = show.first_air_date
date = date.split("-")[0]
print("Real date",date)
return showname,date
elif type == "StreamingWatch":
date = show.first_air_date
date = date.split("-")[0]
print("Real date",date)
return showname,date
elif type == "DDLStream":
return showname
@@ -63,10 +56,7 @@ def get_info_tmdb(tmbda,ismovie,type):
showname= show.title
#Get all release dates
if type == "Filmpertutti":
date = show.release_dates
date = date.split("-")[0]
#GET US RELEASE DATE because filmpertutti somewhy uses US release date
return showname,date
return showname
elif type == "StreamingCommunity":
date = show.release_date
date = date.split("-")[0]
@@ -76,7 +66,6 @@ def get_info_tmdb(tmbda,ismovie,type):
elif type == "Tantifilm":
date = show.release_date
date = date.split("-")[0]
print("Real date",date)
return showname,date
elif type == "TantifilmFS":
return showname
@@ -85,12 +74,10 @@ def get_info_tmdb(tmbda,ismovie,type):
elif type == "LordChannel":
date = show.release_date
date = date.split("-")[0]
print("Real date",date)
return showname,date
elif type == "StreamingWatch":
date = show.release_date
date = date.split("-")[0]
print("Real date",date)
return showname,date
elif type == "DDLStream":
return showname
@@ -109,10 +96,7 @@ async def get_info_imdb(imdb_id, ismovie, type,client):
if ismovie == 0:
showname = data['tv_results'][0]['name']
if type == "Filmpertutti":
date= data['tv_results'][0]['first_air_date']
date = date.split("-")[0]
print("Real date",date)
return showname, date
return showname
elif type == "StreamingCommunity":
date = data['tv_results'][0]['first_air_date']
date = date.split("-")[0]
@@ -141,9 +125,7 @@ async def get_info_imdb(imdb_id, ismovie, type,client):
elif ismovie == 1:
showname= data['movie_results'][0]['title']
if type == "Filmpertutti":
date = data['movie_results'][0]['release_date']
date = date.split("-")[0]
return showname,date
return showname
elif type == "StreamingCommunity":
date = data['movie_results'][0]['release_date']
date = date.split("-")[0]
@@ -179,7 +161,7 @@ async def get_info_kitsu(kitsu_id,client):
def is_movie(imdb_id):
async def is_movie(imdb_id):
if "tmdb:" in imdb_id:
imdb_id = imdb_id.replace("tmdb:","")
if ":" in imdb_id:
Src/Utilities/loadenv.py
+1 -2
View File
@@ -2,7 +2,6 @@ import os
import Src.Utilities.config as config
MYSTERIUS = config.MYSTERIUS
dotenv = config.dotenv
HF = config.HF
Public_Instance = config.Public_Instance
#You need to keep dotenv disabled on remote servers
if dotenv == "1":
@@ -16,7 +15,7 @@ def load_env():
if MYSTERIUS == "1":
env_vars['MYSTERIUS_KEY'] = os.getenv('MYSTERIUS_KEY')
env_vars['PROXY_CREDENTIALS'] = os.getenv('PROXY')
env_vars['MEDIAFLOW_PASS'] = os.getenv('MEDIAFLOW_PASS')
if Public_Instance == "1":
env_vars['ALTERNATIVE_LINK'] = os.getenv('ALTERNATIVE_LINK')
env_vars['ForwardProxy'] = os.getenv('FORWARDPROXY')
return env_vars
Src/Utilities/m3u8.py
+40
View File
@@ -0,0 +1,40 @@
from urllib.parse import urlparse, parse_qs,unquote,quote
from fastapi import APIRouter, HTTPException, Response
from curl_cffi.requests import AsyncSession
router = APIRouter()
User_Agent= "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36 OPR/111.0.0.0"
async def fetch_m3u8(url):
"""Fetch the M3U8 file from the URL and store it in memory."""
async with AsyncSession() as client:
response = await client.get(url,headers = {"User-Agent": User_Agent, "user-agent": User_Agent})
response.raise_for_status() # Raise an error for bad responses
return response.text
@router.get("/clone/manifest.m3u8")
async def clone_m3u8(d: str = None):
if d:
try:
d = unquote(d)
m3u8_content = await fetch_m3u8(d)
return Response(content=m3u8_content, media_type='application/vnd.apple.mpegurl')
except Exception as e:
print(f"Failed to fetch M3U8 file: {e}")
raise HTTPException(status_code=404, detail="M3U8 content not found")
#This is just for testing, ignore that endpoint and the func
@router.api_route("/vixcloud/manifest.m3u8", methods=["GET", "HEAD"])
async def clone2_m3u8(d:str,token:str,expires:str,h:str = None, b:str = None):
try:
m3u8 = f'{d}?token={token}&expires={expires}'
if h:
m3u8 = m3u8 + f'&h={h}'
if b:
m3u8 = m3u8 + f'&b={b}'
m3u8_content = await fetch_m3u8(m3u8)
return Response(content=m3u8_content, media_type='application/vnd.apple.mpegurl')
except Exception as e:
print(f"Failed to fetch M3U8 file: {e}")
raise HTTPException(status_code=404, detail="M3U8 content not found")