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

MammaMia 2.2.0

Browse Source
This commit is contained in:
UrloMythus
2026-05-15 21:53:13 +02:00 Unverified
parent 5d8f1249fe
commit 0f3f144874
21 changed files with 410 additions and 155 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Src/API/cb01.py
+1 -1
View File
@@ -255,7 +255,7 @@ async def cb01(streams,id,MFP,MFP_CREDENTIALS,client):
async def test_animeworld():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
test_id = "tt12324366:2:2" # This is an example ID format
test_id = "tt11378946" # This is an example ID format
MFP = "0"
results = await cb01({'streams': []},test_id,MFP,['test','test'],client)
print(results)
Src/API/extractors/deltabit.py
+1 -1
View File
@@ -66,7 +66,7 @@ async def deltabit(page_url,client,streams,site_name,proxies,ForwardProxy,langua
if link:
link = link[0]
elif extractor == 'Turbovid':
link = await eval_solver(response.text,proxies, ForwardProxy, client)
link = await eval_solver(response.text,proxies, ForwardProxy, client,r'sources:\s*\["([^"]+)"')
if not link:
i = 0
if not i >= 3:
Src/API/extractors/mixdrop.py
+1 -1
View File
@@ -25,7 +25,7 @@ async def mixdrop(url,client,MFP,MFP_CREDENTIALS,streams,site_name,proxies,Forwa
else:
headers = {"accept-language": "en-US,en;q=0.5"}
url = await eval_solver(url, proxies , ForwardProxy, client)
url = await eval_solver(url, proxies , ForwardProxy, client,r'MDCore.wurl ?= ?"(.*?)"')
if url:
url = "https:" + url
logger.info(f"{site_name} on Mixdrop found results for the current ID")
Src/API/extractors/streamhg.py
+35
View File
@@ -0,0 +1,35 @@
import Src.Utilities.config as config
from Src.Utilities.eval import eval_solver
import logging
from Src.Utilities.config import setup_logging
level = config.LEVEL
logger = setup_logging(level)
Icon = config.Icon
Name = config.Name
async def streamhg(link,client,streams, site_name, proxies,ForwardProxy):
link = 'https://audinifer.com/e/' + link.split('/e/')[1]
url = await eval_solver(link,proxies, ForwardProxy, client,r'"hls2":"([^"]+)"')
if url:
streams['streams'].append({'name': f"{Name}",'title': f'{Icon}{site_name}\n▶️ StreamHG', 'url': url, 'behaviorHints': {'bingeGroup': f'{site_name.lower()}'}})
logger.info(f"{site_name} on StreamHG found results for the current ID")
status = True
else:
status = False
return streams,status
async def test_streamhg():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
results = await streamhg("https://dhcplay.com/e/zctlgpn3zoll",client,{'streams': []}, "StreamHG", {},"")
print(results)
if __name__ == "__main__":
import asyncio
asyncio.run(test_streamhg())
Src/API/extractors/supervideo.py
+1 -1
View File
@@ -11,7 +11,7 @@ Name = config.Name
async def supervideo(supervideo_link,client,streams, site_name, proxies,ForwardProxy):
url = await eval_solver(supervideo_link,proxies, ForwardProxy, client)
url = await eval_solver(supervideo_link,proxies, ForwardProxy, client,r'file:"(.*?)"')
if url:
streams['streams'].append({'name': f"{Name}",'title': f'{Icon}{site_name}\n▶️ Supervideo', 'url': url, 'behaviorHints': {'bingeGroup': f'{site_name.lower()}'}})
logger.info(f"{site_name} on SuperVideo found results for the current ID")
Src/API/extractors/uprot.py
+17 -3
View File
@@ -5,7 +5,8 @@ import logging
import Src.Utilities.config as config
from Src.Utilities.config import setup_logging
from Src.Utilities.loadenv import load_env
from fake_headers import Headers
random_headers = Headers()
UT_PROXY = config.UT_PROXY
UT_ForwardProxy = config.UT_ForwardProxy
level = config.LEVEL
@@ -31,6 +32,8 @@ else:
ForwardProxy = ""
url = 'https://uprot.net/msf/r4hcq47tarq8'
headers = {
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
@@ -118,8 +121,14 @@ async def bypass_uprot(client,link):
if 'msfi' not in link:
if 'msf' in link:
link = link.replace('msf','mse')
headers = random_headers.generate()
headers['Origin'] = url.split('msf')[0]
headers['Referer'] = url
response = await client.get(ForwardProxy + link, headers=headers, impersonate = 'chrome', proxies = proxies)
max_stream = await find_link(response.text,client)
if response.status_code != 403:
max_stream = await find_link(response.text,client)
else:
logger.info("Uprot blocked the request: 403")
else:
if 'mse' in link:
link = link.replace('mse','msf')
@@ -131,7 +140,12 @@ async def bypass_uprot(client,link):
cookies = json.loads(parts[0].replace("'", '"'))
data = json.loads(parts[1].replace("'", '"'))
response = await client.post(ForwardProxy + link, cookies=cookies, headers=headers, data =data, impersonate = 'chrome', proxies = proxies)
max_stream = await get_maxstream_link(response.text,client)
if response.status_code != 403:
max_stream = await get_maxstream_link(response.text,client)
if response.status_code == 403:
max_stream = False
return max_stream
except Exception as e:
Src/API/extractors/uprot.txt
+2 -2
View File
@@ -1,2 +1,2 @@
{'PHPSESSID': 'bm1qtf1pe0dnr4u4nhvofitk8u', 'captcha': 'ca5612d51c9bbc60f8d76051693d3315'}
{'captcha': '853'}
{}
{}
Src/API/extractors/vidxgo.py
+77
View File
@@ -0,0 +1,77 @@
import Src.Utilities.config as config
Icon = config.Icon
Name = config.Name
import re
from fake_headers import Headers
random_headers = Headers()
import logging
from Src.Utilities.config import setup_logging
level = config.LEVEL
logger = setup_logging(level)
import base64
from bs4 import BeautifulSoup,SoupStrainer
VD_DOMAIN = config.VD_DOMAIN
headers = {
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64; rv:150.0) Gecko/20100101 Firefox/150.0',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Accept-Language': 'en-US,en;q=0.9',
# 'Accept-Encoding': 'gzip, deflate, br, zstd',
'Sec-GPC': '1',
'Alt-Used': 'v.vidxgo.co',
'Connection': 'keep-alive',
'Upgrade-Insecure-Requests': '1',
'Sec-Fetch-Dest': 'iframe',
'Sec-Fetch-Mode': 'navigate',
'Sec-Fetch-Site': 'none',
'DNT': '1',
'Referer': 'https://altadefinizione.you/',
'Sec-Fetch-Storage-Access': 'none',
'-': '-',
'Priority': 'u=0, i',
}
async def vidxgo(link,client,streams,proxies,ForwardProxy):
#headers = random_headers.generate()
response = await client.get(link, allow_redirects=True, headers = headers)
soup = BeautifulSoup(response.text,'lxml',parse_only=SoupStrainer('script'))
scripts = soup.find_all('script')
text = scripts[5]
match = re.search(r"var .+'(.*)',d=atob\('(.*)'",text.text)
if match:
key = match.group(1)
base64_text = match.group(2)
decoded = base64.b64decode(base64_text)
u = bytearray(len(decoded))
for i in range(len(decoded)):
u[i] = decoded[i] ^ ord(key[i % len(key)])
decrypted_code = u.decode('utf-8')
match = re.search(r'currentSrc.+"(https:[^";]+)',decrypted_code)
if match:
url = match.group(1).replace("\\","")
streams['streams'].append({'name': f"{Name}",'title': f'{Icon}Vidxgo\n ▶️ Vidxgo', 'url': url, 'behaviorHints': {'proxyHeaders': {"request": {"User-Agent": 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36',"Accept": "*/*", "Accept-Language":"en-US,en;q=0.9","Referer":VD_DOMAIN+"/","Origin":VD_DOMAIN,"Sec-GPC": "1","Connection": "keep-alive","Sec-Fetch-Dest":"empty","Sec-Fetch-Mode":"cors","Sec-Fetch-Site":"cross-site","DNT": "1","Priority":"u=0"}}, 'notWebReady': True, 'bingeGroup': f'vidxgo'}})
logger.info(f"Vidxgo found results for the current ID")
return streams
#Testing
async def test_vidxgo():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
results = await vidsrc("https://v.vidxgo.co/tt34437972/1/1",client,{'streams': []}, "Vidxgo", {})
print(results)
if __name__ == "__main__":
import asyncio
asyncio.run(test_vidxgo())
# XOR decryption
Src/API/guardaflix.py
+1 -1
View File
@@ -44,7 +44,7 @@ async def new_domain(url,data,headers,proxies,ForwardProxy,client):
response = await client.post(ForwardProxy + f'{url}/wp-admin/admin-ajax.php', data = data, headers = headers, proxies = proxies)
if response.status_code != 200:
logger.warning(f"Guardaflix Failed to fetch search results: {response.status_code}")
result = update_site(client,['guardaplay','Guardaflix'])
result = await update_site(client,['guardaplay','Guardaflix'])
return response.text
async def get_player(page_link,MFP, MFP_CREDENTIALS,client, headers,streams):
Src/API/guardahd.py
+15 -11
View File
@@ -4,7 +4,8 @@ import re
import Src.Utilities.config as config
from fake_headers import Headers
from Src.Utilities.loadenv import load_env
from Src.API.extractors.supervideo import supervideo
from Src.API.extractors.streamhg import streamhg
from Src.API.extractors.mixdrop import mixdrop
import json, random
import logging
from Src.Utilities.config import setup_logging
@@ -41,24 +42,27 @@ async def search(clean_id,client):
logger.warning(f"GuardaHD Failed to fetch search results: {response.status_code}")
soup = BeautifulSoup(response.text,'lxml',parse_only=SoupStrainer('li'))
li_tags = soup.find_all('li')
results = {}
for tag in li_tags:
if tag.get('data-link'):
if 'supervideo' in tag.get('data-link'):
href = 'https:' + tag['data-link']
return href
return href
if tag.get('data-link') and (('mixdrop') in tag.text or ('streamhg') in tag.text):
results[tag.text.strip()] = 'https:' + tag['data-link']
return results
async def guardahd(streams,id,client):
async def guardahd(streams,id,client,MFP,MFP_CREDENTIALS):
try:
general = await is_movie(id)
ismovie = general[0]
clean_id = general[1]
if ismovie == 0:
return streams
supervideo_link = await search(clean_id,client)
streams = await supervideo(supervideo_link, client,streams,"GuardaHD",proxies, ForwardProxy)
results = await search(clean_id,client)
status = False
if 'streamhg' in results:
streams,status = await streamhg(results['streamhg'], client,streams,"GuardaHD",proxies, ForwardProxy)
if 'mixdrop' in results and not status:
streams,status = await mixdrop(results['mixdrop'],client,MFP,MFP_CREDENTIALS,streams,"GuardaHD",proxies,ForwardProxy,'')
return streams
except Exception as e:
logger.warning(f"MammaMia: GuardaHD Failed {e}")
@@ -68,8 +72,8 @@ 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 = "tt7181546" # This is an example ID format
results = await guardahd({'streams': []},test_id, client)
test_id = "tt16426418" # This is an example ID format
results = await guardahd({'streams': []},test_id, client,'0','')
print(results)
if __name__ == "__main__":
Src/API/guardaserie.py
+1 -1
View File
@@ -117,7 +117,7 @@ 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 = "tt12324366:2:2" # This is an example ID format tt0460649
test_id = "tt33318655:2:5" # This is an example ID format tt0460649
results = await guardaserie({'streams': []},test_id, client)
print(results)
Src/API/legacy/onlineserietv.py → Src/API/onlineserietv.py
+52 -53
View File
@@ -7,10 +7,18 @@ import json, random
import re
from Src.Utilities.eval import eval_solver
import urllib.parse
from Src.API.extractors.maxstream import maxstream
from Src.API.extractors.uprot import bypass_uprot
OST_DOMAIN = config.OST_DOMAIN
OST_PROXY = config.OST_PROXY
env_vars = load_env()
proxies = {}
random_headers = Headers()
import logging
from Src.Utilities.config import setup_logging
level = config.LEVEL
logger = setup_logging(level)
if OST_PROXY == "1":
PROXY_CREDENTIALS = env_vars.get('PROXY_CREDENTIALS')
proxy_list = json.loads(PROXY_CREDENTIALS)
@@ -27,25 +35,12 @@ if OST_ForwardProxy == "1":
ForwardProxy = env_vars.get('ForwardProxy')
else:
ForwardProxy = ""
headers = {
'User-Agent': 'Mozilla/5.0 (Linux; Android 12) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.5060.71 Mobile Safari/537.36',
'Accept': '*/*',
'Accept-Language': 'en-US,en;q=0.5',
# 'Accept-Encoding': 'gzip, deflate, br, zstd',
'X-Requested-With': 'XMLHttpRequest',
'DNT': '1',
'Sec-GPC': '1',
'Connection': 'keep-alive',
'Referer': f'https://onlineserietv.{OST_DOMAIN}/',
# 'Cookie': 'player_opt=fx',
'Sec-Fetch-Dest': 'empty',
'Sec-Fetch-Mode': 'cors',
'Sec-Fetch-Site': 'same-origin',
'Pragma': 'no-cache',
'Cache-Control': 'no-cache',
}
Name = config.Name
Icon = config.Icon
async def search(showname,date,client,ismovie,episode,season):
headers = random_headers.generate()
headers['Referer'] = f'{OST_DOMAIN}/'
cookies = {
'player_opt': 'fx',
}
@@ -57,64 +52,69 @@ async def search(showname,date,client,ismovie,episode,season):
'origin_id': '50141',
'searchwp_live_search_client_nonce': 'undefined',
}
response = await client.get(ForwardProxy + f"https://onlineserietv.{OST_DOMAIN}/wp-admin/admin-ajax.php?s={urllib.parse.quote(showname)}&action=searchwp_live_search&swpengine=default&swpquery={urllib.parse.quote(showname)}&origin_id=50141&searchwp_live_search_client_nonce=undefined", headers=headers, cookies=cookies, impersonate = "chrome124", proxies = proxies)
response = await client.get(ForwardProxy + f"{OST_DOMAIN}/wp-admin/admin-ajax.php?s={urllib.parse.quote(showname)}&action=searchwp_live_search&swpengine=default&swpquery={urllib.parse.quote(showname)}&origin_id=50141&searchwp_live_search_client_nonce=undefined", headers=headers, cookies=cookies, impersonate = "chrome", proxies = proxies)
if response.status_code != 200:
print("IP Blocked by OnlineserieTV",response)
logger.info("IP Blocked by OnlineserieTV",response)
soup = BeautifulSoup(response.text, 'lxml', parse_only=SoupStrainer('a'))
a_tags_with_href = soup.find_all("a", href=True)
for a_tag in a_tags_with_href:
href = a_tag.get("href")
if ismovie == 1:
if "film" in href:
response = await client.get(ForwardProxy + href, headers=headers, impersonate = "chrome124", proxies = proxies)
response = await client.get(ForwardProxy + href, headers=headers, impersonate = "chrome", proxies = proxies)
if response.status_code != 200:
print("IP Blocked by OnlineserieTV",response)
logger.info("IP Blocked by OnlineserieTV",response)
year_match = re.search(r'Anno: <i>(\d{4})</i>', response.text)
year = year_match.group(1) if year_match else None
if year == date:
pattern = r'https://uprot\.net/fxf/[^\s"<>]+'
pattern = r'https://uprot\.net/msf/[^\s"<>]+'
match = re.search(pattern, response.text)
if match:
name = a_tag.text.replace("\t","").replace("\n","")
flexy_link = match.group()
return flexy_link,name
maxstream_link = match.group()
return maxstream_link,name
else:
print("No flexy link found.")
logger.info("No Maxstream link found.")
else:
continue
elif ismovie == 0:
if "serietv" in href:
response = await client.get(ForwardProxy + href, headers=headers, impersonate = "chrome124", proxies = proxies)
response = await client.get(ForwardProxy + href, headers=headers, impersonate = "chrome", proxies = proxies)
if response.status_code != 200:
print("IP Blocked by OnlineserieTV",response)
logger.info("IP Blocked by OnlineserieTV",response)
year_match = re.search(r'Anno: <i>(\d{4})</i>', response.text)
year = year_match.group(1) if year_match else None
if year == date:
season = season.zfill(2)
episode = episode.zfill(2)
if season == "0000000000000":
print(True,season,episode)
#pattern = rf'{season}x{episode}.*?<a href=[\'"](https://uprot\.net/fxf/[^\'"]+)|0{episode}.*?<a href=[\'"](https://uprot\.net/fxf/[^\'"]+)'
pattern = rf'0{episode}.*?<a href=[\'"](https://uprot\.net/fxf/[^\'"]+)'
match1 = re.findall(pattern, response.text)
match = re.search(pattern, response.text)
print(match)
else:
pattern = rf'{season}x{episode}.*?<a href=[\'"](https://uprot\.net/fxf/[^\'"]+)'
pattern = rf'{season}x{episode}.*?<a href=[\'"](https://uprot\.net/msf/[^\'"]+)'
match = re.search(pattern, response.text, re.DOTALL)
if not match:
pattern = rf'0{episode}.*?<a href=[\'"](https://uprot\.net/msf/[^\'"]+)'
match = re.search(pattern, response.text, re.DOTALL)
if match:
name = a_tag.text.replace("\t","").replace("\n","")
flexy_link = match.group(1)
print(match,flexy_link)
return flexy_link,name
else:
print("No flexy link found.")
maxstream_link = match.group(1)
return maxstream_link,name
else:
logger.info("No maxstream link found.")
else:
continue
return None,None
async def get_maxstream(uprot_link,streams,language,client):
maxstream_link = await bypass_uprot(client,uprot_link)
if maxstream_link:
streams = await maxstream(maxstream_link,client,streams,'OnlineSerieTV',language,proxies,ForwardProxy)
else:
if maxstream_link == False:
return streams
else:
streams['streams'].append({'name': f"{Name}",'title': f'{Icon}OnlineSerieTV\n▶️ Please do the captcha at /uprot in order to be able to play this content! \n Remember to refresh the sources!\nIf you recently did the captcha then dont worry, just refresh the sources', 'url': 'https://github.com/UrloMythus/MammaMia', 'behaviorHints': { 'bingeGroup': 'cb01'}})
return streams
async def onlineserietv(id,client):
async def onlineserietv(streams,id,client):
try:
general = await is_movie(id)
ismovie = general[0]
@@ -131,22 +131,21 @@ async def onlineserietv(id,client):
else:
showname,date = get_info_tmdb(clean_id,ismovie,type)
showname = showname.replace("'"," ")
flexy_link,name = await search(showname,date,client,ismovie,episode,season)
flexy_link = flexy_link.replace("fxf","fxe")
real_url = await client.head(ForwardProxy + flexy_link, headers=headers, impersonate = "chrome124", proxies = proxies)
real_url = real_url.url
final_url = await eval_solver(real_url,proxies, ForwardProxy, client)
return final_url,name
uprot_link,name = await search(showname,date,client,ismovie,episode,season)
if uprot_link != None:
streams = await get_maxstream(uprot_link,streams,'',client)
return streams
except Exception as e:
print("MammaMia: Onlineserietv Failed",e)
return None,None
logger.info("MammaMia: Onlineserietv Failed")
logger.info(e)
return streams
async def test_animeworld():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
test_id = "tt0206516:2:1" # This is an example ID format
results = await onlineserietv(test_id, client)
test_id = "tt16426418" # This is an example ID format
results = await onlineserietv({'streams': []},test_id, client)
print(results)
if __name__ == "__main__":
Src/API/vidxgoalta.py
+65
View File
@@ -0,0 +1,65 @@
from Src.Utilities.info import is_movie
import Src.Utilities.config as config
from Src.Utilities.loadenv import load_env
from Src.API.extractors.vidxgo import vidxgo
import json, random
import logging
from Src.Utilities.config import setup_logging
level = config.LEVEL
logger = setup_logging(level)
env_vars = load_env()
VD_PROXY = config.VD_PROXY
proxies = {}
if VD_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
}
VD_ForwardProxy = config.GH_ForwardProxy
if VD_ForwardProxy == "1":
ForwardProxy = env_vars.get('ForwardProxy')
else:
ForwardProxy = ""
VD_DOMAIN = config.VD_DOMAIN
async def vidxgoalta(streams,id,client):
try:
general = await is_movie(id)
ismovie = general[0]
clean_id = general[1]
type = "Vidxgo"
if "tmdb" in id:
clean_id = await get_IMDB_id_from_TMDb_id(clean_id,client)
if ismovie == 0 :
season = str(general[2])
episode = str(general[3])
link = f'{VD_DOMAIN}/{clean_id}/{season}/{episode}'
elif ismovie == 1:
link = f'{VD_DOMAIN}/{clean_id}'
streams = await vidxgo(link,client,streams,proxies,ForwardProxy)
return streams
except Exception as e:
logger.info("Vidxgo failed due to ")
logger.info(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 = "tt7181546" # This is an example ID format
results = await vidxgoalta({'streams': []},test_id, client)
print(results)
if __name__ == "__main__":
import asyncio
asyncio.run(test_script())