273 Commits
51 changed files with 315 additions and 2618 deletions
-121
View File
@@ -1,121 +0,0 @@
name: Conditional Sync Upstream, Build, and Push Docker Image
on:
schedule:
- cron: '0 * * * *' # ogni ora
workflow_dispatch:
push:
branches: [main]
jobs:
sync-build-push:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Backup files and force sync upstream
id: sync
run: |
git config user.name "github-actions"
git config user.email "[email protected]"
git remote get-url upstream || git remote add upstream https://github.com/UrloMythus/MammaMia.git
git fetch upstream
# Controlla se ci sono differenze PRIMA del reset
if git diff --quiet HEAD upstream/main; then
echo "Repository already up to date"
echo "no_change=true" >> $GITHUB_OUTPUT
else
echo "Changes detected, proceeding with sync and build"
backup_dir=".git/backup_files"
mkdir -p "$backup_dir/.github/workflows"
mkdir -p "$backup_dir/Dockerfiles"
if [ -f config.json ]; then
cp config.json "$backup_dir/config.json"
else
echo "⚠️ config.json non trovato, skipping backup"
fi
if [ -f update_domains.py ]; then
cp update_domains.py "$backup_dir/update_domains.py"
else
echo "⚠️ update_domains.py non trovato, skipping backup"
fi
if [ -d .github/workflows ]; then
cp -r .github/workflows/* "$backup_dir/.github/workflows/" || echo "⚠️ No workflows to backup"
else
echo "⚠️ Cartella .github/workflows non trovata, skipping backup"
fi
if [ -d Dockerfiles ]; then
cp -r Dockerfiles/* "$backup_dir/Dockerfiles/" || echo "⚠️ No Dockerfiles to backup"
else
echo "⚠️ Cartella Dockerfiles non trovata, skipping backup"
fi
git checkout main || git checkout -b main
git reset --hard upstream/main
if [ -f "$backup_dir/config.json" ]; then
cp "$backup_dir/config.json" config.json
else
echo "⚠️ File config.json non trovato nel backup, skipping restore"
fi
if [ -f "$backup_dir/update_domains.py" ]; then
cp "$backup_dir/update_domains.py" update_domains.py
else
echo "⚠️ File update_domains.py non trovato nel backup, skipping restore"
fi
mkdir -p .github/workflows
if [ -d "$backup_dir/.github/workflows" ]; then
cp -r "$backup_dir/.github/workflows/"* .github/workflows/ || echo "⚠️ No workflows to restore"
else
echo "⚠️ Backup workflows folder mancante, skipping restore"
fi
mkdir -p Dockerfiles
if [ -d "$backup_dir/Dockerfiles" ]; then
cp -r "$backup_dir/Dockerfiles/"* Dockerfiles/ || echo "⚠️ No Dockerfiles to restore"
else
echo "⚠️ Backup Dockerfiles folder mancante, skipping restore"
fi
git add .
git commit -m "Restore files after forced sync" || echo "No changes needed"
git push origin main --force
echo "no_change=false" >> $GITHUB_OUTPUT
rm -rf "$backup_dir"
fi
- name: Check if build is needed
run: echo "Changes detected, proceed with build."
if: steps.sync.outputs.no_change == 'false'
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
if: steps.sync.outputs.no_change == 'false'
- name: Login to Docker Hub
uses: docker/login-action@v3
with:
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
if: steps.sync.outputs.no_change == 'false'
- name: Build and push Docker image
uses: docker/build-push-action@v6
with:
context: .
file: ./Dockerfile
push: true
tags: dula117/mammamia:latest
if: steps.sync.outputs.no_change == 'false'
-37
View File
@@ -1,37 +0,0 @@
name: Update Domains
on:
schedule:
- cron: "0 20 * * *" # Esegui alle 20:00 ogni giorno
workflow_dispatch: # Permette di avviare manualmente il workflow
jobs:
update-domains:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.12'
- name: Install dependencies
run: |
pip install requests
- name: Update domains in config.json
run: |
python update_domains.py
- name: Commit and push changes
run: |
git config --global user.email "[email protected]"
git config --global user.name "Your Name"
git add config.json
git commit -m "Aggiornamento domini in config.json"
git push
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-1
View File
@@ -1,4 +1,3 @@
.env
__pycache__/
.vercel
beamup.json
+1 -9
View File
@@ -2,16 +2,8 @@ FROM python:3.11-slim-bullseye
WORKDIR /app
ADD . /app
RUN apt-get update && \
apt-get install -y tesseract-ocr libtesseract-dev && \
apt-get clean && \
rm -rf /var/lib/apt/lists/*
RUN pip install --no-cache-dir -r requirements.txt
RUN pip install --no-cache-dir pytesseract
EXPOSE 8080
# Run run.py when the container launches
ENTRYPOINT ["python","run.py"]
ENTRYPOINT ["python", "run.py"]
+8
View File
@@ -1,15 +1,23 @@
##Use an official Python runtime as a parent image
FROM python:3.11-slim-bullseye
# Set the working directory in the container to /app
WORKDIR /app
# Install git
RUN apt-get update && apt-get install -y git
# Clone the repository
RUN git clone https://github.com/UrloMythus/MammaMia.git .
# Copy the local config.json file to the container
COPY config.json /app/config.json
# Install any needed packages specified in requirements.txt
RUN pip install --no-cache-dir -r requirements.txt
# Expose the port, for now default is 8080 cause it's the only one really allowed by HuggingFace
EXPOSE 8080
# Run run.py when the container launches
ENTRYPOINT ["python", "run.py"]
@@ -1,14 +1,21 @@
#Use an official Python runtime as a parent image
FROM python:3.11-slim-bullseye
# Set the working directory in the container to /app
WORKDIR /app
# Install git
RUN apt-get update && apt-get install -y git
# Clone the repository
RUN git clone https://github.com/UrloMythus/MammaMia.git .
# Install any needed packages specified in requirements.txt
RUN pip install --no-cache-dir -r requirements.txt
# Expose the port, for now default is 8080 cause it's the only one really allowed by HuggingFace
EXPOSE 8080
# Run run.py when the container launches
ENTRYPOINT ["python", "run.py"]
+2 -2
View File
@@ -7,7 +7,7 @@ RUN apt-get update && \
apt-get clean && \
rm -rf /var/lib/apt/lists/*
RUN git clone https://github.com/vitouchiha/MammaMia.git .
RUN git clone https://github.com/UrloMythus/MammaMia.git .
RUN pip install --no-cache-dir -r requirements.txt
@@ -15,4 +15,4 @@ RUN pip install --no-cache-dir pytesseract
EXPOSE 8080
ENTRYPOINT ["python", "run.py"]
ENTRYPOINT ["python", "run.py"]
-1
View File
@@ -1 +0,0 @@
web: uvicorn run:app --host=0.0.0.0 --port=$PORT --workers 2
+2 -2
View File
@@ -171,8 +171,7 @@ Whatever you want
To install this project on [Render](https://render.com/) follow these steps:
[!WARNING] Render might ban your account. Modify the static.py file might help with that
@@ -346,3 +345,4 @@ We hereby issue this notice to inform you that this Add-On just function like an
Thank You.
+6 -7
View File
@@ -52,7 +52,6 @@ showname_replace = {
"Shippuuden": "Shippuden",
" ": "+",
"Solo+Leveling+2": "Solo+Leveling+2:",
"Case+Book": "Casebook",
"-": ""
}
async def security_cookie (response):
@@ -78,16 +77,16 @@ async def get_mp4(anime_url,ismovie,episode,client,i,streams):
if ismovie == 0:
episode_page = soup.find('a', {'data-episode-num':episode })
if episode_page is None:
return streams
episode_page = f'{AW_DOMAIN}/api/episode/serverPlayerAnimeWorld?id={episode_page["href"].split("/")[-1]}'
return None
episode_page = f'{AW_DOMAIN}{episode_page["href"]}'
response = await client.get(ForwardProxy + episode_page,allow_redirects=True, cookies = cookies,impersonate = "chrome124", proxies=proxies)
if response.status_code == 202:
cookies = await security_cookie(response)
response = await client.get(ForwardProxy + episode_page,allow_redirects=True, cookies = cookies,impersonate = "chrome124", proxies=proxies)
soup = BeautifulSoup(response.text,'lxml')
a_tag = soup.find('source')
url = a_tag['src']
a_tag = soup.find('a', {'id': 'alternativeDownloadLink', 'class': 'm-1 btn btn-sm btn-primary'})
url = a_tag['href']
response = await client.head(url)
if response.status_code == 404:
url = None
@@ -167,8 +166,8 @@ async def animeworld(streams,id,client):
async def test_animeworld():
async with AsyncSession() as client:
# Replace with actual id, for example 'anime_id:episode' format
test_id = "kitsu:12:1" # This is an example ID format
results = await animeworld({'streams': []},test_id, client)
test_id = "kitsu:46474:1" # This is an example ID format
results = await animeworld(test_id, client)
print(results)
if __name__ == "__main__":
+23 -59
View File
@@ -6,8 +6,6 @@ from fake_headers import Headers
import Src.Utilities.config as config
from Src.Utilities.loadenv import load_env
from Src.API.extractors.mixdrop import mixdrop
from Src.API.extractors.maxstream import maxstream
from Src.API.extractors.uprot import bypass_uprot
import logging
from Src.Utilities.config import setup_logging
level = config.LEVEL
@@ -15,8 +13,6 @@ logger = setup_logging(level)
CB_DOMAIN = config.CB_DOMAIN
CB_PROXY = config.CB_PROXY
MX_PROXY = config.MX_PROXY
Icon = config.Icon
Name = config.Name
proxies = {}
proxies2 = {}
env_vars = load_env()
@@ -74,21 +70,7 @@ async def get_stayonline(link,client):
real_url = response.json()['data']['value']
return real_url
async def get_maxstream(link,streams,client):
if 'stayonline' in link:
uprot_link = await get_stayonline(link,client)
else:
uprot_link = link
maxstream_link = await bypass_uprot(client,uprot_link)
if maxstream_link:
streams = await maxstream(maxstream_link,client,streams,'CB01','',proxies2,ForwardProxy)
else:
if maxstream_link == False:
return streams
else:
streams['streams'].append({'name': f"{Name}",'title': f'{Icon}CB01\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 search_movie(showname,date,client):
try:
showname = showname.replace(" ","+").replace("ò","o").replace("è","e").replace("à","a").replace("ù","u").replace("ì","i")
@@ -112,7 +94,7 @@ async def search_movie(showname,date,client):
match = year_pattern.search(date_text)
if match:
year = match.group(0)
if year == date :
if year == date : # Check if the year is 2011
return href
except Exception as e:
logger.warning(f'MammaMia: Error in search_series cb01: {e}')
@@ -134,6 +116,7 @@ async def search_series(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:
@@ -142,7 +125,7 @@ async def search_series(showname,date,client):
match = year_pattern.search(date_text)
if match:
year = match.group(0)
if (int(year) - int(date)) == 0 or (int(year)-int(date)) == 1 or (int(year)-int(date)) == -1 : # Check if the year is 2011
if year == date : # Check if the year is 2011
return href
except Exception as e:
logger.warning(f'MammaMia: Error in search_series cb01: {e}')
@@ -153,55 +136,35 @@ async def movie_redirect_url(link,client,MFP,MFP_CREDENTIALS,streams):
# Extract the redirect URL from the HTML
soup = BeautifulSoup(response.text, "lxml",parse_only=SoupStrainer('div'))
redirect_url = soup.find("div", id="iframen2").get("data-src")
redirect_url_2 = soup.find("div", id="iframen1").get("data-src")
if "stayonline" in redirect_url:
mixdrop_real_link = await get_stayonline(redirect_url,client)
streams,status = await mixdrop(mixdrop_real_link,client,MFP,MFP_CREDENTIALS,streams,"CB01",proxies,ForwardProxy,"")
if status == False:
streams = await get_maxstream(redirect_url_2,streams,client)
return streams
streams = await mixdrop(mixdrop_real_link,client,MFP,MFP_CREDENTIALS,streams,"CB01",proxies,ForwardProxy,"")
async def series_search_streams(text,search_text,response_text,headers,season,episode,client,MFP,MFP_CREDENTIALS,streams):
if "A" in text:
captcha = re.search(rf'{search_text}.*?href="([^"]+)',response_text,re.DOTALL)
if captcha:
response = await client.get(ForwardProxy + captcha.group(1), headers = headers, allow_redirects = True, proxies = proxies)
season = season.zfill(2)
match = re.search(rf"(S{season}E{episode}|{season}x{episode}).+?href='([^']+)",response.text)
print(match.group(2))
if match:
streams = await get_maxstream(match.group(2),streams,client)
else:
match = re.search(rf'{season}&#215;{episode}\s*&#8211;\s*<a[^>]*href="[^"]*"[^>]*>[^<]*</a>\s&#8211; <a[^>]*href="[^"]*.*</a>',response_text)
if match:
links = match.group(0)
soup = BeautifulSoup(links,'lxml',parse_only=SoupStrainer('a'))
host_links = soup.find_all('a')
mixdrop_real_link = await get_stayonline(host_links[1]['href'],client)
status = False
streams,status = await mixdrop(mixdrop_real_link,client,MFP,MFP_CREDENTIALS,streams,"CB01",proxies,ForwardProxy,"")
if status == False:
streams = await get_maxstream(host_links[0]['href'],streams,client)
return streams
return streams
async def series_redirect_url(link,season,episode,client,MFP,MFP_CREDENTIALS,streams):
episode = episode.zfill(2)
if len(episode) == 1:
episode = f'0{episode}'
headers = fake_headers.generate()
response = await client.get(ForwardProxy + link, headers=headers, allow_redirects=True, timeout=10, proxies = proxies)
soup = BeautifulSoup(response.text, "lxml")
soup = BeautifulSoup(response.text, "lxml")
seasons_text = soup.find_all('div', class_='sp-head')
for season_text in seasons_text:
text = season_text.text
search_text = season_text.text
if f'STAGIONE' in text and f'{season}' in text:
text = text.replace("STAGIONE","").replace("ITA", "")
streams = await series_search_streams(text,search_text,response.text,headers,season,episode,client,MFP,MFP_CREDENTIALS,streams)
elif f'STAGIONE' in text and 'A' in text.replace("STAGIONE","").replace("ITA", ""):
parts = text.split('STAGIONE ')[1].split(' A ')
num1 = int(parts[0])
num2 = int(parts[1][0])
if int(season) in range(num1,num2):
streams = await series_search_streams(text,search_text,response.text,headers,season,episode,client,MFP,MFP_CREDENTIALS,streams)
return streams
if "A" in text:
return streams
else:
episode = episode.zfill(2)
match = re.search(rf'{season}&#215;{episode}\s*&#8211;\s*<a[^>]*href="[^"]*"[^>]*>[^<]*</a>\s&#8211; <a[^>]*href="[^"]*',response.text)
if match:
links = match.group(0)
soup = BeautifulSoup(links,'lxml',parse_only=SoupStrainer('a'))
mixdrop_a = soup.find_all('a')[1]
mixdrop_link = mixdrop_a['href']
mixdrop_real_link = await get_stayonline(mixdrop_link,client)
streams = await mixdrop(mixdrop_real_link,client,MFP,MFP_CREDENTIALS,streams,"CB01",proxies,ForwardProxy,"")
return streams
@@ -255,7 +218,8 @@ 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 = "tt11378946" # This is an example ID format
# Replace with actual id, for example 'anime_id:episode' format
test_id = "tt32360916:1:1" # This is an example ID format
MFP = "0"
results = await cb01({'streams': []},test_id,MFP,['test','test'],client)
print(results)
+1 -1
View File
@@ -1 +1 @@
{}
{'PHPSESSID': 'aq7rf8gu6nd9fuk0r711899blp', 'captch4': 'bc3c5526ddc7d0a3ad827e7848f5678f'}
+22 -59
View File
@@ -5,8 +5,6 @@ from urllib.parse import quote
from Src.Utilities.loadenv import load_env
from Src.API.extractors.mixdrop import mixdrop
from Src.API.extractors.deltabit import deltabit
from Src.API.extractors.maxstream import maxstream
from Src.API.extractors.uprot import bypass_uprot
import re
import logging
from Src.Utilities.config import setup_logging
@@ -51,18 +49,6 @@ else:
ForwardProxy = ""
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,'Eurostreaming',language,proxies,ForwardProxy)
else:
if maxstream_link == False:
return streams
else:
streams['streams'].append({'name': f"{Name}",'title': f'{Icon}Eurostreaming\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
def convert_numbers(base64_data):
image_data = base64.b64decode(base64_data)
@@ -74,9 +60,8 @@ def convert_numbers(base64_data):
async def get_numbers(safego_url,client):
headers = random_headers.generate()
headers['User-Agent'] = 'Mozilla/5.0 (X11; Linux x86_64; rv:146.0) Gecko/20100101 Firefox/146.0'
headers['Accept'] = 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8'
response = await client.get(ForwardProxy + safego_url,headers = headers, proxies = proxies, impersonate = 'chrome')
headers['User-Agent'] = 'Mozilla/5.0 (X11; Linux x86_64; rv:138.0) Gecko/20100101 Firefox/138.0'
response = await client.get(ForwardProxy + safego_url,headers = headers, proxies = proxies)
cookies = (response.cookies.get_dict())
soup = BeautifulSoup(response.text,'lxml',parse_only=SoupStrainer('img'))
numbers = soup.img['src'].split(',')[1]
@@ -89,8 +74,7 @@ async def real_page(safego_url,client):
headers = random_headers.generate()
headers['Origin'] = 'https://safego.cc'
headers['Referer'] = safego_url
headers['User-Agent'] = 'Mozilla/5.0 (X11; Linux x86_64; rv:146.0) Gecko/20100101 Firefox/146.0'
headers['User-Agent'] = 'Mozilla/5.0 (X11; Linux x86_64; rv:138.0) Gecko/20100101 Firefox/138.0'
with open(file_path, 'r') as file:
cookies = file.read()
cookies = json.loads(cookies.replace("'", '"'))
@@ -98,11 +82,11 @@ async def real_page(safego_url,client):
soup = BeautifulSoup(response.text,'lxml', parse_only=SoupStrainer('a'))
if len(soup)>= 1:
return soup.a['href']
elif 'The requested URL was not found on this server.' not in response.text:
else:
logger.info("Getting numbers")
numbers,cookies = await get_numbers(safego_url,client)
numbers = convert_numbers(numbers)
data = {'captch5': numbers}
data = {'captch4': numbers}
response = await client.post(ForwardProxy + safego_url,headers=headers, data=data, cookies = cookies, proxies = proxies)
cap4 = response.headers.get('set-cookie').split(';')[0]
cookies[cap4.split('=')[0]] = cap4.split('=')[1]
@@ -117,34 +101,27 @@ async def get_host_link(pattern,atag,client):
headers = random_headers.generate()
if match:
href_value = match.group(1)
k = 0
while 'safego' not in href_value:
response = await client.get(ForwardProxy + href_value, headers={**headers, 'Range': 'bytes=0-0'}, proxies = proxies)
href_value = response.url
if '%20' in href_value:
href_value = href_value.replace('%20','')
k+=1
if k == 5:
break
response = await client.head(ForwardProxy + href_value, headers={**headers, 'Range': 'bytes=0-0'}, proxies = proxies)
href_value = response.url
href = await real_page(href_value,client)
return href
async def scraping_links(atag,MFP,MFP_CREDENTIALS,client,streams,language):
#Check which hosts are avaiable and extract the links from one of them. Turbovid uses Cloudflare therefore is not avaiable. Maxstream has a captcha.
if "MixDrop" in atag and "DeltaBit" in atag:
pattern = r'<a\s+href="([^"]+)"[^>]*[^>]*>DeltaBit</a>'
pattern = r'<a\s+href="([^"]+)"[^>]*rel="noopener"[^>]*>DeltaBit</a>'
href = await get_host_link(pattern,atag,client)
try:
streams = await deltabit(href,client,streams,"Eurostreaming",proxies,ForwardProxy,language,'Deltabit')
streams = await deltabit(href,client,streams,"Eurostreaming",proxies,ForwardProxy,language)
except Exception as e:
pattern = r'<a\s+href="([^"]+)"[^>]*rel="noopener"[^>]*>MixDrop</a>'
href = await get_host_link(pattern,atag,client)
streams,status = await mixdrop(href,client,MFP,MFP_CREDENTIALS,streams,"Eurostreaming",proxies,ForwardProxy,language)
streams = await mixdrop(href,client,MFP,MFP_CREDENTIALS,streams,"Eurostreaming",proxies,ForwardProxy,language)
return streams
if "MixDrop" in atag and "DeltaBit" not in atag:
try:
pattern = r'<a\s+href="([^"]+)"[^>]*rel="noopener"[^>]*>MixDrop</a>'
href = await get_host_link(pattern,atag,client)
streams,status = await mixdrop(href,client,MFP,MFP_CREDENTIALS,streams,"Eurostreaming",proxies,ForwardProxy,language)
streams = await mixdrop(href,client,MFP,MFP_CREDENTIALS,streams,"Eurostreaming",proxies,ForwardProxy,language)
return streams
except Exception as e:
return streams
@@ -152,36 +129,18 @@ async def scraping_links(atag,MFP,MFP_CREDENTIALS,client,streams,language):
try:
pattern = r'<a\s+href="([^"]+)"[^>]*rel="noopener"[^>]*>DeltaBit</a>'
href = await get_host_link(pattern,atag,client)
streams = await deltabit(href,client,streams,"Eurostreaming",proxies,ForwardProxy,language,'Deltabit')
streams = await deltabit(href,client,streams,"Eurostreaming",proxies,ForwardProxy,language)
return streams
except Exception as e:
return streams
if 'DeltaBit' not in atag and 'MixDrop' not in atag and 'MaxStream' in atag:
try:
pattern = r'<a\s+href="([^"]+)"[^>]*rel="noopener"[^>]*>MaxStream</a>'
match = re.search(pattern, atag)
if match:
href_value = match.group(1)
streams = await get_maxstream(href_value,streams,language,client)
return streams
except Exception as e:
return streams
if 'Deltabit' not in atag and 'Mixdrop' not in atag and 'Maxstream' not in atag and 'Turbovid' in atag:
try:
pattern = r'<a\s+href="([^"]+)"[^>]*rel="noopener"?[^>]*>Turbovid</a>'
href = await get_host_link(pattern,atag,client)
streams = await deltabit(href,client,streams,"Eurostreaming",proxies,ForwardProxy,language,'Turbovid')
return streams
except Exception as e:
return streams
if 'DeltaBit' not in atag and 'MixDrop' not in atag and 'MaxStream' not in atag:
if 'DeltaBit' not in atag and 'MixDrop' not in atag:
logger.info("Just give up")
return streams
async def episodes_find(description,season,episode,MFP,MFP_CREDENTIALS,client,streams):
t = 0
episode = episode.zfill(2)
pattern = rf'\b{season}&#215;{episode}\s*(.*?)(?=<br\s*/?>)'
pattern = rf'{season}&#215;{episode}\s*(.*?)(?=<br\s*/?>)'
match = re.findall(pattern, description)
if match:
for episode_details in match:
@@ -195,7 +154,7 @@ async def episodes_find(description,season,episode,MFP,MFP_CREDENTIALS,client,st
return streams
async def search(showname,date,season,episode,MFP,MFP_CREDENTIALS,client,streams):
headers = random_headers.generate()
showname = showname.replace("'"," ")
response = await client.get(ForwardProxy + f"{ES_DOMAIN}/wp-json/wp/v2/search?search={quote(showname)}&_fields=id", proxies = proxies, headers = headers)
results = response.json()
for i in results:
@@ -206,7 +165,7 @@ async def search(showname,date,season,episode,MFP,MFP_CREDENTIALS,client,streams
title = description['title']['rendered']
description = description['content']['rendered']
ratio = difflib.SequenceMatcher(None, title, showname).ratio()
if ratio >=0.96 or (showname in title):
if ratio >=0.96:
streams = await episodes_find(description,season,episode,MFP,MFP_CREDENTIALS,client,streams)
return streams
else:
@@ -224,7 +183,6 @@ async def search(showname,date,season,episode,MFP,MFP_CREDENTIALS,client,streams
match = year_pattern.search(response_2.text)
if match:
year = match.group(0)
print(abs(int(year)),int(date))
if abs(int(year) - int(date)) <=1:
streams = await episodes_find(description,season,episode,MFP,MFP_CREDENTIALS,client,streams)
return streams
@@ -260,9 +218,14 @@ async def eurostreaming(streams,id,client,MFP,MFP_CREDENTIALS):
async def test_euro():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
results = await eurostreaming({'streams': []},"tt2701582:7:2",client,"0",['test','test'])
results = await eurostreaming({'streams': []},"tt9253284:1:4",client,"0",['test','test'])
print(results)
async def test_deltabit():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
results = await deltabit("https://deltabit.co/fgeki2456ab1",client)
print(results)
if __name__ == "__main__":
import asyncio
+13 -37
View File
@@ -5,7 +5,6 @@ import re
import time
import logging
from Src.Utilities.config import setup_logging
from Src.Utilities.eval import eval_solver
level = config.LEVEL
logger = setup_logging(level)
Icon = config.Icon
@@ -16,8 +15,9 @@ random_headers = Headers()
async def deltabit(page_url,client,streams,site_name,proxies,ForwardProxy,language,extractor):
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()
@@ -46,49 +46,25 @@ async def deltabit(page_url,client,streams,site_name,proxies,ForwardProxy,langua
name = input.get('name')
value = input.get('value')
data[name] = value
if extractor == 'Deltabit':
imhuman = ''
elif extractor == 'Turbovid':
imhuman = 'Proceed+to+video'
data['imhuman'] = imhuman
data['imhuman'] = ''
data['referer']= page_url
#We need to wait 2.5/5 seconds or the request will fail
if extractor == 'Deltabit':
time.sleep(2.5)
elif extractor == 'Turbovid':
time.sleep(5)
#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)
if extractor == 'Deltabit':
link = re.findall(r'sources:\s*\["([^"]+)"', response.text, re.DOTALL)
if link:
link = link[0]
elif extractor == 'Turbovid':
link = await eval_solver(response.text,proxies, ForwardProxy, client,r'sources:\s*\["([^"]+)"')
link = re.findall(r'sources:\s*\["([^"]+)"', response.text, re.DOTALL)
if not link:
i = 0
if not i >= 3:
link,fname = await deltabit(page_url,client,streams,site_name,proxies,ForwardProxy,language,extractor)
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▶️ {extractor}\n{fname}', 'url': link, 'behaviorHints': {'bingeGroup': f'{site_name.lower()}'}})
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
#Testing
async def test_deltabit():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
results = await deltabit("https://turbovid.me/squc5gdzptgw",client,{'streams': []},'Deltabit', {},"","",'Turbovid')
print(results)
if __name__ == "__main__":
import asyncio
asyncio.run(test_deltabit())
# https://loadm.cam/#cfnqx
return streams
+3 -8
View File
@@ -72,15 +72,10 @@ async def loadm(player_link,client,streams,referer,site_name,proxies,ForwardProx
#We get the hls
hls = data['cf']
title = data['title']
m3u8 = data['source']
if hls:
streams['streams'].append({'name': f"{Name}",'title': f'{Icon}{site_name}\n▶️ Loadm\nPlayer One\n{title}', 'url': hls, 'behaviorHints': {'proxyHeaders': {"request": {"Referer": player_url}}, 'notWebReady': True, 'bingeGroup': f'{site_name.lower()}hls'}})
logger.info(f"{site_name} on Loadm found results for player one in the current ID")
if m3u8:
streams['streams'].append({'name': f"{Name}",'title': f'{Icon}{site_name}\n▶️ Loadm\nPlayer Two\n{title}', 'url': m3u8, 'behaviorHints': {'proxyHeaders': {"request": {"Referer": player_url}}, 'notWebReady': True, 'bingeGroup': f'{site_name.lower()}m3u8'}})
logger.info(f"{site_name} on Loadm found results for player two in the current ID")
return streams
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
-31
View File
@@ -1,31 +0,0 @@
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
from fake_headers import Headers
random_headers = Headers()
import re
async def maxstream(url,client,streams,site_name,language,proxies,ForwardProxy):
headers = random_headers.generate()
response = await client.get( ForwardProxy + url, allow_redirects=True, timeout=30, headers = headers, proxies = proxies, impersonate = "chrome")
pattern = r'sources\W+src\W+(.*)",'
match = re.search(pattern,response.text)
if match:
url = match.group(1)
logger.info(f"{site_name} on Maxstream found results for the current ID")
streams['streams'].append({'name': f"{Name}{language}",'title': f'{Icon}{site_name}\n▶️ Maxstream', 'url': url, 'behaviorHints': { 'bingeGroup': f'{site_name.lower()}'}})
return streams
async def test_maxstream():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
results = await maxstream("https://maxstream.video/emvvv/a7wc8bb0h68c",client,{'streams': []},'CB01', {},'')
print(results)
if __name__ == "__main__":
import asyncio
asyncio.run(test_maxstream())
+2 -5
View File
@@ -11,7 +11,6 @@ from Src.Utilities.eval import eval_solver
async def mixdrop(url,client,MFP,MFP_CREDENTIALS,streams,site_name,proxies,ForwardProxy,language):
"""Extract Mixdrop URL."""
status = False
if "club" in url:
url = url.replace("club", "cv").split("/2")[0]
if "cfd" in url:
@@ -20,15 +19,13 @@ async def mixdrop(url,client,MFP,MFP_CREDENTIALS,streams,site_name,proxies,Forwa
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")
status = True
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,r'MDCore.wurl ?= ?"(.*?)"')
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")
status = True
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,status
return streams
-45
View File
@@ -1,45 +0,0 @@
import Src.Utilities.config as config
from Src.Utilities.eval import eval_solver
import logging
from Src.Utilities.config import setup_logging
from Src.Utilities.mfp import build_mfp
level = config.LEVEL
logger = setup_logging(level)
Icon = config.Icon
Name = config.Name
async def streamhg(link,client,MFP,MFP_CREDENTIALS,streams, site_name, proxies,ForwardProxy):
link = 'https://audinifer.com/e/' + link.split('/e/')[1]
if MFP == '1':
url = await build_mfp(MFP_CREDENTIALS,link,"StreamHG",client)
logger.info(f"{site_name} on StreamHG found results for the current ID")
if url:
status = True
streams['streams'].append({'name': f"{Name} 🕵️‍♂️",'title': f'{Icon}{site_name}\n▶️ StreamHG', 'url': url, 'behaviorHints':{'bingeGroup': f'{site_name.lower()}'}})
else:
status = False
else:
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,"1",['',''],{'streams': []}, "StreamHG", {},"")
print(results)
if __name__ == "__main__":
import asyncio
asyncio.run(test_streamhg())
+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,r'file:"(.*?)"')
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")
-157
View File
@@ -1,157 +0,0 @@
import os
from bs4 import BeautifulSoup,SoupStrainer
import json
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
env_vars = load_env()
import random
logger = setup_logging(level)
#Set up proxies
proxies = {}
if UT_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
}
if UT_ForwardProxy == "1":
ForwardProxy = env_vars.get('ForwardProxy')
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',
'Accept-Language': 'en-US,en;q=0.5',
'Content-Type': 'application/x-www-form-urlencoded',
'Origin': url.split('msf')[0],
'Sec-GPC': '1',
'Connection': 'keep-alive',
'Referer': url,
'Upgrade-Insecure-Requests': '1',
'Sec-Fetch-Dest': 'document',
'Sec-Fetch-Mode': 'navigate',
'Sec-Fetch-Site': 'same-origin',
'Sec-Fetch-User': '?1',
'DNT': '1',
'Priority': 'u=0, i',
}
async def get_uprot_numbers(client):
try:
response = await client.post(ForwardProxy + url, headers=headers, proxies = proxies, impersonate = 'chrome')
set_cookie = response.headers.get('set-cookie')
parts = set_cookie.split(';')[0].split('=')
key = parts[0]
value = parts[1]
cookies = {
key:value
}
soup = BeautifulSoup(response.text,'lxml',parse_only=SoupStrainer('img'))
img = soup.find('img')['src']
return img,cookies
except Exception as e:
logger.info(f'Uprot failed to generate numbers/cookies: {e}')
return None,None
async def generate_uprot_txt(numbers,cookies,client):
try:
data = {
'captcha': str(numbers),
}
response = await client.post(ForwardProxy + url, cookies=cookies, headers=headers, data=data, proxies = proxies, impersonate = 'chrome')
if response.status_code != 200:
return False
set_cookie = response.headers.get('set-cookie')
parts = set_cookie.split(';')[0].split('=')
key = parts[0]
value = parts[1]
cookies[key] = value
current_directory = os.path.dirname(os.path.abspath(__file__))
file_path = os.path.join(current_directory, 'uprot.txt')
with open(file_path,'w') as file:
file.write(f'{str(cookies)}\n{str(data)}')
return True
return False
except Exception as e:
logger.info(f'Uprot failed to generate captcha cookie: {e}')
return False
async def find_link(text,client):
soup = BeautifulSoup(text,'lxml',parse_only=SoupStrainer('a'))
a_tags = soup.find_all('a')
max_stream = None
for tag in a_tags:
if 'C O N T I N U E' in tag.text.upper() or 'CONTINUE' in tag.text.upper() :
max_stream = tag['href']
return max_stream
async def get_maxstream_link(text,client):
max_stream = await find_link(text,client)
if max_stream:
redirect = max_stream
time = 0
while 'uprots' in redirect:
response = await client.head(ForwardProxy + max_stream,headers=headers,impersonate = 'chrome', allow_redirects = True, proxies = proxies)
redirect = response.url
time+=1
if time == 10:
return False
max_stream = 'https://maxstream.video/emvvv/' + response.url.split('watchfree/')[1].split('/')[1]
return max_stream
async def bypass_uprot(client,link):
try:
if 'msfi' not in link:
if 'msf' in link:
link = link.replace('msf','mse')
headers2 = random_headers.generate()
headers2['Origin'] = url.split('msf')[0]
headers2['Referer'] = url
response = await client.get(ForwardProxy + link, headers=headers2, impersonate = 'chrome', proxies = proxies)
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')
current_directory = os.path.dirname(os.path.abspath(__file__))
file_path = os.path.join(current_directory, 'uprot.txt')
with open(file_path,'r') as file:
text = file.read()
parts = text.split('\n')
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)
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:
logger.info(f'Uprot failed: {e}')
if '[Errno 2] No such file or directory' in e:
return None
return False
-2
View File
@@ -1,2 +0,0 @@
{}
{}
+1 -1
View File
@@ -15,7 +15,7 @@ async def uqload(player_link,client,MFP,MFP_CREDENTIALS,streams,site_name,proxie
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}{site_name}\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'}})
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)
-128
View File
@@ -1,128 +0,0 @@
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
from base64 import b64encode,b64decode
level = config.LEVEL
logger = setup_logging(level)
import base64
from bs4 import BeautifulSoup,SoupStrainer
VD_DOMAIN = config.VD_DOMAIN
from curl_cffi.requests import AsyncSession
import json
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.VD_ForwardProxy
if VD_ForwardProxy == "1":
ForwardProxy = env_vars.get('ForwardProxy')
else:
ForwardProxy = ""
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',
}
headers2 = {
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64; rv:150.0) Gecko/20100101 Firefox/150.0',
'Accept': '*/*',
'Accept-Language': 'en-US,en;q=0.9',
'Referer': 'https://v.vidxgo.co/tt34437972',
'Sec-GPC': '1',
'Connection': 'keep-alive',
'Sec-Fetch-Dest': 'empty',
'Sec-Fetch-Mode': 'cors',
'Sec-Fetch-Site': 'same-origin',
'DNT': '1',
'Priority': 'u=4',
}
async def vidxgo_refresh(link,id):
async with AsyncSession() as client:
headers['Referer'] = f'{VD_DOMAIN}/t/{id}'
response = await client.get(ForwardProxy + f'{VD_DOMAIN}/t/{id}', allow_redirects=True, headers = headers2, proxies = proxies)
return response.json()['url'].replace('\\','')
async def vidxgo(link,client,streams,instance_url):
#headers = random_headers.generate()
response = await client.get(ForwardProxy + link, allow_redirects=True, headers = headers, proxies = proxies)
soup = BeautifulSoup(response.text,'lxml',parse_only=SoupStrainer('script'))
scripts = soup.find_all('script')
text = scripts[6]
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)
submatch = re.search(r'window.__EXTERNAL_SUBS\s+=\s+(\[.*]);',decrypted_code)
suburlmatch = re.search(r'window.__SUBS_ORIGIN\s+=\s*"(.*)";',decrypted_code)
if match:
url = match.group(1).replace("\\","")
subtitles = []
if submatch and suburlmatch:
subtitles = json.loads(submatch.group(1))
for item in subtitles:
if item['forced'] == True:
item['id'] = item['lang'] + '-forced'
else:
item['id'] = item['lang']
del item['forced']
del item['file']
item['url'] = suburlmatch.group(1).replace('\\','') + item['url']
streams['streams'].append({'name': f"{Name}",'title': f'{Icon}Vidxgo\n ▶️ Vidxgo', 'url': f'{instance_url}/clone/manifest.m3u8?d={b64encode(url.encode("utf-8")).decode("utf-8")}', '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'},'subtitles' : subtitles})
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 vidxgo("https://v.vidxgo.co/tt0816692",client,{'streams': []},"localhost")
print(results)
if __name__ == "__main__":
import asyncio
asyncio.run(test_vidxgo())
# XOR decryption
+27 -37
View File
@@ -19,44 +19,34 @@ async def vixcloud(link,client,MFP,MFP_CREDENTIALS,streams,site_name,proxies,For
final_url = await build_mfp(MFP_CREDENTIALS, link, "VixCloud",client)
quality = ""
mfp_icon = '🕵️‍♂️'
if final_url:
streams['streams'].append({"name":f'{Name} {mfp_icon}\n{quality}', 'title': f'{Icon} StreamingCommunity\n▶️ Vixcloud','url': final_url,'behaviorHints': {'bingeGroup': f'{site_name.lower()}{quality}'}})
return streams
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
# I will get real link from API
parts = link.split(SC_DOMAIN)
response = await client.get(ForwardProxy + SC_DOMAIN + '/api' + parts[1],proxies=proxies)
link = SC_DOMAIN + '/' + response.json()['src']
response = await client.get(ForwardProxy + link,headers=headers, proxies = proxies)
if response.status_code != 200:
logger.warning(f"Failed to extract URL components from VixSRC, Invalid Request: {response.status_code}")
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"
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")
#We add .m3u8 after the VixID in order to make the video playable by ExoPlayer
parts_final_url = final_url.split("?")
first_part_final_url = parts_final_url[0] + ".m3u8"
final_url = first_part_final_url + "?" + parts_final_url[1]
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
@@ -73,7 +63,7 @@ async def vixcloud(link,client,MFP,MFP_CREDENTIALS,streams,site_name,proxies,For
async def test_vixcloud():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
results = await vixcloud("https://vixsrc.to/tv/76479/5/6/",client,"0",['',''],{'streams': []},'guardoserie.live', {},'')
results = await vixcloud("https://vixsrc.to/tv/204082/1/3/",client,"0",['',''],{'streams': []},'guardoserie.live', {},'')
print(results)
if __name__ == "__main__":
import asyncio
-100
View File
@@ -1,100 +0,0 @@
'''
Code taken from https://github.com/mhdzumair/mediaflow-proxy/blob/main/mediaflow_proxy/extractors/voe.py adpted to work with MammaMia
The owner of MammaMia should not be thanked for any code in this script.
'''
import base64
import re
from typing import Dict, Any
from urllib.parse import urljoin
import Src.Utilities.config as config
from Src.Utilities.config import setup_logging
from Src.Utilities.mfp import build_mfp
level = config.LEVEL
logger = setup_logging(level)
Icon = config.Icon
Name = config.Name
from fake_headers import Headers
random_headers = Headers()
async def voe(url,streams,site_name,MFP,MFP_CREDENTIALS,proxies,ForwardProxy,client):
if MFP == '1':
url = await build_mfp(MFP_CREDENTIALS,url,"Voe",client)
if url and "https%3ANone" not in url:
logger.info(f"{site_name} on Voe found results for the current ID")
streams['streams'].append({'name': f"{Name} 🕵️‍♂️",'title': f'{Icon}{site_name}\n▶️ Voe', 'url': url, 'behaviorHints':{'bingeGroup': f'{site_name.lower()}'}})
return streams
headers = random_headers.generate()
response = await client.get(ForwardProxy + url,headers=headers, impersonate = 'chrome', proxies = proxies)
# See https://github.com/Gujal00/ResolveURL/blob/master/script.module.resolveurl/lib/resolveurl/plugins/voesx.py
redirect_pattern = r'''window\.location\.href\s*=\s*'([^']+)'''
redirect_match = re.search(redirect_pattern, response.text, re.DOTALL)
if redirect_match:
if redirected:
logger.info("VOE: too many redirects")
return await voe(redirect_match.group(1))
code_and_script_pattern = r'json">\["([^"]+)"]</script>\s*<script\s*src="([^"]+)'
code_and_script_match = re.search(code_and_script_pattern, response.text, re.DOTALL)
if not code_and_script_match:
logger.info("VOE: unable to locate obfuscated payload or external script URL")
script_response = await client.get(ForwardProxy+urljoin(url, code_and_script_match.group(2)),headers=headers, proxies = proxies)
luts_pattern = r"(\[(?:'\W{2}'[,\]]){1,9})"
luts_match = re.search(luts_pattern, script_response.text, re.DOTALL)
if not luts_match:
logger.info("VOE: unable to locate LUTs in external script")
data = voe_decode(code_and_script_match.group(1), luts_match.group(1))
final_url = data.get('source')
if not final_url:
logger.info("VOE: failed to extract video URL")
return streams
if final_url:
logger.info(f"{site_name} on Voe found results for the current ID")
streams['streams'].append({'name': f"{Name}",'title': f'{Icon}{site_name}\n▶️ Voe', 'url': final_url, 'behaviorHints': {'proxyHeaders': {"request": {"Referer": url}}, 'notWebReady': True, 'bingeGroup': f'{site_name.lower()}'}})
return streams
def voe_decode(ct: str, luts: str) -> Dict[str, Any]:
import json
lut = [''.join([('\\' + x) if x in '.*+?^${}()|[]\\' else x for x in i]) for i in luts[2:-2].split("','")]
txt = ''
for i in ct:
x = ord(i)
if 64 < x < 91:
x = (x - 52) % 26 + 65
elif 96 < x < 123:
x = (x - 84) % 26 + 97
txt += chr(x)
for i in lut:
txt = re.sub(i, '', txt)
ct = base64.b64decode(txt).decode('utf-8')
txt = ''.join([chr(ord(i) - 3) for i in ct])
txt = base64.b64decode(txt[::-1]).decode('utf-8')
return json.loads(txt)
async def test_voe():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
results = await voe("https://crystaltreatmenteast.com/wnnov1m62xzy",{'streams': []},'ToonItalia','1','[https:idk.com,a]',{},'',client)
print(results)
if __name__ == "__main__":
import asyncio
asyncio.run(test_voe())
+4 -20
View File
@@ -9,9 +9,6 @@ from Src.API.extractors.uqload import uqload
from Src.API.extractors.dropload import dropload
import logging
from Src.Utilities.config import setup_logging
from Src.Utilities.update_config import update_site
import json
import random
level = config.LEVEL
logger = setup_logging(level)
env_vars = load_env()
@@ -37,15 +34,7 @@ if GF_ForwardProxy == "1":
else:
ForwardProxy = ""
async def new_domain(url,data,headers,proxies,ForwardProxy,client):
url = url.split('/wp-admin')[0]
headers['Origin'] = url
headers['Referer'] = f'{url}/'
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 = await update_site(client,['guardaplay','Guardaflix'])
return response.text
async def get_player(page_link,MFP, MFP_CREDENTIALS,client, headers,streams):
response = await client.get(ForwardProxy + page_link, headers= headers, proxies = proxies)
@@ -66,19 +55,14 @@ async def search(showname,date,MFP,MFP_CREDENTIALS,client,streams):
headers['Origin'] = f'{GF_DOMAIN}'
headers['Referer'] = f'{GF_DOMAIN}/'
headers['Content-Type'] = 'application/x-www-form-urlencoded; charset=UTF-8'
showname = quote(showname)
data = {
'action': 'action_tr_search_suggest',
'nonce': '20115729b4',
'term': showname,
}
response = await client.post(ForwardProxy + f'{GF_DOMAIN}/wp-admin/admin-ajax.php', headers=headers, data=data, proxies = proxies)
text = response.text
if response.status_code != 200:
if response.status_code == 400:
text = await new_domain(response.url,data,headers,proxies,ForwardProxy,client)
if len(text) < 10:
logger.warning(f"Guardaflix Failed to fetch search results: {response.status_code}")
soup = BeautifulSoup(text,'lxml', parse_only= SoupStrainer('a'))
soup = BeautifulSoup(response.text,'lxml', parse_only= SoupStrainer('a'))
a_tags = soup.find_all('a')
for a in a_tags:
href = a['href']
@@ -125,7 +109,7 @@ async def guardaflix(streams,id,client,MFP,MFP_CREDENTIALS):
async def test_guardaflix():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
results = await guardaflix({'streams': []},"tt4045450",client,"0",['test','test'])
results = await guardaflix({'streams': []},"tt16426418",client,"0",['test','test'])
print(results)
+8 -16
View File
@@ -4,8 +4,7 @@ import re
import Src.Utilities.config as config
from fake_headers import Headers
from Src.Utilities.loadenv import load_env
from Src.API.extractors.streamhg import streamhg
from Src.API.extractors.mixdrop import mixdrop
from Src.API.extractors.supervideo import supervideo
import json, random
import logging
from Src.Utilities.config import setup_logging
@@ -41,28 +40,21 @@ async def search(clean_id,client):
if response.status_code != 200:
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') and (('mixdrop') in tag.text or ('streamhg') in tag.text):
results[tag.text.strip()] = 'https:' + tag['data-link']
return results
li_tag = soup.find('li')
href = "https:" + li_tag['data-link']
return href
async def guardahd(streams,id,client,MFP,MFP_CREDENTIALS):
async def guardahd(streams,id,client):
try:
general = await is_movie(id)
ismovie = general[0]
clean_id = general[1]
if ismovie == 0:
return streams
results = await search(clean_id,client)
status = False
if 'streamhg' in results:
streams,status = await streamhg(results['streamhg'], client,MFP,MFP_CREDENTIALS,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,'')
supervideo_link = await search(clean_id,client)
streams = await supervideo(supervideo_link, client,streams,"GuardaHD",proxies, ForwardProxy)
return streams
except Exception as e:
logger.warning(f"MammaMia: GuardaHD Failed {e}")
@@ -73,7 +65,7 @@ async def test_script():
async with AsyncSession() as client:
# Replace with actual id, for example 'anime_id:episode' format
test_id = "tt16426418" # This is an example ID format
results = await guardahd({'streams': []},test_id, client,'0','')
results = await guardahd({'streams': []},test_id, client)
print(results)
if __name__ == "__main__":
+3 -3
View File
@@ -51,7 +51,7 @@ async def search_imdb(clean_id,client):
except Exception as e:
logger.warning(f'GS {e}')
return None
'''
async def search(showname,date,client):
try:
headers = random_headers.generate()
@@ -71,7 +71,7 @@ async def search(showname,date,client):
except Exception as e:
return None
'''
async def player_url(page_url, season, episode,client):
@@ -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 = "tt33318655:2:5" # This is an example ID format tt0460649
test_id = "tt10919420:1:1" # This is an example ID format tt0460649
results = await guardaserie({'streams': []},test_id, client)
print(results)
+3 -21
View File
@@ -10,10 +10,7 @@ from Src.API.extractors.loadm import loadm
from Src.API.extractors.dropload import dropload
import logging
from Src.Utilities.config import setup_logging
from Src.Utilities.update_config import update_site
level = config.LEVEL
import json
import random
logger = setup_logging(level)
env_vars = load_env()
random_headers = Headers()
@@ -33,21 +30,12 @@ if GO_PROXY == "1":
"https": proxy
}
GO_ForwardProxy = config.GO_ForwardProxy
if GO_ForwardProxy == "1":
ForwardProxy = env_vars.get('ForwardProxy')
else:
ForwardProxy = ""
async def new_domain(url,data,headers,proxies,ForwardProxy,client):
url = url.split('/wp-admin')[0]
headers['Origin'] = url
headers['Referer'] = f'{url}/'
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"Guardoserie Failed to fetch search results: {response.status_code}")
result = await update_site(client,['guardoserie_2','Guardoserie'])
return response.text
async def get_player(page_link,MFP,MFP_CREDENTIALS,client, headers, streams):
response = await client.get(ForwardProxy + page_link, headers= headers, proxies = proxies)
@@ -73,13 +61,7 @@ async def search(showname,date,season,episode,MFP, MFP_CREDENTIALS,ismovie,clien
showname = quote(showname)
data = f's={showname}&action=searchwp_live_search&swpengine=default&swpquery={showname}'
response = await client.post(ForwardProxy + f'{GO_DOMAIN}/wp-admin/admin-ajax.php', data = data, headers = headers, proxies = proxies)
text = response.text
if response.status_code != 200:
if response.status_code == 400:
text = await new_domain(response.url,data,headers,proxies,ForwardProxy,client)
if len(text) < 10:
logger.warning(f"Guardoserie Failed to fetch search results: {response.status_code}")
soup = BeautifulSoup(text,'lxml', parse_only= SoupStrainer('a', class_='ss-title'))
soup = BeautifulSoup(response.text,'lxml', parse_only= SoupStrainer('a', class_='ss-title'))
a_tags = soup.find_all('a')
for a in a_tags:
href = a['href']
@@ -148,7 +130,7 @@ async def guardoserie(streams,id,client,MFP,MFP_CREDENTIALS):
async def test_guardoserie():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
results = await guardoserie({'streams': []},"tt9165438:3:1",client,"1",['test','test'])
results = await guardoserie({'streams': []},"tt16426418",client,"1",['test','test'])
print(results)
@@ -7,18 +7,10 @@ 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)
@@ -35,12 +27,25 @@ if OST_ForwardProxy == "1":
ForwardProxy = env_vars.get('ForwardProxy')
else:
ForwardProxy = ""
Name = config.Name
Icon = config.Icon
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',
}
async def search(showname,date,client,ismovie,episode,season):
headers = random_headers.generate()
headers['Referer'] = f'{OST_DOMAIN}/'
cookies = {
'player_opt': 'fx',
}
@@ -52,69 +57,64 @@ async def search(showname,date,client,ismovie,episode,season):
'origin_id': '50141',
'searchwp_live_search_client_nonce': 'undefined',
}
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)
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)
if response.status_code != 200:
logger.info("IP Blocked by OnlineserieTV",response)
print("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 = "chrome", proxies = proxies)
response = await client.get(ForwardProxy + href, headers=headers, impersonate = "chrome124", proxies = proxies)
if response.status_code != 200:
logger.info("IP Blocked by OnlineserieTV",response)
print("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/msf/[^\s"<>]+'
pattern = r'https://uprot\.net/fxf/[^\s"<>]+'
match = re.search(pattern, response.text)
if match:
name = a_tag.text.replace("\t","").replace("\n","")
maxstream_link = match.group()
return maxstream_link,name
flexy_link = match.group()
return flexy_link,name
else:
logger.info("No Maxstream link found.")
print("No flexy link found.")
else:
continue
elif ismovie == 0:
if "serietv" in href:
response = await client.get(ForwardProxy + href, headers=headers, impersonate = "chrome", proxies = proxies)
response = await client.get(ForwardProxy + href, headers=headers, impersonate = "chrome124", proxies = proxies)
if response.status_code != 200:
logger.info("IP Blocked by OnlineserieTV",response)
print("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)
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/[^\'"]+)'
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/[^\'"]+)'
match = re.search(pattern, response.text, re.DOTALL)
if match:
name = a_tag.text.replace("\t","").replace("\n","")
maxstream_link = match.group(1)
return maxstream_link,name
else:
logger.info("No maxstream link found.")
flexy_link = match.group(1)
print(match,flexy_link)
return flexy_link,name
else:
print("No flexy 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(streams,id,client):
async def onlineserietv(id,client):
try:
general = await is_movie(id)
ismovie = general[0]
@@ -131,21 +131,22 @@ async def onlineserietv(streams,id,client):
else:
showname,date = get_info_tmdb(clean_id,ismovie,type)
showname = showname.replace("'"," ")
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
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
except Exception as e:
logger.info("MammaMia: Onlineserietv Failed")
logger.info(e)
return streams
print("MammaMia: Onlineserietv Failed",e)
return None,None
async def test_animeworld():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
test_id = "tt16426418" # This is an example ID format
results = await onlineserietv({'streams': []},test_id, client)
test_id = "tt0206516:2:1" # This is an example ID format
results = await onlineserietv(test_id, client)
print(results)
if __name__ == "__main__":
-341
View File
@@ -1,341 +0,0 @@
from bs4 import BeautifulSoup,SoupStrainer
from Src.Utilities.convert import get_TMDb_id_from_IMDb_id
from Src.Utilities.info import get_info_tmdb, is_movie, get_info_imdb
import Src.Utilities.config as config
import json
import random
import re
from urllib.parse import urlparse, parse_qs
from fake_headers import Headers
from Src.Utilities.loadenv import load_env
from Src.API.extractors.vixcloud import vixcloud
import urllib.parse
User_Agent= "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:127.0) Gecko/20100101 Firefox/127.0"
env_vars = load_env()
SC_PROXY = config.SC_PROXY
VX_PROXY = config.VX_PROXY
proxies = {}
proxies2 = {}
if VX_PROXY == "1":
PROXY_CREDENTIALS = env_vars.get('PROXY_CREDENTIALS')
proxy_list = json.loads(PROXY_CREDENTIALS)
proxy = random.choice(proxy_list)
if proxy == "":
proxies2 = {}
else:
proxies2 = {
"http": proxy,
"https": proxy
}
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
}
if VX_PROXY == "1":
proxies2 = proxies
SC_ForwardProxy = config.SC_ForwardProxy
VX_ForwardProxy = config.VX_ForwardProxy
if SC_ForwardProxy == "1":
ForwardProxy = env_vars.get('ForwardProxy')
elif VX_ForwardProxy == "1":
ForwardProxy = env_vars.get('ForwardProxy')
else:
ForwardProxy = ""
#Get domain
SC_DOMAIN = config.SC_DOMAIN
Public_Instance = config.Public_Instance
Alternative_Link = env_vars.get('ALTERNATIVE_LINK')
headers = Headers()
#GET VERSION OF STREAMING COMMUNITY:
async def get_version(client):
#Extract the version from the main page of the site
#You can extract it from any page of the site but I chose one of the lightiest
try:
random_headers = headers.generate()
random_headers['Referer'] = f"{SC_DOMAIN}/"
random_headers['Origin'] = f"{SC_DOMAIN}"
base_url = f'{SC_DOMAIN}/richiedi-un-titolo'
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")
# Extract version
version = json.loads(soup.find("div", {"id": "app"}).get("data-page"))['version']
return version
except Exception as e:
print("Couldn't find the version",e)
version = "65e52dcf34d64173542cd2dc6b8bb75b"
return version
async def search(query,date,ismovie, client,SC_FAST_SEARCH,movie_id):
'''
This function is used to search for a movie or a show in the Streaming Community website using the API
After it checks if the imdb id given is the same as the one in the website
We need to convert imdb id to tmdb id because often imdb ids are missing
'''
random_headers = headers.generate()
random_headers['Referer'] = f"{SC_DOMAIN}/"
random_headers['Origin'] = f"{SC_DOMAIN}"
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(ForwardProxy + query, headers = random_headers, allow_redirects=True, impersonate = "chrome124", proxies = proxies)
print(response)
response = response.json()
for item in response['data']:
tid = item['id']
slug = item['slug']
type = item['type']
if type == "tv":
type = 0
elif type == "movie":
type = 1
if type == ismovie:
#Added a Check to see if the result is what it is supposed to be
if SC_FAST_SEARCH == "0":
random_headers = headers.generate()
random_headers['Referer'] = f"{SC_DOMAIN}/"
random_headers['Origin'] = f"{SC_DOMAIN}"
response = await client.get ( ForwardProxy + f'{SC_DOMAIN}/it/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']
if "tt" in movie_id:
movie_id = str(await get_TMDb_id_from_IMDb_id(movie_id,client))
print(movie_id)
#Here we need to convert because the IMDB ID is often bugged
tmdb_id = str(data['props']['title']['tmdb_id'])
if tmdb_id == movie_id:
return tid,slug,version
elif SC_FAST_SEARCH == "1":
version = await get_version(client)
return tid,slug,version
else:
print("Couldn't find anything")
async def get_film(tid,version,client,MFP):
''''
This function is used to get the link of the m3u8 from the Streaming Community player,vixcloud
'''
random_headers = headers.generate()
random_headers['Referer'] = f"{SC_DOMAIN}/"
random_headers['Origin'] = f"{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
#Access the iframe
url = f'{SC_DOMAIN}/it/iframe/{tid}'
if MFP == "1":
url = f'{SC_DOMAIN}/it/iframe/{tid}'
quality = "Unknown"
return url,quality
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")
#Get the ID containted in the src of iframe
vixid = iframe.split("/embed/")[1].split("?")[0]
parsed_url = urlparse(iframe)
query_params = parse_qs(parsed_url.query)
random_headers = headers.generate()
random_headers['Referer'] = f"{SC_DOMAIN}/"
random_headers['Origin'] = f"{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(ForwardProxy + iframe, headers = random_headers, allow_redirects=True,impersonate= "chrome124", proxies = proxies2)
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?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
elif "&b" in url and quality == "1080":
url = url.replace("&b=1","&h=1")
elif quality == "1080" and "&b" and "&h" not in url:
url = url + "&h=1"
else:
url = url + f"&token={token}"
'''
return url,quality
async def get_season_episode_id(tid,slug,season,episode,version,client):
'''
This function is used to get the ID of the episode in the Streaming Community website
'''
random_headers = headers.generate()
random_headers['Referer'] = f"{SC_DOMAIN}/"
random_headers['Origin'] = f"{SC_DOMAIN}"
random_headers['x-inertia'] = "true"
random_headers['x-inertia-version'] = version
random_headers['accept'] = 'text/html, application/xhtml+xml'
random_headers['accept-language'] = 'it-IT,it;q=0.9,en-US;q=0.8,en;q=0.7'
#Set some basic headers for the request
#Get episode ID
response = await client.get(ForwardProxy + f'{SC_DOMAIN}/it/titles/{tid}-{slug}/season-{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:
if dict_episode['number'] == episode:
return dict_episode['id']
async def get_episode_link(episode_id,tid,version,client,MFP):
''''
This function is used to get the link of the m3u8 from the Streaming Community player,vixcloud
'''
#The parameters for the request
random_headers = headers.generate()
random_headers['Referer'] = f"{SC_DOMAIN}/"
random_headers['Origin'] = f"{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
if MFP == "1":
url = f'{SC_DOMAIN}/it/iframe/{tid}?episode_id={episode_id}&next_episode=1'
quality = "Unknown"
return url,quality
response = await client.get(ForwardProxy + f"{SC_DOMAIN}/it/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'] = f"{SC_DOMAIN}/"
random_headers['Origin'] = f"{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(ForwardProxy + iframe, headers = random_headers, allow_redirects=True, impersonate = "chrome124", proxies = proxies2)
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?token={token}&expires={expires}'
if 'canPlayFHD' in query_params:
canPlayFHD = 'h=1'
url += "&h=1"
if 'b=1' in base_url:
b = 'b=1'
url += "&b=1"
'''
if quality == "1080":
if "&h" in url:
url = url
elif "&b" in url and quality == "1080":
url = url.replace("&b=1","&h=1")
elif quality == "1080" and "&b" and "&h" not in url:
url = url + "&h=1"
else:
url = url + f"&token={token}"
'''
return url,quality
async def streamingcommunity_site(imdb,client,SC_FAST_SEARCH,MFP):
general = await is_movie(imdb)
ismovie = general[0]
imdb_id = general[1]
if ismovie == 0 :
season = int(general[2])
episode = int(general[3])
#Check if fast search is enabled or disabled
if SC_FAST_SEARCH == "1":
type = "StreamingCommunityFS"
if "tt" in imdb:
#Get showname
showname = await get_info_imdb(imdb_id,ismovie,type,client)
date = None
else:
#I just set n season to None to avoid bugs, but it is not needed if Fast search is enabled
date = None
#else just equals them
tmdba = imdb_id.replace("tmdb:","")
showname = get_info_tmdb(tmdba,ismovie,type)
elif SC_FAST_SEARCH == "0":
type = "StreamingCommunity"
if "tt" in imdb:
tmdba = await get_TMDb_id_from_IMDb_id(imdb_id,client)
showname,date = get_info_tmdb(tmdba,ismovie,type)
#HERE THE CASE IF IT IS A MOVIE
else:
if SC_FAST_SEARCH == "1":
type = "StreamingCommunityFS"
if "tt" in imdb:
#Get showname
date = None
showname = await get_info_imdb(imdb_id,ismovie,type,client)
else:
date = None
tmdba = imdb_id.replace("tmdb:","")
showname = get_info_tmdb(tmdba,ismovie,type)
elif SC_FAST_SEARCH == "0":
type = "StreamingCommunity"
if "tt" in imdb:
#Get showname
showname,date = await get_info_imdb(imdb_id,ismovie,type,client)
else:
tmdba = imdb_id.replace("tmdb:","")
showname,date = get_info_tmdb(tmdba,ismovie,type)
showname = showname.replace(" ", "+").replace("–", "+").replace("—","+")
showname = urllib.parse.quote_plus(showname)
query = f'{SC_DOMAIN}/api/search?q={showname}'
tid,slug,version = await search(query,date,ismovie,client,SC_FAST_SEARCH,imdb_id)
if ismovie == 1:
#TID means temporaly ID
url,quality = await get_film(tid,version,client,MFP)
print("MammaMia found results for StreamingCommunity")
return url,quality,slug
if ismovie == 0:
#Uid = URL ID
episode_id = await get_season_episode_id(tid,slug,season,episode,version,client)
url,quality = await get_episode_link(episode_id,tid,version,client,MFP)
return url,quality,slug
-298
View File
@@ -1,298 +0,0 @@
from Src.Utilities.info import get_info_tmdb, is_movie, get_info_imdb
import Src.Utilities.config as config
from fake_headers import Headers
from Src.Utilities.loadenv import load_env
from urllib.parse import quote
import logging
from Src.Utilities.config import setup_logging
level = config.LEVEL
import random
Icon = config.Icon
Name = config.Name
logger = setup_logging(level)
env_vars = load_env()
random_headers = Headers()
RT_PROXY = config.RT_PROXY
proxies = {}
if RT_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
}
RT_ForwardProxy = config.RT_ForwardProxy
if RT_ForwardProxy == "1":
ForwardProxy = env_vars.get('ForwardProxy')
else:
ForwardProxy = ""
endpoints = {
'it' : 'https://public.aurora.enhanced.live',
'dplay' : 'https://eu1-prod.disco-api.com'
}
async def search(showname,date,client):
showname = showname.split('-')[0]
showname = quote(showname)
response = await client.get(ForwardProxy + f'https://public.aurora.enhanced.live/site/search/page/?include=default&filter[environment]=realtime&v=2&q={showname}&page[number]=1&page[size]=20', proxies = proxies)
data = response.json()
for item in data['data']:
return item['slug']
async def program_info(slug,season,episode,client):
headers = {
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64; rv:143.0) Gecko/20100101 Firefox/143.0',
'Accept': '*/*',
'Accept-Language': 'en-US,en;q=0.5',
'Referer': 'https://realtime.it/',
'Origin': 'https://realtime.it',
'Sec-GPC': '1',
'Connection': 'keep-alive',
'Sec-Fetch-Dest': 'empty',
'Sec-Fetch-Mode': 'cors',
'Sec-Fetch-Site': 'cross-site',
'DNT': '1',
'Priority': 'u=4',
}
link = f'https://public.aurora.enhanced.live/site/page/{slug}/?include=default&filter[environment]=realtime&v=2&parent_slug=programmi-real-time'
response = await client.get(ForwardProxy + link,headers = headers, proxies = proxies)
data = response.json()
for item in reversed(data['blocks'][1]['items']):
if season == item['seasonNumber'] and episode == item['episodeNumber']:
if 'aurora' in data['blocks'][0]['item']['poster']['src']:
platform = 'IT'
elif 'eu1-prod' in data['blocks'][0]['item']['poster']['src']:
platform = 'DPLAY'
x_realm_it = ''
x_realm_dplay = ''
if 'X-REALM-IT' in data['userMeta']['realm']:
x_realm_it = data['userMeta']['realm']['X-REALM-IT']
if 'X-REALM-DPLAY' in data['userMeta']['realm']:
x_realm_dplay = data['userMeta']['realm']['X-REALM-DPLAY']
return item['id'],x_realm_it,x_realm_dplay, platform
return None,None,None,None
async def get_token(client):
headers = {
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64; rv:143.0) Gecko/20100101 Firefox/143.0',
'Accept': '*/*',
'Accept-Language': 'en-US,en;q=0.5',
'Referer': 'https://realtime.it/',
'Content-Type': 'application/json',
'X-disco-client': 'WEB:UNKNOWN:wbdatv:2.1.9',
'X-disco-params': 'realm=dplay',
'X-Device-Info': 'STONEJS/1 (Unknown/Unknown; Linux/undefined; Unknown)',
'Origin': 'https://realtime.it',
'Sec-GPC': '1',
'Connection': 'keep-alive',
'Sec-Fetch-Dest': 'empty',
'Sec-Fetch-Mode': 'cors',
'Sec-Fetch-Site': 'cross-site',
'DNT': '1',
'Priority': 'u=4',
}
response = await client.get(ForwardProxy + 'https://public.aurora.enhanced.live/site/page/casa-a-prima-vista/?include=default&filter[environment]=realtime&v=2&parent_slug=programmi-real-time', headers = headers, proxies = proxies)
data = response.json()
x_realm_it = data['userMeta']['realm']['X-REALM-IT']
x_realm_dplay = data['userMeta']['realm']['X-REALM-DPLAY']
return x_realm_it,x_realm_dplay
async def get_url(id, endpoint, x_realm_it,x_realm_dplay,streams,client):
if 'IT' in endpoint:
base_url = endpoints['it']
token = x_realm_it
elif 'DPLAY' in endpoint:
base_url = endpoints['dplay']
token = x_realm_dplay
headers = {
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64; rv:143.0) Gecko/20100101 Firefox/143.0',
'Accept': '*/*',
'Accept-Language': 'en-US,en;q=0.5',
'Referer': 'https://realtime.it/',
'Content-Type': 'application/json',
'X-disco-client': 'WEB:UNKNOWN:wbdatv:2.1.9',
'X-disco-params': 'realm=dplay',
'X-Device-Info': 'STONEJS/1 (Unknown/Unknown; Linux/undefined; Unknown)',
'Authorization': f'Bearer {token}',
'Origin': 'https://realtime.it',
'Sec-GPC': '1',
'Connection': 'keep-alive',
'Sec-Fetch-Dest': 'empty',
'Sec-Fetch-Mode': 'cors',
'Sec-Fetch-Site': 'cross-site',
'DNT': '1',
'Priority': 'u=4',
}
json_data = {
'deviceInfo': {
'adBlocker': False,
'drmSupported': True,
'hdrCapabilities': [
'SDR',
],
'hwDecodingCapabilities': [],
'soundCapabilities': [
'STEREO',
],
},
'wisteriaProperties': {},
'videoId': id,
}
response = await client.post(ForwardProxy + f'{base_url}/playback/v3/videoPlaybackInfo', headers=headers, json=json_data, proxies = proxies)
data = response.json()
for item in data['data']['attributes']['streaming']:
if item['type'] == 'hls':
m3u8 = item['url']
streams['streams'].append({'name': f"{Name}",'title': f'{Icon}▶️ Realtime\n HLS', 'url': m3u8, 'behaviorHints': {'bingeGroup': f'realtimehls'}})
elif item['type'] == 'dash':
mpd = item['url']
streams['streams'].append({'name': f"{Name}",'title': f'{Icon}▶️ Realtime\n MPD', 'url': mpd, 'behaviorHints': {'bingeGroup': f'realtimempd'}})
return streams
async def search_catalog(query,catalog,client):
showname = quote(query)
response = await client.get(f'https://public.aurora.enhanced.live/site/search/page/?include=default&filter[environment]=realtime&v=2&q={showname}&page[number]=1&page[size]=20')
data = response.json()
for item in data['data']:
title = item['title']
description = item['subtitle']
date = item['datePublished'].split('-')[0]
id = item['slug']
image = item['image']['url']
typeof = item['type']
catalog['metas'].append({'id': f'realtime{typeof}:'+id, 'type': "series",'name': f'{title}', 'description': description, 'releaseInfo': date, 'background': image, 'poster': image })
return catalog
async def meta_catalog(id,catalog,client):
try:
headers = {
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64; rv:143.0) Gecko/20100101 Firefox/143.0',
'Accept': '*/*',
'Accept-Language': 'en-US,en;q=0.5',
'Referer': 'https://realtime.it/',
'Origin': 'https://realtime.it',
'Sec-GPC': '1',
'Connection': 'keep-alive',
'Sec-Fetch-Dest': 'empty',
'Sec-Fetch-Mode': 'cors',
'Sec-Fetch-Site': 'cross-site',
'DNT': '1',
'Priority': 'u=4',
}
parts = id.split(':')
slug = parts[1]
parts2 = parts[0].split('realtime')
typeof= parts2[1]
if typeof == 'showpage':
link = f'https://public.aurora.enhanced.live/site/page/{slug}/?include=default&filter[environment]=realtime&v=2&parent_slug=programmi-real-time'
elif typeof == 'article':
link = f'https://public.aurora.enhanced.live/site/page/{slug}/?include=default&filter[environment]=realtime&v=2'
response = await client.get(ForwardProxy + link,headers = headers, proxies = proxies)
data = response.json()
title = data['title']
subtitle = data['subtitle']
if data['type'] == 'articlepage':
if data['blocks'][1]['sonicOverrideEnabled'] == True:
platform = 'IT'
else:
platform = 'DPLAY'
elif data['type'] == 'showpage':
if 'aurora' in data['blocks'][0]['item']['poster']['src']:
platform = 'IT'
elif 'eu1-prod' in data['blocks'][0]['item']['poster']['src']:
platform = 'DPLAY'
if typeof == 'showpage':
for item in reversed(data['blocks'][1]['items']):
id = item['id']
description = item['description']
episode = item['episodeNumber']
season = item['seasonNumber']
poster = item['poster']
date = item['publishStart']
catalog['meta']['videos'].append({'title': f'S{season}'+ f'E{episode}','season': season, 'episode': episode,'firstAired':date,'overview': description, 'thumbnail': poster['src'], 'id': f'realtime{platform}:id:'+id})
elif typeof == 'article':
item = data['blocks'][1]['item']
id = item['id']
description = item['description']
episode = item['episodeNumber']
season = item['seasonNumber']
poster = item['poster']
date = item['publishStart']
catalog['meta']['videos'].append({'title': f'{title}','season': season, 'episode': episode,'firstAired':date,'overview': description, 'thumbnail': poster['src'], 'id': f'realtime{platform}:id:'+id})
catalog['meta']['name'] = title
catalog['meta']['description'] = subtitle
catalog['meta']['releaseInfo'] = '-' + data['datePublished'].split('-')[0]
catalog['meta']['background'] = data['metaMedia'][0]['media']['url']
return catalog
except Exception as e:
catalog['meta']['name'] = title
catalog['meta']['description'] = subtitle
catalog['meta']['releaseInfo'] = '-' + data['datePublished'].split('-')[0]
catalog['meta']['background'] = data['metaMedia'][0]['media']['url']
return catalog
async def realtime(streams,id,client):
try:
if 'realtime' in id:
parts = id.split('id:')
id = parts[1]
endpoint = parts[0]
x_realm_it,x_realm_dplay = await get_token(client)
streams = await get_url(id, endpoint, x_realm_it,x_realm_dplay,streams,client)
else:
general = await is_movie(id)
ismovie = general[0]
clean_id = general[1]
type = "Realtime"
if ismovie == 0 :
season = int(general[2])
episode = int(general[3])
elif ismovie == 1:
season = None
episode = None
if "tmdb" in id:
showname,date = get_info_tmdb(clean_id,ismovie,type)
else:
showname,date = await get_info_imdb(clean_id,ismovie,type,client)
logger.info(f'RT {showname}')
slug = await search(showname,date,client)
id,x_realm_it,x_realm_dplay,platform = await program_info(slug,season,episode,client)
streams = await get_url(id,platform,x_realm_it,x_realm_dplay,streams,client)
return streams
except Exception as e:
logger.warning(f'RT {e}')
return streams
async def test_realtime():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
results = await realtime({'streams': []},"tt5684368:13:10",client)
print(results)
if __name__ == "__main__":
import asyncio
asyncio.run(test_realtime())
+1 -1
View File
@@ -87,7 +87,7 @@ async def test_vixsrc():
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
results = await streaming_community({'streams': []},test_id, client,"0",['test','welp'])
results = await streaming_community(test_id, client,"0","0")
print(results)
if __name__ == "__main__":
-123
View File
@@ -1,123 +0,0 @@
from fake_headers import Headers
random_headers = Headers()
import Src.Utilities.config as config
from Src.Utilities.info import get_info_imdb,get_info_tmdb
from urllib.parse import quote
from Src.Utilities.convert import get_IMDB_id_from_TMDb_id
from Src.Utilities.info import get_info_tmdb, is_movie, get_info_imdb
from Src.API.extractors.voe import voe
from Src.API.extractors.maxstream import maxstream
import logging
from Src.Utilities.config import setup_logging
level = config.LEVEL
import re
import json
import random
logger = setup_logging(level)
from Src.Utilities.loadenv import load_env
env_vars = load_env()
TI_DOMAIN = config.TI_DOMAIN
TI_PROXY = config.TI_PROXY
proxies = {}
if TI_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
}
TI_ForwardProxy = config.TI_ForwardProxy
if TI_ForwardProxy == "1":
ForwardProxy = env_vars.get('ForwardProxy')
else:
ForwardProxy = ""
async def search(showname,date,season,episode,MFP,MFP_CREDENTIALS,ismovie,client,streams):
headers = random_headers.generate()
response = await client.get(ForwardProxy + f'{TI_DOMAIN}/wp-json/wp/v2/search?search={quote(showname)}',headers=headers, proxies = proxies)
data = response.json()
for item in data:
response = await client.get(ForwardProxy + item['_links']['self'][0]['href'],headers=headers, proxies = proxies)
info = response.json()
html = info['content']['rendered']
if 'Stagione' in html and ismovie == 1:
continue
if 'Stagione' not in html and ismovie == 0:
continue
if ismovie == 0:
episode = episode.zfill(2)
pattern = rf'{season}&#215;{episode}.*?<a[^>]*href="([^"]*)'
if ismovie == 1:
pattern = r'Link.*href="([^"]*)(?=.*VOE)'
match = re.search(pattern,html)
if match:
url = match.group(1)
if 'maxstream' in url:
parts = url.split('.video/')
url = 'https://maxstream.video/emvvv/' + parts[1]
streams = await maxstream(url,client,streams,'ToonItalia','',proxies,ForwardProxy)
else:
streams = await voe(url,streams,'ToonItalia',MFP,MFP_CREDENTIALS,proxies,ForwardProxy,client)
return streams
return streams
async def toonitalia(streams,id,client,MFP,MFP_CREDENTIALS):
try:
general = await is_movie(id)
ismovie = general[0]
clean_id = general[1]
type = "Toonitalia"
if ismovie == 0 :
season = str(general[2])
episode = str(general[3])
elif ismovie == 1:
season = None
episode = None
if "tmdb" in id:
showname,date = get_info_tmdb(clean_id,ismovie,type)
else:
showname,date = await get_info_imdb(clean_id,ismovie,type,client)
streams = await search(showname,date, season,episode,MFP,MFP_CREDENTIALS,ismovie,client,streams)
return streams
except Exception as e:
logger.warning(f'Toonitalia Error: {e}')
return streams
async def test_toonitalia():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
results = await toonitalia({'streams': []},"tt0096697:1:1",client,"0",['test','test'])
print(results)
if __name__ == "__main__":
import asyncio
asyncio.run(test_toonitalia())
-65
View File
@@ -1,65 +0,0 @@
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.VD_ForwardProxy
if VD_ForwardProxy == "1":
ForwardProxy = env_vars.get('ForwardProxy')
else:
ForwardProxy = ""
VD_DOMAIN = config.VD_DOMAIN
async def vidxgoalta(streams,id,client,instance_url):
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,instance_url)
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())
+5 -21
View File
@@ -5,9 +5,11 @@ import logging
with open('config.json') as f:
# Load JSON data from file
config = json.load(f)
# Accessing SC_DOMAIN
SITE = config["Siti"]
SC_DOMAIN = SITE["StreamingCommunity"]['url']
SW_DOMAIN = SITE["StreamingWatch"]['url']
AW_DOMAIN = SITE['AnimeWorld']['url']
CB_DOMAIN = SITE['CB01']['url']
GS_DOMAIN = SITE['Guardaserie']['url']
@@ -15,11 +17,8 @@ GHD_DOMAIN = SITE['GuardaHD']['url']
ES_DOMAIN = SITE['Eurostreaming']['url']
GF_DOMAIN = SITE['Guardaflix']['url']
GO_DOMAIN = SITE['Guardoserie']['url']
RT_DOMAIN = SITE['Realtime']['url']
TI_DOMAIN = SITE['Toonitalia']['url']
OST_DOMAIN = SITE['OnlineSerieTV']['url']
VD_DOMAIN = SITE['Vidxgo']['url']
SC = SITE['StreamingCommunity']['enabled']
SW = SITE['StreamingWatch']['enabled']
AW = SITE['AnimeWorld']['enabled']
CB = SITE['CB01']['enabled']
GS = SITE['Guardaserie']['enabled']
@@ -27,10 +26,6 @@ GHD = SITE['GuardaHD']['enabled']
ES= SITE['Eurostreaming']['enabled']
GF = SITE['Guardaflix']['enabled']
GO = SITE['Guardoserie']['enabled']
RT = SITE['Realtime']['enabled']
TI = SITE['Toonitalia']['enabled']
OST = SITE['OnlineSerieTV']['enabled']
VD = SITE['Vidxgo']['enabled']
SC_ForwardProxy = SITE['StreamingCommunity']["SC_ForwardProxy"]
GS_ForwardProxy = SITE['Guardaserie']["GS_ForwardProxy"]
GH_ForwardProxy = SITE['GuardaHD']["GH_ForwardProxy"]
@@ -41,10 +36,7 @@ CB_ForwardProxy = SITE['CB01']["CB_ForwardProxy"]
ES_ForwardProxy = SITE['Eurostreaming']['ES_ForwardProxy']
GF_ForwardProxy = SITE['Guardaflix']['GF_ForwardProxy']
GO_ForwardProxy = SITE['Guardoserie']['GO_ForwardProxy']
RT_ForwardProxy = SITE['Realtime']['RT_ForwardProxy']
OST_ForwardProxy = SITE['OnlineSerieTV']['OST_ForwardProxy']
TI_ForwardProxy = SITE['Toonitalia']['TI_ForwardProxy']
VD_ForwardProxy = SITE['Vidxgo']['VD_ForwardProxy']
SW_ForwardProxy = SITE['StreamingWatch']['SW_ForwardProxy']
GS_PROXY = SITE['Guardaserie']["GS_PROXY"]
GH_PROXY = SITE['GuardaHD']["GH_PROXY"]
CB_PROXY = SITE['CB01']["CB_PROXY"]
@@ -55,15 +47,7 @@ MX_PROXY = SITE['CB01']["MX_PROXY"]
ES_PROXY = SITE['Eurostreaming']['ES_PROXY']
GF_PROXY = SITE['Guardaflix']['GF_PROXY']
GO_PROXY = SITE['Guardoserie']['GO_PROXY']
RT_PROXY = SITE['Realtime']['RT_PROXY']
TI_PROXY = SITE['Toonitalia']['TI_PROXY']
OST_PROXY = SITE['OnlineSerieTV']['OST_PROXY']
VD_PROXY = SITE['Vidxgo']['VD_PROXY']
#Captcha
CAPTCHA = config['Captcha']
UT_ForwardProxy = CAPTCHA['Uprot']['UT_ForwardProxy']
UT_PROXY = CAPTCHA['Uprot']['UT_PROXY']
SW_PROXY = SITE['StreamingWatch']['SW_PROXY']
#General
GENERAL = config['General']
dotenv = GENERAL["load_env"]
+4 -12
View File
@@ -8,7 +8,7 @@ STREAM = {
"name": "HD/FHD",
"genres": ["La7"],
"poster": "https://upload.wikimedia.org/wikipedia/commons/thumb/0/02/LA7_-_Logo_2011.svg/1280px-LA7_-_Logo_2011.svg.png",
"url": "https://d1chghleocc9sm.cloudfront.net/v1/master/3722c60a815c199d9c0ef36c5b73da68a62b09d1/cc-evfku205gqrtf/Live.m3u8"
"url": "https://d3749synfikwkv.cloudfront.net/v1/master/3722c60a815c199d9c0ef36c5b73da68a62b09d1/cc-74ylxpgd78bpb/Live.m3u8"
},
{
"id": "rai-1",
@@ -413,11 +413,6 @@ STREAM = {
]
}
REALTIME = {}
extra_sources = {
"rai-1": ["https://ilglobotv-live.akamaized.net/channels/RAI1/Live.m3u8"],
"rai-2": ["https://ilglobotv-live.akamaized.net/channels/RAI2/Live.m3u8"],
@@ -443,8 +438,9 @@ extra_sources = {
provider_map = {
'SC': 'STREAMINGCOMMUNITY',
'SC_MFP': 'SC_MFP',
'SC_FS': 'SC_FAST_SEARCH',
'LC': 'LORDCHANNEL',
'SW': 'STREAMINGWATCH',
'TF_FS': 'TF_FAST_SEARCH',
'TF': 'TANTIFILM', # Assuming 'TF' should also be handled
'FT': 'FILMPERTUTTI',
@@ -456,9 +452,5 @@ provider_map = {
"OST": "ONLINESERIETV",
"ES": "EUROSTREAMING",
"GO": "GUARDOSERIE",
"RT": "REALTIME",
"TI": "TOONITALIA",
"GF": "GUARDAFLIX",
"OST": "ONLINESERIETV",
"VD": "VIDXGO"
"GF": "GUARDAFLIX"
}
+9 -8
View File
@@ -141,22 +141,23 @@ class UnpackingError(Exception):
random_headers = Headers()
async def eval_solver(stream_link,proxies, ForwardProxy, client,pattern):
async def eval_solver(stream_link,proxies, ForwardProxy, client):
try:
headers = random_headers.generate()
headers["User-Agent"] = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36"
headers["accept-language"] = 'en-US,en;q=0.5'
if len(stream_link) <= 200:
response = await client.get( ForwardProxy + stream_link, allow_redirects=True, timeout=30, headers = headers, proxies = proxies, impersonate = "chrome124")
text = response.text
logger.debug("Eval:",response.status_code)
else:
text = stream_link
soup = BeautifulSoup(text, "lxml",parse_only=SoupStrainer("script"))
response = await client.get( ForwardProxy + stream_link, allow_redirects=True, timeout=30, headers = headers, proxies = proxies, impersonate = "chrome124")
logger.debug("Eval:",response.status_code)
soup = BeautifulSoup(response.text, "lxml",parse_only=SoupStrainer("script"))
script_all = soup.find_all("script")
for i in script_all:
if detect(i.text):
unpacked_code = unpack(i.text)
if "mixdrop" in stream_link:
pattern = r'MDCore.wurl ?= ?"(.*?)"'
else:
pattern = r'file:"(.*?)"'
match = re.search(pattern, unpacked_code)
if match:
m3u8_url = match.group(1)
+1 -29
View File
@@ -61,13 +61,6 @@ def get_info_tmdb(tmbda,ismovie,type):
date = show.first_air_date
date = date.split("-")[0]
return showname,date
elif type == 'Realtime':
date = show.first_air_date
return showname,date
elif type == 'Toonitalia':
date = show.first_air_date
date = date.split("-")[0]
return showname,date
elif ismovie == 1:
movie = Movie()
@@ -116,13 +109,7 @@ def get_info_tmdb(tmbda,ismovie,type):
date = show.release_date
date = date.split("-")[0]
return showname,date
elif type == 'Realtime':
date = show.release_date
return showname,date
elif type == "Toonitalia":
date = show.release_date
date = date.split("-")[0]
return showname,date
async def get_info_imdb(imdb_id, ismovie, type,client):
resp = await client.get(f'https://api.themoviedb.org/3/find/{imdb_id}?api_key={TMDB_KEY}&language=it&external_source=imdb_id')
data = resp.json()
@@ -170,13 +157,6 @@ async def get_info_imdb(imdb_id, ismovie, type,client):
date = data['tv_results'][0]['first_air_date']
date = date.split("-")[0]
return showname,date
elif type == 'Realtime':
return showname,data['tv_results'][0]['first_air_date']
elif type == "Toonitalia":
date = data['tv_results'][0]['first_air_date']
date = date.split("-")[0]
return showname,date
elif ismovie == 1:
showname= data['movie_results'][0]['title']
if type == "Filmpertutti":
@@ -215,14 +195,6 @@ async def get_info_imdb(imdb_id, ismovie, type,client):
date = data['movie_results'][0]['release_date']
date = date.split("-")[0]
return showname,date
elif type == 'Realtime':
date = data['movie_results'][0]['release_date']
return showname,date
elif type == 'Toonitalia':
date = data['movie_results'][0]['release_date']
date = date.split("-")[0]
return showname,date
async def get_info_kitsu(kitsu_id,client):
-104
View File
@@ -1,104 +0,0 @@
from fastapi import APIRouter, HTTPException, Response, Request
from fastapi.responses import RedirectResponse
from curl_cffi.requests import AsyncSession
from urllib.parse import urlparse, parse_qs,unquote,quote
from base64 import b64decode,b64encode
import re
from Src.API.extractors.vidxgo import vidxgo_refresh
import time
import Src.Utilities.config as config
from Src.Utilities.config import setup_logging
level = config.LEVEL
logger = setup_logging(level)
VD_DOMAIN = config.VD_DOMAIN
router = APIRouter()
headers = {
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64; rv:150.0) Gecko/20100101 Firefox/150.0',
'Accept': '*/*',
'Accept-Language': 'en-US,en;q=0.9',
'Referer': f'{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',
}
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 = headers)
return response.text,response.status_code
@router.get("/clone/manifest.m3u8")
async def clone_m3u8(d: str = None):
if d:
try:
url = b64decode(d).decode('utf-8')
m3u8_content,status = await fetch_m3u8(url)
if status != 200:
match = re.search(r'hls[\w/]*/(\d{2}\d*[/]*\d*[/]*\d*)[\w/]+master',url)
if match:
id = match.group(1)
url = await vidxgo_refresh("",id)
m3u8_content,status =await fetch_m3u8(url)
m3u8_content = re.sub(r'(&b=\d+)',rf'\1&base={d}',m3u8_content)
return Response(content=m3u8_content, media_type='application/vnd.apple.mpegurl')
except Exception as e:
logger.info(f"Failed to fetch M3U8 file: {e}")
raise HTTPException(status_code=404, detail="M3U8 content not found")
@router.get("/clone/index{suffix}")
async def index_route(suffix: str, base: str):
try:
text2 = b64decode(base).decode('utf-8')
text = text2.replace("master.m3u8",f'index{suffix}')
m3u8_content,status = await fetch_m3u8(text)
match = re.search(r'hls[\w/]*/(\d{2}\d*[/]*\d*[/]*\d*)[\w/]+master',text2)
if match:
id = match.group(1)
else:
raise HTTPException(status_code=502, detail="ID not found")
if status != 200:
text = await vidxgo_refresh("",id)
text = text.replace("master.m3u8",f'index{suffix}')
m3u8_content,status =await fetch_m3u8(text)
current_date = time.strftime('%Y-%m-%d %H:%M')
m3u8_content = re.sub(r'(b=\d+)', rf'\1&id={b64encode(id.encode("utf-8")).decode("utf-8")}&date={b64encode(current_date.encode("utf-8")).decode("utf-8")}', m3u8_content)
return Response(content=m3u8_content, media_type='application/vnd.apple.mpegurl')
except Exception as e:
logger.info(f"Failed to fetch M3U8 file: {e}")
raise HTTPException(status_code=502, detail="M3U8 content not found")
@router.get("/clone/seg{suffix}")
async def ts_route(suffix:str,id:str,date: str,t:str,e:str,b:str):
try:
url = await vidxgo_refresh("",b64decode(id).decode('utf-8'))
parsed = urlparse(url)
params = parse_qs(parsed.query)
url = url.split("?")[0]
text = url.replace("master.m3u8",f'seg{suffix}?t={params["t"][0]}&e={params["e"][0]}&b={b}')
return RedirectResponse(text,headers=headers)
except Exception as e:
logger.info(f"Failed to redirect TS file: {e}")
raise HTTPException(status_code=502, detail="Failed to redirect TS file")
async def test_fetch():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
results = await index_route('-v1-a1','aHR0cHM6Ly9jZG4udjMubWVkaWEtNjg0LmQyYi55b3UvcHJveHkvbWVkaWEtNDAyL2hscy8xNjQyNjQxOC9tYXN0ZXIubTN1OD90PTJiYjg0MGYwMzFjY2MzNmYmZT0xNzgwNTA2OTk2Njk5JmI9Mjk3Ng====')
print(results)
if __name__ == "__main__":
import asyncio
asyncio.run(test_fetch())
+4 -11
View File
@@ -6,18 +6,11 @@ import urllib.parse
level = config.LEVEL
logger = setup_logging(level)
async def transform_mfp(mfp_stream_url,MFP_url,MFP_password,client):
async def transform_mfp(mfp_stream_url,client):
try:
response = await client.get(mfp_stream_url)
data = response.json()
if 'mediaflow_endpoint' in data:
if data['mediaflow_endpoint'] == 'hls_manifest_proxy':
MFP_url += '/proxy/hls/manifest.m3u8'
else:
MFP_url += '/proxy/stream'
url = MFP_url + "?api_password=" + MFP_password + "&d=" + urllib.parse.quote(data['destination_url'])
else:
url = data['mediaflow_proxy_url'] + "?api_password=" + data['query_params']['api_password'] + "&d=" + urllib.parse.quote(data['destination_url'])
url = data['mediaflow_proxy_url'] + "?api_password=" + data['query_params']['api_password'] + "&d=" + urllib.parse.quote(data['destination_url'])
for i in data['request_headers']:
url += f"&h_{i}={urllib.parse.quote(data['request_headers'][i])}"
return url
@@ -30,5 +23,5 @@ async def build_mfp(MFP_CREDENTIALS,url,host,client):
MFP_password = MFP_CREDENTIALS[1]
url_mfp = f'{MFP_url}/extractor/video?api_password={MFP_password}&d={url}&host={host}&redirect_stream=false'
#To build pieces together
url_mfp = await transform_mfp(url_mfp,MFP_url,MFP_password,client)
return url_mfp
url_mfp = await transform_mfp(url_mfp,client)
return url_mfp
-66
View File
@@ -1,66 +0,0 @@
link = 'https://raw.githubusercontent.com/AstraeLabs/Domains/main/domains.json'
import json
import os
import logging
from Src.Utilities.config import setup_logging
import Src.Utilities.config as config
level = config.LEVEL
logger = setup_logging(level)
async def fetch_domain(client,site_name):
response = await client.get(link)
data = response.json()
domain = data[site_name[0]]['full_url'][:-1]
return domain
def write_config(domain,site_name):
try:
json_file_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), 'config.json')
with open(json_file_path, 'r') as json_file:
config_data = json.load(json_file)
try:
config_data['Siti'][site_name[1]]['url'] = domain
with open(json_file_path, 'w') as json_file:
json.dump(config_data, json_file, indent=4)
return True
except Exception as e:
with open(json_file_path, 'w') as json_file:
json.dump(config_data, json_file, indent=4)
logger.info('Failed to write into config: ',e)
return False
except Exception as e:
logger.info('Failed to write into config: ',e)
return False
async def update_site(client,site_name):
try:
domain = await fetch_domain(client,site_name)
result = write_config(domain,site_name)
return result
except Exception as e:
return False
async def update_all_sites(client):
try:
sites = [['cb01new','CB01'],['guardaserie','Guardaserie'],['eurostreaming','Eurostreaming'],['guardaplay','Guardaflix'],['guardoserie_2','Guardoserie'],['animeworld','AnimeWorld'],['toonitalia','Toonitalia']]
for item in sites:
domain = await fetch_domain(client,item)
write_config(domain,item)
return True
except Exception as e:
logger.info('Failed to update domains: ',e)
return False
async def test_update_site():
from curl_cffi.requests import AsyncSession
async with AsyncSession() as client:
result = await update_site(client,['guardoserie_2','Guardoserie'])
#result = await update_all_sites(client)
#domain = await update_site(client,['eurostreamings','Eurostreaming'])
#write_config(domain,['eurostreamings','Eurostreaming'])
if __name__ == "__main__":
import asyncio
asyncio.run(test_update_site())
+2 -2
View File
@@ -1,4 +1,4 @@
{
"projectName": "mammamia",
"lastCommit": "58af5b8"
"projectName": "MammaMia",
"lastCommit": "10f381b"
}
+60 -84
View File
@@ -1,95 +1,71 @@
{
"Siti": {
"StreamingCommunity": {
"url": "https://vixsrc.to",
"SC_ForwardProxy": "0",
"SC_PROXY": "0",
"VX_ForwardProxy": "0",
"VX_PROXY": "0",
"enabled": "1"
},
"CB01": {
"url": "https://cb01uno.click",
"CB_PROXY": "0",
"CB_ForwardProxy": "0",
"MX_ForwardProxy": "0",
"MX_PROXY": "0",
"enabled": "1"
},
"Guardaserie": {
"url": "https://guarda-serie.click",
"enabled": "1",
"GS_ForwardProxy": "0",
"GS_PROXY": "0"
},
"GuardaHD": {
"url": "https://mostraguarda.stream",
"enabled": "1",
"GH_ForwardProxy": "0",
"GH_PROXY": "0"
},
"Eurostreaming": {
"url": "https://eurostream.ing",
"StreamingCommunity": {
"url": "https://vixsrc.to",
"SC_ForwardProxy": "0",
"SC_PROXY": "0",
"VX_ForwardProxy": "0",
"VX_PROXY": "0",
"enabled": "1"
},
"StreamingWatch":{
"url":"https://www.streamingwatch.org",
"SW_PROXY": "0",
"SW_ForwardProxy": "0",
"enabled":"1"
},
"CB01":{
"url":"https://cb01.uno",
"CB_PROXY":"0",
"CB_ForwardProxy":"0",
"MX_ForwardProxy":"0",
"MX_PROXY":"0",
"enabled":"1"
},
"Guardaserie":{
"url":"https://guardaserie.qpon",
"enabled":"1",
"GS_ForwardProxy":"0",
"GS_PROXY":"0"
},
"GuardaHD":{
"url":"https://mostraguarda.stream",
"enabled":"1",
"GH_ForwardProxy":"0",
"GH_PROXY":"0"
},
"Eurostreaming":{
"url": "https://eurostreaming.garden",
"enabled": "1",
"ES_ForwardProxy": "0",
"ES_PROXY": "0"
},
"Guardaflix": {
"url": "https://guardaplay.live",
},
"Guardaflix":{
"url": "https://guardaflix.live",
"enabled": "1",
"GF_ForwardProxy": "0",
"GF_PROXY": "0"
},
"Guardoserie": {
"url": "https://guardoserie.run",
},
"Guardoserie":{
"url": "https://guardoserie.live",
"enabled": "1",
"GO_ForwardProxy": "0",
"GO_PROXY": "0"
},
"AnimeWorld": {
"url": "https://www.animeworld.ac",
"enabled": "1",
"AW_ForwardProxy": "0",
"AW_PROXY": "0"
},
"Realtime": {
"url": "https://realtime.it",
"enabled": "1",
"RT_ForwardProxy": "0",
"RT_PROXY": "0"
},
"Toonitalia": {
"url": "https://toonitalia.xyz",
"enabled": "1",
"TI_ForwardProxy": "0",
"TI_PROXY": "0"
},
"OnlineSerieTV": {
"url": "https://onlineserietv.lol",
"enabled": "1",
"OST_ForwardProxy": "0",
"OST_PROXY": "0"
},
"Vidxgo": {
"url": "https://v.vidxgo.co",
"enabled": "1",
"VD_ForwardProxy": "0",
"VD_PROXY": "0"
}
},
"Captcha": {
"Uprot": {
"UT_ForwardProxy": "0",
"UT_PROXY": "0"
}
},
"General": {
"load_env": "0",
"HOST": "0.0.0.0",
"PORT": 8080,
"Name": "MammaMia",
"Icon": "\ud83c\udf55",
"level": "info",
"Global_Proxy": "0"
}
}
},
"AnimeWorld":{
"url":"https://www.animeworld.so",
"enabled":"1",
"AW_ForwardProxy":"0",
"AW_PROXY":"0"
}
},
"General":{
"load_env": "0",
"HOST": "0.0.0.0",
"PORT": 8080,
"Name": "MammaMia",
"Icon": "🍕️",
"level": "info",
"Global_Proxy": "0"
}
}
+1 -3
View File
@@ -10,6 +10,4 @@ slowapi
curl_cffi
fake-headers
pycryptodome
rich
python-multipart
jinja2
rich
+22 -130
View File
@@ -1,12 +1,14 @@
from fastapi import FastAPI, HTTPException,Request, Form,Cookie
from fastapi import FastAPI, HTTPException,Request
from fastapi.responses import JSONResponse,RedirectResponse,HTMLResponse
from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.middleware import SlowAPIMiddleware
from static.static import HTML
from static.configure import CONFIGURE
from fastapi.templating import Jinja2Templates
from Src.API.streamingcommunity import streaming_community
from Src.API.streamingwatch import streamingwatch
from Src.API.cb01 import cb01
from Src.API.guardaserie import guardaserie
from Src.API.guardahd import guardahd
@@ -14,16 +16,10 @@ from Src.API.animeworld import animeworld
from Src.API.guardaflix import guardaflix
from Src.API.guardoserie import guardoserie
from Src.API.eurostreaming import eurostreaming
from Src.API.onlineserietv import onlineserietv
from Src.API.vidxgoalta import vidxgoalta
from Src.API.toonitalia import toonitalia
from Src.API.realtime import search_catalog as realtime
from Src.API.realtime import meta_catalog as meta_catalog_realtime
from Src.API.realtime import realtime as streams_realtime
from Src.Utilities.dictionaries import STREAM,extra_sources,provider_map
from Src.Utilities.update_config import update_all_sites
from Src.API.epg import tivu, tivu_get,epg_guide,convert_bho_1,convert_bho_2,convert_bho_3
from Src.API.extractors.uprot import get_uprot_numbers,generate_uprot_txt
import logging
from urllib.parse import unquote
from curl_cffi.requests import AsyncSession
@@ -34,7 +30,6 @@ level = config.LEVEL
logger = setup_logging(level)
from Src.Utilities.loadenv import load_env
env_vars = load_env()
from Src.Utilities.manifest import router as clone
#Configure Env Vars
Global_Proxy = config.Global_Proxy
@@ -50,6 +45,7 @@ else:
# Configure config
SC = config.SC
SC_DOMAIN = config.SC_DOMAIN
SW = config.SW
AW = config.AW
CB = config.CB
GS = config.GS
@@ -57,25 +53,18 @@ GHD = config.GHD
ES = config.ES
GF = config.GF
GO = config.GO
RT = config.RT
TI = config.TI
OST = config.OST
VD = config.VD
HOST = config.HOST
PORT = int(config.PORT)
if env_vars.get('PORT_ENV'):
PORT = int(env_vars.get('PORT_ENV'))
Icon = config.Icon
Name = config.Name
#Cool code to set the hugging face if the service is hosted there.
app = FastAPI()
app.include_router(clone)
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_middleware(SlowAPIMiddleware)
User_Agent= "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36"
#Tells Where to look for static files
static = Jinja2Templates(directory="static")
MANIFEST = {
"id": "org.stremio.mammamia",
"version": "2.0.0",
@@ -95,17 +84,6 @@ MANIFEST = {
"options": ["Rai", "Mediaset", "Sky", "Euronews", "La7", "Warner Bros", "FIT", "Sportitalia","RSI","DAZN", "Rakuten", "Pluto", "A+E", "Paramount", "Chill"]
}
]
},
{
'id': 'realtime',
'type': "series",
"name": "realtime",
'extra' :[
{
"name": "search",
"isRequired": True
}
]
}
],
"resources": ["stream", "catalog", "meta"],
@@ -135,25 +113,17 @@ def config(request: Request):
def addon_manifest(config: str):
manifest_copy = MANIFEST.copy()
config = base64.b64decode(config).decode('utf-8')
if "LIVETV" not in config:
if "LIVETV" in config:
return respond_with(manifest_copy)
elif "LIVETV" not in config:
manifest_copy["catalogs"] = []
if "catalog" in manifest_copy["resources"]:
for item in manifest_copy['catalogs']:
if item['id'] == 'tv_channels':
manifest_copy['catalogs'].remove(item)
if not any(manifest_copy['catalogs']):
manifest_copy["resources"].remove('catalog')
if 'RT' not in config or RT == '0':
if "catalog" in manifest_copy["resources"]:
for item in manifest_copy['catalogs']:
if item['id'] == 'realtime':
manifest_copy['catalogs'].remove(item)
if not any(manifest_copy['catalogs']):
manifest_copy["resources"].remove('catalog')
return respond_with(manifest_copy)
manifest_copy["resources"].remove("catalog")
return respond_with(manifest_copy)
@app.get('/manifest.json')
def manifest():
return RedirectResponse(url="/|SC|LC|/manifest.json")
return RedirectResponse(url="/|SC|LC|SW|/manifest.json")
@app.get('/', response_class=HTMLResponse)
def root(request: Request):
@@ -194,14 +164,6 @@ async def first_catalog(type: str, id: str, genre: str = None):
catalogs = await addon_catalog(type, id,genre)
return respond_with(catalogs)
@app.get('/{config:path}/catalog/{type}/{id}/search={query}.json')
async def realtime_catalog(type:str,id:str,query: str = None):
if type != 'series':
raise HTTPException(status_code=404)
catalogs = {"query": query,'cacheMaxAge': 86400,"metas": []}
async with AsyncSession(proxies = proxies) as client:
catalogs = await realtime(query, catalogs,client)
return respond_with(catalogs)
@app.get('/{config:path}/meta/tv/{id}.json')
@limiter.limit("20/second")
async def addon_meta(request: Request,id: str):
@@ -237,14 +199,7 @@ async def addon_meta(request: Request,id: str):
meta['meta']['url'] = channel['url'] # Using the stream URL as a website link
return respond_with(meta)
@app.get('/{config:path}/meta/series/{id}.json')
async def addon_meta(request: Request,id: str):
meta = {'meta': {'videos':[], 'status': 'Continuing', 'type': 'series', 'id': id}}
async with AsyncSession(proxies = proxies) as client:
if 'realtime' not in id:
raise HTTPException(status_code=404)
meta = await meta_catalog_realtime(id,meta,client)
return respond_with(meta)
@app.get('/{config:path}/stream/{type}/{id}.json')
@limiter.limit("5/second")
async def addon_stream(request: Request,config, type, id,):
@@ -278,10 +233,6 @@ async def addon_stream(request: Request,config, type, id,):
else:
MFP = "0"
MFP_CREDENTIALS = ['','']
#Get instance URL to pass it to vidxgo
forwarded_proto = request.headers.get("x-forwarded-proto")
scheme = forwarded_proto if forwarded_proto else request.url.scheme
instance_url = f"{scheme}://{request.url.netloc}"
async with AsyncSession(proxies = proxies) as client:
if type == "tv":
for channel in STREAM["channels"]:
@@ -298,9 +249,6 @@ async def addon_stream(request: Request,config, type, id,):
if not streams['streams']:
raise HTTPException(status_code=404)
return respond_with(streams)
elif "realtime" in id and RT == '1':
streams = await streams_realtime(streams,id,client)
return respond_with(streams)
elif "tt" in id or "tmdb" in id or "kitsu" in id:
logger.info(f"Handling movie or series: {id}")
if "kitsu" in id:
@@ -308,84 +256,28 @@ async def addon_stream(request: Request,config, type, id,):
streams = await animeworld(streams,id,client)
else:
if provider_maps['STREAMINGCOMMUNITY'] == "1" and SC == "1":
if provider_maps['SC_MFP'] != "0":
SC_MFP = "1"
else:
SC_MFP = '0'
streams = await streaming_community(streams,id,client,SC_MFP,MFP_CREDENTIALS)
streams = await streaming_community(streams,id,client,MFP,MFP_CREDENTIALS)
if provider_maps['STREAMINGWATCH'] == "1" and SW == "1":
streams = await streamingwatch(streams,id,client)
if provider_maps['CB01'] == "1" and CB == "1":
streams = await cb01(streams,id,MFP,MFP_CREDENTIALS,client)
if provider_maps['GUARDASERIE'] == "1" and GS == "1":
streams = await guardaserie(streams,id,client)
if provider_maps['GUARDAHD'] == "1" and GHD == "1":
streams = await guardahd(streams,id,client,MFP,MFP_CREDENTIALS)
streams = await guardahd(streams,id,client)
if provider_maps['EUROSTREAMING'] == "1" and ES == "1":
streams = await eurostreaming(streams,id,client,MFP,MFP_CREDENTIALS)
if provider_maps['GUARDAFLIX'] == "1" and GF == "1":
streams = await guardaflix(streams,id,client,MFP,MFP_CREDENTIALS)
if provider_maps['GUARDOSERIE'] == "1" and GO == "1":
streams = await guardoserie(streams,id,client,MFP,MFP_CREDENTIALS)
if provider_maps['REALTIME'] == '1' and RT == '1':
streams = await streams_realtime(streams,id,client)
if provider_maps['TOONITALIA'] == '1' and TI == '1':
streams = await toonitalia(streams,id,client,MFP,MFP_CREDENTIALS)
if provider_maps['ONLINESERIETV'] == '1' and OST == '1':
streams = await onlineserietv(streams,id,client)
if provider_maps['VIDXGO'] == '1' and VD == '1':
streams = await vidxgoalta(streams,id,client,instance_url)
return respond_with(streams)
if not streams['streams']:
raise HTTPException(status_code=404)
return respond_with(streams)
@app.get('/uprot')
async def uprot(request: Request):
async with AsyncSession(proxies = proxies) as client:
image, cookies = await get_uprot_numbers(client)
context = {
'request': request,
'image_url': image
}
response = static.TemplateResponse(name='uprot.html',context=context,request=request)
if cookies:
response.set_cookie(key='PHPSESSID', value=cookies.get('PHPSESSID'),httponly=True)
return response
@app.post("/uprot")
async def execute_uprot(request: Request,user_input = Form(...),PHPSESSID: str = Cookie(None)):
async with AsyncSession(proxies = proxies) as client:
cookies = {
'PHPSESSID': PHPSESSID
}
status = await generate_uprot_txt(user_input,cookies,client)
if status == True:
context = {
'request': request,
'image_url': 'https://tinyurl.com/doneokdone'
}
return static.TemplateResponse(name='uprot.html',context=context,request=request)
elif status == False:
context = {
'request': request,
'image_url': 'https://tinyurl.com/tryagaindumb'
}
return static.TemplateResponse('uprot.html',context=context,request=request)
@app.get('/update')
async def update(request: Request):
async with AsyncSession(proxies = proxies) as client:
result = await update_all_sites(client)
print(result)
if result:
return JSONResponse(content={"message": "200"})
if not result:
return JSONResponse(content={"message": "Failed"})
if __name__ == '__main__':
import uvicorn
uvicorn.run("run:app", host=HOST, port=PORT, log_level=level)
uvicorn.run("run:app", host=HOST, port=PORT, log_level=level)
+8 -35
View File
@@ -75,7 +75,6 @@ CONFIGURE = """
border-radius: 5px;
overflow: hidden;
width: 100%;
}
.provider-group label {
display: flex;
@@ -91,11 +90,6 @@ CONFIGURE = """
width: 4vh;
height: 4vh;
}
.provider-group input[type="checkbox"][id$="_mfp"] {
margin-left: 2.2vh;
width: 4vh;
height: 4vh;
}
.parent-container {
display: flex;
align-items: center;
@@ -180,13 +174,18 @@ CONFIGURE = """
<img src="https://creazilla-store.fra1.digitaloceanspaces.com/emojis/49647/pizza-emoji-clipart-md.png" alt="Logo">
</div>
<h1 class="name">Mamma Mia</h1>
<h2 class="version">v2.2.0</h2>
<h2 class="version">v1.5.0</h2>
<div id="additionalText">
<h2>This addon provides Movie, Series, Anime, and Live TV HTTPS Streams.<br> https://github.com/UrloMythus/MammaMia/</h2>
</div>
<p class="description">🕵️‍♂️ = Mediaflowproxy might be needed <br>Select the box with this icon if you want to enable MFP for that source<br></p>
<p class="description">🕵️‍♂️ = Mediaflowproxy might be needed</p>
<h3 class="gives">Select Providers:</h3>
<form class="pure-form" id="provider-form">
<div class="provider-group">
<label for="streamingwatch" class="provider-label">
<input type="checkbox" id="streamingwatch"> StreamingWatch
</label>
</div>
<div class="provider-group">
<label for="guardoserie" class="provider-label">
<input type="checkbox" id="guardoserie"> Guardoserie
@@ -219,20 +218,9 @@ CONFIGURE = """
<input type="checkbox" id="cb01"> CB01 🕵️‍♂️
</label>
</div>
<div class="provider-group">
<label for="onlineserietv" class="provider-label">
<input type="checkbox" id="onlineserietv"> OnlineSerieTV
</label>
</div>
<div class="provider-group">
<label for="vidxgo" class="provider-label">
<input type="checkbox" id="vidxgo"> Vidxgo
</label>
</div>
<div class="provider-group">
<label for="streamingcommunity" class="provider-label">
<input type="checkbox" id="streamingcommunity"> StreamingCommunity 🕵️‍♂️
<input type="checkbox" id="streamingcommunity_mfp" checked> 🕵️‍♂️
</label>
</div>
<div class="provider-group">
@@ -240,17 +228,6 @@ CONFIGURE = """
<input type="checkbox" id="eurostreaming"> Eurostreaming 🕵️‍♂️
</label>
</div>
<div class="provider-group">
<label for="realtime" class="provider-label">
<input type="checkbox" id="toonitalia"> Toonitalia 🕵️‍♂️
</label>
</div>
<div class="provider-group">
<label for="realtime" class="provider-label">
<input type="checkbox" id="realtime"> Realtime
</label>
</div>
<div class="provider-group">
<label for="livetv" class="provider-label">
<input type="checkbox" id="livetv"> LiveTV
@@ -287,6 +264,7 @@ CONFIGURE = """
let manifest = "|";
const providers = {
"streamingcommunity": "SC",
"streamingwatch": "SW",
"animeworld": "AW",
"livetv": "LIVETV",
"cb01": "CB",
@@ -295,11 +273,6 @@ CONFIGURE = """
"guardoserie": "GO",
"guardaflix": "GF",
"eurostreaming": "ES",
"realtime": "RT",
"toonitalia": "TI",
"onlineserietv": "OST",
"vidxgo": "VD",
"streamingcommunity_mfp": "SC_MFP",
"mediaflowproxy": "MFP"
};
+8 -35
View File
@@ -75,7 +75,6 @@ HTML = """
border-radius: 5px;
overflow: hidden;
width: 100%;
}
.provider-group label {
display: flex;
@@ -91,11 +90,6 @@ HTML = """
width: 4vh;
height: 4vh;
}
.provider-group input[type="checkbox"][id$="_mfp"] {
margin-left: 2.2vh;
width: 4vh;
height: 4vh;
}
.parent-container {
display: flex;
align-items: center;
@@ -180,13 +174,18 @@ HTML = """
<img src="https://creazilla-store.fra1.digitaloceanspaces.com/emojis/49647/pizza-emoji-clipart-md.png" alt="Logo">
</div>
<h1 class="name">Mamma Mia</h1>
<h2 class="version">v2.2.0</h2>
<h2 class="version">v1.5.0</h2>
<div id="additionalText">
<h2>This addon provides Movie, Series, Anime, and Live TV HTTPS Streams.<br> https://github.com/UrloMythus/MammaMia/</h2>
</div>
<p class="description">🕵️‍♂️ = Mediaflowproxy might be needed <br>Select the box with this icon if you want to enable MFP for that source<br></p>
<p class="description">🕵️‍♂️ = Mediaflowproxy might be needed</p>
<h3 class="gives">Select Providers:</h3>
<form class="pure-form" id="provider-form">
<div class="provider-group">
<label for="streamingwatch" class="provider-label">
<input type="checkbox" id="streamingwatch"> StreamingWatch
</label>
</div>
<div class="provider-group">
<label for="guardoserie" class="provider-label">
<input type="checkbox" id="guardoserie"> Guardoserie
@@ -219,20 +218,9 @@ HTML = """
<input type="checkbox" id="cb01"> CB01 🕵️‍♂️
</label>
</div>
<div class="provider-group">
<label for="onlineserietv" class="provider-label">
<input type="checkbox" id="onlineserietv"> OnlineSerieTV
</label>
</div>
<div class="provider-group">
<label for="vidxgo" class="provider-label">
<input type="checkbox" id="vidxgo"> Vidxgo
</label>
</div>
<div class="provider-group">
<label for="streamingcommunity" class="provider-label">
<input type="checkbox" id="streamingcommunity"> StreamingCommunity 🕵️‍♂️
<input type="checkbox" id="streamingcommunity_mfp" checked> 🕵️‍♂️
</label>
</div>
<div class="provider-group">
@@ -240,17 +228,6 @@ HTML = """
<input type="checkbox" id="eurostreaming"> Eurostreaming 🕵️‍♂️
</label>
</div>
<div class="provider-group">
<label for="realtime" class="provider-label">
<input type="checkbox" id="toonitalia"> Toonitalia 🕵️‍♂️
</label>
</div>
<div class="provider-group">
<label for="realtime" class="provider-label">
<input type="checkbox" id="realtime"> Realtime
</label>
</div>
<div class="provider-group">
<label for="livetv" class="provider-label">
<input type="checkbox" id="livetv"> LiveTV
@@ -287,6 +264,7 @@ HTML = """
let manifest = "|";
const providers = {
"streamingcommunity": "SC",
"streamingwatch": "SW",
"animeworld": "AW",
"livetv": "LIVETV",
"cb01": "CB",
@@ -295,11 +273,6 @@ HTML = """
"guardoserie": "GO",
"guardaflix": "GF",
"eurostreaming": "ES",
"realtime": "RT",
"toonitalia": "TI",
"onlineserietv": "OST",
"vidxgo": "VD",
"streamingcommunity_mfp": "SC_MFP",
"mediaflowproxy": "MFP"
};
-128
View File
@@ -1,128 +0,0 @@
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>Mamma Mia</title>
<link rel="icon" href="https://creazilla-store.fra1.digitaloceanspaces.com/emojis/49647/pizza-emoji-clipart-md.png" type="image/x-icon">
<title>Fast Search Example</title>
<style>
* {
box-sizing: border-box;
}
body, html {
margin: 0;
padding: 0;
width: 100%;
height: 100%;
font-size: 2.2vh;
font-family: 'Open Sans', Arial, sans-serif;
color: white;
background: url('https://i.postimg.cc/ry3p76HY/italian-seamless-free-vector-pattern3.png') center center repeat;
background-size: cover;
display: flex;
align-items: flex-start;
justify-content: center;
overflow-y: auto;
}
#captcha {
background: rgba(0, 0, 0, 0.8);
padding: 0.5vh;
border-radius: 10px;
width: 65vh;
max-width: 100%;
text-align: center;
margin-top: 10vh;
}
.logo {
width: 12vh;
margin: 0 auto;
margin-bottom: 3vh;
margin-top: -3vh;
}
.logo img {
width: 100%;
height: auto;
}
.captchaimg img {
width: 80%;
height: auto;
object-fit: contain;
display: block;
margin: 0 auto;
}
h1, h2, h3 {
margin: 0;
text-shadow: 0 0 1vh rgba(0, 0, 0, 0.15);
}
h1 {
font-size: 4.5vh;
font-weight: 700;
}
h2 {
font-size: 2vh;
font-weight: normal;
font-style: italic;
opacity: 0.8;
margin-bottom: 20px;
}
h3 {
font-size: 2.2vh;
margin-bottom: 10px;
}
#additionalText {
margin-top: 2vh;
font-size: 1.8vh;
text-align: left;
}
/* Responsive adjustments for smaller screens */
@media (max-width: 600px) {
.provider-group label {
font-size: 2vh;
white-space: nowrap;
}
}
form {
border: 0;
outline: 0;
color: white;
background: #8A5AAB;
padding: 1.2vh 3.5vh;
text-align: center;
font-family: 'Open Sans', Arial, sans-serif;
font-size: 2.2vh;
font-weight: 600;
cursor: pointer;
display: block;
box-shadow: 0 0.5vh 1vh rgba(0, 0, 0, 0.2);
transition: box-shadow 0.1s ease-in-out;
width: 80%;
max-width: 35vh;
margin: 1vh auto;
}
form:hover {
box-shadow: none;
}
</style>
</head>
<body>
<div id="captcha">
<div class="logo">
<img src="https://creazilla-store.fra1.digitaloceanspaces.com/emojis/49647/pizza-emoji-clipart-md.png" alt="Logo">
</div>
<h1 class="name">Uprot Captcha</h1>
<h2 class="version">Per abilitare MaxStream</h2>
<div id="additionalText">
<h2>Inserire i numeri in foto per abilitare la fonte MaxStream per la durata di un giorno<br> https://github.com/UrloMythus/MammaMia/</h2>
<div class="captchaimg">
<img src="{{ image_url }}" alt="Failed to get the captcha">
</div>
<form id = 'form' action="/uprot" method="post">
<input type="text" name="user_input" required>
<button type="submit">Submit</button>
</form>
</body>
</html>
-159
View File
@@ -1,159 +0,0 @@
#!/usr/bin/env python3
"""
Merge remote URLs into local config.json, replacing ONLY the "url" lines.
- Scarica config.json remoto (raw GitHub).
- Per ogni sito in "Siti", segue i redirect e trova l'URL finale.
- Apre il config.json locale come testo, trova le righe "url" relative ai siti
e sostituisce SOLO il valore dentro le virgolette, preservando tutto il resto.
- Crea un backup del config.json locale prima di sovrascriverlo.
"""
from pathlib import Path
from datetime import datetime
import shutil
import re
import requests
import urllib3
from urllib.parse import urlparse
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# --- CONFIGURAZIONE ---
REMOTE_RAW_URL = "https://raw.githubusercontent.com/UrloMythus/MammaMia/main/config.json"
LOCAL_CONFIG_PATH = Path("config.json")
HEADERS = {"User-Agent": "Mozilla/5.0 (compatible; DomainMerger/1.0)"}
REQUEST_TIMEOUT = 12
# -----------------------
def backup_file(path: Path) -> Path | None:
if not path.exists():
return None
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
bak = path.with_name(f"{path.name}.bak_{ts}")
shutil.copy2(path, bak)
return bak
def normalize_url(u: str) -> str:
parsed = urlparse(u)
scheme = parsed.scheme if parsed.scheme else "https"
netloc = parsed.netloc or parsed.path
return f"{scheme}://{netloc}"
def follow_redirect(url: str) -> str | None:
if not url.startswith(("http://", "https://")):
test = "http://" + url
else:
test = url
try:
r = requests.get(test, allow_redirects=True, headers=HEADERS,
timeout=REQUEST_TIMEOUT, verify=False)
return normalize_url(r.url)
except requests.RequestException as e:
print(f"⚠️ Redirect fallito per '{url}': {e}")
return None
def load_remote_urls(remote_raw_url: str) -> dict:
"""Scarica il JSON remoto e ritorna mappa site_key -> url (o None se mancante)."""
try:
r = requests.get(remote_raw_url, headers=HEADERS, timeout=REQUEST_TIMEOUT)
r.raise_for_status()
remote_json = r.json()
except Exception as e:
raise RuntimeError(f"Impossibile scaricare/parsing remote config: {e}")
siti = remote_json.get("Siti", {})
result = {}
for site_key, site_info in siti.items():
url = site_info.get("url")
if url:
result[site_key] = url
return result
def merge_remote_into_local(remote_map: dict, local_path: Path):
if not local_path.exists():
raise FileNotFoundError(f"File locale non trovato: {local_path}")
# backup
bak = backup_file(local_path)
if bak:
print(f"📦 Backup creato: {bak.name}")
text = local_path.read_text(encoding="utf-8")
lines = text.splitlines()
# regex: match apertura blocco sito es. "CB01":{
site_open_re = re.compile(r'^\s*"(?P<site>[^"]+)"\s*:\s*\{\s*$')
# regex per trovare esattamente la riga "url": "valore" preservando indentazione e virgola finale
url_line_re = re.compile(r'^(\s*)"url"\s*:\s*"([^"]*)"( *)(,?)\s*$')
current_site = None
updated_lines = []
changes = []
for idx, line in enumerate(lines):
site_match = site_open_re.match(line)
if site_match:
current_site = site_match.group("site")
updated_lines.append(line)
continue
url_match = url_line_re.match(line)
if url_match:
indent = url_match.group(1)
old_value = url_match.group(2)
trailing_space = url_match.group(3) or ""
trailing_comma = url_match.group(4) or ""
site_key = current_site or "unknown"
# se il sito esiste nella mappa remota, proviamo ad aggiornare
remote_url = remote_map.get(site_key)
if remote_url:
final_url = follow_redirect(remote_url)
if final_url and final_url != normalize_url(old_value):
# sostituisco solo la parte tra virgolette
new_line = f'{indent}"url": "{final_url}"{trailing_space}{trailing_comma}'
updated_lines.append(new_line)
changes.append((site_key, old_value, final_url))
print(f"✅ {site_key}: {old_value} → {final_url}")
else:
# nessuna modifica (redirect fallito o identico)
updated_lines.append(line)
print(f"ℹ️ {site_key}: nessuna modifica per '{old_value}'")
else:
# il sito non è nella lista remota: non tocchiamo la riga
updated_lines.append(line)
print(f"ℹ️ {site_key}: nessun URL remoto per questo sito, riga lasciata invariata")
continue
# linea normale
updated_lines.append(line)
# Scrivo solo se ci sono cambiamenti (ma anche se non ci sono, sovrascrivo per coerenza)
new_text = "\n".join(updated_lines)
# preserva newline finale se il file originale ce l'aveva
if text.endswith("\n") and not new_text.endswith("\n"):
new_text += "\n"
local_path.write_text(new_text, encoding="utf-8")
print("💾 Merge completato: solo le righe 'url' del file locale sono state eventualmente aggiornate.")
if not changes:
print("🔎 Nessun cambiamento effettuato.")
else:
print("📄 Riepilogo modifiche:")
for s, old, new in changes:
print(f" - {s}: {old} → {new}")
def main():
try:
remote_map = load_remote_urls(REMOTE_RAW_URL)
except Exception as e:
print(f"❌ Errore: {e}")
return
try:
merge_remote_into_local(remote_map, LOCAL_CONFIG_PATH)
except Exception as e:
print(f"❌ Errore durante il merge: {e}")
if __name__ == "__main__":
main()