告别臃肿面板!手写一个极简 Telegram 端口流量查询机器人
对于喜欢折腾 VPS、做端口转发或与朋友合租节点的人来说,分配端口和限制流量是刚需。市面上主流的方案通常是安装庞大的网页面板(如 3x-ui 或各类转发面板),但如果你和我一样有“代码洁癖”,只想要一个干干净净的纯命令行环境,同时又想给小伙伴提供一个高逼格的自助查询服务,那么这篇教程就是为你量身定制的。
💡 我们的终极架构:前后端分离玩法
- 后端(底层限速): 使用开源的
Port Traffic Dog(端口流量狗),在系统底层通过iptables/nftables精准统计流量,并在超量时铁面无私地切断端口。 - 前端(客户交互): 自己跑一个轻量级的 Python Telegram 机器人。它不干涉系统规则,只负责去读取“流量狗”的本地 JSON 数据,并在 TG 里给客户展示漂亮的“通行证”报表。
第一步:安装“端口流量狗” (底层核心)
首先,我们需要在服务器上安装大佬开源的轻量级流量控制脚本。
SSH 登录你的 VPS,执行以下一键安装命令:
Bash
1wget -O port-traffic-dog.sh https://raw.githubusercontent.com/zywe03/realm-xwPF/main/port-traffic-dog.sh && chmod +x port-traffic-dog.sh && ./port-traffic-dog.sh
(注:如果国内机访问 GitHub 慢,可以在网址前加上加速前缀)
安装完成后,以后只需在终端输入 dog 回车,即可呼出中文可视化菜单。你可以直接在里面添加端口、设置流量限额(支持GB/TB/无限)、设置每月重置日期。 设置好之后,底层的工作就完全交给流量狗去默默守护了。
第二步:申请 Telegram 机器人 (你的传话筒)
我们需要去 Telegram 官方申请一个专属的机器人账号:
- 在 TG 搜索
@BotFather。 - 发送
/newbot,按照提示给机器人起个名字和用户名(必须以_bot结尾)。 - 成功后,你会得到一串 Bot Token(类似
123456:ABCDefg...),妥善保存。 - 顺便搜索
@userinfobot,发送任意消息,获取你自己的 纯数字 ID(一串数字,这就是你的管理员凭证)。
第三步:部署我们的 Python 查询机器人
这是本文的核心!我们需要让机器人读取流量狗的数据并优雅地展示给用户。
1. 安装依赖库
Bash
1apt update && apt install python3 python3-pip -y
安装 TG 机器人通信库(如果你的系统是较新的 Debian 12 或 Ubuntu 24,可能会报错,请务必加上 --break-system-packages):
Bash
1pip3 install pyTelegramBotAPI --break-system-packages
2. 创建机器人代码文件
Bash
nano dog_tg_bot.py
将下面的代码完整粘贴进去,并替换成你自己的 BOT_TOKEN 和 ADMIN_ID:
Python
1import os
2import json
3import telebot
4
5# ================= 核心配置区 =================
6BOT_TOKEN = "你的_BOT_TOKEN_填在这里"
7ADMIN_ID = 123456789 # 填入你的管理员 TG ID (纯数字)
8
9# 路径配置
10TG_BIND_DB = "tg_bind.json"
11DOG_CONFIG_FILE = "/etc/port-traffic-dog/config.json"
12DOG_DATA_FILE = "/etc/port-traffic-dog/traffic_data.json"
13# ==========================================
14
15bot = telebot.TeleBot(BOT_TOKEN)
16
17def load_bind_db():
18 if not os.path.exists(TG_BIND_DB):
19 return {}
20 with open(TG_BIND_DB, 'r') as f:
21 return json.load(f)
22
23def save_bind_db(data):
24 with open(TG_BIND_DB, 'w') as f:
25 json.dump(data, f, indent=4)
26
27def get_dog_data(port):
28 """跨文件读取流量狗的底层配置与实时流量"""
29 port_str = str(port)
30 limit_gb = 0.0
31 used_gb = 0.0
32 is_unlimited = False
33 reset_day = 1
34
35 # 解析限额配置
36 try:
37 if os.path.exists(DOG_CONFIG_FILE):
38 with open(DOG_CONFIG_FILE, 'r', encoding='utf-8') as f:
39 cfg = json.load(f)
40 quota = cfg.get("ports", {}).get(port_str, {}).get("quota", {})
41 if quota.get("enabled"):
42 reset_day = quota.get("reset_day", 1)
43 limit_str = str(quota.get("monthly_limit", "0")).upper()
44 if limit_str == "UNLIMITED":
45 is_unlimited = True
46 elif "GB" in limit_str:
47 limit_gb = float(limit_str.replace("GB", ""))
48 elif "TB" in limit_str:
49 limit_gb = float(limit_str.replace("TB", "")) * 1024
50 elif "MB" in limit_str:
51 limit_gb = float(limit_str.replace("MB", "")) / 1024
52 else:
53 limit_gb = float(limit_str)
54 except Exception as e:
55 pass
56
57 # 解析实时已用流量
58 try:
59 if os.path.exists(DOG_DATA_FILE):
60 with open(DOG_DATA_FILE, 'r', encoding='utf-8') as f:
61 data = json.load(f)
62 port_data = data.get(port_str, {})
63 if port_data:
64 in_bytes = int(port_data.get("input", 0))
65 out_bytes = int(port_data.get("output", 0))
66 used_gb = (in_bytes + out_bytes) / (1024 ** 3)
67 except Exception as e:
68 pass
69
70 return limit_gb, round(used_gb, 4), is_unlimited, reset_day
71
72# ================= 管理员命令:人员开通 =================
73@bot.message_handler(commands=['bind'])
74def admin_bind_user(message):
75 if message.from_user.id != ADMIN_ID:
76 return bot.reply_to(message, "⛔ 权限不足。")
77
78 try:
79 args = message.text.split()
80 user_id = str(args[1])
81 port = int(args[2])
82
83 db = load_bind_db()
84 db[user_id] = port
85 save_bind_db(db)
86
87 limit, used, is_unlim, reset_day = get_dog_data(port)
88 limit_show = "无限容量" if is_unlim else f"{limit} GB"
89
90 reply_msg = (f"✅ **用户开通成功!**\n"
91 f"👤 客户 ID: `{user_id}`\n"
92 f"🔗 绑定通道: `{port}`\n"
93 f"📦 初始限额: `{limit_show}`\n"
94 f"🔄 重置规则: 每月 `{reset_day}` 日")
95 bot.reply_to(message, reply_msg, parse_mode="Markdown")
96 except Exception as e:
97 bot.reply_to(message, "❌ 格式错误!请使用: `/bind 用户ID 通道号`", parse_mode="Markdown")
98
99# ================= 普通用户:自助查询 =================
100@bot.message_handler(commands=['start', 'traffic', 'cx'])
101def user_query_traffic(message):
102 user_id = str(message.from_user.id)
103 db = load_bind_db()
104
105 if user_id in db:
106 port = db[user_id]
107 limit_gb, used_gb, is_unlim, reset_day = get_dog_data(port)
108
109 if is_unlim:
110 limit_show = "无限容量"
111 remain_show = "无限"
112 status_icon = "🟢 运行良好"
113 percent_show = ""
114 else:
115 limit_show = f"{limit_gb} GB"
116 remain = round(limit_gb - used_gb, 2)
117 remain_show = f"{remain if remain > 0 else 0} GB"
118 status_icon = "🟢 运行良好" if used_gb < limit_gb else "🔴 流量已耗尽"
119 percent = round((used_gb / limit_gb) * 100, 1) if limit_gb > 0 else 0
120 percent_show = f" ({percent}%)"
121
122 report = (
123 f"✨ **专属网络通行证** ✨\n"
124 f"━━━━━━━━━━━━━━━━━━\n"
125 f"📦 **总计额度:** `{limit_show}`\n"
126 f"📤 **当前已用:** `{used_gb} GB` `{percent_show}`\n"
127 f"📉 **剩余可用:** `{remain_show}`\n"
128 f"━━━━━━━━━━━━━━━━━━\n"
129 f"💡 **服务状态:** {status_icon}\n"
130 f"🔄 **数据结算:** `每月 {reset_day} 日自动刷新流量`"
131 )
132 bot.reply_to(message, report, parse_mode="Markdown")
133 else:
134 bot.reply_to(message, f"⛔ 您的账号尚未激活。\n请将您的 ID `{user_id}` 发送给管理员获取通行证。")
135
136print("🚀 尊享版机器人已启动!")
137bot.infinity_polling()
3. 后台永久运行
保存退出后,让它在后台默默打工:
Bash
1nohup python3 dog_tg_bot.py > bot.log 2>&1 &
第四步:如何使用?(极其优雅)
整个使用流程完全告别了复杂的配置:
- 添加底层规则: 你在 SSH 终端用
dog命令给朋友开一个端口,比如10003,限额500GB。 - 在 TG 绑定账号: 朋友进入机器人获取他的数字 ID 发给你。你用自己的号向机器人发送
/bind 朋友的ID 10003。 - 朋友自助查询: 朋友给机器人发送
/cx或/traffic,立刻就能收到一张漂亮、脱敏、隐藏了端口号和技术细节的“专属网络通行证”报表!
大功告成!这套系统既保留了 Linux 纯命令行的干净清爽,又提供了媲美商业面板的客户体验,简直是强迫症患者的福音。