mirror of
https://github.com/Viren070/MediaFusion.git
synced 2025-12-01 23:21:11 +01:00
109 KiB
109 KiB
In [ ]:
!pip install ipywidgets requests
!apt-get -qq install aria2
!wget -qO torrenttools.AppImage https://github.com/fbdtemme/torrenttools/releases/download/v0.6.2/torrenttools-0.6.2-linux-x86_64.AppImage
!chmod +x torrenttools.AppImage
!./torrenttools.AppImage --appimage-extract > /dev/nullIn [ ]:
import os
import subprocess
import base64
from typing import Dict, Optional, List
from urllib.parse import urlparse, unquote
import requests
import ipywidgets as widgets
from IPython.display import display, HTML, Javascript
class MediaTorrentCreator:
# Constants for media information
RESOLUTIONS = ['480p', '576p', '720p', '1080p', '1440p', '2160p', '4K']
SOURCES = ['BluRay', 'BluRay REMUX', 'BRRip', 'BDRip', 'WEB-DL', 'HDRip', 'DVDRip', 'HDTV', 'CAM', 'TeleSync', 'SCR']
AUDIO_CODECS = ['AAC', 'AC3', 'DTS', 'DTS-HD MA', 'TrueHD', 'Atmos', 'DD+']
VIDEO_CODECS = ['x264', 'x265', 'H.264', 'H.265', 'HEVC', 'AVC', 'MPEG-2', 'MPEG-4', 'VP9']
AUDIO_CHANNELS = ['2.0', '5.1', '7.1']
FILE_TYPES = ['mp4', 'mkv', 'avi', 'webm']
LANGUAGES = [
'English', 'Spanish', 'French', 'German', 'Italian', 'Russian', 'Japanese',
'Korean', 'Chinese', 'Hindi', 'Tamil', 'Telugu', 'Malayalam', 'Kannada',
'Portuguese', 'Turkish', 'Arabic', 'Thai', 'Vietnamese', 'Indonesian'
]
LANGUAGE_CODES = {
'English': 'ENG', 'Spanish': 'SPA', 'French': 'FRE', 'German': 'GER',
'Italian': 'ITA', 'Russian': 'RUS', 'Japanese': 'JPN', 'Korean': 'KOR',
'Chinese': 'CHI', 'Hindi': 'HIN', 'Tamil': 'TAM', 'Telugu': 'TEL',
'Malayalam': 'MAL', 'Kannada': 'KAN', 'Portuguese': 'POR', 'Turkish': 'TUR',
'Arabic': 'ARA', 'Thai': 'THA', 'Vietnamese': 'VIE', 'Indonesian': 'IND'
}
def __init__(self, base_path: str = "./media_storage"):
"""Initialize the creator with local storage paths"""
self.base_path = base_path
self.create_storage_directories()
def create_storage_directories(self):
"""Create necessary directories for local storage"""
directories = ['downloads', 'torrents']
for dir_name in directories:
path = os.path.join(self.base_path, dir_name)
os.makedirs(path, exist_ok=True)
print(f"Created directory: {path}")
def _fetch_best_trackers(self) -> List[str]:
"""Fetch best trackers from ngosang's trackerslist"""
try:
response = requests.get(
"https://raw.githubusercontent.com/ngosang/trackerslist/master/trackers_best.txt",
timeout=30
)
if response.status_code == 200:
return [tracker.strip() for tracker in response.text.split("\n") if tracker.strip()]
return []
except Exception as e:
print(f"Warning: Failed to fetch trackers: {str(e)}")
return []
def _get_filename_from_url(self, url: str, custom_filename: Optional[str] = None) -> str:
"""Get filename from URL or custom input"""
if custom_filename:
return custom_filename
try:
# Try to get filename from Content-Disposition header
response = requests.head(url, allow_redirects=True, timeout=10)
if 'Content-Disposition' in response.headers:
content_disp = response.headers['Content-Disposition']
if 'filename=' in content_disp:
return content_disp.split('filename=')[-1].strip('"\'')
# Try to get filename from URL path
parsed_url = urlparse(url)
url_path = unquote(parsed_url.path)
filename = os.path.basename(url_path)
if filename and '.' in filename:
return filename
# Use Content-Type for extension if available
if 'Content-Type' in response.headers:
content_type = response.headers['Content-Type'].split(';')[0]
ext_map = {
'video/mp4': '.mp4',
'video/x-matroska': '.mkv',
'video/webm': '.webm',
'application/x-bittorrent': '.torrent'
}
ext = ext_map.get(content_type, '')
if ext:
return f'download{ext}'
except Exception as e:
print(f"Warning: Error in filename detection: {str(e)}")
return 'download'
def _generate_torrent_name(self, media_info: Dict) -> str:
"""Generate structured torrent name from media information"""
parts = []
# Add title and year
title = media_info.get('title', '').replace(' ', '.')
year = media_info.get('year', '')
parts.append(f"{title}{f'.{year}' if year else ''}")
# Add season and episode for series
if media_info.get('media_type') == 'series':
season = media_info.get('season')
episode = media_info.get('episode')
if season and episode:
parts.append(f"S{int(season):02d}E{int(episode):02d}")
# Add quality info
resolution = media_info.get('resolution', '')
source = media_info.get('source', '')
if resolution and source:
parts.append(f"{resolution}.{source}")
elif resolution:
parts.append(resolution)
elif source:
parts.append(source)
# Add language info
languages = media_info.get('languages', [])
if languages:
# Convert full language names to codes and join them
language_codes = [self.LANGUAGE_CODES.get(lang, lang) for lang in languages]
parts.append('-'.join(language_codes))
# Add codec info
video_codec = media_info.get('video_codec', '')
if video_codec:
parts.append(video_codec)
# Add audio info
audio_codec = media_info.get('audio_codec', '')
audio_channels = media_info.get('audio_channels', '')
if audio_codec and audio_channels:
parts.append(f"{audio_codec}.{audio_channels}")
elif audio_codec:
parts.append(audio_codec)
return '.'.join(parts)
def create_torrent(
self,
source_url: str,
media_info: Dict,
custom_filename: Optional[str] = None,
cleanup: bool = False
) -> Dict:
"""Create a torrent with web seeding for content"""
try:
download_dir = os.path.join(self.base_path, "downloads")
trackers = self._fetch_best_trackers()
# Get appropriate filename for download
download_filename = self._get_filename_from_url(source_url, custom_filename)
safe_download_filename = "".join(c for c in download_filename if c.isalnum() or c in "._- ").strip()
local_file_path = os.path.join(download_dir, safe_download_filename)
# Generate torrent name
torrent_name = self._generate_torrent_name(media_info)
# Download with aria2c
print(f"Downloading using aria2c to: {safe_download_filename}")
aria2c_command = [
"aria2c",
"-x", "16",
"-s", "16",
"-k", "1M",
"-o", safe_download_filename,
"-d", download_dir,
source_url
]
process = subprocess.run(
aria2c_command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
if process.returncode != 0:
raise Exception(f"Failed to download file: {process.stderr}")
if not os.path.exists(local_file_path):
raise Exception(f"Downloaded file not found: {local_file_path}")
file_size = os.path.getsize(local_file_path) # Size in bytes
if file_size < 10 * 1024 * 1024: # 10 MB in bytes
raise Exception(f"Downloaded file size is less than 10MB (size: {file_size / (1024 * 1024):.2f} MB)")
# Create torrent file
output_path = os.path.join(self.base_path, "torrents", f"{torrent_name}.torrent")
create_command = [
"./squashfs-root/AppRun",
"create",
local_file_path,
"-v", "1",
"-n", torrent_name,
"-o", output_path,
"-w", source_url,
"--announce", *trackers,
"-l", "auto",
"-s", "WebSeedTorrentCreator",
"--created-by", "WebSeedTorrentCreator"
]
process = subprocess.run(
create_command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
if process.returncode != 0:
raise Exception(f"Failed to create torrent: {process.stderr}")
if cleanup:
try:
os.remove(local_file_path)
print(f"Cleaned up downloaded file: {local_file_path}")
except Exception as e:
print(f"Warning: Failed to clean up file: {str(e)}")
return {
"torrent_path": output_path,
"web_seed_url": source_url,
"local_file": local_file_path,
"filename": safe_download_filename,
"torrent_name": torrent_name,
"trackers": trackers,
"command_output": process.stdout
}
except Exception as e:
raise Exception(f"Error in create_torrent: {str(e)}")
def create_torrent_gui(self):
"""Create GUI with structured naming options"""
# Create all input widgets
url_input = widgets.Text(
value='',
placeholder='Enter URL',
description='URL:',
style={'description_width': 'initial'},
layout=widgets.Layout(width='80%', margin='10px 0px')
)
naming_mode = widgets.ToggleButtons(
options=['Direct Filename', 'Use Metadata'],
description='Naming Mode:',
style={'description_width': 'initial'},
layout=widgets.Layout(margin='10px 0px')
)
direct_filename = widgets.Text(
value='',
placeholder='Enter complete filename (e.g., Movie.Name.2024.1080p.BluRay.x264)',
description='Filename:',
style={'description_width': 'initial'},
layout=widgets.Layout(width='80%', margin='10px 0px')
)
file_type = widgets.Dropdown(
options=self.FILE_TYPES,
value='mkv',
description='File Type:',
style={'description_width': 'initial'},
layout=widgets.Layout(width='200px', margin='10px 0px')
)
media_type = widgets.Dropdown(
options=['movie', 'series'],
value='movie',
description='Type:',
style={'description_width': 'initial'},
layout=widgets.Layout(width='200px', margin='10px 0px')
)
title_input = widgets.Text(
value='',
placeholder='Enter title',
description='Title:',
style={'description_width': 'initial'},
layout=widgets.Layout(width='80%', margin='10px 0px')
)
year_input = widgets.Text(
value='',
placeholder='Optional: Year',
description='Year:',
style={'description_width': 'initial'},
layout=widgets.Layout(width='200px', margin='10px 0px')
)
season_input = widgets.IntText(
value=1,
description='Season:',
style={'description_width': 'initial'},
layout=widgets.Layout(width='150px', margin='10px 0px', display='none')
)
episode_input = widgets.IntText(
value=1,
description='Episode:',
style={'description_width': 'initial'},
layout=widgets.Layout(width='150px', margin='10px 0px', display='none')
)
resolution_input = widgets.Dropdown(
options=[''] + self.RESOLUTIONS,
value='',
description='Resolution:',
style={'description_width': 'initial'},
layout=widgets.Layout(width='200px', margin='10px 0px')
)
source_input = widgets.Dropdown(
options=[''] + self.SOURCES,
value='',
description='Source:',
style={'description_width': 'initial'},
layout=widgets.Layout(width='200px', margin='10px 0px')
)
video_codec_input = widgets.Dropdown(
options=[''] + self.VIDEO_CODECS,
value='',
description='Video Codec:',
style={'description_width': 'initial'},
layout=widgets.Layout(width='200px', margin='10px 0px')
)
audio_codec_input = widgets.Dropdown(
options=[''] + self.AUDIO_CODECS,
value='',
description='Audio Codec:',
style={'description_width': 'initial'},
layout=widgets.Layout(width='200px', margin='10px 0px')
)
audio_channels_input = widgets.Dropdown(
options=[''] + self.AUDIO_CHANNELS,
value='',
description='Audio Channels:',
style={'description_width': 'initial'},
layout=widgets.Layout(width='200px', margin='10px 0px')
)
languages_select = widgets.SelectMultiple(
options=self.LANGUAGES,
value=[],
description='Languages:',
style={'description_width': 'initial'},
layout=widgets.Layout(width='300px', height='150px', margin='10px 0px')
)
language_description = widgets.HTML(
value='<small>Hold Ctrl/Cmd to select multiple languages</small>',
layout=widgets.Layout(margin='0px 0px 10px 10px')
)
create_button = widgets.Button(
description='Create Torrent',
button_style='primary',
tooltip='Click to create torrent',
icon='check',
layout=widgets.Layout(width='200px', margin='20px 0px')
)
output_area = widgets.Output(
layout={'border': '1px solid black', 'margin': '10px 0px', 'padding': '10px'}
)
download_area = widgets.HTML(
value='',
layout=widgets.Layout(display='none', margin='10px 0px')
)
loading_indicator = widgets.HTML(
value='<div class="loader"></div>',
layout=widgets.Layout(display='none')
)
# Add loading spinner CSS
display(HTML("""
<style>
.loader {
border: 4px solid #f3f3f3;
border-top: 4px solid #3498db;
border-radius: 50%;
width: 30px;
height: 30px;
animation: spin 1s linear infinite;
margin: 10px;
}
@keyframes spin {
0% { transform: rotate(0deg); }
100% { transform: rotate(360deg); }
}
</style>
"""))
def on_naming_mode_change(change):
if change['new'] == 'Direct Filename':
metadata_section.layout.display = 'none'
direct_filename.layout.display = 'block'
else:
metadata_section.layout.display = 'block'
direct_filename.layout.display = 'none'
def on_media_type_change(change):
if change['new'] == 'series':
season_input.layout.display = 'block'
episode_input.layout.display = 'block'
else:
season_input.layout.display = 'none'
episode_input.layout.display = 'none'
def create_download_button(torrent_path: str, filename: str):
"""Create a download button for the torrent file"""
try:
with open(torrent_path, 'rb') as f:
torrent_data = f.read()
b64_data = base64.b64encode(torrent_data).decode('utf-8')
download_html = f"""
<a href="data:application/x-bittorrent;base64,{b64_data}"
download="{filename}"
class="jupyter-button jupyter-widget btn btn-primary"
style="text-decoration: none; padding: 8px 16px; margin: 10px 0;">
📥 Download Torrent File
</a>
"""
download_area.value = download_html
download_area.layout.display = 'block'
except Exception as e:
print(f"Error creating download button: {str(e)}")
def on_create_button_click(b):
# Disable button and show loading
create_button.disabled = True
loading_indicator.layout.display = 'block'
with output_area:
output_area.clear_output()
print("Starting torrent creation process...")
try:
# Determine torrent name based on mode
if naming_mode.value == 'Direct Filename':
media_info = {
'title': direct_filename.value
}
torrent_name = f"{direct_filename.value}.{file_type.value}"
else:
media_info = {
'media_type': media_type.value,
'title': title_input.value,
'year': year_input.value,
'resolution': resolution_input.value,
'source': source_input.value,
'video_codec': video_codec_input.value,
'audio_codec': audio_codec_input.value,
'audio_channels': audio_channels_input.value,
'languages': list(languages_select.value)
}
if media_type.value == 'series':
media_info.update({
'season': season_input.value,
'episode': episode_input.value
})
torrent_name = self._generate_torrent_name(media_info)
torrent_name = f"{torrent_name}.{file_type.value}"
result = self.create_torrent(
source_url=url_input.value,
media_info={'title': torrent_name}, # Pass the complete name
custom_filename=f"{torrent_name}"
)
print("\n✅ Torrent created successfully!")
print(f"📁 Torrent path: {result['torrent_path']}")
print(f"🏷️ Torrent name: {result['torrent_name']}")
print(f"📝 Download filename: {result['filename']}")
print(f"🔗 Web seed URL: {result['web_seed_url']}")
# Create download button
create_download_button(result['torrent_path'], os.path.basename(result['torrent_path']))
except Exception as e:
print(f"❌ Error creating torrent: {str(e)}")
print("\nPlease check:")
print("1. Is the URL valid and accessible?")
print("2. Have you filled in all required fields?")
print("3. Do you have write permissions in the output directory?")
finally:
# Re-enable button and hide loading
create_button.disabled = False
loading_indicator.layout.display = 'none'
# Connect event handlers
naming_mode.observe(on_naming_mode_change, names='value')
media_type.observe(on_media_type_change, names='value')
create_button.on_click(on_create_button_click)
# Create layout sections
quality_info = widgets.VBox([
widgets.HTML("<h4>Quality Information</h4>"),
widgets.HBox([resolution_input, source_input]),
widgets.HBox([video_codec_input, audio_codec_input, audio_channels_input]),
widgets.HTML("<h4>Language Information</h4>"),
languages_select,
language_description
])
# Metadata section contains all the structured input fields
metadata_section = widgets.VBox([
widgets.HTML("<h4>Media Information</h4>"),
media_type,
title_input,
year_input,
season_input,
episode_input,
quality_info
])
# Initial layout setup
metadata_section.layout.display = 'none' # Start with direct filename mode
# Main container
container = widgets.VBox([
widgets.HTML("<h3>WebSeed Torrent Creator</h3>"),
url_input,
naming_mode,
direct_filename,
file_type,
metadata_section,
create_button,
loading_indicator,
output_area,
download_area
], layout=widgets.Layout(
width='100%',
padding='20px',
border='1px solid #ddd',
border_radius='5px'
))
display(container)In [ ]:
# Create and display the GUI
creator = MediaTorrentCreator()
creator.create_torrent_gui()