Finish add ddl ...

This commit is contained in:
Lovi 2024-06-14 10:15:25 +02:00
parent f3090c330c
commit 5337a7561a
19 changed files with 599 additions and 133 deletions

View File

@ -8,7 +8,6 @@ from Src.Util.console import console, msg
from .site import ( from .site import (
title_search, title_search,
get_select_title, get_select_title,
manager_clear
) )
from .film import download_film from .film import download_film
@ -20,10 +19,10 @@ def search():
""" """
# Make request to site to get content that corrsisponde to that string # Make request to site to get content that corrsisponde to that string
film_search = msg.ask("\n[purple]Insert word to search in all site").strip() string_to_search = msg.ask("\n[purple]Insert word to search in all site").strip()
len_database = title_search(film_search) len_database = title_search(string_to_search)
if len_database != 0: if len_database > 0:
# Select title from list # Select title from list
select_title = get_select_title() select_title = get_select_title()
@ -34,3 +33,5 @@ def search():
url=select_title.url url=select_title.url
) )
else:
console.print(f"\n[red]Nothing matching was found for[white]: [purple]{string_to_search}")

View File

@ -122,18 +122,3 @@ def get_select_title(type_filter: list = None) -> MediaItem:
else: else:
console.print("\n[red]Wrong index") console.print("\n[red]Wrong index")
sys.exit(0) sys.exit(0)
def manager_clear():
"""
Clears the data lists managed by media_search_manager and table_show_manager.
This function clears the data lists managed by global variables media_search_manager
and table_show_manager. It removes all the items from these lists, effectively
resetting them to empty lists.
"""
global media_search_manager, table_show_manager
# Clear list of data
media_search_manager.clear()
table_show_manager.clear()

View File

@ -3,6 +3,7 @@
# Internal utilities # Internal utilities
from Src.Util.console import console, msg from Src.Util.console import console, msg
# Logic class # Logic class
from .site import title_search, get_select_title from .site import title_search, get_select_title
from .anime import donwload_film, donwload_series from .anime import donwload_film, donwload_series
@ -14,7 +15,7 @@ def search():
string_to_search = msg.ask("\n[purple]Insert word to search in all site").strip() string_to_search = msg.ask("\n[purple]Insert word to search in all site").strip()
len_database = title_search(string_to_search) len_database = title_search(string_to_search)
if len_database != 0: if len_database > 0:
# Select title from list # Select title from list
select_title = get_select_title() select_title = get_select_title()
@ -31,6 +32,5 @@ def search():
title_name=select_title.slug title_name=select_title.slug
) )
# If no media find
else: else:
console.print("[red]Cant find a single element") console.print(f"\n[red]Nothing matching was found for[white]: [purple]{string_to_search}")

View File

@ -239,19 +239,3 @@ def get_select_title(type_filter: list = None) -> MediaItem:
else: else:
console.print("\n[red]Wrong index") console.print("\n[red]Wrong index")
sys.exit(0) sys.exit(0)
def manager_clear():
"""
Clears the data lists managed by media_search_manager and table_show_manager.
This function clears the data lists managed by global variables media_search_manager
and table_show_manager. It removes all the items from these lists, effectively
resetting them to empty lists.
"""
global media_search_manager, table_show_manager
# Clear list of data
media_search_manager.clear()
table_show_manager.clear()

View File

@ -0,0 +1,85 @@
# 13.06.24
import sys
import logging
from typing import List, Dict
# External libraries
import httpx
from bs4 import BeautifulSoup
# Internal utilities
from Src.Util.headers import get_headers
from Src.Util._jsonConfig import config_manager
# Logic class
from .SearchType import MediaItem
class GetSerieInfo:
def __init__(self, dict_serie: MediaItem) -> None:
"""
Initializes the GetSerieInfo object with default values.
Args:
dict_serie (MediaItem): Dictionary containing series information (optional).
"""
self.headers = {'user-agent': get_headers()}
self.cookies = config_manager.get_dict('REQUESTS', 'index')
self.url = dict_serie.url
self.tv_name = None
self.list_episodes = None
def get_episode_number(self) -> List[Dict[str, str]]:
"""
Retrieves the number of episodes for a specific season.
Args:
n_season (int): The season number.
Returns:
List[Dict[str, str]]: List of dictionaries containing episode information.
"""
# Make an HTTP request to the series URL
try:
response = httpx.get(self.url + "?area=online", cookies=self.cookies, headers=self.headers)
response.raise_for_status()
except Exception as e:
logging.error(f"Insert: {'ips4_IPSSessionFront': 'your_code', 'ips4_member_id': 'your_code'} in config file \ REQUESTS \ index, instead of user-agent. Use browser debug and cookie request with a valid account. Error: {e}")
sys.exit(0)
# Parse HTML content of the page
soup = BeautifulSoup(response.text, "html.parser")
# Get tv name
self.tv_name = soup.find("span", class_= "ipsType_break").get_text(strip=True)
# Find the container of episodes for the specified season
table_content = soup.find('div', class_='ipsMargin_bottom:half')
list_dict_episode = []
for episode_div in table_content.find_all('a', href=True):
# Get text of episode
part_name = episode_div.get_text(strip=True)
if part_name:
link = episode_div['href']
obj_episode = {
'name': part_name,
'url': link
}
list_dict_episode.append(obj_episode)
self.list_episodes = list_dict_episode
return list_dict_episode

View File

@ -0,0 +1,60 @@
# 13.06.24
from typing import List
class MediaItem:
def __init__(self, data: dict):
self.name: str = data.get('name')
self.type: str = data.get('type')
self.url: int = data.get('url')
def __str__(self):
return f"MediaItem(name='{self.name}', type='{self.type}', url={self.url})"
class MediaManager:
def __init__(self):
self.media_list: List[MediaItem] = []
def add_media(self, data: dict) -> None:
"""
Add media to the list.
Args:
data (dict): Media data to add.
"""
self.media_list.append(MediaItem(data))
def get(self, index: int) -> MediaItem:
"""
Get a media item from the list by index.
Args:
index (int): The index of the media item to retrieve.
Returns:
MediaItem: The media item at the specified index.
"""
return self.media_list[index]
def get_length(self) -> int:
"""
Get the number of media find with research
Returns:
int: Number of episodes.
"""
return len(self.media_list)
def clear(self) -> None:
"""
This method clears the medias list.
Args:
self: The object instance.
"""
self.media_list.clear()
def __str__(self):
return f"MediaManager(num_media={len(self.media_list)})"

View File

@ -0,0 +1,83 @@
# 14.06.24
import sys
import logging
# External libraries
import httpx
from bs4 import BeautifulSoup
# Internal utilities
from Src.Util.headers import get_headers
from Src.Util._jsonConfig import config_manager
class VideoSource:
def __init__(self) -> None:
"""
Initializes the VideoSource object with default values.
Attributes:
headers (dict): A dictionary to store HTTP headers.
cookie (dict): A dictionary to store cookies.
"""
self.headers = {'user-agent': get_headers()}
self.cookie = config_manager.get_dict('REQUESTS', 'index')
def setup(self, url: str) -> None:
"""
Sets up the video source with the provided URL.
Args:
url (str): The URL of the video source.
"""
self.url = url
def make_request(self, url: str) -> str:
"""
Make an HTTP GET request to the provided URL.
Args:
url (str): The URL to make the request to.
Returns:
str: The response content if successful, None otherwise.
"""
try:
response = httpx.get(url, headers=self.headers, cookies=self.cookie)
response.raise_for_status()
return response.text
except httpx.HTTPStatusError as http_err:
logging.error(f"HTTP error occurred: {http_err}")
except Exception as err:
logging.error(f"An error occurred: {err}")
return None
def get_playlist(self):
"""
Retrieves the playlist URL from the video source.
Returns:
tuple: The mp4 link if found, None otherwise.
"""
try:
text = self.make_request(self.url)
if text:
soup = BeautifulSoup(text, "html.parser")
source = soup.find("source")
if source:
mp4_link = source.get("src")
return mp4_link
else:
logging.error("No <source> tag found in the HTML.")
else:
logging.error("Failed to retrieve content from the URL.")
except Exception as e:
logging.error(f"An error occurred while parsing the playlist: {e}")

View File

@ -0,0 +1,71 @@
# 02.05.24
import logging
from typing import List
# Internal utilities
from Src.Util._jsonConfig import config_manager
from Src.Util.os import remove_special_characters
# Config
MAP_EPISODE = config_manager.get('DEFAULT', 'map_episode_name')
def manage_selection(cmd_insert: str, max_count: int) -> List[int]:
"""
Manage user selection for seasons to download.
Args:
- cmd_insert (str): User input for season selection.
- max_count (int): Maximum count of seasons available.
Returns:
list_season_select (List[int]): List of selected seasons.
"""
list_season_select = []
logging.info(f"Command insert: {cmd_insert}, end index: {max_count + 1}")
# For a single number (e.g., '5')
if cmd_insert.isnumeric():
list_season_select.append(int(cmd_insert))
# For a range (e.g., '[5-12]')
elif "[" in cmd_insert:
start, end = map(int, cmd_insert[1:-1].split('-'))
list_season_select = list(range(start, end + 1))
# For all seasons
elif cmd_insert == "*":
list_season_select = list(range(1, max_count+1))
# Return list of selected seasons)
logging.info(f"List return: {list_season_select}")
return list_season_select
def map_episode_title(tv_name: str, number_season: int, episode_number: int, episode_name: str) -> str:
"""
Maps the episode title to a specific format.
Args:
tv_name (str): The name of the TV show.
number_season (int): The season number.
episode_number (int): The episode number.
episode_name (str): The original name of the episode.
Returns:
str: The mapped episode title.
"""
map_episode_temp = MAP_EPISODE
map_episode_temp = map_episode_temp.replace("%(tv_name)", remove_special_characters(tv_name))
map_episode_temp = map_episode_temp.replace("%(season)", str(number_season))
map_episode_temp = map_episode_temp.replace("%(episode)", str(episode_number))
map_episode_temp = map_episode_temp.replace("%(episode_name)", remove_special_characters(episode_name))
# Additional fix
map_episode_temp = map_episode_temp.replace(".", "_")
logging.info(f"Map episode string return: {map_episode_temp}")
return map_episode_temp

View File

@ -1,3 +1,39 @@
# 09.06.24 # 09.06.24
from .site import search import sys
import logging
# Internal utilities
from Src.Util.console import console, msg
# Logic class
from .site import title_search, get_select_title
from .series import download_thread
def search():
"""
Main function of the application for film and series.
"""
# Make request to site to get content that corrsisponde to that string
string_to_search = msg.ask("\n[purple]Insert word to search in all site").strip()
len_database = title_search(string_to_search)
if len_database > 0:
# Select title from list
select_title = get_select_title()
# Download only film
if "Serie TV" in str(select_title.type):
download_thread(select_title)
else:
logging.error(f"Not supported: {select_title.type}")
sys.exit(0)
else:
console.print(f"\n[red]Nothing matching was found for[white]: [purple]{string_to_search}")

View File

@ -2,3 +2,4 @@
MAIN_FOLDER = "ddlstreamitaly" MAIN_FOLDER = "ddlstreamitaly"
MOVIE_FOLDER = "Movie" MOVIE_FOLDER = "Movie"
SERIES_FOLDER = "Serie"

View File

@ -0,0 +1,140 @@
# 13.06.24
import os
import sys
import logging
from urllib.parse import urlparse
# Internal utilities
from Src.Util.color import Colors
from Src.Util.console import console, msg
from Src.Util.os import create_folder, can_create_file
from Src.Util._jsonConfig import config_manager
from Src.Util.table import TVShowManager
from Src.Util.message import start_message
from Src.Lib.Hls.download_mp4 import MP4_downloader
# Logic class
from .Core.Class.SearchType import MediaItem
from .Core.Class.ScrapeSerie import GetSerieInfo
from .Core.Util.manage_ep import manage_selection, map_episode_title
from .Core.Player.ddl import VideoSource
# Config
ROOT_PATH = config_manager.get('DEFAULT', 'root_path')
# Variable
table_show_manager = TVShowManager()
from .costant import MAIN_FOLDER, SERIES_FOLDER
video_source = VideoSource()
def donwload_video(scape_info_serie: GetSerieInfo, index_episode_selected: int) -> None:
"""
Download a single episode video.
Args:
- tv_name (str): Name of the TV series.
- index_episode_selected (int): Index of the selected episode.
"""
start_message()
# Get info about episode
obj_episode = scape_info_serie.list_episodes[index_episode_selected - 1]
console.print(f"[yellow]Download: [red]{obj_episode.get('name')}")
print()
# Define filename and path for the downloaded video
mp4_name = f"{map_episode_title(scape_info_serie.tv_name, None, index_episode_selected, obj_episode.get('name'))}.mp4"
mp4_path = os.path.join(ROOT_PATH, MAIN_FOLDER, SERIES_FOLDER, scape_info_serie.tv_name)
# Check if can create file output
create_folder(mp4_path)
if not can_create_file(mp4_name):
logging.error("Invalid mp4 name.")
sys.exit(0)
# Setup video source
video_source.setup(obj_episode.get('url'))
# Get m3u8 master playlist
master_playlist = video_source.get_playlist()
# Parse start page url
start_message()
parsed_url = urlparse(obj_episode.get('url'))
path_parts = parsed_url.path.split('/')
MP4_downloader(
url = master_playlist,
path = os.path.join(mp4_path, mp4_name),
referer = f"{parsed_url.scheme}://{parsed_url.netloc}/",
add_desc=f"{Colors.MAGENTA}video"
)
def download_thread(dict_serie: MediaItem):
"""Download all episode of a thread"""
# Start message and set up video source
start_message()
# Init class
scape_info_serie = GetSerieInfo(dict_serie)
# Collect information about thread
list_dict_episode = scape_info_serie.get_episode_number()
episodes_count = len(list_dict_episode)
# Display episodes list and manage user selection
last_command = display_episodes_list(list_dict_episode)
list_episode_select = manage_selection(last_command, episodes_count)
# Download selected episodes
if len(list_episode_select) == 1 and last_command != "*":
donwload_video(scape_info_serie, list_episode_select[0])
# Download all other episodes selecter
else:
for i_episode in list_episode_select:
donwload_video(scape_info_serie, i_episode)
def display_episodes_list(obj_episode_manager) -> str:
"""
Display episodes list and handle user input.
Returns:
last_command (str): Last command entered by the user.
"""
# Set up table for displaying episodes
table_show_manager.set_slice_end(10)
# Add columns to the table
column_info = {
"Index": {'color': 'red'},
"Name": {'color': 'magenta'},
}
table_show_manager.add_column(column_info)
# Populate the table with episodes information
for i, media in enumerate(obj_episode_manager):
table_show_manager.add_tv_show({
'Index': str(i+1),
'Name': media.get('name'),
})
# Run the table and handle user input
last_command = table_show_manager.run()
if last_command == "q":
console.print("\n[red]Quit [white]...")
sys.exit(0)
return last_command

View File

@ -1,9 +1,7 @@
# 09.06.24 # 09.06.24
import os
import sys import sys
import logging import logging
from urllib.parse import urlparse
# External libraries # External libraries
@ -12,71 +10,121 @@ from bs4 import BeautifulSoup
# Internal utilities # Internal utilities
from Src.Util.message import start_message from Src.Util.table import TVShowManager
from Src.Util.color import Colors
from Src.Util.console import console, msg from Src.Util.console import console, msg
from Src.Util.os import create_folder, can_create_file
from Src.Util._jsonConfig import config_manager from Src.Util._jsonConfig import config_manager
from Src.Util.headers import get_headers from Src.Util.headers import get_headers
from Src.Lib.Hls.download_mp4 import MP4_downloader
# Logic class
from .Core.Class.SearchType import MediaManager, MediaItem
# Config # Config
ROOT_PATH = config_manager.get('DEFAULT', 'root_path') SITE_NAME = "ddlstreamitaly"
from .costant import MAIN_FOLDER, MOVIE_FOLDER DOMAIN_NOW = config_manager.get('SITE', SITE_NAME)
# Variable # Variable
cookie_index = config_manager.get_dict('REQUESTS', 'index') cookie_index = config_manager.get_dict('REQUESTS', 'index')
media_search_manager = MediaManager()
table_show_manager = TVShowManager()
def search() -> int:
def title_search(word_to_search) -> int:
""" """
Search for titles based on a search query. Search for titles based on a search query.
""" """
try:
print()
url_search = msg.ask(f"[cyan]Insert url title")
# Send request to search for titles # Send request to search for titles
try: response = httpx.get(f"https://ddlstreamitaly.{DOMAIN_NOW}/search/?&q={word_to_search}&quick=1&type=videobox_video&nodes=11", headers={'user-agent': get_headers()})
response = httpx.get(url_search, headers={'user-agent': get_headers()}, cookies=cookie_index)
response.raise_for_status() response.raise_for_status()
except Exception as e: # Create soup and find table
logging.error("Insert: {'ips4_IPSSessionFront': 'your_code', 'ips4_member_id': 'your_code'} in config file \ REQUESTS \ index, instead of user-agent. Use browser debug and cookie request with a valid account.")
sys.exit(0)
# Create soup and mp4 video
soup = BeautifulSoup(response.text, "html.parser") soup = BeautifulSoup(response.text, "html.parser")
souce = soup.find("source") table_content = soup.find('ol', class_="ipsStream")
# Get url and filename if table_content:
for title_div in table_content.find_all('li', class_='ipsStreamItem'):
try: try:
mp4_link = souce.get("src") title_type = title_div.find("p", class_="ipsType_reset").find_all("a")[-1].get_text(strip=True)
name = title_div.find("span", class_="ipsContained").find("a").get_text(strip=True)
link = title_div.find("span", class_="ipsContained").find("a").get("href")
title_info = {
'name': name,
'url': link,
'type': title_type
}
media_search_manager.add_media(title_info)
except Exception as e: except Exception as e:
logging.error("Insert: {'ips4_IPSSessionFront': 'your_code', 'ips4_member_id': 'your_code'} in config file \ REQUESTS \ index, instead of user-agent. Use browser debug and cookie request with a valid account.") logging.error(f"Error processing title div: {e}")
# Return the number of titles found
return media_search_manager.get_length()
else:
logging.error("No table content found.")
return -999
except Exception as err:
logging.error(f"An error occurred: {err}")
return -9999
def get_select_title(type_filter: list = None) -> MediaItem:
"""
Display a selection of titles and prompt the user to choose one.
Args:
- type_filter (list): A list of media types to filter. Can include 'film', 'tv', 'ova'. Ex. ['tv', 'film']
Returns:
MediaItem: The selected media item.
"""
# Set up table for displaying titles
table_show_manager.set_slice_end(10)
# Add columns to the table
column_info = {
"Index": {'color': 'red'},
"Name": {'color': 'magenta'},
"Type": {'color': 'yellow'},
}
table_show_manager.add_column(column_info)
# Populate the table with title information
for i, media in enumerate(media_search_manager.media_list):
# Filter for only a list of category
if type_filter is not None:
if str(media.type) not in type_filter:
continue
table_show_manager.add_tv_show({
'Index': str(i),
'Name': media.name,
'Type': media.type,
})
# Run the table and handle user input
last_command = table_show_manager.run(force_int_input=True, max_int_input=len(media_search_manager.media_list))
table_show_manager.clear()
# Handle user's quit command
if last_command == "q":
console.print("\n[red]Quit [white]...")
sys.exit(0) sys.exit(0)
parsed_url = urlparse(url_search) # Check if the selected index is within range
path_parts = parsed_url.path.split('/') if 0 <= int(last_command) <= len(media_search_manager.media_list):
mp4_name = path_parts[-2] if path_parts[-1] == '' else path_parts[-1] + ".mp4" return media_search_manager.get(int(last_command))
else:
# Create destination folder console.print("\n[red]Wrong index")
mp4_path = os.path.join(ROOT_PATH, MAIN_FOLDER, MOVIE_FOLDER)
# Check if can create file output
create_folder(mp4_path)
if not can_create_file(mp4_name):
logging.error("Invalid mp4 name.")
sys.exit(0) sys.exit(0)
# Start download
start_message()
MP4_downloader(
url = mp4_link,
path = os.path.join(mp4_path, mp4_name),
referer = f"{parsed_url.scheme}://{parsed_url.netloc}/",
add_desc=f"{Colors.MAGENTA}video"
)

View File

@ -33,7 +33,6 @@ class GetSerieInfo:
self.url = dict_serie.url self.url = dict_serie.url
self.tv_name = None self.tv_name = None
self.list_episodes = None self.list_episodes = None
self.list_episodes = None
def get_seasons_number(self) -> int: def get_seasons_number(self) -> int:
""" """

View File

@ -59,4 +59,3 @@ class MediaManager:
def __str__(self): def __str__(self):
return f"MediaManager(num_media={len(self.media_list)})" return f"MediaManager(num_media={len(self.media_list)})"

View File

@ -15,13 +15,16 @@ def search():
""" """
# Make request to site to get content that corrsisponde to that string # Make request to site to get content that corrsisponde to that string
film_search = msg.ask("\n[purple]Insert word to search in all site").strip() string_to_search = msg.ask("\n[purple]Insert word to search in all site").strip()
len_database = title_search(film_search) len_database = title_search(string_to_search)
if len_database != 0: if len_database > 0:
# Select title from list # Select title from list
select_title = get_select_title() select_title = get_select_title()
# Download only film # Download only film
download_series(select_title) download_series(select_title)
else:
console.print(f"\n[red]Nothing matching was found for[white]: [purple]{string_to_search}")

View File

@ -122,18 +122,3 @@ def get_select_title(type_filter: list = None) -> MediaItem:
else: else:
console.print("\n[red]Wrong index") console.print("\n[red]Wrong index")
sys.exit(0) sys.exit(0)
def manager_clear():
"""
Clears the data lists managed by media_search_manager and table_show_manager.
This function clears the data lists managed by global variables media_search_manager
and table_show_manager. It removes all the items from these lists, effectively
resetting them to empty lists.
"""
global media_search_manager, table_show_manager
# Clear list of data
media_search_manager.clear()
table_show_manager.clear()

View File

@ -8,8 +8,7 @@ from Src.Util.console import console, msg
from .site import ( from .site import (
get_version_and_domain, get_version_and_domain,
title_search, title_search,
get_select_title, get_select_title
manager_clear
) )
from .film import download_film from .film import download_film
@ -25,10 +24,10 @@ def search():
site_version, domain = get_version_and_domain() site_version, domain = get_version_and_domain()
# Make request to site to get content that corrsisponde to that string # Make request to site to get content that corrsisponde to that string
film_search = msg.ask("\n[purple]Insert word to search in all site").strip() string_to_search = msg.ask("\n[purple]Insert word to search in all site").strip()
len_database = title_search(film_search, domain) len_database = title_search(string_to_search, domain)
if len_database != 0: if len_database > 0:
# Select title from list # Select title from list
select_title = get_select_title() select_title = get_select_title()
@ -50,6 +49,5 @@ def search():
domain=domain domain=domain
) )
# If no media find
else: else:
console.print("[red]Cant find a single element") console.print(f"\n[red]Nothing matching was found for[white]: [purple]{string_to_search}")

View File

@ -203,18 +203,3 @@ def get_select_title(type_filter: list = None) -> MediaItem:
else: else:
console.print("\n[red]Wrong index") console.print("\n[red]Wrong index")
sys.exit(0) sys.exit(0)
def manager_clear():
"""
Clears the data lists managed by media_search_manager and table_show_manager.
This function clears the data lists managed by global variables media_search_manager
and table_show_manager. It removes all the items from these lists, effectively
resetting them to empty lists.
"""
global media_search_manager, table_show_manager
# Clear list of data
media_search_manager.clear()
table_show_manager.clear()

View File

@ -12,7 +12,9 @@
"REQUESTS": { "REQUESTS": {
"timeout": 5, "timeout": 5,
"verify_ssl": false, "verify_ssl": false,
"index": {"user-agent": ""}, "index": {
"user-agent": ""
},
"proxy_start_min": 0.6, "proxy_start_min": 0.6,
"proxy_start_max": 1.0 "proxy_start_max": 1.0
}, },
@ -46,6 +48,7 @@
"streamingcommunity": "foo", "streamingcommunity": "foo",
"animeunity": "to", "animeunity": "to",
"altadefinizione": "vodka", "altadefinizione": "vodka",
"guardaserie": "ceo" "guardaserie": "ceo",
"ddlstreamitaly": "co"
} }
} }