MammaMia version 2.0.0
This commit is contained in:
UrloMythus
+57
-75
@@ -6,9 +6,14 @@ from fake_headers import Headers
|
||||
from Src.Utilities.loadenv import load_env
|
||||
import json, random
|
||||
from Src.Utilities.info import get_info_imdb,get_info_tmdb
|
||||
from Src.Utilities.eval import eval_solver
|
||||
import urllib.parse
|
||||
from Src.Utilities.convert import get_IMDB_id_from_TMDb_id
|
||||
from Src.API.extractors.supervideo import supervideo
|
||||
import logging
|
||||
from Src.Utilities.config import setup_logging
|
||||
level = config.LEVEL
|
||||
logger = setup_logging(level)
|
||||
|
||||
env_vars = load_env()
|
||||
GS_PROXY = config.GS_PROXY
|
||||
proxies = {}
|
||||
@@ -32,60 +37,28 @@ else:
|
||||
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, proxies = proxies)
|
||||
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_imdb(clean_id,client):
|
||||
try:
|
||||
headers = random_headers.generate()
|
||||
response = await client.get(ForwardProxy + f'{GS_DOMAIN}/?story={clean_id}&do=search&subaction=search', allow_redirects=True, impersonate = "chrome124", headers = headers, proxies = proxies)
|
||||
response = await client.get(ForwardProxy + f'{GS_DOMAIN}/?story={clean_id}&do=search&subaction=search', allow_redirects=True, headers = headers, proxies = proxies)
|
||||
if response.status_code != 200:
|
||||
print(f"Guardaserie Failed to fetch search results: {response.status_code}")
|
||||
logger.warning(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:
|
||||
logger.warning(f'GS {e}')
|
||||
return None
|
||||
|
||||
async def search(showname,date,client):
|
||||
try:
|
||||
headers = random_headers.generate()
|
||||
showname = urllib.parse.quote(showname)
|
||||
response = await client.get(ForwardProxy + f'{GS_DOMAIN}/?story={showname}&do=search&subaction=search', allow_redirects=True, impersonate = "chrome124", headers = headers, proxies = proxies)
|
||||
response = await client.get(ForwardProxy + f'{GS_DOMAIN}/?story={showname}&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}")
|
||||
logger.warning(f"Guardaserie Failed to fetch search results: {response.status_code}")
|
||||
soup = BeautifulSoup(response.text,'lxml',parse_only=SoupStrainer('div'))
|
||||
div_mlnew = soup.find_all('div', class_='mlnew')
|
||||
for mlnew in div_mlnew:
|
||||
@@ -104,7 +77,7 @@ async def search(showname,date,client):
|
||||
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)
|
||||
response = await client.get(ForwardProxy + page_url, allow_redirects=True, 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']
|
||||
@@ -116,19 +89,55 @@ async def player_url(page_url, season, episode,client):
|
||||
|
||||
|
||||
|
||||
async def guardaserie(id,client):
|
||||
async def guardaserie(streams,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 streams
|
||||
if ismovie == 0:
|
||||
season = general[2]
|
||||
episode = general[3]
|
||||
type = "Guardaserie"
|
||||
page_url = await search_imdb(clean_id,client)
|
||||
supervideo_link =await player_url(page_url,season,episode,client)
|
||||
if supervideo_link:
|
||||
streams = await supervideo(supervideo_link, client, streams,"Guardaserie",proxies, ForwardProxy)
|
||||
return streams
|
||||
else:
|
||||
logger.warning(" GS Couldn't find iframe source")
|
||||
return streams
|
||||
except Exception as e:
|
||||
logger.warning(f"MammaMia: Guardaserie Failed{e}")
|
||||
return streams
|
||||
|
||||
|
||||
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 = "tt10919420:1:1" # This is an example ID format tt0460649
|
||||
results = await guardaserie({'streams': []},test_id, client)
|
||||
print(results)
|
||||
|
||||
if __name__ == "__main__":
|
||||
import asyncio
|
||||
asyncio.run(test_script())
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
'''
|
||||
if "tt" not in clean_id:
|
||||
clean_id = await get_IMDB_id_from_TMDb_id(clean_id,client)
|
||||
print(clean_id)
|
||||
'''
|
||||
type = "Guardaserie"
|
||||
'''
|
||||
'''
|
||||
if "tt" in id:
|
||||
showname,date = await get_info_imdb(clean_id,ismovie,type,client)
|
||||
else:
|
||||
@@ -136,34 +145,7 @@ async def guardaserie(id,client):
|
||||
showname = showname.replace("'"," ")
|
||||
if "Guru" in showname:
|
||||
showname = showname.split("-")[0]
|
||||
'''
|
||||
if ismovie == 1:
|
||||
return None
|
||||
'''
|
||||
page_url = await search_imdb(clean_id,client)
|
||||
'''
|
||||
'''
|
||||
'''
|
||||
page_url = await search(showname,date,client)
|
||||
'''
|
||||
supervideo_link =await player_url(page_url,season,episode,client)
|
||||
if supervideo_link:
|
||||
final_url = await eval_solver(supervideo_link,proxies, ForwardProxy, client)
|
||||
return final_url
|
||||
else:
|
||||
print("Couldn't find iframe source")
|
||||
return None
|
||||
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 = "tt6226232:3:18" # This is an example ID format tt0460649
|
||||
results = await guardaserie(test_id, client)
|
||||
print(results)
|
||||
|
||||
if __name__ == "__main__":
|
||||
import asyncio
|
||||
asyncio.run(test_script())
|
||||
'''
|
||||
Reference in New Issue
Block a user