feat(transmission): add protocol support

This commit is contained in:
InfinityPacer
2025-01-07 00:51:07 +08:00
parent d7300777cb
commit 430702abd3
2 changed files with 50 additions and 10 deletions

View File

@@ -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