MammaMia version 2.0.0
This commit is contained in:
UrloMythus
@@ -0,0 +1,70 @@
|
||||
from fake_headers import Headers
|
||||
import Src.Utilities.config as config
|
||||
from bs4 import BeautifulSoup,SoupStrainer
|
||||
import re
|
||||
import time
|
||||
import logging
|
||||
from Src.Utilities.config import setup_logging
|
||||
level = config.LEVEL
|
||||
logger = setup_logging(level)
|
||||
Icon = config.Icon
|
||||
Name = config.Name
|
||||
random_headers = Headers()
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
async def deltabit(page_url,client,streams,site_name,proxies,ForwardProxy,language):
|
||||
"""Extract Deltabit URL."""
|
||||
i = 0
|
||||
#Set up some headers
|
||||
headers = random_headers.generate()
|
||||
headers2 = random_headers.generate()
|
||||
|
||||
#Get the redirected link, so the actual deltabit link.
|
||||
page_url_response = await client.get(ForwardProxy + page_url,headers={**headers, 'Range': 'bytes=0-0'}, proxies=proxies)
|
||||
page_url = page_url_response.url
|
||||
logger.info(f"Page_URl is {page_url}")
|
||||
|
||||
#Change the referer and user agent of the generated headers
|
||||
headers2['referer'] = 'https://safego.cc/'
|
||||
headers2['user-agent'] = 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36'
|
||||
|
||||
#Request to the deltabit link
|
||||
response = await client.get(ForwardProxy + page_url, headers = headers2, allow_redirects = True, proxies = proxies)
|
||||
page_url = response.url
|
||||
#Often deltabit redirects to another deltabit link. We need the redirected link to put it in the headers.
|
||||
origin = page_url.split('/')[2]
|
||||
headers['origin'] = f'https://{origin}'
|
||||
headers['referer'] = page_url
|
||||
headers['user-agent'] = 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36'
|
||||
#We check the page for the data such as hash and fname.
|
||||
soup = BeautifulSoup(response.text, 'lxml', parse_only=SoupStrainer('input'))
|
||||
data = {}
|
||||
for input in soup:
|
||||
name = input.get('name')
|
||||
value = input.get('value')
|
||||
data[name] = value
|
||||
data['imhuman'] = ''
|
||||
data['referer']= page_url
|
||||
#We need to wait 2.5 seconds or the request will fail
|
||||
time.sleep(2.5)
|
||||
|
||||
|
||||
fname = data['fname']
|
||||
#Request to the deltabit link
|
||||
response = await client.post(ForwardProxy + page_url, data = data, headers = headers, proxies = proxies)
|
||||
link = re.findall(r'sources:\s*\["([^"]+)"', response.text, re.DOTALL)
|
||||
if not link:
|
||||
if not i >= 3:
|
||||
page_url = 'https://deltabit.co/q4r8w4assff4'
|
||||
link,fname = await deltabit(page_url,client)
|
||||
i +=1
|
||||
else:
|
||||
link = link[0]
|
||||
if link:
|
||||
streams['streams'].append({'name': f"{Name}{language}",'title': f'{Icon}{site_name}\n▶️ Deltabit\n{fname}', 'url': link, 'behaviorHints': {'bingeGroup': f'{site_name.lower()}'}})
|
||||
|
||||
|
||||
return streams
|
||||
@@ -0,0 +1,18 @@
|
||||
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 dropload(player_link,client,streams,site_name, proxies, ForwardProxy):
|
||||
url = await eval_solver(player_link, proxies,ForwardProxy, client)
|
||||
if url:
|
||||
streams['streams'].append({'name': f"{Name}",'title': f'{Icon}{site_name}\n▶️ Dropload', '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'}}, 'notWebReady': True, 'bingeGroup': f'{site_name.lower()}'}})
|
||||
logger.info(f"{site_name} on Dropload found results for the current ID")
|
||||
|
||||
return streams
|
||||
@@ -0,0 +1,41 @@
|
||||
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)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
async def hdplayer(hdplayer,client,streams,Referer,site_name,proxies,ForwardProxy):
|
||||
headers = random_headers.generate()
|
||||
response = await client.get(hdplayer, allow_redirects=True, headers = headers)
|
||||
match = re.search(r'sources:\s*\[\s*\{\s*file\s*:\s*"([^"]*)"', response.text)
|
||||
url = match.group(1)
|
||||
if url:
|
||||
#Fix for Exoplayer which needs .m3u8 in the url to play the file
|
||||
url = url + ".m3u8"
|
||||
streams['streams'].append({'name': f"{Name}\n720/1080p",'title': f'{Icon}{site_name}\n ▶️ HDPlayer', 'url': url, 'behaviorHints': {'proxyHeaders': {"request": {"Referer": Referer}}, 'notWebReady': True, 'bingeGroup': f'{site_name.lower()}'}})
|
||||
logger.info(f"{site_name} on HDPlayer found results for the current ID")
|
||||
|
||||
return streams
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
#Testing
|
||||
async def test_hdplayer():
|
||||
from curl_cffi.requests import AsyncSession
|
||||
async with AsyncSession() as client:
|
||||
results = await hdplayer("https://hdplayer.gives/embed/udnjutxECQGuwsU",client,{'streams': []},'https://hdplayer.gives/', "StreamingWatch", {},'')
|
||||
print(results)
|
||||
if __name__ == "__main__":
|
||||
import asyncio
|
||||
asyncio.run(test_hdplayer())
|
||||
@@ -0,0 +1,94 @@
|
||||
import requests
|
||||
from Crypto.Cipher import AES
|
||||
import json
|
||||
import re
|
||||
from rich.prompt import Prompt
|
||||
from rich.console import Console
|
||||
from fake_headers import Headers
|
||||
import Src.Utilities.config as config
|
||||
import logging
|
||||
from Src.Utilities.config import setup_logging
|
||||
level = config.LEVEL
|
||||
logger = setup_logging(level)
|
||||
Icon = config.Icon
|
||||
Name = config.Name
|
||||
'''
|
||||
This script was done mostly by @Arrowar, creator of https://github.com/Arrowar/StreamingCommunity/
|
||||
It was modified for the use in MammaMia
|
||||
'''
|
||||
|
||||
|
||||
random_headers = Headers()
|
||||
|
||||
KEY = b'kiemtienmua911ca' # 16 byte
|
||||
IV = b'1234567890oiuytr' # 16 byte
|
||||
|
||||
|
||||
async def fetch_response(client,referer,player_url,url,id, proxies, ForwardProxy):
|
||||
headers = random_headers.generate()
|
||||
headers['referer'] = player_url
|
||||
data = {
|
||||
'id': id,
|
||||
'w': '2560',
|
||||
'h': '1440',
|
||||
'r': referer,
|
||||
}
|
||||
response = await client.get(ForwardProxy + url, headers=headers, params = data, proxies = proxies)
|
||||
return response.text
|
||||
|
||||
def hex_to_bytes(hex_str):
|
||||
hex_str = re.sub(r'[^0-9a-fA-F]', '', hex_str)
|
||||
|
||||
if len(hex_str) % 2 != 0:
|
||||
hex_str = '0' + hex_str
|
||||
|
||||
return bytes(int(hex_str[i:i+2], 16) for i in range(0, len(hex_str), 2))
|
||||
|
||||
def decrypt_aes_cbc(ciphertext_bytes, key, iv):
|
||||
cipher = AES.new(key, AES.MODE_CBC, iv)
|
||||
decrypted = cipher.decrypt(ciphertext_bytes)
|
||||
|
||||
pad_len = decrypted[-1]
|
||||
if 1 <= pad_len <= 16:
|
||||
decrypted = decrypted[:-pad_len]
|
||||
|
||||
return decrypted
|
||||
|
||||
|
||||
|
||||
async def loadm(player_link,client,streams,referer,site_name,proxies,ForwardProxy):
|
||||
parts = player_link.split('#')
|
||||
id = parts[1]
|
||||
player_url = parts[0]
|
||||
api = player_url + 'api/v1/video'
|
||||
#It downloads m (crypted informations) from the API"
|
||||
response_hex = await fetch_response(client,referer,player_url,api,id, proxies, ForwardProxy)
|
||||
#Convert Hex in Bytes
|
||||
ciphertext_bytes = hex_to_bytes(response_hex)
|
||||
#We decrypt AES-CBC encryption with Key,IV and the bytes
|
||||
decrypted_bytes = decrypt_aes_cbc(ciphertext_bytes, KEY, IV)
|
||||
decrypted_str = decrypted_bytes.decode('utf-8')
|
||||
data = json.loads(decrypted_str)
|
||||
#We get the hls
|
||||
hls = data['cf']
|
||||
title = data['title']
|
||||
if hls:
|
||||
streams['streams'].append({'name': f"{Name}",'title': f'{Icon}{site_name}\n▶️ Loadm\n{title}', 'url': hls, 'behaviorHints': {'proxyHeaders': {"request": {"Referer": player_url}}, 'notWebReady': True, 'bingeGroup': f'{site_name.lower()}'}})
|
||||
logger.info(f"{site_name} on Loadm found results for the current ID")
|
||||
return streams
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
#Testing
|
||||
async def test_loadm():
|
||||
from curl_cffi.requests import AsyncSession
|
||||
async with AsyncSession() as client:
|
||||
results = await loadm("https://loadm.cam/#6wx1p",client,{'streams': []},'guardoserie.live', 'Guardoserie', {},"")
|
||||
print(results)
|
||||
if __name__ == "__main__":
|
||||
import asyncio
|
||||
asyncio.run(test_loadm())
|
||||
|
||||
# https://loadm.cam/#cfnqx
|
||||
@@ -0,0 +1,31 @@
|
||||
import Src.Utilities.config as config
|
||||
Icon = config.Icon
|
||||
Name = config.Name
|
||||
from Src.Utilities.mfp import build_mfp
|
||||
import logging
|
||||
from Src.Utilities.config import setup_logging
|
||||
level = config.LEVEL
|
||||
logger = setup_logging(level)
|
||||
from Src.Utilities.eval import eval_solver
|
||||
|
||||
|
||||
async def mixdrop(url,client,MFP,MFP_CREDENTIALS,streams,site_name,proxies,ForwardProxy,language):
|
||||
"""Extract Mixdrop URL."""
|
||||
if "club" in url:
|
||||
url = url.replace("club", "cv").split("/2")[0]
|
||||
if "cfd" in url:
|
||||
url = url.replace("cfd", "cv").replace("emb","e").split("/2")[0]
|
||||
if MFP == "1":
|
||||
url = await build_mfp(MFP_CREDENTIALS,url,"Mixdrop",client)
|
||||
if url and "https%3ANone" not in url:
|
||||
logger.info(f"{site_name} on Mixdrop found results for the current ID")
|
||||
streams['streams'].append({'name': f"{Name} 🕵️♂️\n{language}",'title': f'{Icon}{site_name}\n▶️ MixDrop', 'url': url, 'behaviorHints':{'bingeGroup': f'{site_name.lower()}'}})
|
||||
|
||||
else:
|
||||
headers = {"accept-language": "en-US,en;q=0.5"}
|
||||
url = await eval_solver(url, proxies , ForwardProxy, client)
|
||||
if url:
|
||||
url = "https:" + url
|
||||
logger.info(f"{site_name} on Mixdrop found results for the current ID")
|
||||
streams['streams'].append({'name': f"{Name}{language}",'title': f'{Icon}{site_name}\n▶️ MixDrop', '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'}}, 'notWebReady': True, 'bingeGroup': f'{site_name.lower()}'}})
|
||||
return streams
|
||||
@@ -0,0 +1,18 @@
|
||||
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 supervideo(supervideo_link,client,streams, site_name, proxies,ForwardProxy):
|
||||
url = await eval_solver(supervideo_link,proxies, ForwardProxy, client)
|
||||
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")
|
||||
return streams
|
||||
@@ -0,0 +1,28 @@
|
||||
from fake_headers import Headers
|
||||
import Src.Utilities.config as config
|
||||
Icon = config.Icon
|
||||
Name = config.Name
|
||||
from Src.Utilities.mfp import build_mfp
|
||||
from fake_headers import Headers
|
||||
import re
|
||||
import logging
|
||||
from Src.Utilities.config import setup_logging
|
||||
level = config.LEVEL
|
||||
logger = setup_logging(level)
|
||||
random_headers = Headers()
|
||||
|
||||
async def uqload(player_link,client,MFP,MFP_CREDENTIALS,streams,site_name,proxies,ForwardProxy):
|
||||
headers = random_headers.generate()
|
||||
if MFP == "1":
|
||||
url = await build_mfp(MFP_CREDENTIALS, player_link, "Uqload",client)
|
||||
streams['streams'].append({'name': f"{Name} 🕵️♂️",'title': f'{Icon}Guardaflix\n▶️ Uqload', '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'}}, 'notWebReady': True, 'bingeGroup': 'guardaflix'}})
|
||||
return streams
|
||||
response = await client.get(ForwardProxy + player_link, headers = headers, proxies = proxies)
|
||||
video_url_match = re.search(r'sources: \["(.*?)"]', response.text)
|
||||
title = re.search(r'title: "(.*?)"', response.text).group(1)
|
||||
if video_url_match:
|
||||
url = video_url_match.group(1)
|
||||
if url:
|
||||
streams['streams'].append({'name': f"{Name}",'title': f'{Icon}{site_name}\n▶️ Uqload\n{title}', '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'}}, 'notWebReady': True, 'bingeGroup': f'{site_name.lower()}'}})
|
||||
logger.info(f"{site_name} on Uqload found results for the current ID")
|
||||
return streams
|
||||
@@ -0,0 +1,71 @@
|
||||
from fake_headers import Headers
|
||||
import re
|
||||
from bs4 import BeautifulSoup,SoupStrainer
|
||||
random_headers = Headers()
|
||||
import Src.Utilities.config as config
|
||||
Icon = config.Icon
|
||||
Name = config.Name
|
||||
from Src.Utilities.mfp import build_mfp
|
||||
import logging
|
||||
from Src.Utilities.config import setup_logging
|
||||
level = config.LEVEL
|
||||
logger = setup_logging(level)
|
||||
SC_DOMAIN = config.SC_DOMAIN
|
||||
User_Agent= "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36"
|
||||
|
||||
|
||||
async def vixcloud(link,client,MFP,MFP_CREDENTIALS,streams,site_name,proxies,ForwardProxy):
|
||||
if MFP == "1":
|
||||
final_url = await build_mfp(MFP_CREDENTIALS, link, "VixCloud",client)
|
||||
quality = ""
|
||||
mfp_icon = '🕵️♂️'
|
||||
else:
|
||||
mfp_icon = ''
|
||||
headers = random_headers.generate()
|
||||
headers['Referer'] = f"{SC_DOMAIN}/"
|
||||
headers['Origin'] = f"{SC_DOMAIN}"
|
||||
headers['User-Agent'] = User_Agent
|
||||
headers['user-agent'] = User_Agent
|
||||
response = await client.get(link)
|
||||
if response.status_code != 200:
|
||||
logger.warning("Failed to extract URL components, Invalid Request")
|
||||
soup = BeautifulSoup(response.text, "lxml", parse_only=SoupStrainer("body"))
|
||||
if soup:
|
||||
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)
|
||||
server_url = re.search(r"url:\s*'([^']+)'", script).group(1)
|
||||
try:
|
||||
quality = re.search(r'"quality":(\d+)', script).group(1)
|
||||
except:
|
||||
quality = ""
|
||||
if "?b=1" in server_url:
|
||||
final_url = f'{server_url}&token={token}&expires={expires}'
|
||||
else:
|
||||
final_url = f"{server_url}?token={token}&expires={expires}"
|
||||
if "window.canPlayFHD = true" in script:
|
||||
final_url += "&h=1"
|
||||
if final_url:
|
||||
logger.info(f"Vixcloud Found Results for the current ID")
|
||||
streams['streams'].append({"name":f'{Name} {mfp_icon}\n{quality}', 'title': f'{Icon} StreamingCommunity\n▶️ Vixcloud','url': final_url,'behaviorHints': {'proxyHeaders': {"request": {"user-agent": User_Agent}}, 'notWebReady': True, 'bingeGroup': f'{site_name.lower()}{quality}'}})
|
||||
|
||||
return streams
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
#Testing
|
||||
async def test_vixcloud():
|
||||
from curl_cffi.requests import AsyncSession
|
||||
async with AsyncSession() as client:
|
||||
results = await vixcloud("https://vixsrc.to/tv/204082/1/3/",client,"0",['',''],{'streams': []},'guardoserie.live', {},'')
|
||||
print(results)
|
||||
if __name__ == "__main__":
|
||||
import asyncio
|
||||
asyncio.run(test_vixcloud())
|
||||
|
||||
Reference in New Issue
Block a user