diff --git a/app/modules/transmission/transmission.py b/app/modules/transmission/transmission.py index 79b91d6a..a7075f6a 100644 --- a/app/modules/transmission/transmission.py +++ b/app/modules/transmission/transmission.py @@ -1,14 +1,15 @@ -from typing import Optional, Union, Tuple, List +from typing import Optional, Union, Tuple, List, Literal import transmission_rpc from transmission_rpc import Client, Torrent, File from transmission_rpc.session import SessionStats, Session from app.log import logger -from app.utils.string import StringUtils +from app.utils.url import UrlUtils class Transmission: + _protocol: Literal["http", "https"] = "http" _host: str = None _port: int = None _username: str = None @@ -28,9 +29,14 @@ class Transmission: 若不设置参数,则创建配置文件设置的下载器 """ if host and port: - self._host, self._port = host, port + self._protocol, self._host, self._port = kwargs.get("protocol", self._protocol), host, port elif host: - self._host, self._port = StringUtils.get_domain_address(address=host, prefix=False) + result = UrlUtils.parse_url_params(url=host) + if result: + self._protocol, self._host, self._port, path = result + else: + logger.error("Transmission配置不正确!") + return else: logger.error("Transmission配置不完整!") return @@ -46,8 +52,9 @@ class Transmission: """ try: # 登录 - logger.info(f"正在连接 transmission:{self._host}:{self._port}") - trt = transmission_rpc.Client(host=self._host, + logger.info(f"正在连接 transmission:{self._protocol}://{self._host}:{self._port}") + trt = transmission_rpc.Client(protocol=self._protocol, + host=self._host, port=self._port, username=self._username, password=self._password, diff --git a/app/utils/url.py b/app/utils/url.py index 5e932332..3ef6333a 100644 --- a/app/utils/url.py +++ b/app/utils/url.py @@ -1,6 +1,6 @@ import mimetypes from pathlib import Path -from typing import Optional, Union +from typing import Optional, Union, Tuple from urllib import parse from urllib.parse import parse_qs, urlencode, urljoin, urlparse, urlunparse @@ -27,7 +27,7 @@ class UrlUtils: @staticmethod def adapt_request_url(host: str, endpoint: str) -> Optional[str]: """ - 基于传入的host,适配请求的URL,确保每个请求的URL是完整的,用于在发送请求前自动处理和修正请求的URL。 + 基于传入的host,适配请求的URL,确保每个请求的URL是完整的,用于在发送请求前自动处理和修正请求的URL :param host: 主机头 :param endpoint: 端点 :return: 完整的请求URL字符串 @@ -42,7 +42,7 @@ class UrlUtils: @staticmethod def combine_url(host: str, path: Optional[str] = None, query: Optional[dict] = None) -> Optional[str]: """ - 使用给定的主机头、路径和查询参数组合生成完整的URL。 + 使用给定的主机头、路径和查询参数组合生成完整的URL :param host: str, 主机头,例如 https://example.com :param path: Optional[str], 包含路径和可能已经包含的查询参数的端点,例如 /path/to/resource?current=1 :param query: Optional[dict], 可选,额外的查询参数,例如 {"key": "value"} @@ -101,9 +101,42 @@ class UrlUtils: def quote(s: str) -> str: """ 将字符串编码为 URL 安全的格式 - 这将确保路径中的特殊字符(如空格、中文字符等)被正确编码,以便在 URL 中传输 :param s: 要编码的字符串 :return: 编码后的字符串 """ return parse.quote(s) + + @staticmethod + def parse_url_params(url: str) -> Optional[Tuple[str, str, int, str]]: + """ + 解析给定的 URL,并提取协议、主机名、端口和路径信息 + + :param url: str + 需要解析的 URL 字符串 + 可以是完整的 URL(例如:"http://example.com:8080/path")或不带协议的地址(例如:"example.com:1234") + :return: Optional[Tuple[str, str, int, str]] + - str: 协议(例如:"http", "https") + - str: 主机名或 IP 地址(例如:"example.com", "192.168.1.1") + - int: 端口号(例如:80, 443) + - str: URL 的路径部分(例如:"/", "/path") + 如果输入地址无效或无法解析,则返回 None + """ + try: + if not url: + return None + + url = UrlUtils.standardize_base_url(host=url) + parsed = urlparse(url) + + if not parsed.hostname: + return None + protocol = parsed.scheme + hostname = parsed.hostname + port = parsed.port or (443 if protocol == "https" else 80) + path = parsed.path or "/" + + return protocol, hostname, port, path + except Exception as e: + logger.debug(f"Error parse_url_params: {e}") + return None