162 lines
3.8 KiB
Python
162 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
工具函数模块
|
|
提供项目所需的通用工具函数
|
|
"""
|
|
|
|
import socket
|
|
import re
|
|
from typing import Optional
|
|
|
|
|
|
def validate_ip_address(ip: str) -> bool:
|
|
"""
|
|
验证IP地址格式是否正确
|
|
|
|
Args:
|
|
ip: IP地址字符串
|
|
|
|
Returns:
|
|
bool: 如果IP地址格式正确返回True,否则返回False
|
|
"""
|
|
# IPv4正则表达式
|
|
ipv4_pattern = r'^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$'
|
|
|
|
# IPv6正则表达式(简化版)
|
|
ipv6_pattern = r'^(?:[0-9a-fA-F]{1,4}:){7}[0-9a-fA-F]{1,4}$|^::$|^:(?::)+(?:[0-9a-fA-F]{1,4})?$'
|
|
|
|
return bool(re.match(ipv4_pattern, ip) or re.match(ipv6_pattern, ip))
|
|
|
|
|
|
def validate_hostname(hostname: str) -> bool:
|
|
"""
|
|
验证主机名格式是否正确
|
|
|
|
Args:
|
|
hostname: 主机名字符串
|
|
|
|
Returns:
|
|
bool: 如果主机名格式正确返回True,否则返回False
|
|
"""
|
|
if len(hostname) > 253:
|
|
return False
|
|
|
|
if hostname[-1] == ".":
|
|
hostname = hostname[:-1]
|
|
|
|
allowed = re.compile(r"^(?!-)[A-Z0-9-]{1,63}(?<!-)$", re.IGNORECASE)
|
|
return all(allowed.match(label) for label in hostname.split("."))
|
|
|
|
|
|
def resolve_hostname(hostname: str) -> Optional[str]:
|
|
"""
|
|
解析主机名为IP地址
|
|
|
|
Args:
|
|
hostname: 主机名
|
|
|
|
Returns:
|
|
IP地址字符串,解析失败返回None
|
|
"""
|
|
try:
|
|
ip_address = socket.gethostbyname(hostname)
|
|
return ip_address
|
|
except socket.gaierror:
|
|
return None
|
|
|
|
|
|
def is_port_open(host: str, port: int, timeout: float = 2.0) -> bool:
|
|
"""
|
|
检查指定主机的端口是否开放
|
|
|
|
Args:
|
|
host: 主机地址
|
|
port: 端口号
|
|
timeout: 超时时间(秒)
|
|
|
|
Returns:
|
|
bool: 端口开放返回True,否则返回False
|
|
"""
|
|
try:
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
sock.settimeout(timeout)
|
|
result = sock.connect_ex((host, port))
|
|
sock.close()
|
|
return result == 0
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def format_bytes(bytes_size: int) -> str:
|
|
"""
|
|
格式化字节大小为可读字符串
|
|
|
|
Args:
|
|
bytes_size: 字节数
|
|
|
|
Returns:
|
|
格式化后的字符串
|
|
"""
|
|
for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
|
|
if bytes_size < 1024.0:
|
|
return f"{bytes_size:.2f} {unit}"
|
|
bytes_size /= 1024.0
|
|
return f"{bytes_size:.2f} PB"
|
|
|
|
|
|
def format_duration(seconds: float) -> str:
|
|
"""
|
|
格式化时长为可读字符串
|
|
|
|
Args:
|
|
seconds: 秒数
|
|
|
|
Returns:
|
|
格式化后的字符串
|
|
"""
|
|
if seconds < 60:
|
|
return f"{seconds:.1f}秒"
|
|
elif seconds < 3600:
|
|
minutes = seconds / 60
|
|
return f"{minutes:.1f}分钟"
|
|
elif seconds < 86400:
|
|
hours = seconds / 3600
|
|
return f"{hours:.1f}小时"
|
|
else:
|
|
days = seconds / 86400
|
|
return f"{days:.1f}天"
|
|
|
|
|
|
def safe_divide(numerator: float, denominator: float, default: float = 0.0) -> float:
|
|
"""
|
|
安全除法,避免除以零
|
|
|
|
Args:
|
|
numerator: 分子
|
|
denominator: 分母
|
|
default: 除零时的默认返回值
|
|
|
|
Returns:
|
|
除法结果或默认值
|
|
"""
|
|
try:
|
|
return numerator / denominator if denominator != 0 else default
|
|
except Exception:
|
|
return default
|
|
|
|
|
|
def clamp(value: float, min_value: float, max_value: float) -> float:
|
|
"""
|
|
将数值限制在指定范围内
|
|
|
|
Args:
|
|
value: 输入值
|
|
min_value: 最小值
|
|
max_value: 最大值
|
|
|
|
Returns:
|
|
限制后的值
|
|
"""
|
|
return max(min_value, min(value, max_value))
|