Fix altadefinizione and streamingcomunity dont work.

This commit is contained in:
Lovi-0 2024-07-25 20:58:29 +02:00
parent bbed4d9d12
commit 492684d4d5
5 changed files with 15 additions and 12 deletions

View File

@ -26,13 +26,14 @@ AUTO_UPDATE_DOMAIN = config_manager.get_bool('DEFAULT', 'auto_update_domain')
def check_url_for_content(url: str, content: str, timeout: int = 1) -> bool: def check_url_for_content(url: str, content: str, follow_redirects: bool = True, timeout: int = 1, ) -> bool:
""" """
Check if a URL contains specific content. Check if a URL contains specific content.
Parameters: Parameters:
- url (str): The URL to check. - url (str): The URL to check.
- content (str): The content to search for in the response. - content (str): The content to search for in the response.
- follow_redirects (bool): To follow redirect url or not.
- timeout (int): Timeout for the request in seconds. - timeout (int): Timeout for the request in seconds.
Returns: Returns:
@ -40,7 +41,7 @@ def check_url_for_content(url: str, content: str, timeout: int = 1) -> bool:
""" """
try: try:
response = httpx.get(url, timeout=timeout, headers={'user-agent': get_headers()}, follow_redirects=True) response = httpx.get(url, timeout=timeout, headers={'user-agent': get_headers()}, follow_redirects=follow_redirects)
logging.info(f"Testing site to extract domain: {url}, response: {response.status_code}") logging.info(f"Testing site to extract domain: {url}, response: {response.status_code}")
# Raise an error if the status is not successful # Raise an error if the status is not successful
@ -62,7 +63,7 @@ def check_url_for_content(url: str, content: str, timeout: int = 1) -> bool:
return False return False
def get_top_level_domain(base_url: str, target_content: str, max_workers: int = os.cpu_count(), timeout: int = 2, retries: int = 1) -> str: def get_top_level_domain(base_url: str, target_content: str, max_workers: int = os.cpu_count(), timeout: int = 2, retries: int = 1, follow_redirects: bool = True) -> str:
""" """
Get the top-level domain (TLD) from a list of URLs. Get the top-level domain (TLD) from a list of URLs.
@ -72,6 +73,7 @@ def get_top_level_domain(base_url: str, target_content: str, max_workers: int =
- max_workers (int): Maximum number of threads. - max_workers (int): Maximum number of threads.
- timeout (int): Timeout for the request in seconds. - timeout (int): Timeout for the request in seconds.
- retries (int): Number of retries for failed requests. - retries (int): Number of retries for failed requests.
- follow_redirects (bool): To follow redirect url or not.
Returns: Returns:
str: The found TLD, if any. str: The found TLD, if any.
@ -108,7 +110,7 @@ def get_top_level_domain(base_url: str, target_content: str, max_workers: int =
if stop_event.is_set(): if stop_event.is_set():
return None return None
if check_url_for_content(url, target_content, timeout): if check_url_for_content(url, target_content, follow_redirects, timeout):
stop_event.set() stop_event.set()
progress_bar.update(1) progress_bar.update(1)
return url.split(".")[-1] return url.split(".")[-1]
@ -148,7 +150,7 @@ def get_top_level_domain(base_url: str, target_content: str, max_workers: int =
return None return None
def search_domain(site_name: str, target_content: str, base_url: str): def search_domain(site_name: str, target_content: str, base_url: str, follow_redirects: bool = True):
""" """
Search for a valid domain for the given site name and base URL. Search for a valid domain for the given site name and base URL.
@ -156,6 +158,7 @@ def search_domain(site_name: str, target_content: str, base_url: str):
- site_name (str): The name of the site to search the domain for. - site_name (str): The name of the site to search the domain for.
- target_content (str): The content to search for in the response. - target_content (str): The content to search for in the response.
- base_url (str): The base URL to construct complete URLs. - base_url (str): The base URL to construct complete URLs.
- follow_redirects (bool): To follow redirect url or not.
Returns: Returns:
tuple: The found domain and the complete URL. tuple: The found domain and the complete URL.
@ -169,7 +172,7 @@ def search_domain(site_name: str, target_content: str, base_url: str):
try: try:
# Test the current domain # Test the current domain
response = httpx.get(f"{base_url}.{domain}", headers={'user-agent': get_headers()}, timeout=5, follow_redirects=True) response = httpx.get(f"{base_url}.{domain}", headers={'user-agent': get_headers()}, timeout=5, follow_redirects=follow_redirects)
console.print(f"[cyan]Test response site[white]: [red]{response.status_code}") console.print(f"[cyan]Test response site[white]: [red]{response.status_code}")
response.raise_for_status() response.raise_for_status()
@ -185,7 +188,7 @@ def search_domain(site_name: str, target_content: str, base_url: str):
if AUTO_UPDATE_DOMAIN: if AUTO_UPDATE_DOMAIN:
console.print("[red]Extract new DOMAIN from TLD list.") console.print("[red]Extract new DOMAIN from TLD list.")
new_domain = get_top_level_domain(base_url=base_url, target_content=target_content) new_domain = get_top_level_domain(base_url=base_url, target_content=target_content, follow_redirects=follow_redirects)
if new_domain is not None: if new_domain is not None:

View File

@ -39,7 +39,7 @@ def title_search(title_search: str) -> int:
domain_to_use, _ = search_domain(SITE_NAME, '<meta name="generator" content="altadefinizione">', f"https://{SITE_NAME}") domain_to_use, _ = search_domain(SITE_NAME, '<meta name="generator" content="altadefinizione">', f"https://{SITE_NAME}")
# Send request to search for titles # Send request to search for titles
response = httpx.get(f"https://{SITE_NAME}.{domain_to_use}/page/1/?story={unidecode(title_search.replace(' ', '+'))}&do=search&subaction=search&titleonly=3", headers={'user-agent': get_headers()}) response = httpx.get(f"https://{SITE_NAME}.{domain_to_use}/page/1/?story={unidecode(title_search.replace(' ', '+'))}&do=search&subaction=search&titleonly=3", headers={'user-agent': get_headers()}, follow_redirects=True)
response.raise_for_status() response.raise_for_status()
# Create soup and find table # Create soup and find table

View File

@ -83,7 +83,7 @@ def get_version_and_domain():
""" """
# Find new domain if prev dont work # Find new domain if prev dont work
domain_to_use, base_url = search_domain(SITE_NAME, '<meta name="author" content="StreamingCommunity">', f"https://{SITE_NAME}") domain_to_use, base_url = search_domain(SITE_NAME, '<meta name="author" content="StreamingCommunity">', f"https://{SITE_NAME}", False)
# Extract version from the response # Extract version from the response
version, list_title_top_10 = get_version(httpx.get(base_url, headers={'user-agent': get_headers()}).text) version, list_title_top_10 = get_version(httpx.get(base_url, headers={'user-agent': get_headers()}).text)

View File

@ -17,5 +17,5 @@ from Src.Lib.Downloader import MP4_downloader
# Test # Test
MP4_downloader( MP4_downloader(
"", "",
"EP_2.mp4", "EP_1.mp4",
) )

View File

@ -24,7 +24,7 @@
"proxy_start_min": 0.1, "proxy_start_min": 0.1,
"proxy_start_max": 0.5 "proxy_start_max": 0.5
}, },
"BROWSER" : { "BROWSER": {
"headless": true "headless": true
}, },
"M3U8_DOWNLOAD": { "M3U8_DOWNLOAD": {
@ -60,7 +60,7 @@
"streamingcommunity": { "streamingcommunity": {
"video_workers": 2, "video_workers": 2,
"audio_workers": 2, "audio_workers": 2,
"domain": "boston" "domain": "love"
}, },
"animeunity": { "animeunity": {
"video_workers": 2, "video_workers": 2,