mirror of
https://github.com/Viren070/MediaFusion.git
synced 2025-12-01 23:21:11 +01:00
mhdzumair
120 lines
3.1 KiB
Python
120 lines
3.1 KiB
Python
import logging
|
|
|
|
import requests
|
|
from urllib.parse import urlparse, quote
|
|
from requests.exceptions import RequestException
|
|
|
|
from db import schemas
|
|
|
|
|
|
def is_valid_url(url: str) -> bool:
|
|
parsed_url = urlparse(url)
|
|
return all([parsed_url.scheme, parsed_url.netloc])
|
|
|
|
|
|
def does_url_exist(url: str) -> bool:
|
|
try:
|
|
response = requests.head(url, allow_redirects=True, timeout=5)
|
|
return response.status_code == 200
|
|
except RequestException:
|
|
return False
|
|
|
|
|
|
def validate_image_url(url: str) -> bool:
|
|
return is_valid_url(url) and does_url_exist(url)
|
|
|
|
|
|
def validate_m3u8_url(url: str, behaviour_hint: dict) -> bool:
|
|
if not is_valid_url(url):
|
|
return False
|
|
|
|
try:
|
|
headers = behaviour_hint.get("proxyHeaders", {}).get("request", {})
|
|
response = requests.get(url, allow_redirects=True, headers=headers, timeout=30)
|
|
content_type = response.headers.get("Content-Type", "").lower()
|
|
return (
|
|
"application/vnd.apple.mpegurl" in content_type
|
|
or "application/x-mpegurl" in content_type
|
|
)
|
|
except RequestException as err:
|
|
logging.error(err)
|
|
return False
|
|
|
|
|
|
class ValidationError(Exception):
|
|
pass
|
|
|
|
|
|
def validate_yt_id(yt_id: str) -> bool:
|
|
image_url = f"https://img.youtube.com/vi/{yt_id}/mqdefault.jpg"
|
|
response = requests.head(image_url, allow_redirects=True, timeout=5)
|
|
return response.status_code == 200
|
|
|
|
|
|
def validate_tv_metadata(metadata: schemas.TVMetaData) -> list[schemas.TVStreams]:
|
|
if (
|
|
not validate_image_url(metadata.poster)
|
|
or (metadata.logo and not validate_image_url(metadata.logo))
|
|
or (metadata.background and not validate_image_url(metadata.background))
|
|
):
|
|
raise ValidationError("Invalid image URL provided.")
|
|
|
|
# Validate stream URLs at least 1 stream should be valid
|
|
valid_streams = [
|
|
stream
|
|
for stream in metadata.streams
|
|
if (
|
|
stream.url
|
|
and validate_m3u8_url(
|
|
stream.url, stream.behaviorHints.model_dump(exclude_none=True)
|
|
)
|
|
)
|
|
or (stream.ytId and validate_yt_id(stream.ytId))
|
|
]
|
|
if not valid_streams:
|
|
raise ValidationError("Invalid stream URLs provided.")
|
|
|
|
unique_streams = {stream.url or stream.ytId: stream for stream in valid_streams}
|
|
return list(unique_streams.values())
|
|
|
|
|
|
def is_video_file(filename: str) -> bool:
|
|
return filename.lower().endswith(
|
|
(
|
|
".3g2",
|
|
".3gp",
|
|
".amv",
|
|
".asf",
|
|
".avi",
|
|
".drc",
|
|
".flv",
|
|
".gif",
|
|
".gifv",
|
|
".m2v",
|
|
".m4p",
|
|
".m4v",
|
|
".mkv",
|
|
".mng",
|
|
".mov",
|
|
".mp2",
|
|
".mp4",
|
|
".mpe",
|
|
".mpeg",
|
|
".mpg",
|
|
".mpv",
|
|
".mxf",
|
|
".nsv",
|
|
".ogg",
|
|
".ogv",
|
|
".qt",
|
|
".rm",
|
|
".rmvb",
|
|
".roq",
|
|
".svi",
|
|
".vob",
|
|
".webm",
|
|
".wmv",
|
|
".yuv",
|
|
)
|
|
)
|