下行消耗脚本
73
0
0
简单记录一下
初始环境
python3.11 -m venv venv
./venv/bin/pip install -i http://mirrors.aliyun.com/pypi/simple --trusted-host curl_cffi
./venv/bin/python ./speed.py
speed.py 代码如下
import os
import json
import time
import random
import string
import threading
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
from curl_cffi import requests
from rich.progress import (
Progress,
BarColumn,
DownloadColumn,
TransferSpeedColumn,
TimeRemainingColumn,
)
from rich.console import Console
from rich.live import Live
# ==================== 测试配置 ====================
URL = "https://www.lava3dp.com/wp-content/uploads/stl_files_uploads/ul.stl"
MAX_WORKERS = 10 # 同时保持的并发线程数（根据您的带宽调整）
FEISHU_WEBHOOK_URL = "https://open.feishu.cn/open-apis/bot/v2/hook/xxxxxxxxx"
SYNC_INTERVAL_SECONDS = 3600 # 飞书同步间隔：1小时
STATS_FILE = os.path.join(os.path.dirname(os.path.abspath('.')), "speedtest_stats.json")
# ==================================================
console = Console()
def load_stats():
"""从本地文件加载累计流量统计"""
if os.path.exists(STATS_FILE):
try:
with open(STATS_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except (json.JSONDecodeError, IOError):
pass
return {
"total_bytes": 0,
"total_tasks": 0,
"start_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"last_sync_time": None,
}
def save_stats(stats):
"""将累计流量统计写入本地文件（调用者需持有锁）"""
return
try:
with open(STATS_FILE, "w", encoding="utf-8") as f:
json.dump(stats, f, ensure_ascii=False, indent=2)
except IOError as e:
console.print(f"[red]写入统计文件失败: {e}[/red]")
def accumulate_bytes(bytes_count, stats, stats_lock):
"""累加下载字节数并持久化到本地文件"""
with stats_lock:
stats["total_bytes"] += bytes_count
stats["total_tasks"] += 1
save_stats(stats)
def send_feishu_notification(stats):
"""发送飞书机器人通知，汇报累计流量"""
total_mb = stats["total_bytes"] / (1024 * 1024)
total_gb = total_mb / 1024
now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
if total_gb >= 1:
traffic_display = f"**{total_gb:.2f} GB**"
else:
traffic_display = f"**{total_mb:.2f} MB**"
card = {
"msg_type": "interactive",
"card": {
"header": {
"title": {
"tag": "plain_text",
"content": "📊 带宽测试流量同步报告",
},
"template": "blue",
},
"elements": [
{
"tag": "div",
"fields": [
{
"is_short": True,
"text": {
"tag": "lark_md",
"content": f"**累计流量：**\n{traffic_display}",
},
},
{
"is_short": True,
"text": {
"tag": "lark_md",
"content": f"**累计任务数：**\n{stats['total_tasks']}",
},
},
{
"is_short": True,
"text": {
"tag": "lark_md",
"content": f"**测试开始时间：**\n{stats['start_time']}",
},
},
{
"is_short": True,
"text": {
"tag": "lark_md",
"content": f"**本次同步时间：**\n{now_str}",
},
},
],
},
{"tag": "hr"},
{
"tag": "note",
"elements": [
{
"tag": "plain_text",
"content": f"数据文件: {STATS_FILE}",
}
],
},
],
},
}
try:
resp = requests.post(
FEISHU_WEBHOOK_URL,
json=card,
impersonate="chrome",
timeout=10,
)
if resp.status_code == 200:
result = resp.json()
if result.get("code") == 0:
console.print(f"[green][飞书同步] 成功 | 累计流量: {traffic_display} | 时间: {now_str}[/green]")
else:
console.print(f"[red][飞书同步] 接口返回错误: {result}[/red]")
else:
console.print(f"[red][飞书同步] HTTP {resp.status_code}[/red]")
except Exception as e:
console.print(f"[red][飞书同步] 请求失败: {e}[/red]")
def feishu_sync_daemon(stats, stats_lock, stop_event):
"""后台守护线程：每小时同步一次累计流量到飞书机器人"""
while not stop_event.is_set():
if stop_event.wait(timeout=SYNC_INTERVAL_SECONDS):
break
with stats_lock:
stats_copy = dict(stats)
stats["last_sync_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
save_stats(stats)
send_feishu_notification(stats_copy)
def generate_random_params():
"""生成 1 到 5 个随机防缓存参数"""
params = {}
for _ in range(random.randint(1, 5)):
key = ''.join(random.choices(string.ascii_lowercase, k=5))
val = ''.join(random.choices(string.digits, k=5))
params[key] = val
return params
def generate_random_ip():
"""生成一个随机的公网 IPv4 地址（排除私有/保留段）"""
while True:
ip = f"{random.randint(1, 223)}.{random.randint(0, 255)}.{random.randint(0, 255)}.{random.randint(1, 254)}"
first = int(ip.split('.')[0])
second = int(ip.split('.')[1])
# 排除私有地址段：10.x.x.x / 172.16-31.x.x / 192.168.x.x / 127.x.x.x
if first == 10 or first == 127:
continue
if first == 172 and 16 <= second <= 31:
continue
if first == 192 and second == 168:
continue
return ip
def generate_fake_ip_headers():
"""
生成伪装真实 IP 的请求头（与 wp-config.php 中解析逻辑对应）
优先级顺序：CF-Connecting-IP > X-Forwarded-For > X-Real-IP > X-Forwarded > Client-IP
"""
real_ip = generate_random_ip()
headers = {
"CF-Connecting-IP": real_ip,
"X-Forwarded-For": f"{real_ip}, {generate_random_ip()}",
"X-Real-IP": real_ip,
"X-Forwarded": real_ip,
"Client-IP": real_ip,
}
return headers
def download_worker(task_id, progress, task_lock, stats, stats_lock):
"""单个下载任务：请求文件、流式读取、丢弃数据，并实时更新进度条"""
params = generate_random_params()
headers = generate_fake_ip_headers()
downloaded_bytes = 0
progress_task_id = progress.add_task(
f"[cyan]任务 {task_id}",
total=None,
start=False,
)
try:
response = requests.post(
URL,
params=params,
# headers=headers,
impersonate="chrome",
stream=True,
timeout=15,
)
if response.status_code == 200:
total_size = int(response.headers.get('content-length', 0))
progress.start_task(progress_task_id)
progress.update(progress_task_id, total=total_size if total_size > 0 else None)
for chunk in response.iter_content(chunk_size=1024 * 64):
if chunk:
downloaded_bytes += len(chunk)
progress.update(progress_task_id, completed=downloaded_bytes)
if downloaded_bytes > 0:
progress.update(
progress_task_id,
description=f"[green]任务 {task_id} ✓",
completed=downloaded_bytes,
total=downloaded_bytes,
)
# 累加到本地持久化统计
accumulate_bytes(downloaded_bytes, stats, stats_lock)
return downloaded_bytes
else:
progress.update(
progress_task_id,
description=f"[red]任务 {task_id} ✗ (HTTP {response.status_code})",
)
except Exception as e:
progress.update(
progress_task_id,
description=f"[red]任务 {task_id} ✗ 错误",
)
finally:
time.sleep(0.5)
with task_lock:
try:
progress.remove_task(progress_task_id)
except Exception:
pass
return 0
def start_infinite_test():
stats = load_stats()
stats_lock = threading.Lock()
# 启动飞书定时同步守护线程
stop_event = threading.Event()
sync_thread = threading.Thread(
target=feishu_sync_daemon,
args=(stats, stats_lock, stop_event),
daemon=True,
)
sync_thread.start()
prev_total_mb = stats["total_bytes"] / (1024 * 1024)
console.print("=" * 60, style="bold blue")
console.print("开始【无限循环版】多线程带宽稳定性测试...", style="bold green")
console.print(f"当前并发线程数: [yellow]{MAX_WORKERS}[/yellow] | 底层伪装: [yellow]Chrome 浏览器[/yellow]")
console.print(f"历史累计流量: [cyan]{prev_total_mb:.2f} MB[/cyan] | 飞书同步间隔: [cyan]1小时[/cyan]")
console.print(f"统计文件: [dim]{STATS_FILE}[/dim]")
console.print("数据直接从内存释放（不写磁盘）。按下 [red]Ctrl+C[/red] 可随时停止测试。")
console.print("=" * 60)
console.print()
global_start_time = time.time()
session_downloaded_bytes = 0
task_counter = 0
task_lock = threading.Lock()
progress = Progress(
"[progress.description]{task.description}",
BarColumn(bar_width=30),
"[progress.percentage]{task.percentage:>3.0f}%",
DownloadColumn(),
TransferSpeedColumn(),
TimeRemainingColumn(),
console=console,
transient=False,
refresh_per_second=10,
)
with Live(progress, console=console, refresh_per_second=10):
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
futures = {}
for i in range(1, MAX_WORKERS + 1):
future = executor.submit(download_worker, i, progress, task_lock, stats, stats_lock)
futures[future] = i
task_counter = MAX_WORKERS
try:
while futures:
for future in as_completed(futures):
bytes_downloaded = future.result()
session_downloaded_bytes += bytes_downloaded
futures.pop(future)
task_counter += 1
new_future = executor.submit(download_worker, task_counter, progress, task_lock, stats, stats_lock)
futures[new_future] = task_counter
break
except KeyboardInterrupt:
console.print("\n\n[yellow]正在接收终止信号，请稍候（等待当前进行中的线程安全退出）...[/yellow]")
for f in futures:
f.cancel()
# 停止飞书同步线程
stop_event.set()
sync_thread.join(timeout=3)
# 退出前触发一次飞书同步，确保最新数据及时上报
with stats_lock:
stats_copy = dict(stats)
send_feishu_notification(stats_copy)
total_duration = time.time() - global_start_time
session_mb = session_downloaded_bytes / (1024 * 1024)
total_mb_all = stats["total_bytes"] / (1024 * 1024)
avg_speed = session_mb / total_duration if total_duration > 0 else 0
console.print()
console.print("=" * 60, style="bold blue")
console.print("[bold]最终测试报告[/bold]", style="bold green")
console.print("=" * 60, style="bold blue")
console.print(f"本次测试时长: [cyan]{total_duration:.2f}[/cyan] 秒")
console.print(f"本次下载流量: [cyan]{session_mb:.2f}[/cyan] MB")
console.print(f"本次平均吞吐: [cyan]{avg_speed:.2f}[/cyan] MB/s (约 [cyan]{avg_speed*8:.2f}[/cyan] Mbps)")
console.print(f"本次完成任务: [cyan]{stats['total_tasks']}[/cyan] 次")
console.print(f"历史累计流量: [cyan]{total_mb_all:.2f}[/cyan] MB")
console.print("=" * 60, style="bold blue")
if __name__ == "__main__":
start_infinite_test()