别再傻傻用网页测速了！用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(threads4) # 多线程提升测试准确度 st.upload(threads4, pre_allocateFalse) 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_pathnetwork.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_pathnetwork.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] DescriptionNetwork Speed Monitor Afternetwork.target [Service] ExecStart/usr/bin/python3 /opt/network_monitor/main.py Restartalways Userroot [Install] WantedBymulti-user.target配合crontab定时任务*/30 * * * * /usr/bin/systemctl start network-monitor4. 可视化与报警系统原始数据需要转化为可操作的洞察。我开发了三种报告形式1. 终端实时看板import pandas as pd def show_realtime(db_pathnetwork.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(indexFalse))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 border1 trth指标/thth平均值/thth最低值/th/tr trtd下载速度(Mbps)/tdtd{stats[avg_down]:.2f}/tdtd{stats[min_down]:.2f}/td/tr trtd上传速度(Mbps)/tdtd{stats[avg_up]:.2f}/tdtd-/td/tr trtd延迟(ms)/tdtd{stats[avg_ping]:.2f}/tdtd-/td/tr /table # 发送邮件逻辑 msg MIMEText(html, html) msg[Subject] f网络质量报告 {datetime.now().date()} msg[From] monitorexample.com msg[To] adminexample.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_tokenYOUR_TOKEN headers {Content-Type: application/json} data { msgtype: text, text: { content: f网络异常报警:\n{message} } } requests.post(webhook, headersheaders, jsondata)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%的费用减免。现在它已经成为我智能家居系统中不可或缺的基础设施组件。