别再傻傻用网页测速了！用Python的speedtest库写个自动测速脚本，还能定时发报告
用Python构建智能网络质量监控系统：从测速到自动化报告
家里WiFi时快时慢？办公室网络总在关键时刻卡顿？别再依赖手动网页测速了。作为一位常年被网络问题困扰的技术爱好者，我花了三个月时间迭代出一个全自动的网络质量监控方案，现在每天早晨都能在钉钉收到前一天的网速报告，还能自动生成周报对比运营商是否达标。本文将分享如何用Python的speedtest-cli库为核心，打造一个零人工干预的智能监控系统。
1. 环境准备与基础测速
在开始构建自动化系统前，我们需要先确保基础测速功能正常工作。推荐使用Python 3.8+环境，这是我在多个项目中验证过的最稳定版本。
安装必备库：
pip install speedtest-cli pandas schedule requests基础测速脚本是系统的核心传感器，以下是我优化过的测速函数，增加了异常处理和单位转换：
import speedtest from datetime import datetime def network_benchmark(): try: st = speedtest.Speedtest() st.get_best_server() # 测试配置调优 st.download(threads=4) # 多线程提升测试准确度 st.upload(threads=4, pre_allocate=False) return { "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "download": round(st.results.download / 1e6, 2), # Mbps "upload": round(st.results.upload / 1e6, 2), "ping": round(st.results.ping, 2), "server": st.results.server["host"] } except Exception as e: print(f"测速失败: {str(e)}") return None这个版本相比基础实现有三个改进：
- 明确指定测试线程数，避免默认设置可能导致的资源争用
- 关闭上传测试的内存预分配，减少对小内存设备的压力
- 返回结构化数据方便后续处理
2. 数据存储方案设计
单纯的实时测速价值有限，我们需要历史数据来分析趋势。以下是三种经过验证的存储方案：
| 方案类型 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| CSV文件 | 零配置，易查看 | 大数据量时查询慢 | 个人家庭使用 |
| SQLite | 轻量级，支持SQL查询 | 需要基础数据库知识 | 小型办公室 |
| InfluxDB | 专业时序数据库 | 需要额外安装服务 | 企业级监控 |
我最终选择了SQLite方案，因为它完美平衡了易用性和功能性。以下是数据库操作模块：
import sqlite3 from contextlib import closing def init_db(db_path="network.db"): with closing(sqlite3.connect(db_path)) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS speedtests ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp DATETIME NOT NULL, download REAL NOT NULL, upload REAL NOT NULL, ping REAL NOT NULL, server TEXT NOT NULL ) """) conn.commit() def save_result(result, db_path="network.db"): if not result: return with closing(sqlite3.connect(db_path)) as conn: conn.execute(""" INSERT INTO speedtests (timestamp, download, upload, ping, server) VALUES (?, ?, ?, ?, ?) """, (result["timestamp"], result["download"], result["upload"], result["ping"], result["server"])) conn.commit()重要提示：数据库文件建议放在持久化存储位置，避免被系统清理。我在/var/lib/network_monitor/目录下存放数据库，并设置了自动备份。
3. 定时任务与自动化
实现定时测速有两种主流方案，各有适用场景：
- schedule库：纯Python实现，适合需要精细控制的任务
- 系统crontab：操作系统级定时，适合长期稳定运行的服务
我推荐开发阶段使用schedule便于调试，部署后改用crontab。以下是schedule的实现示例：
import schedule import time def job(): result = network_benchmark() if result: save_result(result) print(f"测试完成: {result}") # 每30分钟测试一次 schedule.every(30).minutes.do(job) while True: schedule.run_pending() time.sleep(1)生产环境建议使用systemd服务管理crontab任务。创建/etc/systemd/system/network-monitor.service：
[Unit] Description=Network Speed Monitor After=network.target [Service] ExecStart=/usr/bin/python3 /opt/network_monitor/main.py Restart=always User=root [Install] WantedBy=multi-user.target配合crontab定时任务：
*/30 * * * * /usr/bin/systemctl start network-monitor4. 可视化与报警系统
原始数据需要转化为可操作的洞察。我开发了三种报告形式：
1. 终端实时看板
import pandas as pd def show_realtime(db_path="network.db"): with closing(sqlite3.connect(db_path)) as conn: df = pd.read_sql(""" SELECT datetime(timestamp) as time, download, upload, ping FROM speedtests ORDER BY time DESC LIMIT 10 """, conn) print("\n最新10次测速结果:") print(df.to_string(index=False))2. 邮件日报模板
import smtplib from email.mime.text import MIMEText def send_daily_report(): # 获取24小时数据 with closing(sqlite3.connect(db_path)) as conn: stats = pd.read_sql(""" SELECT AVG(download) as avg_down, MIN(download) as min_down, AVG(upload) as avg_up, AVG(ping) as avg_ping FROM speedtests WHERE timestamp > datetime('now', '-1 day') """, conn).iloc[0].to_dict() # 生成HTML报告 html = f""" <h3>网络质量日报 {datetime.now().date()}</h3> <table border="1"> <tr><th>指标</th><th>平均值</th><th>最低值</th></tr> <tr><td>下载速度(Mbps)</td><td>{stats['avg_down']:.2f}</td><td>{stats['min_down']:.2f}</td></tr> <tr><td>上传速度(Mbps)</td><td>{stats['avg_up']:.2f}</td><td>-</td></tr> <tr><td>延迟(ms)</td><td>{stats['avg_ping']:.2f}</td><td>-</td></tr> </table> """ # 发送邮件逻辑 msg = MIMEText(html, 'html') msg['Subject'] = f"网络质量报告 {datetime.now().date()}" msg['From'] = "monitor@example.com" msg['To'] = "admin@example.com" with smtplib.SMTP('smtp.example.com', 587) as server: server.login('user', 'password') server.send_message(msg)3. 钉钉机器人报警
当网络指标低于阈值时自动触发：
def dingding_alert(message): webhook = "https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN" headers = {"Content-Type": "application/json"} data = { "msgtype": "text", "text": { "content": f"网络异常报警:\n{message}" } } requests.post(webhook, headers=headers, json=data)5. 高级功能扩展
基础系统运行稳定后，我陆续添加了这些增强功能：
多服务器测试对比
def multi_server_test(): st = speedtest.Speedtest() servers = st.get_servers() results = [] for server in servers[:3]: # 测试前三个最佳服务器 st.get_best_server([server["id"]]) res = network_benchmark() res["server_name"] = server["name"] results.append(res) return results网络质量评分算法
def calculate_score(download, upload, ping): # 基于实际业务需求调整权重 download_score = min(download / 100, 1.0) * 0.5 # 100Mbps为满分 upload_score = min(upload / 50, 1.0) * 0.3 # 50Mbps为满分 ping_score = (1 - min(ping / 100, 1.0)) * 0.2 # 100ms为最低分 return (download_score + upload_score + ping_score) * 100ISP运营商对比报告
SELECT strftime('%W', timestamp) as week, AVG(download) as avg_down, AVG(upload) as avg_up FROM speedtests GROUP BY week ORDER BY week这个系统在我家运行半年后，成功发现了运营商在晚高峰时段的带宽缩水问题，通过数据据理力争获得了30%的费用减免。现在它已经成为我智能家居系统中不可或缺的基础设施组件。