公众号旧文
胖子百科自建的测速服务迎来新升级
http://156.246.91.170:18080/100M.bin http://156.231.112.139:18080/100M.bin http://39.96.187.9:18080/100M.bin http://[2602:fa4f:a30:52b9:ca88:abc5:efcf:e71]:18080/100M.bin http://[2604:4300:a:492:215:17ff:fe92:300a]:18080/100M.bin
这些是我之前发在文章我发现有人每天在拿我的服务器测速评论区的
用 python 起的简单测速服务，把 URL 复制粘贴到浏览器地址栏里，按下回车即可开始下载对应的测速文件，但是这个方法十分的不优雅
因为如果你想知道下载速度，就必须得一直盯着浏览器的下载页面看才行，于是今天就做了一次小升级，把各项数据都记录了下来，这样就不用再盯着啦
顺便我还放了同款按钮进去，如果速度理想，也需要来一台，直接点击就可以
下面给出现在的测速服务的 URL
http://156.231.112.139:18080/
http://156.246.91.170:18080/
http://45.202.240.112:18080/如果有必要，之后会把手上的大部分机器，都做成这种，暂时只做三台看看反响，反正这些机器闲着也是闲着，活着不就是为了折腾和体验嘛
最后还得把源代码也贴出来，方便大家在自己的机器上部署
#!/usr/bin/env python3
import json
import time
from datetime import datetime
from pathlib import Path
from flask import Flask, Response, abort, jsonify, render_template_string, request, stream_with_context
BASE_DIR = Path("/opt/speedtest")
LOG_FILE = BASE_DIR / "download_history.jsonl"
CHUNK_SIZE = 1024 * 1024
app = Flask(__name__)
INDEX_HTML = """
<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>测速文件下载</title>
<style>
body { font-family: Arial, "Microsoft YaHei", sans-serif; margin: 40px; background: #f7f7f7; color: #222; }
.box { max-width: 1200px; margin: 0 auto; background: #fff; padding: 24px 28px; border-radius: 12px; box-shadow: 0 4px 18px rgba(0,0,0,.08); }
h1, h2 { margin-top: 0; }
a { text-decoration: none; }
.file { margin: 10px 0; padding: 12px 14px; background: #f0f4ff; border-radius: 8px; }
.muted { color: #666; font-size: 14px; }
table { width: 100%; border-collapse: collapse; margin-top: 20px; }
th, td { border-bottom: 1px solid #eee; padding: 10px 8px; text-align: left; font-size: 14px; }
th { background: #fafafa; }
.section { margin-top: 28px; }
</style>
</head>
<body>
<div class="box">
<h1>测速文件下载</h1>
<p class="muted">下载时会记录 IP、文件名、传输字节数、耗时和平均速度。</p>
{% for f in files %}
<div class="file">
<a href="/download/{{ f.name }}"><strong>{{ f.name }}</strong></a>
<div class="muted">大小：{{ "%.2f"|format(f.size / 1024 / 1024) }} MB</div>
</div>
{% endfor %}
<div class="section">
<h2>最近下载记录</h2>
{% if rows %}
<table>
<thead>
<tr>
<th>时间</th>
<th>IP</th>
<th>文件</th>
<th>已传输</th>
<th>耗时</th>
<th>平均速度</th>
<th>状态</th>
</tr>
</thead>
<tbody>
{% for r in rows %}
<tr>
<td>{{ r.time }}</td>
<td>{{ r.ip }}</td>
<td>{{ r.file }}</td>
<td>{{ r.bytes_human }}</td>
<td>{{ r.duration_str }}</td>
<td>{{ r.speed_human }}</td>
<td>{{ r.status }}</td>
</tr>
{% endfor %}
</tbody>
</table>
{% else %}
<p class="muted">暂无下载记录</p>
{% endif %}
</div>
</div>
</body>
</html>
"""
HISTORY_HTML = """
<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>下载历史</title>
<style>
body { font-family: Arial, "Microsoft YaHei", sans-serif; margin: 40px; background: #f7f7f7; color: #222; }
.box { max-width: 1200px; margin: 0 auto; background: #fff; padding: 24px 28px; border-radius: 12px; box-shadow: 0 4px 18px rgba(0,0,0,.08); }
h1 { margin-top: 0; }
table { width: 100%; border-collapse: collapse; margin-top: 20px; }
th, td { border-bottom: 1px solid #eee; padding: 10px 8px; text-align: left; font-size: 14px; }
th { background: #fafafa; }
.muted { color: #666; font-size: 14px; }
</style>
</head>
<body>
<div class="box">
<h1>下载历史</h1>
<p class="muted">显示最近 {{ rows|length }} 条记录。</p>
<table>
<thead>
<tr>
<th>时间</th>
<th>IP</th>
<th>文件</th>
<th>已传输</th>
<th>耗时</th>
<th>平均速度</th>
<th>状态</th>
<th>User-Agent</th>
</tr>
</thead>
<tbody>
{% for r in rows %}
<tr>
<td>{{ r.time }}</td>
<td>{{ r.ip }}</td>
<td>{{ r.file }}</td>
<td>{{ r.bytes_human }}</td>
<td>{{ r.duration_str }}</td>
<td>{{ r.speed_human }}</td>
<td>{{ r.status }}</td>
<td>{{ r.user_agent }}</td>
</tr>
{% endfor %}
</tbody>
</table>
<p><a href="/">返回下载页</a></p>
</div>
</body>
</html>
"""
def human_bytes(n: float) -> str:
units = ["B", "KB", "MB", "GB", "TB"]
x = float(n)
for unit in units:
if x < 1024 or unit == units[-1]:
return f"{x:.2f} {unit}"
x /= 1024
return f"{n:.2f} B"
def get_client_ip() -> str:
xff = request.headers.get("X-Forwarded-For", "").strip()
if xff:
ip = xff.split(",")[0].strip()
else:
ip = request.remote_addr or "-"
if ip.startswith("::ffff:"):
ip = ip[7:]
return ip
def append_log(record: dict) -> None:
with LOG_FILE.open("a", encoding="utf-8") as f:
f.write(json.dumps(record, ensure_ascii=False) + "\n")
def load_logs(limit: int = 200):
if not LOG_FILE.exists():
return []
rows = []
with LOG_FILE.open("r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
rows.append(json.loads(line))
except json.JSONDecodeError:
continue
return rows[-limit:][::-1]
def format_rows(raw):
rows = []
for r in raw:
duration = float(r.get("duration_sec", 0))
rows.append({
**r,
"bytes_human": human_bytes(float(r.get("bytes_sent", 0))),
"duration_str": f"{duration:.3f} s",
"speed_human": human_bytes(float(r.get("avg_speed_Bps", 0))) + "/s",
})
return rows
@app.route("/")
def index():
files = []
for p in sorted(BASE_DIR.iterdir()):
if p.is_file() and p.suffix == ".bin":
files.append({"name": p.name, "size": p.stat().st_size})
rows = format_rows(load_logs(10))
return render_template_string(INDEX_HTML, files=files, rows=rows)
@app.route("/download/<path:filename>")
def download(filename: str):
safe_path = (BASE_DIR / filename).resolve()
if not safe_path.is_file() or BASE_DIR.resolve() not in safe_path.parents:
abort(404)
file_size = safe_path.stat().st_size
client_ip = get_client_ip()
user_agent = request.headers.get("User-Agent", "-")
start = time.time()
sent = 0
status = "completed"
def generate():
nonlocal sent, status
try:
with safe_path.open("rb") as f:
while True:
chunk = f.read(CHUNK_SIZE)
if not chunk:
break
sent += len(chunk)
yield chunk
except (BrokenPipeError, ConnectionResetError):
status = "aborted"
return
except Exception:
status = "error"
raise
finally:
duration = max(time.time() - start, 0.001)
speed = sent / duration
record = {
"time": datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC"),
"ip": client_ip,
"file": filename,
"bytes_sent": sent,
"duration_sec": round(duration, 3),
"avg_speed_Bps": round(speed, 2),
"status": status,
"user_agent": user_agent,
"file_size": file_size,
}
append_log(record)
headers = {
"Content-Disposition": f'attachment; filename="{safe_path.name}"',
"Content-Length": str(file_size),
"Content-Type": "application/octet-stream",
}
return Response(stream_with_context(generate()), headers=headers)
@app.route("/history")
def history():
rows = format_rows(load_logs(200))
return render_template_string(HISTORY_HTML, rows=rows)
@app.route("/history.json")
def history_json():
return jsonify(load_logs(200))
if __name__ == "__main__":
app.run(host="::", port=18080, threaded=True)