diff --git a/Src/Api/Altadefinizione/Core/Player/supervideo.py b/Src/Api/Altadefinizione/Core/Player/supervideo.py index 9be0249..48e8155 100644 --- a/Src/Api/Altadefinizione/Core/Player/supervideo.py +++ b/Src/Api/Altadefinizione/Core/Player/supervideo.py @@ -1,15 +1,12 @@ # 26.05.24 import re -import os import sys -import time import logging -import subprocess # External libraries -from Src.Lib.Request import requests +import requests from bs4 import BeautifulSoup diff --git a/Src/Api/Altadefinizione/site.py b/Src/Api/Altadefinizione/site.py index 958cc8d..1a53609 100644 --- a/Src/Api/Altadefinizione/site.py +++ b/Src/Api/Altadefinizione/site.py @@ -6,7 +6,7 @@ import logging # External libraries -from Src.Lib.Request import requests +import requests from bs4 import BeautifulSoup from unidecode import unidecode diff --git a/Src/Api/Animeunity/Core/Util/get_domain.py b/Src/Api/Animeunity/Core/Util/get_domain.py index 36e02cd..6730179 100644 --- a/Src/Api/Animeunity/Core/Util/get_domain.py +++ b/Src/Api/Animeunity/Core/Util/get_domain.py @@ -5,8 +5,8 @@ import threading import logging -# Internal libraries -from Src.Lib.Request import requests +# External libraries +import requests # Internal utilities diff --git a/Src/Api/Animeunity/Core/Vix_player/player.py b/Src/Api/Animeunity/Core/Vix_player/player.py index bcac59a..c977888 100644 --- a/Src/Api/Animeunity/Core/Vix_player/player.py +++ b/Src/Api/Animeunity/Core/Vix_player/player.py @@ -6,7 +6,7 @@ from urllib.parse import urljoin, urlparse, parse_qs, urlencode, urlunparse # External libraries -from Src.Lib.Request import requests +import requests from bs4 import BeautifulSoup diff --git a/Src/Api/Animeunity/site.py b/Src/Api/Animeunity/site.py index 28ee558..b885afd 100644 --- a/Src/Api/Animeunity/site.py +++ b/Src/Api/Animeunity/site.py @@ -5,7 +5,7 @@ import logging # External libraries -from Src.Lib.Request import requests +import requests from bs4 import BeautifulSoup from unidecode import unidecode diff --git a/Src/Api/Streamingcommunity/Core/Util/get_domain.py b/Src/Api/Streamingcommunity/Core/Util/get_domain.py index 0faada4..4017c43 100644 --- a/Src/Api/Streamingcommunity/Core/Util/get_domain.py +++ b/Src/Api/Streamingcommunity/Core/Util/get_domain.py @@ -5,8 +5,11 @@ import threading import logging +# External library +import requests + + # Internal utilities -from Src.Lib.Request import requests from Src.Lib.Google import search as google_search diff --git a/Src/Api/Streamingcommunity/Core/Vix_player/player.py b/Src/Api/Streamingcommunity/Core/Vix_player/player.py index bd72041..cb26be4 100644 --- a/Src/Api/Streamingcommunity/Core/Vix_player/player.py +++ b/Src/Api/Streamingcommunity/Core/Vix_player/player.py @@ -6,7 +6,7 @@ from urllib.parse import urljoin, urlparse, parse_qs, urlencode, urlunparse # External libraries -from Src.Lib.Request import requests +import requests from bs4 import BeautifulSoup diff --git a/Src/Api/Streamingcommunity/site.py b/Src/Api/Streamingcommunity/site.py index 606e77c..bdd5ec9 100644 --- a/Src/Api/Streamingcommunity/site.py +++ b/Src/Api/Streamingcommunity/site.py @@ -8,12 +8,12 @@ from typing import Tuple # External libraries +import requests from bs4 import BeautifulSoup from unidecode import unidecode # Internal utilities -from Src.Lib.Request import requests from Src.Util.headers import get_headers from Src.Util._jsonConfig import config_manager from Src.Util.console import console diff --git a/Src/Lib/Google/page.py b/Src/Lib/Google/page.py index 2185ad7..54e976f 100644 --- a/Src/Lib/Google/page.py +++ b/Src/Lib/Google/page.py @@ -8,7 +8,7 @@ from typing import Generator, Optional # External libraries -from Src.Lib.Request import requests +import requests from bs4 import BeautifulSoup diff --git a/Src/Lib/Hls/downloader.py b/Src/Lib/Hls/downloader.py index 9ef3ff3..26bd443 100644 --- a/Src/Lib/Hls/downloader.py +++ b/Src/Lib/Hls/downloader.py @@ -8,7 +8,7 @@ from concurrent.futures import ThreadPoolExecutor # External libraries -from Src.Lib.Request import requests +import requests from unidecode import unidecode diff --git a/Src/Lib/Hls/segments.py b/Src/Lib/Hls/segments.py index 8dfbf05..050706e 100644 --- a/Src/Lib/Hls/segments.py +++ b/Src/Lib/Hls/segments.py @@ -12,7 +12,7 @@ from urllib.parse import urljoin, urlparse, urlunparse # External libraries -from Src.Lib.Request import requests +import requests from tqdm import tqdm @@ -22,6 +22,7 @@ from Src.Util.headers import get_headers from Src.Util.color import Colors from Src.Util._jsonConfig import config_manager + # Logic class from ..M3U8 import ( M3U8_Decryption, @@ -30,6 +31,11 @@ from ..M3U8 import ( M3U8_UrlFix ) + +# Warning +import urllib3 +urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + # Config TQDM_MAX_WORKER = config_manager.get_int('M3U8_DOWNLOAD', 'tdqm_workers') TQDM_USE_LARGE_BAR = config_manager.get_int('M3U8_DOWNLOAD', 'tqdm_use_large_bar') @@ -265,7 +271,7 @@ class M3U8_Segments: # Make request and calculate time duration start_time = time.time() - response = requests.get(ts_url, headers=headers_segments, verify=REQUEST_VERIFY_SSL, timeout=30) + response = requests.get(ts_url, headers=headers_segments, verify=REQUEST_VERIFY_SSL, timeout=15) duration = time.time() - start_time logging.info(f"Make request to get segment: [{index} - {len(self.segments)}] in: {duration}, len data: {len(response.content)}") @@ -275,7 +281,7 @@ class M3U8_Segments: segment_content = response.content # Update bar - self.class_ts_estimator.update_progress_bar(segment_content, duration, progress_bar) + self.class_ts_estimator.update_progress_bar(int(response.headers.get('Content-Length', 0)), duration, progress_bar) # Decrypt the segment content if decryption is needed if self.decryption is not None: diff --git a/Src/Lib/M3U8/estimator.py b/Src/Lib/M3U8/estimator.py index e0e6596..e762a40 100644 --- a/Src/Lib/M3U8/estimator.py +++ b/Src/Lib/M3U8/estimator.py @@ -31,7 +31,7 @@ class M3U8_Ts_Estimator: """ self.ts_file_sizes = [] self.now_downloaded_size = 0 - self.average_over = 6 + self.average_over = 3 self.list_speeds = deque(maxlen=self.average_over) self.smoothed_speeds = [] self.total_segments = total_segments @@ -52,7 +52,7 @@ class M3U8_Ts_Estimator: # Calculate speed outside of the lock try: - speed_mbps = (size_download * 16) / (duration * 1_000_000) + speed_mbps = (size_download * 8) / (duration * 1_000_000) except ZeroDivisionError as e: logging.error("Division by zero error while calculating speed: %s", e) return @@ -114,16 +114,15 @@ class M3U8_Ts_Estimator: """ return format_size(self.now_downloaded_size) - def update_progress_bar(self, segment_content: bytes, duration: float, progress_counter: tqdm) -> None: + def update_progress_bar(self, total_downloaded: int, duration: float, progress_counter: tqdm) -> None: """ Updates the progress bar with information about the TS segment download. Args: - segment_content (bytes): The content of the downloaded TS segment. + total_downloaded (int): The len of the content of the downloaded TS segment. duration (float): The duration of the segment download in seconds. progress_counter (tqdm): The tqdm object representing the progress bar. """ - total_downloaded = len(segment_content) # Add the size of the downloaded segment to the estimator self.add_ts_file(total_downloaded * self.total_segments, total_downloaded, duration) diff --git a/Src/Lib/M3U8/parser.py b/Src/Lib/M3U8/parser.py index a29a82f..bf364eb 100644 --- a/Src/Lib/M3U8/parser.py +++ b/Src/Lib/M3U8/parser.py @@ -8,7 +8,7 @@ from .lib_parser import load # External libraries -from Src.Lib.Request import requests +import requests # Costant diff --git a/Src/Lib/Request/__init__.py b/Src/Lib/Request/__init__.py deleted file mode 100644 index f6af28d..0000000 --- a/Src/Lib/Request/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -# 04.4.24 - -from .my_requests import requests \ No newline at end of file diff --git a/Src/Lib/Request/my_requests.py b/Src/Lib/Request/my_requests.py deleted file mode 100644 index 405d892..0000000 --- a/Src/Lib/Request/my_requests.py +++ /dev/null @@ -1,541 +0,0 @@ -# 04.4.24 - -import os -import sys -import base64 -import json -import logging -import ssl -import time -import re -import subprocess -import urllib.parse -import urllib.request -import urllib.error - -from typing import Dict, Optional, Union, Any - - -try: - from typing import Unpack, TypedDict -except ImportError: - # (Python <= 3.10), - try: - from typing_extensions import Unpack, TypedDict # type: ignore - except ImportError: - raise ImportError("Unable to import Unpack from typing or typing_extensions. " - "Please make sure you have the necessary libraries installed.") - - -# External library -from bs4 import BeautifulSoup - - -# Internal utilities -from Src.Util._jsonConfig import config_manager - - -# Default settings -HTTP_TIMEOUT = config_manager.get_int('REQUESTS', 'timeout') -HTTP_RETRIES = config_manager.get_int('REQUESTS', 'max_retry') -HTTP_DELAY = 1 - - - -class RequestError(Exception): - """Custom exception class for request errors.""" - - def __init__(self, message: str, original_exception: Optional[Exception] = None) -> None: - """ - Initialize a RequestError instance. - - Args: - - message (str): The error message. - - original_exception (Optional[Exception], optional): The original exception that occurred. Defaults to None. - """ - super().__init__(message) - self.original_exception = original_exception - - def __str__(self) -> str: - """Return a string representation of the exception.""" - if self.original_exception: - return f"{super().__str__()} Original Exception: {type(self.original_exception).__name__}: {str(self.original_exception)}" - else: - return super().__str__() - - -def parse_http_error(error_string: str): - """ - Parse the HTTP error string to extract the error code and message. - - Args: - - error_string (str): The error string from an HTTP response. - - Returns: - dict: A dictionary with 'error_code' and 'message' if the string is parsed successfully, or None if parsing fails. - """ - - # Regular expression to match the error pattern - error_pattern = re.compile(r"HTTP Error (\d{3}): (.+)") - match = error_pattern.search(error_string) - - if match: - error_code = match.group(1) - message = match.group(2) - return {'error_code': error_code, 'message': message} - - else: - logging.error(f"Error string does not match expected format: {error_string}") - return None - - -class Response: - """ - Class representing an HTTP response. - """ - def __init__( - self, - status: int, - text: str, - is_json: bool = False, - content: bytes = b"", - headers: Optional[Dict[str, str]] = None, - cookies: Optional[Dict[str, str]] = None, - redirect_url: Optional[str] = None, - response_time: Optional[float] = None, - timeout: Optional[float] = None, - ): - """ - Initialize a Response object. - - Args: - - status (int): The HTTP status code of the response. - - text (str): The response content as text. - - is_json (bool, optional): Indicates if the response content is JSON. Defaults to False. - - content (bytes, optional): The response content as bytes. Defaults to b"". - - headers (Optional[Dict[str, str]], optional): The response headers. Defaults to None. - - cookies (Optional[Dict[str, str]], optional): The cookies set in the response. Defaults to None. - - redirect_url (Optional[str], optional): The URL if a redirection occurred. Defaults to None. - - response_time (Optional[float], optional): The time taken to receive the response. Defaults to None. - - timeout (Optional[float], optional): The request timeout. Defaults to None. - """ - self.status_code = status - self.text = text - self.is_json = is_json - self.content = content - self.headers = headers or {} - self.cookies = cookies or {} - self.redirect_url = redirect_url - self.response_time = response_time - self.timeout = timeout - self.ok = 200 <= status < 300 - - def raise_for_status(self): - """ - Raise an error if the response status code is not in the 2xx range. - """ - if not self.ok: - raise RequestError(f"Request failed with status code {self.status_code}") - - def json(self): - """ - Return the response content as JSON if it is JSON. - - Returns: - dict or list or None: A Python dictionary or list parsed from JSON if the response content is JSON, otherwise None. - """ - if self.is_json: - return json.loads(self.text) - else: - return None - - def get_redirects(self): - """ - Extracts unique site URLs from HTML elements within the
section. - - Returns: - list or None: A list of unique site URLs if found, otherwise None. - """ - - site_find = [] - - if self.text: - soup = BeautifulSoup(self.text, "html.parser") - - for links in soup.find("head").find_all('link'): - if links is not None: - parsed_url = urllib.parse.urlparse(links.get('href')) - site = parsed_url.scheme + "://" + parsed_url.netloc - - if site not in site_find: - site_find.append(site) - - if site_find: - return site_find - else: - return None - - -class ManageRequests: - """ - Class for managing HTTP requests. - """ - def __init__( - self, - url: str, - method: str = 'GET', - headers: Optional[Dict[str, str]] = None, - timeout: float = HTTP_TIMEOUT, - retries: int = HTTP_RETRIES, - params: Optional[Dict[str, str]] = None, - verify: bool = True, - auth: Optional[tuple] = None, - proxy: Optional[str] = None, - cookies: Optional[Dict[str, str]] = None, - json_data: Optional[Dict[str, Any]] = None, - redirection_handling: bool = True, - ): - """ - Initialize a ManageRequests object. - - Args: - - url (str): The URL to which the request will be sent. - - method (str, optional): The HTTP method to be used for the request. Defaults to 'GET'. - - headers (Optional[Dict[str, str]], optional): The request headers. Defaults to None. - - timeout (float, optional): The request timeout. Defaults to HTTP_TIMEOUT. - - retries (int, optional): The number of retries in case of request failure. Defaults to HTTP_RETRIES. - - params (Optional[Dict[str, str]], optional): The query parameters for the request. Defaults to None. - - verify (bool, optional): Indicates whether SSL certificate verification should be performed. Defaults to True. - - auth (Optional[tuple], optional): Tuple containing the username and password for basic authentication. Defaults to None. - - proxy (Optional[str], optional): The proxy URL. Defaults to None. - - cookies (Optional[Dict[str, str]], optional): The cookies to be included in the request. Defaults to None. - - redirection_handling (bool, optional): Indicates whether redirections should be followed. Defaults to True. - """ - self.url = url - self.method = method - self.headers = headers or {} - self.timeout = timeout - self.retries = retries - self.params = params - self.verify_ssl = verify - self.auth = auth - self.proxy = proxy - self.cookies = cookies - self.json_data = json_data - self.redirection_handling = redirection_handling - - def add_header(self, key: str, value: str) -> None: - """ - Add a header to the request. - """ - self.headers[key] = value - - def send(self) -> Response: - """ - Send the HTTP request. - """ - - start_time = time.time() - self.attempt = 0 - redirect_url = None - - while self.attempt < self.retries: - try: - req = self._build_request() - response = self._perform_request(req) - - return self._process_response(response, start_time, redirect_url) - - except (urllib.error.URLError, urllib.error.HTTPError) as e: - self._handle_error(e) - self.attempt += 1 - - def log_request(self): - """ - Constructs a log message based on the request parameters and logs it. - """ - log_message = "Request: (" - - if self.url: - log_message += f"'url': {self.url}, " - if self.headers: - log_message += f"'headers': {self.headers}, " - if self.cookies: - log_message += f"'cookies': {self.cookies}, " - if self.json_data: - log_message += f"'body': {json.dumps(self.json_data).encode('utf-8')}, " - - # Remove the trailing comma and add parentheses - log_message = log_message.rstrip(", ") + ")" - logging.info(log_message) - - def _build_request(self) -> urllib.request.Request: - """ - Build the urllib Request object. - """ - - # Make a copy of headers to avoid modifying the original dictionary - headers = self.headers.copy() - - # Construct the URL with query parameters if present - if self.params: - url = self.url + '?' + urllib.parse.urlencode(self.params) - else: - url = self.url - - # Create the initial Request object - req = urllib.request.Request(url, headers=headers, method=self.method) - - # Add JSON data if provided - if self.json_data: - req.add_header('Content-Type', 'application/json') - req.data = json.dumps(self.json_data).encode('utf-8') - - # Add authorization header if provided - if self.auth: - req.add_header('Authorization', 'Basic ' + base64.b64encode(f"{self.auth[0]}:{self.auth[1]}".encode()).decode()) - - # Add cookies if provided - if self.cookies: - cookie_str = '; '.join([f"{name}={value}" for name, value in self.cookies.items()]) - req.add_header('Cookie', cookie_str) - - # Add default user agent if not already present - if 'user-agent' not in headers: - default_user_agent = 'Mozilla/5.0' - req.add_header('user-agent', default_user_agent) - - - self.log_request() - return req - - def _perform_request(self, req: urllib.request.Request) -> urllib.response.addinfourl: - """ - Perform the HTTP request. - """ - if self.proxy: - proxy_handler = urllib.request.ProxyHandler({'http': self.proxy, 'https': self.proxy}) - opener = urllib.request.build_opener(proxy_handler) - urllib.request.install_opener(opener) - - if not self.verify_ssl: - - # Create SSL context - ssl_context = ssl.create_default_context() - ssl_context.check_hostname = False - ssl_context.verify_mode = ssl.CERT_NONE - - # Build the request with SSL context - response = urllib.request.urlopen(req, timeout=self.timeout, context=ssl_context) - - else: - response = urllib.request.urlopen(req, timeout=self.timeout, context=ssl.create_default_context()) - - return response - - def _process_response(self, response: urllib.response.addinfourl, start_time: float, redirect_url: Optional[str]) -> Response: - """ - Process the HTTP response. - """ - response_data = response.read() - content_type = response.headers.get('Content-Type', '').lower() - - if self.redirection_handling and response.status in (301, 302, 303, 307, 308): - location = response.headers.get('Location') - logging.info(f"Redirecting to: {location}") - redirect_url = location - self.url = location - return self.send() - - return self._build_response(response, response_data, start_time, redirect_url, content_type) - - def _build_response(self, response: urllib.response.addinfourl, response_data: bytes, start_time: float, redirect_url: Optional[str], content_type: str) -> Response: - """ - Build the Response object. - """ - response_time = time.time() - start_time - response_headers = dict(response.headers) - response_cookies = {} - - for cookie in response.headers.get_all('Set-Cookie', []): - cookie_parts = cookie.split(';') - cookie_name, cookie_value = cookie_parts[0].split('=', 1) # Only the first - response_cookies[cookie_name.strip()] = cookie_value.strip() - - return Response( - status=response.status, - text=response_data.decode('latin-1'), - is_json=("json" in content_type), - content=response_data, - headers=response_headers, - cookies=response_cookies, - redirect_url=redirect_url, - response_time=response_time, - timeout=self.timeout, - ) - - def _handle_error(self, e: Union[urllib.error.URLError, urllib.error.HTTPError]) -> None: - """ - Handle request error. - """ - logging.error(f"Request failed for URL '{self.url}': {parse_http_error(str(e))}") - - if self.attempt < self.retries: - logging.error(f"Retry request for URL '{self.url}' (attempt {self.attempt}/{self.retries})") - time.sleep(HTTP_DELAY) - - else: - logging.error(f"Maximum retries reached for URL '{self.url}'") - raise RequestError(str(e)) - - -class ValidateRequest: - """ - Class for validating request inputs. - """ - @staticmethod - def validate_url(url: str) -> bool: - """Validate URL format.""" - - url_regex = re.compile( - r'^(?:http|ftp)s?://' # http:// or https:// - r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+[A-Z]{2,6}\.?|' # domain... - r'localhost|' # localhost... - r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})', re.IGNORECASE) - return re.match(url_regex, url) is not None - - @staticmethod - def validate_headers(headers: Dict[str, str]) -> bool: - """Validate header values.""" - - for key, value in headers.items(): - if not isinstance(key, str) or not isinstance(value, str): - return False - return True - - -class ValidateResponse: - """ - Class for validating response data. - """ - @staticmethod - def is_valid_json(data: str) -> bool: - """Check if response data is a valid JSON.""" - - try: - json.loads(data) - return True - except ValueError: - return False - - -class SSLHandler: - """Class for handling SSL certificates.""" - @staticmethod - def load_certificate(custom_cert_path: str) -> None: - """Load custom SSL certificate.""" - ssl_context = ssl.create_default_context(cafile=custom_cert_path) - ssl_context.check_hostname = False - ssl_context.verify_mode = ssl.CERT_NONE - - -class KwargsRequest(TypedDict, total = False): - url: str - headers: Optional[Dict[str, str]] = None - timeout: float = HTTP_TIMEOUT - retries: int = HTTP_RETRIES - params: Optional[Dict[str, str]] = None - cookies: Optional[Dict[str, str]] = None - verify_ssl: bool = True - json_data: Optional[Dict[str, Any]] = None - - -class Request: - """ - Class for making HTTP requests. - """ - def __init__(self) -> None: - - # Ensure SSL certificate is set up - self.__setup_ssl_certificate__() - - def __setup_ssl_certificate__(self): - """ - Set up SSL certificate environment variables. - """ - try: - # Determine the Python executable - python_executable = sys.executable - logging.info("Python path: ", python_executable) - - # Check if certifi package is installed, install it if not - if subprocess.run([python_executable, "-c", "import certifi"], capture_output=True).returncode != 0: - subprocess.run(["pip", "install", "certifi"], check=True) - logging.info("Installed certifi package.") - - # Get path to SSL certificate - cert_path = subprocess.run([python_executable, "-c", "import certifi; print(certifi.where())"], capture_output=True, text=True, check=True).stdout.strip() - logging.info("Path cert: ", cert_path) - - if not cert_path: - raise ValueError("Unable to determine the path to the SSL certificate.") - - # Set SSL certificate environment variables - os.environ['SSL_CERT_FILE'] = cert_path - os.environ['REQUESTS_CA_BUNDLE'] = cert_path - - except subprocess.CalledProcessError as e: - raise ValueError(f"Error executing subprocess: {e}") from e - - def get(self, url: str, **kwargs: Unpack[KwargsRequest])-> 'Response': - """ - Send a GET request. - - Args: - - url (str): The URL to which the request will be sent. - **kwargs: Additional keyword arguments for the request. - - Returns: - Response: The response object. - """ - return self._send_request(url, 'GET', **kwargs) - - def post(self, url: str, **kwargs: Unpack[KwargsRequest]) -> 'Response': - """ - Send a POST request. - - Args: - - url (str): The URL to which the request will be sent. - **kwargs: Additional keyword arguments for the request. - - Returns: - Response: The response object. - """ - return self._send_request(url, 'POST', **kwargs) - - def head(self, url: str, **kwargs: Unpack[KwargsRequest]) -> 'Response': - """ - Send a HEAD request. - - Args: - - url (str): The URL to which the request will be sent. - **kwargs: Additional keyword arguments for the request. - - Returns: - Response: The response object. - """ - return self._send_request(url, 'HEAD', **kwargs) - - def _send_request(self, url: str, method: str, **kwargs: Unpack[KwargsRequest]) -> 'Response': - """Send an HTTP request.""" - if not ValidateRequest.validate_url(url): - raise ValueError("Invalid URL format") - - if 'headers' in kwargs and not ValidateRequest.validate_headers(kwargs['headers']): - raise ValueError("Invalid header values") - - return ManageRequests(url, method, **kwargs).send() - - -# Output -requests: Request = Request() \ No newline at end of file diff --git a/Src/Lib/UserAgent/__init__.py b/Src/Lib/UserAgent/__init__.py deleted file mode 100644 index b90fa6e..0000000 --- a/Src/Lib/UserAgent/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -# 04.4.24 - -from .user_agent import ua \ No newline at end of file diff --git a/Src/Lib/UserAgent/user_agent.py b/Src/Lib/UserAgent/user_agent.py deleted file mode 100644 index b7c5b74..0000000 --- a/Src/Lib/UserAgent/user_agent.py +++ /dev/null @@ -1,112 +0,0 @@ -# 04.4.24 - -import logging -import re -import os -import random -import threading -import json -import tempfile - -from typing import Dict, List - - -# Internal libraries -from Src.Lib.Request import requests - - - -def get_browser_user_agents_online(browser: str) -> List[str]: - """ - Retrieve browser user agent strings from a website. - - Args: - - browser (str): The name of the browser (e.g., 'chrome', 'firefox', 'safari'). - - Returns: - List[str]: List of user agent strings for the specified browser. - """ - url = f"https://useragentstring.com/pages/{browser}/" - - try: - - # Make request and find all user agents - html = requests.get(url).text - browser_user_agents = re.findall(r"(.+?)", html, re.UNICODE) - return [ua for ua in browser_user_agents if "more" not in ua.lower()] - - except Exception as e: - logging.error(f"Failed to fetch user agents for '{browser}': {str(e)}") - return [] - - -def update_user_agents(browser_name: str, browser_user_agents: Dict[str, List[str]]) -> None: - """ - Update browser user agents dictionary with new requests. - - Args: - - browser_name (str): Name of the browser. - - browser_user_agents (Dict[str, List[str]]): Dictionary to store browser user agents. - """ - browser_user_agents[browser_name] = get_browser_user_agents_online(browser_name) - - -def create_or_update_user_agent_file() -> None: - """ - Create or update the user agent file with browser user agents. - """ - user_agent_file = os.path.join(tempfile.gettempdir(), 'fake_user_agent.json') - logging.info(f"Upload file: {user_agent_file}") - - if not os.path.exists(user_agent_file): - browser_user_agents: Dict[str, List[str]] = {} - threads = [] - - for browser_name in ['chrome', 'firefox', 'safari']: - t = threading.Thread(target=update_user_agents, args=(browser_name, browser_user_agents)) - threads.append(t) - t.start() - - for t in threads: - t.join() - - with open(user_agent_file, 'w') as f: - json.dump(browser_user_agents, f, indent=4) - logging.info(f"User agent file created at: {user_agent_file}") - - else: - logging.info("User agent file already exists.") - - -class UserAgentManager: - """ - Manager class to access browser user agents from a file. - """ - def __init__(self): - - # Get path to temp file where save all user agents - self.user_agent_file = os.path.join(tempfile.gettempdir(), 'fake_user_agent.json') - logging.info(f"Check file: {self.user_agent_file}") - - # If file dont exist, creaet it - if not os.path.exists(self.user_agent_file): - create_or_update_user_agent_file() - logging.info(f"Create file: {self.user_agent_file}") - - def get_random_user_agent(self, browser: str) -> str: - """ - Get a random user agent for the specified browser. - - Args: - browser (str): The name of the browser ('chrome', 'firefox', 'safari'). - - Returns: - Optional[str]: Random user agent string for the specified browser. - """ - with open(self.user_agent_file, 'r') as f: - browser_user_agents = json.load(f) - return random.choice(browser_user_agents.get(browser.lower(), [])) - - -# Output -ua: UserAgentManager = UserAgentManager() \ No newline at end of file diff --git a/Src/Upload/update.py b/Src/Upload/update.py index 7c6e4fa..6e3e543 100644 --- a/Src/Upload/update.py +++ b/Src/Upload/update.py @@ -10,7 +10,7 @@ from Src.Util.console import console # External library -from Src.Lib.Request import requests +import requests # Variable diff --git a/Src/Util/headers.py b/Src/Util/headers.py index ca76388..b84f4e9 100644 --- a/Src/Util/headers.py +++ b/Src/Util/headers.py @@ -3,8 +3,12 @@ import logging -# Internal utilities -from Src.Lib.UserAgent import ua +# External library +import fake_useragent + + +# Variable +useragent = fake_useragent.UserAgent() def get_headers() -> str: @@ -12,11 +16,8 @@ def get_headers() -> str: Generate a random user agent to use in HTTP requests. Returns: - str: A random user agent string. + - str: A random user agent string. """ # Get a random user agent string from the user agent rotator - random_headers = ua.get_random_user_agent("firefox") - - #logging.info(f"Use headers: {random_headers}") - return random_headers \ No newline at end of file + return useragent.firefox \ No newline at end of file diff --git a/Test/t_get_server_ip_sc.py b/Test/t_get_server_ip_sc.py index 02bebe8..d89ebac 100644 --- a/Test/t_get_server_ip_sc.py +++ b/Test/t_get_server_ip_sc.py @@ -11,7 +11,7 @@ warnings.filterwarnings("ignore", category=urllib3.exceptions.InsecureRequestWar # Variable -url_test = "https://sc-b1-18.scws-content.net/hls/170/3/25/32503b5b-4646-4376-ad47-7766c65be7e2/audio/ita/0004-0100.ts" +url_test = "https://sc-b1-18.scws-content.net/hls/100/b/d3/bd3a430d-0a13-4bec-8fcc-ea41af183555/audio/ita/0010-0100.ts?token=CiEPTIyvEoTkGk3szgDu9g&expires=1722801022" def get_ip_from_url(url): @@ -81,5 +81,6 @@ def main(): print(f"Valid IP addresses: {sorted(valid_ip, reverse=True)}") + if __name__ == '__main__': main() diff --git a/requirements.txt b/requirements.txt index 3f1f64e..d9a07e1 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,7 @@ -bs4 +requests +bs4 certifi tqdm rich -unidecode \ No newline at end of file +unidecode +fake-useragent \ No newline at end of file