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 a3b49bd6f4
commit 0801b3ae93
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