diff --git a/douyinliverecorder/proxy.py b/douyinliverecorder/proxy.py new file mode 100644 index 0000000..28e3148 --- /dev/null +++ b/douyinliverecorder/proxy.py @@ -0,0 +1,92 @@ +import os +import sys +from enum import Enum, auto +from dataclasses import dataclass, field +from .utils import logger + + +class ProxyType(Enum): + HTTP = auto() + HTTPS = auto() + SOCKS = auto() + + +@dataclass(frozen=True) +class ProxyInfo: + ip: str = field(default="", repr=True) + port: str = field(default="", repr=True) + + def __post_init__(self): + if (self.ip and not self.port) or (not self.ip and self.port): + raise ValueError("IP or port cannot be empty") + + if (self.ip and self.port) and (not self.port.isdigit() or not (1 <= int(self.port) <= 65535)): + raise ValueError("Port must be a digit between 1 and 65535") + + +class ProxyDetector: + def __init__(self): + if sys.platform.startswith('win'): + import winreg + self.winreg = winreg + self.__path = r'Software\Microsoft\Windows\CurrentVersion\Internet Settings' + with winreg.ConnectRegistry(None, winreg.HKEY_CURRENT_USER) as key_user: + self.__INTERNET_SETTINGS = winreg.OpenKeyEx(key_user, self.__path, 0, winreg.KEY_ALL_ACCESS) + else: + self.__is_windows = False + + def get_proxy_info(self) -> ProxyInfo: + if sys.platform.startswith('win'): + ip, port = self._get_proxy_info_windows() + else: + ip, port = self._get_proxy_info_linux() + return ProxyInfo(ip, port) + + def is_proxy_enabled(self) -> bool: + if sys.platform.startswith('win'): + return self._is_proxy_enabled_windows() + else: + return self._is_proxy_enabled_linux() + + def _get_proxy_info_windows(self) -> tuple[str, str]: + ip, port = "", "" + if self._is_proxy_enabled_windows(): + try: + ip_port = self.winreg.QueryValueEx(self.__INTERNET_SETTINGS, "ProxyServer")[0] + if ip_port: + ip, port = ip_port.split(":") + except FileNotFoundError as err: + logger.warning("No proxy information found: " + str(err)) + except Exception as err: + logger.error("An error occurred: " + str(err)) + else: + logger.debug("No proxy is enabled on the system") + return ip, port + + def _is_proxy_enabled_windows(self) -> bool: + try: + if self.winreg.QueryValueEx(self.__INTERNET_SETTINGS, "ProxyEnable")[0] == 1: + return True + except FileNotFoundError as err: + print("No proxy information found: " + str(err)) + except Exception as err: + print("An error occurred: " + str(err)) + return False + + @staticmethod + def _get_proxy_info_linux() -> tuple[str, str]: + proxies = { + 'http': os.getenv('http_proxy'), + 'https': os.getenv('https_proxy'), + 'ftp': os.getenv('ftp_proxy') + } + ip = port = "" + for proto, proxy in proxies.items(): + if proxy: + ip, port = proxy.split(':') + break + return ip, port + + def _is_proxy_enabled_linux(self) -> bool: + proxies = self._get_proxy_info_linux() + return any(proxy != '' for proxy in proxies) diff --git a/main.py b/main.py index 7280426..ae3e4ee 100644 --- a/main.py +++ b/main.py @@ -27,8 +27,8 @@ import urllib.request from urllib.error import URLError, HTTPError from typing import Any import configparser -from douyinliverecorder import spider -from douyinliverecorder import stream +from douyinliverecorder import spider, stream +from douyinliverecorder.proxy import ProxyDetector from douyinliverecorder.utils import logger from douyinliverecorder import utils from msg_push import ( @@ -95,7 +95,10 @@ def display_info() -> None: print(f"是否开启代理录制: {'是' if use_proxy else '否'}", end=" | ") if split_video_by_time: print(f"录制分段开启: {split_time}秒", end=" | ") - print(f"是否生成时间文件: {'是' if create_time_file else '否'}", end=" | ") + else: + print(f"录制分段开启: 否", end=" | ") + if create_time_file: + print(f"是否生成时间文件: 是", end=" | ") print(f"录制视频质量为: {video_record_quality}", end=" | ") print(f"录制视频格式为: {video_save_type}", end=" | ") print(f"目前瞬时错误数为: {error_count}", end=" | ") @@ -1514,6 +1517,10 @@ try: response_g = urllib.request.urlopen("https://www.google.com/", timeout=15) global_proxy = True print('\r全局/规则网络代理已开启√') + pd = ProxyDetector() + if pd.is_proxy_enabled(): + proxy_info = pd.get_proxy_info() + print("System Proxy: http://{}:{}".format(proxy_info.ip, proxy_info.port)) except HTTPError as err: print(f"HTTP error occurred: {err.code} - {err.reason}") except URLError as err: @@ -1664,7 +1671,7 @@ while True: logger.warning(f"Disk space remaining is below {disk_space_limit} GB. " f"Exiting program due to the disk space limit being reached.") sys.exit(-1) - + print("") def contains_url(string: str) -> bool: pattern = r"(https?://)?(www\.)?[a-zA-Z0-9-]+(\.[a-zA-Z0-9-]+)+(:\d+)?(/.*)?" @@ -1672,10 +1679,7 @@ while True: try: - url_comments = [] - line_list = [] - url_line_list = [] - + url_comments, line_list, url_line_list = [[] for _ in range(3)] with (open(url_config_file, "r", encoding=text_encoding, errors='ignore') as file): for origin_line in file: if origin_line in line_list: @@ -1868,3 +1872,4 @@ while True: first_run = False time.sleep(3) + \ No newline at end of file