Stremio Universal Add-On for Movies & Series
import json
import logging
from typing import Literal

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from fastapi import FastAPI, Request, Response, BackgroundTasks, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import RedirectResponse, FileResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates

from db import database, crud, schemas
from db.config import settings
from streaming_providers.exceptions import ProviderException
from streaming_providers.realdebrid.api import router as realdebrid_router
from streaming_providers.realdebrid.utils import get_direct_link_from_realdebrid
from streaming_providers.seedr.api import router as seedr_router
from streaming_providers.seedr.utils import get_direct_link_from_seedr
from utils import crypto, torrent
from utils.const import CATALOG_ID_DATA, CATALOG_NAME_DATA

logging.basicConfig(
    format="%(levelname)s::%(asctime)s - %(message)s",
    datefmt="%d-%b-%y %H:%M:%S",
    level=settings.logging_level,
)
app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
app.mount("/static", StaticFiles(directory="resources"), name="static")
TEMPLATES = Jinja2Templates(directory="resources")
headers = {
    "Access-Control-Allow-Origin": "*",
    "Access-Control-Allow-Headers": "*",
    "Cache-Control": "max-age=3600, stale-while-revalidate=3600, stale-if-error=604800, public",
}
no_cache_headers = {
    "Cache-Control": "no-store, no-cache, must-revalidate, max-age=0",
    "Pragma": "no-cache",
    "Expires": "0",
}


@app.on_event("startup")
async def init_db():
    await database.init()


@app.on_event("startup")
async def start_scheduler():
    scheduler = AsyncIOScheduler()
    # scheduler.add_job(tb_scrapper.run_schedule_scrape, CronTrigger(hour="*/3"))
    scheduler.start()
    app.state.scheduler = scheduler


@app.on_event("shutdown")
async def stop_scheduler():
    app.state.scheduler.shutdown(wait=False)


@app.get("/", tags=["home"])
async def get_home(request: Request):
    with open("resources/manifest.json") as file:
        manifest = json.load(file)
    return TEMPLATES.TemplateResponse(
        "html/home.html",
        {
            "request": request,
            "name": manifest.get("name"),
            "version": f"{manifest.get('version')}-{settings.git_rev[:7]}",
            "description": manifest.get("description"),
            "gives": [
                "Tamil Movies & Series",
                "Malayalam Movies & Series",
                "Telugu Movies & Series",
                "Hindi Movies & Series",
                "Kannada Movies & Series",
                "English Movies & Series",
                "Dubbed Movies & Series",
            ],
            "logo": "static/images/mediafusion_logo.png",
        },
    )


@app.get("/favicon.ico")
async def get_favicon():
    return FileResponse(
        "resources/images/mediafusion_logo.png", media_type="image/x-icon"
    )


@app.get("/configure", tags=["configure"])
@app.get("/{secret_str}/configure", tags=["configure"])
async def configure(
    response: Response,
    request: Request,
    user_data: schemas.UserData = Depends(crypto.decrypt_user_data),
):
    response.headers.update(headers)
    response.headers.update(no_cache_headers)
    return TEMPLATES.TemplateResponse(
        "html/configure.html",
        {
            "request": request,
            "user_data": user_data.model_dump(),
            "catalogs": zip(CATALOG_ID_DATA, CATALOG_NAME_DATA),
        },
    )


@app.get("/manifest.json", tags=["manifest"])
@app.get("/{secret_str}/manifest.json", tags=["manifest"])
async def get_manifest(
    response: Response, user_data: schemas.UserData = Depends(crypto.decrypt_user_data)
):
    response.headers.update(headers)
    response.headers.update(no_cache_headers)
    with open("resources/manifest.json") as file:
        manifest = json.load(file)

    filtered_catalogs = [
        cat for cat in manifest["catalogs"] if cat["id"] in user_data.selected_catalogs
    ]
    manifest["catalogs"] = filtered_catalogs
    return manifest


@app.get(
    "/{secret_str}/catalog/{catalog_type}/{catalog_id}.json",
    response_model=schemas.Metas,
    response_model_exclude_none=True,
    response_model_by_alias=False,
    tags=["catalog"],
)
@app.get(
    "/catalog/{catalog_type}/{catalog_id}.json",
    response_model=schemas.Metas,
    response_model_exclude_none=True,
    response_model_by_alias=False,
    tags=["catalog"],
)
@app.get(
    "/{secret_str}/catalog/{catalog_type}/{catalog_id}/skip={skip}.json",
    response_model=schemas.Metas,
    response_model_exclude_none=True,
    response_model_by_alias=False,
    tags=["catalog"],
)
@app.get(
    "/catalog/{catalog_type}/{catalog_id}/skip={skip}.json",
    response_model=schemas.Metas,
    response_model_exclude_none=True,
    response_model_by_alias=False,
    tags=["catalog"],
)
async def get_catalog(
    response: Response,
    catalog_type: Literal["movie", "series"],
    catalog_id: str,
    skip: int = 0,
):
    response.headers.update(headers)
    metas = schemas.Metas()
    metas.metas.extend(await crud.get_meta_list(catalog_type, catalog_id, skip))
    return metas


@app.get(
    "/{secret_str}/catalog/{catalog_type}/tamil_blasters/search={search_query}.json",
    response_model=schemas.Metas,
    tags=["search"],
)
@app.get(
    "/catalog/{catalog_type}/tamil_blasters/search={search_query}.json",
    response_model=schemas.Metas,
    tags=["search"],
)
async def search_movie(
    response: Response,
    catalog_type: Literal["movie", "series"],
    search_query: str,
):
    response.headers.update(headers)
    logging.debug("Searching for %s", search_query)

    return await crud.process_search_query(search_query, catalog_type)


@app.get("/{secret_str}/meta/{catalog_type}/{meta_id}.json", tags=["meta"])
@app.get("/meta/{catalog_type}/{meta_id}.json", tags=["meta"])
async def get_meta(
    catalog_type: Literal["movie", "series"], meta_id: str, response: Response
):
    response.headers.update(headers)
    if catalog_type == "movie":
        return await crud.get_movie_meta(meta_id)
    return await crud.get_series_meta(meta_id)


@app.get(
    "/{secret_str}/stream/{catalog_type}/{video_id}.json",
    response_model=schemas.Streams,
    response_model_exclude_none=True,
    tags=["stream"],
)
@app.get(
    "/stream/{catalog_type}/{video_id}.json",
    response_model=schemas.Streams,
    response_model_exclude_none=True,
    tags=["stream"],
)
@app.get(
    "/{secret_str}/stream/{catalog_type}/{video_id}:{season}:{episode}.json",
    response_model=schemas.Streams,
    response_model_exclude_none=True,
    tags=["stream"],
)
@app.get(
    "/stream/{catalog_type}/{video_id}:{season}:{episode}.json",
    response_model=schemas.Streams,
    response_model_exclude_none=True,
    tags=["stream"],
)
async def get_streams(
    catalog_type: Literal["movie", "series"],
    video_id: str,
    response: Response,
    secret_str: str = None,
    season: int = None,
    episode: int = None,
    user_data: schemas.UserData = Depends(crypto.decrypt_user_data),
):
    response.headers.update(headers)

    if catalog_type == "movie":
        fetched_streams = await crud.get_movie_streams(user_data, secret_str, video_id)
    else:
        fetched_streams = await crud.get_series_streams(
            user_data, secret_str, video_id, season, episode
        )

    return {"streams": fetched_streams}


@app.post("/encrypt-user-data", tags=["user_data"])
async def encrypt_user_data(user_data: schemas.UserData):
    encrypted_str = crypto.encrypt_user_data(user_data)
    return {"encrypted_str": encrypted_str}


@app.get("/{secret_str}/streaming_provider", tags=["streaming_provider"])
async def streaming_provider_endpoint(
    secret_str: str,
    info_hash: str,
    response: Response,
    season: int = None,
    episode: int = None,
):
    response.headers.update(headers)
    response.headers.update(no_cache_headers)

    user_data = crypto.decrypt_user_data(secret_str)
    if not user_data.streaming_provider:
        raise HTTPException(status_code=400, detail="No streaming provider set.")

    stream = await crud.get_stream_by_info_hash(info_hash)
    if not stream:
        raise HTTPException(status_code=400, detail="Stream not found.")

    magnet_link = torrent.convert_info_hash_to_magnet(info_hash, stream.announce_list)

    episode_data = stream.get_episode(season, episode)

    try:
        if user_data.streaming_provider.service == "seedr":
            video_url = await get_direct_link_from_seedr(
                info_hash, magnet_link, user_data, stream, episode_data, 3, 1
            )
        else:
            video_url = get_direct_link_from_realdebrid(
                info_hash, magnet_link, user_data, stream, episode_data, 3, 1
            )
    except ProviderException as error:
        logging.info("Exception occurred: %s", error.message)
        video_url = f"{settings.host_url}/static/exceptions/{error.video_file_name}"

    return RedirectResponse(url=video_url, headers=response.headers)


app.include_router(seedr_router, prefix="/seedr", tags=["seedr"])
app.include_router(realdebrid_router, prefix="/realdebrid", tags=["realdebrid"])

api/main.py
mhdzumair/MediaFusion
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    mongo_uri: str
    git_rev: str = "beta"
    secret_key: str
    host_url: str = "https://882b9915d0fe-mediafusion.baby-beamup.club"
    logging_level: str = "INFO"

    class Config:
        env_file = ".env"
        extra = "ignore"


settings = Settings()

db/config.py
mhdzumair/MediaFusion
import logging
from typing import Optional
from uuid import uuid4

from beanie import WriteRules
from imdb import Cinemagoer, IMDbDataAccessError

from db import schemas
from db.models import (
    MediaFusionMovieMetaData,
    MediaFusionSeriesMetaData,
    Streams,
    Season,
    Episode,
    MediaFusionMetaData,
)
from db.schemas import Stream, MetaIdProjection
from utils.parser import parse_stream_data, get_catalogs

ia = Cinemagoer()


async def get_meta_list(
    catalog_type: str, catalog: str, skip: int = 0, limit: int = 100
) -> list[schemas.Meta]:

    pipeline = [
        # Lookup to join with the Streams collection
        {
            "$lookup": {
                "from": "Streams",
                "localField": "streams.$id",
                "foreignField": "_id",
                "as": "joined_streams",
            }
        },
        # Unwind the joined streams to process each stream separately
        {"$unwind": "$joined_streams"},
        # Match the series based on catalog
        {"$match": {"joined_streams.catalog": catalog}},
        # Group by series ID and determine the latest date
        {
            "$group": {
                "_id": "$_id",
                "latest_date": {"$max": "$joined_streams.created_at"},
                "title": {"$first": "$title"},
                "poster": {"$first": "$poster"},
                "type": {"$first": "$type"},
            }
        },
        # Rename _id to id
        {
            "$project": {
                "id": "$_id",
                "latest_date": 1,
                "title": 1,
                "poster": 1,
                "type": 1,
            }
        },
        # Sort by the latest_date in descending order
        {"$sort": {"latest_date": -1}},
        # Pagination
        {"$skip": skip},
        {"$limit": limit},
    ]

    if catalog_type == "movie":
        meta_list = await MediaFusionMovieMetaData.aggregate(
            pipeline, projection_model=schemas.Meta
        ).to_list()
    else:
        meta_list = await MediaFusionSeriesMetaData.aggregate(
            pipeline, projection_model=schemas.Meta
        ).to_list()
    return meta_list


async def get_movie_data_by_id(
    movie_id: str, fetch_links: bool = False
) -> Optional[MediaFusionMovieMetaData]:
    movie_data = await MediaFusionMovieMetaData.get(movie_id, fetch_links=fetch_links)
    return movie_data


async def get_series_data_by_id(
    series_id: str, fetch_links: bool = False
) -> Optional[MediaFusionSeriesMetaData]:
    series_data = await MediaFusionSeriesMetaData.get(
        series_id, fetch_links=fetch_links
    )
    return series_data


async def get_movie_streams(user_data, secret_str: str, video_id: str) -> list[Stream]:
    movie_data = await get_movie_data_by_id(video_id, True)
    if not movie_data:
        return []

    return parse_stream_data(movie_data.streams, user_data, secret_str)


async def get_series_streams(
    user_data, secret_str: str, video_id: str, season: int, episode: int
) -> list[Stream]:
    series_data = await get_series_data_by_id(video_id, True)
    if not series_data:
        return []

    matched_episode_streams = [
        stream for stream in series_data.streams if stream.get_episode(season, episode)
    ]

    return parse_stream_data(
        matched_episode_streams, user_data, secret_str, season, episode
    )


async def get_movie_meta(meta_id: str):
    movie_data = await get_movie_data_by_id(meta_id)

    if not movie_data:
        return {}

    return {
        "meta": {
            "id": meta_id,
            "type": "movie",
            "name": movie_data.title,
            "poster": movie_data.poster,
            "background": movie_data.poster,
        }
    }


async def get_series_meta(meta_id: str):
    series_data = await get_series_data_by_id(meta_id, True)

    if not series_data:
        return {}

    metadata = {
        "meta": {
            "id": meta_id,
            "type": "series",
            "name": series_data.title,
            "poster": series_data.poster,
            "background": series_data.poster,
            "videos": [],
        }
    }

    # Loop through streams to populate the videos list
    for stream in series_data.streams:
        stream: Streams
        if stream.season:  # Ensure the stream has season data
            for episode in stream.season.episodes:
                metadata["meta"]["videos"].append(
                    {
                        "id": f"{meta_id}:{stream.season.season_number}:{episode.episode_number}",
                        "name": f"S{stream.season.season_number} EP{episode.episode_number}",
                        "season": stream.season.season_number,
                        "episode": episode.episode_number,
                        "released": stream.created_at,
                    }
                )

    return metadata


def search_imdb(title: str, year: int, retry: int = 5) -> dict:
    try:
        result = ia.search_movie(f"{title} {year}")
    except IMDbDataAccessError:
        return search_imdb(title, year, retry - 1) if retry > 0 else {}
    for movie in result:
        if movie.get("year") == year and movie.get("title").lower() in title.lower():
            return {
                "imdb_id": f"tt{movie.movieID}",
                "poster": movie.get("full-size cover url"),
            }
    return {}


async def save_movie_metadata(metadata: dict):
    # Try to get the existing movie
    existing_movie = await MediaFusionMovieMetaData.find_one(
        {"title": metadata["title"], "year": metadata.get("year")}
    )

    if not existing_movie:
        # If the movie doesn't exist in our DB, search for IMDb ID
        imdb_data = search_imdb(metadata["title"], metadata.get("year"))
        meta_id = imdb_data.get("imdb_id")

        if meta_id:
            # Check if the movie with the found IMDb ID already exists in our DB
            existing_movie = await MediaFusionMovieMetaData.get(meta_id)
        else:
            meta_id = f"mf{uuid4().fields[-1]}"
        # Update the poster from IMDb if available
        poster = imdb_data.get("poster") or metadata["poster"]
    else:
        poster = existing_movie.poster
        meta_id = existing_movie.id

    # Determine file index for the main movie file (largest file)
    largest_file = max(
        metadata["torrent_metadata"]["file_data"], key=lambda x: x["size"]
    )

    if "language" in metadata:
        languages = (
            [metadata["language"]]
            if isinstance(metadata["language"], str)
            else metadata["language"]
        )
    else:
        languages = [metadata["scrap_language"]]

    # Create the stream object
    new_stream = Streams(
        id=metadata["torrent_metadata"]["info_hash"],
        torrent_name=metadata["torrent_metadata"]["torrent_name"],
        announce_list=metadata["torrent_metadata"]["announce_list"],
        size=metadata["torrent_metadata"]["total_size"],
        filename=largest_file["filename"],
        file_index=largest_file["index"],
        languages=languages,
        resolution=metadata.get("resolution"),
        codec=metadata.get("codec"),
        quality=metadata.get("quality"),
        audio=metadata.get("audio"),
        encoder=metadata.get("encoder"),
        source=metadata["source"],
        catalog=get_catalogs(metadata["catalog"], languages),
        created_at=metadata["created_at"],
    )

    if existing_movie:
        # Check if the stream with the same info_hash already exists
        await existing_movie.fetch_all_links()
        matching_stream = next(
            (stream for stream in existing_movie.streams if stream.id == new_stream.id),
            None,
        )
        if not matching_stream:
            existing_movie.streams.append(new_stream)
        await existing_movie.save(link_rule=WriteRules.WRITE)
        logging.info("Updated movie %s", existing_movie.title)
    else:
        # If the movie doesn't exist, create a new one
        movie_data = MediaFusionMovieMetaData(
            id=meta_id,
            title=metadata["title"],
            year=metadata["year"],
            poster=poster,
            streams=[new_stream],
        )
        await movie_data.insert(link_rule=WriteRules.WRITE)
        logging.info("Added movie %s", movie_data.title)


async def save_series_metadata(metadata: dict):
    # Try to get the existing series
    series = await MediaFusionSeriesMetaData.find_one({"title": metadata["title"]})

    if not series:
        # If the series doesn't exist in our DB, search for IMDb ID
        imdb_data = search_imdb(metadata["title"], metadata["year"])
        meta_id = imdb_data.get("imdb_id")

        if meta_id:
            # Check if the series with the found IMDb ID already exists in our DB
            series = await MediaFusionSeriesMetaData.get(meta_id)

        if not series:
            meta_id = meta_id or f"mf{uuid4().fields[-1]}"
            poster = imdb_data.get("poster") or metadata["poster"]

            # Create an initial entry for the series
            series = MediaFusionSeriesMetaData(
                id=meta_id,
                title=metadata["title"],
                year=metadata["year"],
                poster=poster,
                streams=[],
            )
            await series.insert()
            logging.info("Added series %s", series.title)

    await series.fetch_all_links()
    existing_stream = next(
        (
            s
            for s in series.streams
            if s.id == metadata["torrent_metadata"]["info_hash"]
        ),
        None,
    )
    if existing_stream:
        # If the stream already exists, return
        logging.info("Stream already exists for series %s", series.title)
        return

    season_number = metadata["season"]

    # Extract episodes
    episodes = [
        Episode(
            episode_number=file["episode"],
            filename=file["filename"],
            size=file["size"],
            file_index=file["index"],
        )
        for file in metadata["torrent_metadata"]["file_data"]
        if file["episode"]
    ]

    # Determine languages
    if "language" in metadata:
        languages = (
            [metadata["language"]]
            if isinstance(metadata["language"], str)
            else metadata["language"]
        )
    else:
        languages = [metadata["scrap_language"]]

    # Create the stream
    stream = Streams(
        id=metadata["torrent_metadata"]["info_hash"],
        torrent_name=metadata["torrent_metadata"]["torrent_name"],
        announce_list=metadata["torrent_metadata"]["announce_list"],
        size=metadata["torrent_metadata"]["total_size"],
        languages=languages,
        resolution=metadata.get("resolution"),
        codec=metadata.get("codec"),
        quality=metadata.get("quality"),
        audio=metadata.get("audio"),
        encoder=metadata.get("encoder"),
        source=metadata["source"],
        catalog=get_catalogs(metadata["catalog"], languages),
        created_at=metadata["created_at"],
        season=Season(season_number=season_number, episodes=episodes),
    )

    # Add the stream to the series
    series.streams.append(stream)

    await series.save(link_rule=WriteRules.WRITE)
    logging.info("Updated series %s", series.title)


async def process_search_query(search_query: str, catalog_type: str) -> dict:
    # Define the query with the text search and catalog type
    query = {"$text": {"$search": search_query}, "type": catalog_type}

    # Perform the search
    search_results = (
        await MediaFusionMetaData.find(query).project(MetaIdProjection).to_list()
    )

    logging.debug(
        "Found %s results for %s in %s", len(search_results), search_query, catalog_type
    )

    metas = []

    for item in search_results:
        # Use the appropriate function to get the meta data
        if catalog_type == "movie":
            meta = await get_movie_meta(item.id)
        else:
            meta = await get_series_meta(item.id)

        if not meta:
            continue

        metas.append(meta["meta"])  # Extract the 'meta' key from the result

    return {"metas": metas}


async def get_stream_by_info_hash(info_hash: str) -> Streams | None:
    stream = await Streams.get(info_hash)
    return stream

db/crud.py
mhdzumair/MediaFusion
import motor.motor_asyncio
from beanie import init_beanie

from db.config import settings
from db.models import MediaFusionSeriesMetaData, MediaFusionMovieMetaData, Streams


async def init():
    # Create Motor client
    client = motor.motor_asyncio.AsyncIOMotorClient(settings.mongo_uri)

    # Init beanie with the Product document class
    await init_beanie(database=client.mediafusion, document_models=[MediaFusionMovieMetaData, MediaFusionSeriesMetaData, Streams])

db/database.py
mhdzumair/MediaFusion
from datetime import datetime
from typing import Optional

import pymongo
from beanie import Document, Link
from pydantic import BaseModel, Field
from pymongo import IndexModel, ASCENDING


class Episode(BaseModel):
    episode_number: int
    filename: str
    size: int
    file_index: int


class Season(BaseModel):
    season_number: int
    episodes: list[Episode]


class Streams(Document):
    id: str
    torrent_name: str
    size: int
    season: Optional[Season] = None
    filename: Optional[str] = None
    file_index: Optional[int] = None
    announce_list: list[str]
    languages: list[str]
    source: str
    catalog: list[str]
    created_at: datetime = Field(default_factory=datetime.now)
    resolution: Optional[str]
    codec: Optional[str]
    quality: Optional[str]
    audio: Optional[str]
    encoder: Optional[str]
    seeders: Optional[int] = None
    cached: Optional[bool] = None

    def get_episode(self, season_number: int, episode_number: int) -> Optional[Episode]:
        """
        Returns the Episode object for the given season and episode number.
        """
        if self.season and self.season.season_number == season_number:
            for episode in self.season.episodes:
                if episode.episode_number == episode_number:
                    return episode
        return None


class MediaFusionMetaData(Document):
    id: str
    title: str
    year: Optional[int]
    poster: str
    streams: list[Link[Streams]]

    class Settings:
        is_root = True
        indexes = [
            IndexModel([("title", ASCENDING), ("year", ASCENDING)], unique=True),
            IndexModel([("title", pymongo.TEXT)]),
        ]


class MediaFusionMovieMetaData(MediaFusionMetaData):
    type: str = "movie"


class MediaFusionSeriesMetaData(MediaFusionMetaData):
    type: str = "series"

db/models.py
mhdzumair/MediaFusion
from typing import Optional, Any, Literal

from pydantic import BaseModel, Field

from utils.const import CATALOG_ID_DATA


class Catalog(BaseModel):
    id: str
    name: str
    type: str


class Meta(BaseModel):
    id: str
    name: str = Field(alias="title")
    type: str = Field(default="movie")
    poster: str
    videos: list | None = None


class Metas(BaseModel):
    metas: list[Meta] = []


class Stream(BaseModel):
    name: str
    description: str
    infoHash: str | None = None
    fileIdx: int | None = None
    url: str | None = None
    behaviorHints: dict[str, Any] | None = None


class Streams(BaseModel):
    streams: Optional[list[Stream]] = []


class StreamingProvider(BaseModel):
    service: Literal["realdebrid", "seedr"]
    token: str

    class Config:
        extra = "ignore"


class UserData(BaseModel):
    streaming_provider: StreamingProvider | None = None
    selected_catalogs: list[str] = Field(default=CATALOG_ID_DATA)

    class Config:
        extra = "ignore"


class AuthorizeData(BaseModel):
    device_code: str


class MetaIdProjection(BaseModel):
    id: str

db/schemas.py
mhdzumair/MediaFusion
import logging

import cloudscraper
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

from db import crud
from utils.torrent import extract_torrent_metadata


def get_scrapper_session(proxy_url=None):
    session = requests.session()
    session.headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36"
    }
    adapter = HTTPAdapter(
        max_retries=Retry(total=10, read=10, connect=10, backoff_factor=0.5)
    )
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    if proxy_url:
        session.proxies = {
            "http": proxy_url,
            "https": proxy_url,
        }
    scraper = cloudscraper.create_scraper(
        browser={"browser": "chrome", "platform": "windows", "mobile": False},
        delay=10,
        sess=session,
    )
    return scraper


async def check_cloudflare_validation(page):
    if await page.title() == "Just a moment...":
        logging.info("Cloudflare validation required")
        await page.wait_for_selector("a#elUserSignIn", timeout=9999999)


async def get_page_content(page, url):
    await page.goto(url)
    await check_cloudflare_validation(page)
    return await page.content()


async def download_and_save_torrent(
    torrent_element,
    metadata: dict,
    media_type: str,
    page_link: str,
    scraper=None,
    page=None,
):
    torrent_link = torrent_element.get("href")
    logging.info(f"Downloading torrent: {torrent_link}")

    if scraper:
        response = scraper.get(torrent_link)
        torrent_metadata = extract_torrent_metadata(response.content)
    elif page:
        async with page.expect_download() as download_info:
            try:
                await page.goto(torrent_link)
            except:
                pass
        download = await download_info.value
        torrent_path = await download.path()
        with open(torrent_path, "rb") as torrent_file:
            torrent_metadata = extract_torrent_metadata(torrent_file.read())

    if not torrent_metadata:
        logging.error(f"Info hash not found for {torrent_link}")
        return False

    metadata.update({"torrent_metadata": torrent_metadata})

    # Saving the metadata
    if media_type == "series":
        if not metadata.get("season"):
            logging.error(f"Season not found for {page_link}")
            return False
        await crud.save_series_metadata(metadata)
    else:
        if metadata.get("season"):
            await crud.save_series_metadata(metadata)
        else:
            await crud.save_movie_metadata(metadata)

    return True

scrappers/helpers.py
mhdzumair/MediaFusion
#!/usr/bin/env python3

import argparse
import asyncio
import logging
import math
import random
import re

import PTN
from bs4 import BeautifulSoup
from dateutil.parser import parse as dateparser
from playwright.async_api import async_playwright
from playwright_stealth import stealth_async

from db import database
from scrappers.helpers import (
    get_page_content,
    get_scrapper_session,
    download_and_save_torrent,
)

HOMEPAGE = "https://www.tamilblasters.life"
TAMIL_BLASTER_LINKS = {
    "tamil": {
        "hdrip": "7-tamil-new-movies-hdrips-bdrips-dvdrips-hdtv",
        "tcrip": "8-tamil-new-movies-tcrip-dvdscr-hdcam-predvd",
        "dubbed": "9-tamil-dubbed-movies-bdrips-hdrips-dvdscr-hdcam-in-multi-audios",
        "series": "63-tamil-new-web-series-tv-shows",
    },
    "malayalam": {
        "tcrip": "75-malayalam-new-movies-tcrip-dvdscr-hdcam-predvd",
        "hdrip": "74-malayalam-new-movies-hdrips-bdrips-dvdrips-hdtv",
        "dubbed": "76-malayalam-dubbed-movies-bdrips-hdrips-dvdscr-hdcam",
        "series": "98-malayalam-new-web-series-tv-shows",
    },
    "telugu": {
        "tcrip": "79-telugu-new-movies-tcrip-dvdscr-hdcam-predvd",
        "hdrip": "78-telugu-new-movies-hdrips-bdrips-dvdrips-hdtv",
        "dubbed": "80-telugu-dubbed-movies-bdrips-hdrips-dvdscr-hdcam",
        "series": "96-telugu-new-web-series-tv-shows",
    },
    "hindi": {
        "tcrip": "87-hindi-new-movies-tcrip-dvdscr-hdcam-predvd",
        "hdrip": "86-hindi-new-movies-hdrips-bdrips-dvdrips-hdtv",
        "dubbed": "88-hindi-dubbed-movies-bdrips-hdrips-dvdscr-hdcam",
        "series": "forums/forum/89-hindi-new-web-series-tv-shows",
    },
    "kannada": {
        "tcrip": "83-kannada-new-movies-tcrip-dvdscr-hdcam-predvd",
        "hdrip": "82-kannada-new-movies-hdrips-bdrips-dvdrips-hdtv",
        "dubbed": "84-kannada-dubbed-movies-bdrips-hdrips-dvdscr-hdcam",
        "series": "103-kannada-new-web-series-tv-shows",
    },
    "english": {
        "tcrip": "52-english-movies-hdcam-dvdscr-predvd",
        "hdrip": "53-english-movies-hdrips-bdrips-dvdrips",
        "series": "92-english-web-series-tv-shows",
    },
}


async def get_search_results(page, keyword, page_number=1):
    search_link = f"{HOMEPAGE}/index.php?/search/&q={keyword}&type=forums_topic&page={page_number}&search_and_or=or&search_in=titles&sortby=relevancy"
    # Get page content and initialize BeautifulSoup
    page_content = await get_page_content(page, search_link)
    soup = BeautifulSoup(page_content, "html.parser")

    return soup


async def process_movie(
    movie,
    scraper=None,
    page=None,
    keyword=None,
    language=None,
    media_type=None,
    supported_forums=None,
):
    if keyword:
        movie_link = movie.find("a", {"data-linktype": "link"})
        forum_link = movie.find("a", href=re.compile(r"forums/forum/")).get("href")
        forum_id = re.search(r"forums/forum/([^/]+)/", forum_link)[1]
        if forum_id not in supported_forums:
            logging.error(f"Unsupported forum {forum_id}")
            return
        # Extracting language and media_type from supported_forums
        language = supported_forums[forum_id]["language"]
        media_type = supported_forums[forum_id]["media_type"]
    else:
        movie_link = movie.find("a")

    if not movie_link:
        logging.error(f"Movie link not found")
        return

    metadata = PTN.parse(re.sub(r"\s+", " ", movie_link.text))
    page_link = movie_link.get("href")

    try:
        if scraper:  # If using the scraper
            response = scraper.get(page_link)
            movie_page_content = response.content
        else:  # If using playwright
            movie_page_content = await get_page_content(page, page_link)

        movie_page = BeautifulSoup(movie_page_content, "html.parser")

        # Extracting other details
        poster_element = movie_page.select_one(
            "div[data-commenttype='forums'] img[data-src]"
        )
        poster = poster_element.get("data-src") if poster_element else None

        datetime_element = movie_page.select_one("time")
        created_at = (
            dateparser(datetime_element.get("datetime")) if datetime_element else None
        )

        # Updating metadata
        metadata.update(
            {
                "catalog": f"{language}_{media_type}",
                "poster": poster,
                "created_at": created_at,
                "scrap_language": language.title(),
                "source": "TamilBlasters",
            }
        )

        # Extracting torrent details
        torrent_elements = movie_page.select("a[data-fileext='torrent']")

        if not torrent_elements:
            logging.error(f"No torrents found for {page_link}")
            return

        for torrent_element in torrent_elements:
            await download_and_save_torrent(
                torrent_element,
                scraper=scraper,
                page=page,
                metadata=metadata,
                media_type=media_type,
                page_link=page_link,
            )

        return True
    except Exception as e:
        logging.error(
            f"Error processing movie {page_link}: {e}", exc_info=True, stack_info=True
        )
        return False


async def scrap_page(url, language, media_type, proxy_url=None):
    scraper = get_scrapper_session(proxy_url)
    response = scraper.get(url)
    if response.status_code == 403:
        logging.error(
            "Cloudflare validation required. Run with --scrap-with-playwright"
        )
        return

    response.raise_for_status()
    tamil_blasters = BeautifulSoup(response.content, "html.parser")
    movies = tamil_blasters.select("li[data-rowid]")

    for movie in movies:
        await process_movie(
            movie, scraper=scraper, language=language, media_type=media_type
        )


async def scrap_page_with_playwright(url, language, media_type, proxy_url=None):
    async with async_playwright() as p:
        # Launch a new browser session
        browser = await p.firefox.launch(
            headless=False,
            proxy={"server": proxy_url} if proxy_url else None,
        )
        page = await browser.new_page()
        await stealth_async(page)
        await asyncio.sleep(2)

        page_content = await get_page_content(page, url)
        tamil_blasters = BeautifulSoup(page_content, "html.parser")

        movies = tamil_blasters.select("li[data-rowid]")

        for movie in movies:
            await process_movie(
                movie, page=page, language=language, media_type=media_type
            )

        await browser.close()


async def scrap_search_keyword(keyword, proxy_url=None):
    supported_forums = {
        TAMIL_BLASTER_LINKS[language][media_type]: {
            "language": language,
            "media_type": media_type,
        }
        for language in TAMIL_BLASTER_LINKS
        for media_type in TAMIL_BLASTER_LINKS[language]
    }

    async with async_playwright() as p:
        # Launch a new browser session
        browser = await p.firefox.launch(
            headless=False,
            proxy={"server": proxy_url} if proxy_url else None,
        )
        page = await browser.new_page()
        await stealth_async(page)
        await asyncio.sleep(2)

        soup = await get_search_results(page, keyword)
        results_element = soup.find("div", {"data-role": "resultsArea"})

        results_count = int(re.search(r"\d+", results_element.find("p").text).group())
        logging.info(f"Found {results_count} results for {keyword}")

        movies = results_element.select("li[data-role='activityItem']")
        if results_count > 25:
            number_of_pages = math.ceil(results_count / 25)
            logging.info(f"Found {number_of_pages} pages for {keyword}")
            for page_number in range(2, number_of_pages + 1):
                soup = await get_search_results(page, keyword, page_number)
                movies.extend(soup.select("li[data-role='activityItem']"))
                await asyncio.sleep(random.randint(2, 5))

        for movie in movies:
            await process_movie(
                movie, page=page, keyword=keyword, supported_forums=supported_forums
            )

        await browser.close()


async def run_scraper(
    language: str = None,
    video_type: str = None,
    pages: int = None,
    start_page: int = None,
    search_keyword: str = None,
    scrap_with_playwright: bool = None,
    proxy_url: str = None,
):
    await database.init()
    if search_keyword:
        await scrap_search_keyword(search_keyword, proxy_url)
        return
    try:
        scrap_link = f"{HOMEPAGE}/index.php?/forums/forum/{TAMIL_BLASTER_LINKS[language][video_type]}"
    except KeyError:
        logging.error(f"Unsupported language or video type: {language}_{video_type}")
        return
    for page in range(start_page, pages + start_page):
        scrap_link = f"{scrap_link}/page/{page}/"
        logging.info(f"Scrap page: {page}")
        if scrap_with_playwright is True:
            await scrap_page_with_playwright(
                scrap_link, language, video_type, proxy_url
            )
        else:
            await scrap_page(scrap_link, language, video_type, proxy_url)

    logging.info(f"Scrap completed for : {language}_{video_type}")


async def run_schedule_scrape(
    pages: int = 1,
    start_page: int = 1,
    scrap_with_playwright: bool = None,
    proxy_url: str = None,
):
    for language in TAMIL_BLASTER_LINKS:
        for video_type in TAMIL_BLASTER_LINKS[language]:
            await run_scraper(
                language,
                video_type,
                pages=pages,
                start_page=start_page,
                scrap_with_playwright=scrap_with_playwright,
                proxy_url=proxy_url,
            )


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Scrap Movie metadata from TamilBlasters"
    )
    parser.add_argument(
        "--all", action="store_true", help="scrap all type of movies & series"
    )
    parser.add_argument(
        "-l",
        "--language",
        help="scrap movie language",
        default="tamil",
        choices=["tamil", "malayalam", "telugu", "hindi", "kannada", "english"],
    )
    parser.add_argument(
        "-t",
        "--video-type",
        help="scrap movie video type",
        default="hdrip",
        choices=["hdrip", "tcrip", "dubbed", "series"],
    )
    parser.add_argument(
        "-p", "--pages", type=int, default=1, help="number of scrap pages"
    )
    parser.add_argument(
        "-s", "--start-pages", type=int, default=1, help="page number to start scrap."
    )
    parser.add_argument(
        "-k",
        "--search-keyword",
        help="search keyword to scrap movies & series. ex: 'bigg boss'",
        default=None,
    )
    parser.add_argument(
        "--scrap-with-playwright", action="store_true", help="scrap with playwright"
    )
    parser.add_argument(
        "--proxy-url",
        help="proxy url to scrap. ex: socks5://127.0.0.1:1080",
        default=None,
    )
    args = parser.parse_args()

    logging.basicConfig(
        format="%(levelname)s::%(asctime)s - %(message)s",
        datefmt="%d-%b-%y %H:%M:%S",
        level=logging.INFO,
    )
    if args.all:
        asyncio.run(
            run_schedule_scrape(
                args.pages, args.start_pages, args.scrap_with_playwright, args.proxy_url
            )
        )
    else:
        asyncio.run(
            run_scraper(
                args.language,
                args.video_type,
                args.pages,
                args.start_pages,
                args.search_keyword,
                args.scrap_with_playwright,
                args.proxy_url,
            )
        )

scrappers/tamil_blasters_scrapper.py
mhdzumair/MediaFusion
class ProviderException(Exception):
    def __init__(self, message, video_file_name):
        self.message = message
        self.video_file_name = video_file_name
        super().__init__(self.message)

streaming_providers/exceptions.py
mhdzumair/MediaFusion
from fastapi import APIRouter
from fastapi.responses import JSONResponse

from db.schemas import AuthorizeData
from streaming_providers.realdebrid.client import RealDebrid

router = APIRouter()
headers = {"Cache-Control": "no-store, no-cache, must-revalidate, max-age=0", "Pragma": "no-cache", "Expires": "0"}


@router.get("/get-device-code")
async def get_device_code():
    rd_client = RealDebrid()
    return JSONResponse(content=rd_client.get_device_code(), headers=headers)


@router.post("/authorize")
async def authorize(data: AuthorizeData):
    rd_client = RealDebrid()
    response = rd_client.authorize(data.device_code)
    return JSONResponse(content=response, headers=headers)

streaming_providers/realdebrid/api.py
mhdzumair/MediaFusion
import traceback
from base64 import b64encode, b64decode
from typing import Any

import requests
from requests import RequestException, JSONDecodeError

from streaming_providers.exceptions import ProviderException


class RealDebrid:
    BASE_URL = "https://api.real-debrid.com/rest/1.0"
    OAUTH_URL = "https://api.real-debrid.com/oauth/v2"
    OPENSOURCE_CLIENT_ID = "X245A4XAIBGVM"

    def __init__(self, encoded_token=None):
        self.encoded_token = encoded_token
        self.headers = {}
        self.initialize_headers()

    def __del__(self):
        if self.encoded_token:
            self.disable_access_token()

    def _make_request(
        self,
        method: str,
        url: str,
        data=None,
        params=None,
        is_return_none=False,
        is_expected_to_fail=False,
    ) -> dict:
        if method == "GET":
            response = requests.get(url, params=params, headers=self.headers)
        elif method == "POST":
            response = requests.post(url, data=data, headers=self.headers)
        elif method == "DELETE":
            response = requests.delete(url, headers=self.headers)
        else:
            raise ValueError(f"Unsupported method: {method}")

        try:
            response.raise_for_status()
        except RequestException as error:
            if is_expected_to_fail:
                pass
            elif error.response.status_code == 401:
                raise ProviderException("Invalid token", "invalid_token.mp4")
            else:
                formatted_traceback = "".join(traceback.format_exception(error))
                raise ProviderException(
                    f"status code: {error.response.status_code}, data: {error.response.content}, trace log:\n {formatted_traceback}",
                    "api_error.mp4",
                )

        if is_return_none:
            return {}
        try:
            return response.json()
        except JSONDecodeError as error:
            formatted_traceback = "".join(traceback.format_exception(error))
            raise ProviderException(
                f"Failed to parse response. content: {response.text}, trace log:\n {formatted_traceback}",
                "api_error.mp4",
            )

    def initialize_headers(self):
        if self.encoded_token:
            token_data = self.decode_token_str(self.encoded_token)
            access_token_data = self.get_token(
                token_data["client_id"], token_data["client_secret"], token_data["code"]
            )
            self.headers = {
                "Authorization": f"Bearer {access_token_data['access_token']}"
            }

    @staticmethod
    def encode_token_data(client_id: str, client_secret: str, code: str):
        token = f"{client_id}:{client_secret}:{code}"
        return b64encode(str(token).encode()).decode()

    @staticmethod
    def decode_token_str(token: str) -> dict[str, str]:
        try:
            client_id, client_secret, code = b64decode(token).decode().split(":")
        except ValueError:
            raise ProviderException("Invalid token", "invalid_token.mp4")
        return {"client_id": client_id, "client_secret": client_secret, "code": code}

    def get_device_code(self):
        return self._make_request(
            "GET",
            f"{self.OAUTH_URL}/device/code",
            params={"client_id": self.OPENSOURCE_CLIENT_ID, "new_credentials": "yes"},
        )

    def get_token(self, client_id, client_secret, device_code):
        return self._make_request(
            "POST",
            f"{self.OAUTH_URL}/token",
            data={
                "client_id": client_id,
                "client_secret": client_secret,
                "code": device_code,
                "grant_type": "http://oauth.net/grant_type/device/1.0",
            },
        )

    def authorize(self, device_code):
        response_data = self._make_request(
            "GET",
            f"{self.OAUTH_URL}/device/credentials",
            params={"client_id": self.OPENSOURCE_CLIENT_ID, "code": device_code},
            is_expected_to_fail=True,
        )

        if "client_secret" not in response_data:
            return response_data

        token_data = self.get_token(
            response_data["client_id"], response_data["client_secret"], device_code
        )

        if "access_token" in token_data:
            token = self.encode_token_data(
                response_data["client_id"],
                response_data["client_secret"],
                token_data["refresh_token"],
            )
            return {"token": token}
        else:
            return token_data

    def add_magent_link(self, magnet_link):
        return self._make_request(
            "POST", f"{self.BASE_URL}/torrents/addMagnet", data={"magnet": magnet_link}
        )

    def get_user_torrent_list(self):
        return self._make_request("GET", f"{self.BASE_URL}/torrents")

    def get_torrent_info(self, torrent_id):
        return self._make_request("GET", f"{self.BASE_URL}/torrents/info/{torrent_id}")

    def get_torrent_instant_availability(self, torrent_hash):
        return self._make_request(
            "GET", f"{self.BASE_URL}/torrents/instantAvailability/{torrent_hash}"
        )

    def disable_access_token(self):
        return self._make_request(
            "GET", f"{self.BASE_URL}/disable_access_token", is_return_none=True
        )

    def start_torrent_download(self, torrent_id, file_ids="all"):
        return self._make_request(
            "POST",
            f"{self.BASE_URL}/torrents/selectFiles/{torrent_id}",
            data={"files": file_ids},
            is_return_none=True,
        )

    def get_available_torrent(self, info_hash) -> dict[str, Any] | None:
        available_torrents = self.get_user_torrent_list()
        for torrent in available_torrents:
            if torrent["hash"] == info_hash:
                return torrent

    def create_download_link(self, link):
        response = self._make_request(
            "POST",
            f"{self.BASE_URL}/unrestrict/link",
            data={"link": link},
            is_expected_to_fail=True,
        )
        if "download" in response:
            return response

        if "error_code" in response:
            if response["error_code"] == 23:
                raise ProviderException(
                    "Exceed remote traffic limit", "exceed_remote_traffic_limit.mp4"
                )
        raise ProviderException(
            f"Failed to create download link. response: {response}", "api_error.mp4"
        )

    def delete_torrent(self, torrent_id):
        return self._make_request(
            "DELETE",
            f"{self.BASE_URL}/torrents/delete/{torrent_id}",
            is_return_none=True,
        )

streaming_providers/realdebrid/client.py
mhdzumair/MediaFusion
import time
from typing import Any

from db.models import Streams, Episode
from db.schemas import UserData
from streaming_providers.exceptions import ProviderException
from streaming_providers.realdebrid.client import RealDebrid


def wait_for_status(
    rd_client,
    torrent_id: str,
    target_status: str,
    max_retries: int,
    retry_interval: int,
):
    """Wait for the torrent to reach a particular status."""
    retries = 0
    while retries < max_retries:
        torrent_info = rd_client.get_torrent_info(torrent_id)
        if torrent_info["status"] == target_status:
            return torrent_info
        time.sleep(retry_interval)
        retries += 1
    raise ProviderException(
        f"Torrent did not reach {target_status} status.", "torrent_not_downloaded.mp4"
    )


def get_direct_link_from_realdebrid(
    info_hash: str,
    magnet_link: str,
    user_data: UserData,
    stream: Streams,
    episode_data: Episode = None,
    max_retries=5,
    retry_interval=5,
) -> str:
    rd_client = RealDebrid(encoded_token=user_data.streaming_provider.token)
    filename = episode_data.filename if episode_data else stream.filename

    # Check if the torrent already exists
    torrent_info = rd_client.get_available_torrent(info_hash)
    if torrent_info:
        torrent_id = torrent_info.get("id")
        if torrent_info["status"] == "downloaded":
            torrent_info = rd_client.get_torrent_info(torrent_id)
            file_index = select_file_index_from_torrent(torrent_info, filename)
            response = rd_client.create_download_link(torrent_info["links"][file_index])
            return response.get("download")
        elif torrent_info["status"] == "magnet_error":
            rd_client.delete_torrent(torrent_id)
            raise ProviderException(
                "Not enough seeders available for parse magnet link",
                "torrent_not_downloaded.mp4",
            )
    else:
        # If torrent doesn't exist, add it
        response_data = rd_client.add_magent_link(magnet_link)
        if "id" not in response_data:
            raise ProviderException(
                "Failed to add magnet link to Real-Debrid", "transfer_error.mp4"
            )
        torrent_id = response_data["id"]

    # Wait for file selection and then start torrent download
    torrent_info = wait_for_status(
        rd_client, torrent_id, "waiting_files_selection", max_retries, retry_interval
    )
    if torrent_info["status"] == "magnet_error":
        rd_client.delete_torrent(torrent_id)
        raise ProviderException(
            "Not enough seeders available for parse magnet link",
            "torrent_not_downloaded.mp4",
        )
    rd_client.start_torrent_download(torrent_id)

    # Wait for download completion and get the direct link
    torrent_info = wait_for_status(
        rd_client, torrent_id, "downloaded", max_retries, retry_interval
    )
    file_index = select_file_index_from_torrent(torrent_info, filename)
    response = rd_client.create_download_link(torrent_info["links"][file_index])

    return response.get("download")


def order_streams_by_instant_availability_and_date(
    streams: list[Streams], user_data: UserData
) -> list[Streams]:
    """Orders the streams by instant availability."""
    rd_client = RealDebrid(encoded_token=user_data.streaming_provider.token)
    for stream in streams:
        try:
            stream.cached = bool(
                rd_client.get_torrent_instant_availability(stream.id)[stream.id]
            )
        except ProviderException:
            return streams

    return sorted(
        streams,
        key=lambda x: (
            x.cached,
            x.created_at,
        ),
        reverse=True,
    )


def select_file_index_from_torrent(torrent_info: dict[str, Any], filename: str) -> int:
    """Select the file index from the torrent info."""
    selected_files = [file for file in torrent_info["files"] if file["selected"] == 1]
    for index, file in enumerate(selected_files):
        if file["path"] == "/" + filename:
            return index
    raise ProviderException(
        "No matching file available for this torrent", "api_error.mp4"
    )

streaming_providers/realdebrid/utils.py
mhdzumair/MediaFusion
from fastapi import APIRouter
from fastapi.responses import JSONResponse
from seedrcc import Login

from db.schemas import AuthorizeData

router = APIRouter()
headers = {"Cache-Control": "no-store, no-cache, must-revalidate, max-age=0", "Pragma": "no-cache", "Expires": "0"}


@router.get("/get-device-code")
async def get_device_code():
    seedr = Login()
    device_code = seedr.getDeviceCode()
    return JSONResponse(content=device_code, headers=headers)


@router.post("/authorize")
async def authorize(data: AuthorizeData):
    seedr = Login()
    response = seedr.authorize(data.device_code)

    if "access_token" in response:
        return JSONResponse(content={"token": seedr.token}, headers=headers)
    else:
        return JSONResponse(content=response, headers=headers)

streaming_providers/seedr/api.py
mhdzumair/MediaFusion
import logging
import time
from datetime import datetime

from seedrcc import Seedr

from db.models import Streams, Episode
from db.schemas import UserData
from streaming_providers.exceptions import ProviderException
from utils.parser import clean_name


def check_torrent_status(seedr, info_hash: str):
    """Checks if a torrent with a given info_hash is currently downloading."""
    folder_content = seedr.listContents()
    torrents = folder_content.get("torrents", [])
    return next((t for t in torrents if t["hash"] == info_hash), None)


def check_folder_status(seedr, folder_name: str):
    """Checks if a torrent with a given folder_name has completed downloading."""
    folder_content = seedr.listContents()
    folders = folder_content.get("folders", [])
    return next((f for f in folders if f["name"] == folder_name), None)


def add_magnet_and_get_torrent(seedr, magnet_link: str, info_hash: str):
    """Adds a magnet link to Seedr and returns the corresponding torrent."""
    transfer = seedr.addTorrent(magnet_link)

    # Handle potential errors from Seedr response
    if "error" in transfer:
        if transfer["error"] == "invalid_token":
            raise ProviderException("Invalid Seedr token", "invalid_token.mp4")
        raise ProviderException(
            "Error transferring magnet link to Seedr", "transfer_error.mp4"
        )

    # Return the appropriate torrent name based on the transfer result
    if transfer["result"] is True and "title" in transfer:
        return transfer["title"]
    elif transfer["result"] is True:
        torrent = check_torrent_status(seedr, info_hash)
        if torrent:
            return torrent["name"]
    elif transfer["result"] in (
        "not_enough_space_added_to_wishlist",
        "not_enough_space_wishlist_full",
    ):
        raise ProviderException(
            "Not enough space in Seedr account to add this torrent",
            "not_enough_space.mp4",
        )

    raise ProviderException(
        "Error transferring magnet link to Seedr", "transfer_error.mp4"
    )


def wait_for_torrent_to_complete(
    seedr, info_hash: str, max_retries: int, retry_interval: int
):
    """Waits for a torrent with the given info_hash to complete downloading."""
    retries = 0
    while retries < max_retries:
        torrent = check_torrent_status(seedr, info_hash)
        if torrent is None:
            return  # Torrent was already downloaded
        if torrent and torrent.get("progress") == "100":
            return
        time.sleep(retry_interval)
        retries += 1

    raise ProviderException("Torrent not downloaded yet.", "torrent_not_downloaded.mp4")


def get_file_details_from_folder(seedr, folder_id: int, filename: str):
    """Gets the details of the file in a given folder."""
    folder_content = seedr.listContents(folder_id)
    return [f for f in folder_content["files"] if f["name"] == filename][0]


async def get_direct_link_from_seedr(
    info_hash: str,
    magnet_link: str,
    user_data: UserData,
    stream: Streams,
    episode_data: Episode = None,
    max_retries=5,
    retry_interval=5,
) -> str:
    """Gets a direct download link from Seedr using a magnet link and token."""
    seedr = Seedr(token=user_data.streaming_provider.token)

    # Check for existing torrent or folder
    torrent = check_torrent_status(seedr, info_hash)
    folder = check_folder_status(seedr, clean_name(stream.torrent_name))

    # Handle the torrent based on its status or if it's already in a folder
    if folder:
        folder_id = folder["id"]
    else:
        if torrent:
            folder_title = torrent["name"]
        else:
            free_up_space(seedr, stream.size)
            folder_title = add_magnet_and_get_torrent(seedr, magnet_link, info_hash)
        if clean_name(stream.torrent_name) != folder_title:
            logging.warning(
                f"Torrent name mismatch: '{clean_name(stream.torrent_name)}' != '{folder_title}'."
            )
        folder = check_folder_status(seedr, folder_title)
        if not folder:
            wait_for_torrent_to_complete(seedr, info_hash, max_retries, retry_interval)
            folder = check_folder_status(seedr, folder_title)
        folder_id = folder["id"]

    selected_file = get_file_details_from_folder(
        seedr,
        folder_id,
        clean_name(episode_data.filename if episode_data else stream.filename, ""),
    )
    video_link = seedr.fetchFile(selected_file["folder_file_id"])["url"]

    return video_link


def free_up_space(seedr, required_space):
    """Frees up space in the Seedr account by deleting folders until the required space is available."""
    contents = seedr.listContents()
    available_space = contents["space_max"] - contents["space_used"]

    if available_space >= required_space:
        return  # There's enough space, no need to delete anything

    folders = sorted(
        contents["folders"],
        key=lambda x: (
            -x["size"],
            datetime.strptime(x["last_update"], "%Y-%m-%d %H:%M:%S"),
        ),
    )

    for folder in folders:
        if available_space >= required_space:
            break
        seedr.deleteFolder(folder["id"])
        available_space += folder["size"]

streaming_providers/seedr/utils.py
mhdzumair/MediaFusion
CATALOG_ID_DATA = [
    "mediafusion_search_movies",
    "mediafusion_search_series",
    "tamil_hdrip",
    "tamil_tcrip",
    "tamil_dubbed",
    "tamil_series",
    "malayalam_tcrip",
    "malayalam_hdrip",
    "malayalam_dubbed",
    "malayalam_series",
    "telugu_tcrip",
    "telugu_hdrip",
    "telugu_dubbed",
    "telugu_series",
    "hindi_tcrip",
    "hindi_hdrip",
    "hindi_dubbed",
    "hindi_series",
    "kannada_tcrip",
    "kannada_hdrip",
    "kannada_dubbed",
    "kannada_series",
    "english_hdrip",
    "english_tcrip",
    "english_series",
]

CATALOG_NAME_DATA = [
    "MediaFusion Search Movies",
    "MediaFusion Search Series",
    "Tamil HD Movies",
    "Tamil TCRip/DVDScr/HDCAM Movies",
    "Tamil Dubbed Movies",
    "Tamil Series",
    "Malayalam TCRip/DVDScr/HDCAM Movies",
    "Malayalam HD Movies",
    "Malayalam Dubbed Movies",
    "Malayalam Series",
    "Telugu TCRip/DVDScr/HDCAM Movies",
    "Telugu HD Movies",
    "Telugu Dubbed Movies",
    "Telugu Series",
    "Hindi TCRip/DVDScr/HDCAM Movies",
    "Hindi HD Movies",
    "Hindi Dubbed Movies",
    "Hindi Series",
    "Kannada TCRip/DVDScr/HDCAM Movies",
    "Kannada HD Movies",
    "Kannada Dubbed Movies",
    "Kannada Series",
    "English HD Movies",
    "English TCRip/DVDScr/HDCAM Movies",
    "English Series",
]

utils/const.py
mhdzumair/MediaFusion
from Crypto.Cipher import AES
from Crypto.Random import get_random_bytes
from base64 import urlsafe_b64encode, urlsafe_b64decode

from db.config import settings
from db.schemas import UserData


def encrypt_user_data(user_data: UserData) -> str:
    data = user_data.model_dump_json(
        exclude_none=True, exclude_defaults=True, exclude_unset=True, round_trip=True
    ).encode("utf-8")
    iv = get_random_bytes(16)
    cipher = AES.new(settings.secret_key.encode("utf-8"), AES.MODE_CBC, iv)
    encrypted_data = cipher.encrypt(data + b"\0" * (16 - len(data) % 16))
    encrypted_str = urlsafe_b64encode(iv + encrypted_data).decode("utf-8")
    return encrypted_str


def decrypt_user_data(secret_str: str | None = None) -> UserData:
    if not secret_str:
        return UserData()
    try:
        encrypted_data = urlsafe_b64decode(secret_str)
        iv = encrypted_data[:16]
        cipher = AES.new(settings.secret_key.encode("utf-8"), AES.MODE_CBC, iv)
        decrypted_data = cipher.decrypt(encrypted_data[16:])
        decrypted_data = decrypted_data.rstrip(b"\0")
        user_data = UserData.model_validate_json(decrypted_data.decode("utf-8"))
    except Exception:
        user_data = UserData()
    return user_data

utils/crypto.py
mhdzumair/MediaFusion
import os


def create_text_video(output_path, text, duration=30, resolution=(1280, 720), fontsize=45, bgcolor="black"):
    # Create a solid color background using FFmpeg
    background_command = (
        f"ffmpeg -f lavfi -i color=c={bgcolor}:s={resolution[0]}x{resolution[1]} -t {duration} -y background.mp4"
    )
    os.system(background_command)

    # Overlay the text with shadow and fade-in and fade-out effects
    text_command = (
        f'ffmpeg -i background.mp4 -vf "drawbox=y=ih/4-10:color=black@0.4:width=iw:height=128:t=2,'
        f"drawtext=text='{text}':fontcolor=white:fontsize={fontsize}:"
        f"x=(w-text_w)/2:y=(h-text_h)/2:shadowx=2:shadowy=2:shadowcolor=black@0.5,"
        f'fade=t=in:st=0:d=1,fade=t=out:st={duration-1}:d=1" -y {output_path}'
    )
    os.system(text_command)

    # Remove the temporary solid color background video
    os.remove("background.mp4")


if __name__ == "__main__":
    create_text_video(
        "../resources/exceptions/transfer_error.mp4",
        "Error transferring magnet link.\nProbably be No seeders available for this torrent.",
    )

utils/exception_video.py
mhdzumair/MediaFusion
import math
import re

from db.config import settings
from db.models import Streams
from db.schemas import Stream, UserData
from streaming_providers.realdebrid.utils import (
    order_streams_by_instant_availability_and_date,
)


def parse_stream_data(
    streams: list[Streams],
    user_data: UserData,
    secret_str: str,
    season: int = None,
    episode: int = None,
) -> list[Stream]:
    stream_list = []

    # filter out streams that are not available in the user's selected catalog
    streams = [
        stream
        for stream in streams
        if any(catalog in stream.catalog for catalog in user_data.selected_catalogs)
    ]

    # sort streams by instant availability and date if realdebrid is selected
    if (
        user_data.streaming_provider
        and user_data.streaming_provider.service == "realdebrid"
    ):
        streams = order_streams_by_instant_availability_and_date(streams, user_data)
    else:
        # Sort the streams by created_at time
        streams = sorted(streams, key=lambda x: x.created_at, reverse=True)

    for stream_data in streams:
        quality_detail = " - ".join(
            filter(
                None,
                [
                    stream_data.quality,
                    stream_data.resolution,
                    stream_data.codec,
                    stream_data.audio,
                ],
            )
        )

        episode_data = stream_data.get_episode(season, episode)

        if user_data.streaming_provider:
            streaming_provider = user_data.streaming_provider.service.title()
            if stream_data.cached:
                streaming_provider += " (Cached)"
        else:
            streaming_provider = "Torrent"

        description_parts = [
            quality_detail,
            convert_bytes_to_readable(
                episode_data.size if episode_data else stream_data.size
            ),
            " + ".join(stream_data.languages),
            stream_data.source,
            streaming_provider,
        ]
        description = ", ".join(filter(lambda x: bool(x), description_parts))

        stream_details = {
            "name": "MediaFusion",
            "description": description,
            "infoHash": stream_data.id,
            "fileIdx": episode_data.file_index
            if episode_data
            else stream_data.file_index,
            "behaviorHints": {"bingeGroup": f"MediaFusion-{quality_detail}"},
        }

        if user_data.streaming_provider:
            base_proxy_url = f"{settings.host_url}/{secret_str}/streaming_provider?info_hash={stream_data.id}"
            if episode_data:
                base_proxy_url += f"&season={season}&episode={episode}"
            stream_details["url"] = base_proxy_url
            stream_details.pop("infoHash")
            stream_details.pop("fileIdx")
            stream_details["behaviorHints"]["notWebReady"] = True

        stream_list.append(Stream(**stream_details))

    return stream_list


def clean_name(name: str, replace: str = " ") -> str:
    # Only allow alphanumeric characters, spaces, and `.,;:_~-[]()`
    cleaned_name = re.sub(r"[^a-zA-Z0-9 .,;:_~\-()\[\]]", replace, name)
    return cleaned_name


def convert_bytes_to_readable(size_bytes: int) -> str:
    """
    Convert a size in bytes into a more human-readable format.
    """
    if size_bytes == 0:
        return "0B"
    size_name = ("B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB")
    i = int(math.floor(math.log(size_bytes, 1024)))
    p = math.pow(1024, i)
    s = round(size_bytes / p, 2)
    return f"{s} {size_name[i]}"


def get_catalogs(catalog: str, languages: list[str]) -> list[str]:
    base_catalogs = ["hdrip", "tcrip", "dubbed", "series"]
    base_catalog = catalog.split("_")[1]

    if base_catalog not in base_catalogs:
        return [catalog]

    # Generate the catalog for each supported language
    return [f"{lang.lower()}_{base_catalog}" for lang in languages]

utils/parser.py
mhdzumair/MediaFusion
import hashlib
import logging
from urllib.parse import quote

import PTN
import bencodepy

from utils.parser import clean_name

TRACKERS = [
    "udp://tracker.openbittorrent.com:80/announce",
    "udp://tracker.opentrackr.org:1337/announce",
    "udp://tracker.pomf.se:80/announce",
    "udp://exodus.desync.com:6969/announce",
    "udp://tracker.internetwarriors.net:1337/announce",
    "udp://tracker.tiny-vps.com:6969/announce",
    "udp://tracker.coppersurfer.tk:6969/announce",
    "udp://tracker.leechers-paradise.org:6969/announce",
    "udp://9.rarbg.to:2710/announce",
    "udp://9.rarbg.me:2710/announce",
    "http://tracker3.itzmx.com:8080/announce",
    "udp://ipv4.tracker.harry.lu:80/announce",
]


def extract_torrent_metadata(content) -> dict:
    try:
        torrent_data = bencodepy.decode(content)

        info = torrent_data[b"info"]
        info_encoded = bencodepy.encode(info)

        m = hashlib.sha1()
        m.update(info_encoded)
        info_hash = m.hexdigest()

        # Extract file size, file list, and announce list
        if b"files" in info:
            total_size = sum(file[b"length"] for file in info[b"files"])
            file_data = []
            for idx, file in enumerate(info[b"files"]):
                filename = file[b"path"][0].decode()
                parsed_data = PTN.parse(filename)
                file_data.append(
                    {
                        "filename": filename,
                        "size": file[b"length"],
                        "index": idx,
                        "season": parsed_data.get("season"),
                        "episode": parsed_data.get("episode"),
                    }
                )
        else:
            total_size = info[b"length"]
            filename = info[b"name"].decode()
            parsed_data = PTN.parse(filename)
            file_data = [
                {
                    "filename": filename,
                    "size": total_size,
                    "index": 0,
                    "season": parsed_data.get("season"),
                    "episode": parsed_data.get("episode"),
                }
            ]

        announce_list = [
            tracker[0].decode() for tracker in torrent_data.get(b"announce-list", [])
        ]
        torrent_name = info.get(b"name", b"").decode() or file_data[0]["filename"]

        return {
            "info_hash": info_hash,
            "announce_list": announce_list,
            "total_size": total_size,
            "file_data": file_data,
            "torrent_name": torrent_name,
        }

    except Exception as e:
        logging.error(f"Error occurred: {e}")
        return {}


def convert_info_hash_to_magnet(info_hash: str, trackers: list[str]) -> str:
    magnet_link = f"magnet:?xt=urn:btih:{info_hash}"
    for tracker in trackers or TRACKERS:
        encoded_tracker = quote(tracker, safe="")
        magnet_link += f"&tr={encoded_tracker}"
    return magnet_link

utils/torrent.py
mhdzumair/MediaFusion