86 lines
2.5 KiB
Python
86 lines
2.5 KiB
Python
import requests
|
|
import json
|
|
import yaml
|
|
import logging
|
|
import time
|
|
from pathlib import Path
|
|
|
|
def download_file(url, save_path, timeout=30):
|
|
"""下载文件"""
|
|
try:
|
|
response = requests.get(url, timeout=timeout, stream=True)
|
|
response.raise_for_status()
|
|
|
|
with open(save_path, 'wb') as f:
|
|
for chunk in response.iter_content(chunk_size=8192):
|
|
f.write(chunk)
|
|
|
|
return True
|
|
except Exception as e:
|
|
logging.error(f"Failed to download file from {url}: {str(e)}")
|
|
return False
|
|
|
|
def fetch_json(url, timeout=15):
|
|
"""获取JSON数据"""
|
|
try:
|
|
response = requests.get(url, timeout=timeout)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
logging.error(f"Failed to fetch JSON from {url}: {str(e)}")
|
|
return None
|
|
|
|
def fetch_yaml(url, timeout=15):
|
|
"""获取YAML数据"""
|
|
try:
|
|
response = requests.get(url, timeout=timeout)
|
|
response.raise_for_status()
|
|
return yaml.safe_load(response.text)
|
|
except Exception as e:
|
|
logging.error(f"Failed to fetch YAML from {url}: {str(e)}")
|
|
return None
|
|
|
|
def check_connectivity(host="8.8.8.8", port=53, timeout=3):
|
|
"""检查网络连接"""
|
|
import socket
|
|
|
|
try:
|
|
socket.setdefaulttimeout(timeout)
|
|
socket.socket(socket.AF_INET, socket.SOCK_STREAM).connect((host, port))
|
|
return True
|
|
except socket.error:
|
|
return False
|
|
|
|
def get_public_ip():
|
|
"""获取公网IP"""
|
|
try:
|
|
response = requests.get('https://api.ipify.org?format=json', timeout=10)
|
|
response.raise_for_status()
|
|
return response.json()['ip']
|
|
except Exception as e:
|
|
logging.error(f"Failed to get public IP: {str(e)}")
|
|
return None
|
|
|
|
def measure_speedtest():
|
|
"""简单测速"""
|
|
try:
|
|
# 下载测试
|
|
start_time = time.time()
|
|
response = requests.get('https://speedtest.tele2.net/10MB.zip', timeout=30, stream=True)
|
|
response.raise_for_status()
|
|
|
|
total_bytes = 0
|
|
for chunk in response.iter_content(chunk_size=8192):
|
|
total_bytes += len(chunk)
|
|
|
|
elapsed_time = time.time() - start_time
|
|
download_speed = (total_bytes * 8) / (elapsed_time * 1024 * 1024) # Mbps
|
|
|
|
return {
|
|
'download_speed': download_speed,
|
|
'elapsed_time': elapsed_time,
|
|
'total_bytes': total_bytes
|
|
}
|
|
except Exception as e:
|
|
logging.error(f"Speed test failed: {str(e)}")
|
|
return None |