如何创建用于代理监控的Telegram机器人:新手的逐步指南
引言
在这本逐步指南中,您将从零开始创建一个完全功能的Telegram机器人,该机器人可以自动监控代理列表,测量响应时间,检测轮换后的IP变化,通过ip-api.com检查地理位置,并在出现问题和降级时在Telegram中发送显著通知。我们将逐步走过从构想到在VPS上部署并自启动的过程,记录到SQLite并提供简单的仪表板,显示正常运行时间和日志导出。
该指南适合初学者,包含一些高级内容。如果您从未编写过机器人,请不要担心:我们将逐步解析每个步骤,解释它的用途,以及如何验证结果。到最后,您将获得一个稳定的监控工具,能够按照计划每5-15分钟启动,能够根据地区和类型对代理进行分组,并且能够在Telegram中发送有用的警报。
在开始之前,了解如何打开终端,安装程序,以及如何创建文件夹和文件是很有帮助的。完成所有步骤大约需要4到8小时,包括环境设置、代码编写、测试和部署。如果您已经拥有VPS并且具备基本的Python技能,时间会更短。
到最后,您将得到:一个基于aiogram 3的工作Telegram机器人,一个使用aiohttp的异步检查模块,一个用于定时任务的APScheduler,一个SQLite数据库记录和统计信息,一个包含主要指标的简单web仪表板,以及维护和扩展功能的说明。
建议:最好依次完成指南,不要跳过步骤。每个步骤后我们都有检查,以便您立即确认一切正常。
预备工作
在编写代码之前,我们需要准备所有必要的工具和环境。这将避免意外错误,并节省时间。
必要的工具、程序和访问权限
- 用于创建机器人的Telegram账户以及接收通知。
- 具有Linux的VPS或本地计算机(最好是Ubuntu 22.04或24.04)。Windows和macOS也可以,但我们将在Ubuntu上展示部署。
- Python 3.11或更新版本。通过命令python3 --version来检查版本。
- 终端和基本的sudo权限,以便安装软件包。
- 要监控的代理列表,格式为host:port,且有必要时提供用户名和密码。支持HTTP、HTTPS和SOCKS5。
系统要求
- 1个CPU和512-1024 MB内存足以处理50-200个代理,间隔为5-15分钟。对于数千个代理,最好使用2个CPU和2 GB内存。
- 磁盘上至少1 GB的可用空间。SQLite轻量,但日志会增大。
- 能够访问互联网。必须能够访问Telegram API和您的代理。ip-api.com需要外部HTTP访问。
需要安装的内容
- Python 3.11+和用于虚拟环境的venv。
- 用于安装依赖的pip。
- 库:aiogram版本3,aiohttp,APScheduler,aiosqlite,python-dotenv(便于配置),uvloop(可选用于Linux),yarl(通常作为依赖项)。
创建备份
在开始时备份并非关键。启动机器人后,一定要设置简单的SQLite数据库备份,包含日志和配置。这使得在错误更新或文件损坏时可以回滚。
⚠️ 注意:SQLite是一个文件。如果在意外断电时文件被损坏,可能会丢失新记录。因此,在VPS上启用自动每日备份文件的cron或systemd定时器。我们将在最后展示一个简单的脚本来实现此功能。
基本概念
为了避免混淆,我们将约定一些关键术语和工作原理。
- 代理(Proxy) — 中介服务器,通过它发送请求。包括HTTP、HTTPS和SOCKS5。代理可能需要用户名:密码的授权,或通过IP绑定工作。
- 延迟(Latency) — 我们通过代理发送请求并收到响应所需的时间。以毫秒为单位。越小越好。
- IP轮换 — 一些代理会定期更换外部IP地址。我们的机器人需要监控这些变化,并在IP意外更换或与预期区域不符时及时通知。
- 地理定位(Geo) — IP地址的国家、城市和运营商。通过服务ip-api.com根据IP获取。这将帮助您确保代理位于所需国家。
- 降级 — 代理响应,但远低于正常速度或频繁错误。我们将区分DOWN(完全不可用)和DEGRADED(正常运行但效果差)。
- aiogram 3 — 现代异步Telegram机器人库。简单、快速,支持2026年Telegram的所有最新功能。
- aiohttp — 带超时和代理支持的异步HTTP请求。
- APScheduler — 任务调度程序,将在N分钟内定期启动检查。
- SQLite — 内置数据库。一个文件,快速且无需服务器安装。非常适合本地日志和统计。
机器人的工作原理是:根据调度,机器人获取代理列表,通过每个代理发起快速测试请求,测量时间,获取外部IP和地理数据,并将结果保存到SQLite。如果代理崩溃或明显降级,机器人会生成一条整齐的消息并发送到Telegram。此外,机器人会响应来自聊天中的命令:显示状态、正常运行时间、最近的错误、导出CSV、添加或禁用代理、修改间隔。
建议: 为了不超过ip-api.com的限制(通常是免费等级每分钟45个请求),我们将实现地理数据缓存和地理定位并发请求的限制。这不会减慢工作速度,但会保护您免于被封锁。
步骤1:在Telegram中创建机器人并配置访问权限
阶段目标
获取Telegram机器人的令牌并创建私人聊天以接收通知,这样机器人就可以在发生事件时发送消息。
逐步说明
- 在您的电话或计算机上打开Telegram应用。
- 在搜索栏中输入BotFather,打开带有蓝色勾的官方帐户。
- 点击“开始”或输入命令/start,然后输入/newbot。
- 想个名字,比如:ProxyMonitorBot。
- 想个独特的用户名,以bot结尾。例如:proxy_monitor_helper_bot。
- 复制发放的令牌。形式如字符和数字的组合,以冒号分隔。
- 创建一个私人聊天或群组,机器人将发送通知到该群组。将机器人添加到其中。
- 在聊天中发送任意消息,使聊天出现在您的账户列表中。
- 打开机器人对话并点击“开始”,以激活它。
重要事项:请务必保管好令牌。不要将其传递给任何人。如果令牌泄露,恶意用户将能够控制您的机器人。如果令牌被泄露,请在BotFather中选择/revoke并获取新的令牌。
建议:创建一个单独的群组用于警报,您和机器人只有2人。这样通知不会在其他消息中丢失。
✅ 检查:您已获得令牌并将机器人添加到所需的聊天中。机器人在个人对话中对/start命令做出响应。
可能的问题和解决方案
- 问题:机器人不响应/start。原因:您未点击“开始”或令牌在代码中输入错误。解决方案:确保机器人已激活,稍后我们将正确指定.env文件中的令牌。
- 问题:未看到聊天ID。原因:我们尚未获取。解决方案:稍后机器人中的命令将自动显示chat_id。
步骤2:准备Python环境和项目结构
阶段目标
创建项目的工作文件夹,Python 3.11+的虚拟环境,安装依赖、配置和目录结构。
逐步说明
- 在您的机器上打开终端或通过SSH连接到VPS。
- 使用命令检查Python版本:python3 --version。如果版本低于3.11,请安装最新版本。
- 在Ubuntu上执行:sudo apt update; sudo apt install -y python3.11 python3.11-venv python3-pip。
- 创建项目文件夹:mkdir proxy-monitor-bot; cd proxy-monitor-bot。
- 创建虚拟环境:python3.11 -m venv .venv。
- 激活环境:source .venv/bin/activate。
- 更新pip:pip install --upgrade pip。
- 安装依赖:pip install aiogram aiohttp aiosqlite APScheduler python-dotenv uvloop。
- 创建文件夹结构:mkdir app app/modules app/db app/bot app/web。
- 创建文件:touch app/__init__.py app/config.py app/main.py app/modules/proxy_checker.py app/db/models.py app/db/migrations.py app/bot/handlers.py app/bot/notifications.py app/scheduler.py app/web/server.py .env .env.example README.md。
基本配置
打开.env文件并添加密钥(根据您的内容替换值):
BOT_TOKEN=您的_BotFather令牌 TELEGRAM_CHAT_ID=数字_或留空 DB_PATH=./data/monitor.db PROBE_TIMEOUT=8 PROBE_RETRIES=2 CHECK_INTERVAL_MINUTES=10 GEO_CACHE_TTL_HOURS=24 IP_API_URL=http://ip-api.com/json PROXY_CONCURRENCY=20 GEO_CONCURRENCY=10 DASHBOARD_HOST=0.0.0.0 DASHBOARD_PORT=8080
将.env复制为.env.example并删除机密,以便于子共享项目而无需令牌。
建议:务必将.env文件保存到版本控制系统之外。将.env加入.gitignore,以防意外上传令牌。
最小代码结构
文件app/config.py将读取环境变量并提供配置。文件app/main.py将启动机器人、调度器和web仪表板。模块proxy_checker.py将执行所有检查。在db文件夹中放置模型和SQLite迁移。在bot模块中编写命令处理和通知发送。在web文件夹中启动简单的web服务器并输出统计信息。
✅ 检查:您拥有虚拟环境、安装的依赖、已创建的.env文件,并且项目结构与列出的结构符合。命令python -c "import aiogram, aiohttp, aiosqlite; print('ok')"输出ok。
可能的问题和解决方案
- 问题:pip报告SSL错误或不可用。原因:没有网络访问。解决方案:检查网络或使用pip的本地镜像。
- 问题:Python 3.11不可用。原因:旧操作系统。解决方案:更新系统或通过deadsnakes PPA或pyenv安装Python。
步骤3:编写代理检查模块:连接、延迟、IP、地理定位
阶段目标
创建一个异步模块,用于快速且可靠地检查代理:能够通过代理发送测试HTTP请求、测量延迟、查找外部IP、发送ip-api.com请求、缓存地理数据和正确处理错误。
关键理念
- 为检查可用性,请向轻量资源发送请求,例如https://api.ipify.org或任何快速的回显端点。我们选择通过服务获取IP,然后通过ip-api.com获取地理位置,也可以反过来。
- 默认超时设置为8秒,重试次数为2。这些参数可以通过.env进行配置。
- aiohttp方便地与代理协作:将proxy参数传递给ClientSession.request。
- 为地理定位缓存数据24小时,以免消耗配额。
- 检验并发请求,但通过信号量限制并发,避免过载连接并避免触及ip-api的限制。
详细说明
- 打开文件app/modules/proxy_checker.py。
- 将代理模型定义为字典:id、label、type(http, https, socks5)、endpoint(host:port)、username、password、region、group、expected_country、last_ip。
- 创建异步函数build_proxy_url,从type、host、port和可选的username:password生成格式为scheme://user:pass@host:port的字符串。
- 创建异步函数fetch_ip(session, proxy_url),用于通过代理获取外部IP。使用url https://api.ipify.org?format=json或类似服务,返回您在JSON中的IP。记录调用前后的时间,以获取延迟。
- 创建地理数据缓存:内存中的字典和SQLite中的表格,以便在重启时不丢失缓存。键为IP,值为国家、城市、组织和缓存时间的JSON。
- 创建函数fetch_geo(session, ip),对ip-api.com/json/{ip}?fields=status,country,countryCode,regionName,city,org,query进行GET请求。如果状态为fail,则返回错误标志;如果成功则返回地理数据。
- 实现函数check_proxy(proxy),逻辑为:构建proxy_url,尝试在超时内获取IP,测量延迟,然后获取地理数据(从缓存或网络),生成结果:可用性(ok)、毫秒级的延迟、外部IP、国家和城市,以及结果类型:UP、DEGRADED或DOWN。
- 定义DEGRADED标准:如果延迟超过您的阈值,例如1500-2000毫秒,或如果最近的检查中错误频率明显。为了简单:如果有响应,但延迟超过2秒 — DEGRADED,否则为UP。如果没有响应 — DOWN。
- 返回详细结果以便记录:时间戳、proxy_id、status、latency_ms或None、ip或None、geo或None、error_reason、changed_ip(如果IP与上一个不同为True或False)、changed_geo(如果国家发生变化)。
- 添加异常处理:asyncio.TimeoutError、aiohttp.ClientConnectorError、aiohttp.ClientHttpProxyError、aiohttp.ClientProxyConnectionError、套接字错误。在错误理由字段中写入简短描述。
代码示例(简化为一行)
逻辑示例紧凑地呈现: import asyncio, time, json, aiohttp; async def check_proxy(proxy, cfg, geo_cache, sem): async with sem: start=time.perf_counter(); purl=f"{proxy['type']}://{proxy.get('username', '')+(':'+proxy.get('password','') if proxy.get('username') else '')+'@' if proxy.get('username') else ''}{proxy['endpoint']}"; timeout=aiohttp.ClientTimeout(total=cfg['PROBE_TIMEOUT']); try: async with aiohttp.ClientSession(timeout=timeout) as s: t0=time.perf_counter(); async with s.get('https://api.ipify.org?format=json', proxy=purl, ssl=False) as r: data=await r.json(); ip=data.get('ip'); latency=int((time.perf_counter()-t0)*1000); if not ip: raise Exception('no-ip'); geo=geo_cache.get(ip); if not geo or geo['ts'] 在实际代码中,我们将添加SQLite记录和地理请求的信号量,以及精确的超时和重试。 ⚠️ 注意:请勿对ip-api.com发起过多的并发请求。如果您有数百个代理,请缓存结果并限制请求的并发。这将保护您免于限制和错误。 ✅ 检查:手动调用check_proxy函数对一个测试代理,并打印结果。您应该会看到状态为UP或DEGRADED,隶属测量的延迟和外部IP。 在SQLite中创建表格,以存储代理列表、检查历史、事件、地理缓存和设置。这将为您提供详细的统计数据和构建图表的能力。 CREATE TABLE IF NOT EXISTS proxies (id INTEGER PRIMARY KEY AUTOINCREMENT, label TEXT, type TEXT, endpoint TEXT, username TEXT, password TEXT, region TEXT, group_name TEXT, expected_country TEXT, last_ip TEXT, enabled INTEGER DEFAULT 1, created_at INTEGER, updated_at INTEGER); CREATE TABLE IF NOT EXISTS checks (id INTEGER PRIMARY KEY AUTOINCREMENT, proxy_id INTEGER, ts INTEGER, status TEXT, latency_ms INTEGER, ip TEXT, country TEXT, city TEXT, org TEXT, error_reason TEXT, FOREIGN KEY(proxy_id) REFERENCES proxies(id)); CREATE INDEX IF NOT EXISTS idx_checks_proxy_ts ON checks(proxy_id, ts); CREATE TABLE IF NOT EXISTS incidents (id INTEGER PRIMARY KEY AUTOINCREMENT, proxy_id INTEGER, opened_at INTEGER, closed_at INTEGER, kind TEXT, last_status TEXT, last_message TEXT, active INTEGER, FOREIGN KEY(proxy_id) REFERENCES proxies(id)); CREATE TABLE IF NOT EXISTS geo_cache (ip TEXT PRIMARY KEY, country TEXT, city TEXT, org TEXT, ts INTEGER); CREATE TABLE IF NOT EXISTS settings (key TEXT PRIMARY KEY, value TEXT); 建议:如果您已经以CSV或JSON格式存储代理列表,请添加方便的导入到数据库的功能:简单脚本读取文件并将记录添加到proxies。 ✅ 检查:运行init_db函数,确保数据库文件根据DB_PATH路径生成。检查是否已创建表,通过执行简单的查询select count(*) from proxies并获取0的响应。 启动一个能够响应命令、能够向您的聊天发送通知并能够从.env文件读取设置的机器人。配置基础命令:/start,/status,/chatid,/addproxy,/list,/enable,/disable,/interval,/export。 在app/bot/notifications.py中制作一个notify_change函数,该函数生成事件消息。消息示例:代理:EU-1 (http) 区域:EU 组:EUROPE 状态:DOWN 原因:timeout 最后IP:203.0.113.45 预期:国家DE 获得:国家NL 时间:2026-02-01 12:34:56 延迟:n/a。对于UP使用UP标签,对于降级使用DEGRADED标签。添加按重要性排序:首先DOWN,然后DEGRADED,然后是IP变化的消息。添加可视化表情符号,例如✅表示UP,⚠️表示DEGRADED,❌表示DOWN。如果TELEGRAM_CHAT_ID未设定,发送到当前聊天——来自于命令的聊天。生产中最好将chat_id固定在.env文件中。 建议:对于群聊,聊天机器人可能需要发送消息的权限。检查群组的设置,以确保机器人是具有写权限的成员。 ✅ 检查:在本地启动机器人并发送/chatid。您应该收到数字ID。将TELEGRAM_CHAT_ID填入.env中。发送/start和/status。机器人应该有回复。目前,/list命令可能会是空的,这是正常的。 设置每5-15分钟定期检查所有启用代理,考虑到当前设置中的间隔,并且并行检查而不超载,记录历史,并在变化时发送警报。 建议:为调度添加0-60秒的随机抖动,以减少峰值负载,同时避免正好同步请求。 ✅ 检查:暂时将CHECK_INTERVAL_MINUTES设置为1并启动应用。日志中您应该会看到每分钟执行的检查,并且新行被添加到表checks中。 完整构建应用:配置、数据库、机器人、检查器、调度器和web仪表板。确保正确启动和停止,处理终止信号。 简化的启动思路为一行: import asyncio, logging; async def main(): await init_db(...); dp=Dispatcher(); dp.include_router(router); scheduler=AsyncIOScheduler(); scheduler.add_job(run_checks, 'interval', minutes=interval); scheduler.start(); await asyncio.gather(dp.start_polling(bot), start_web_server(), wait_for_signals(scheduler)); asyncio.run(main()). 建议:添加INFO和WARN级别的详细日志。这将帮助快速了解启动和检查过程中发生的情况。 ✅ 检查:运行python app/main.py。机器人应对/start做出响应,web仪表板应对此网址http://localhost:8080开放,并且调度程序已开始在设置的间隔下执行检查。 使通知清晰、简洁且信息丰富。配置按区域和类型进行的代理分组,以使消息结构化。 ✅ 检查:暂时用无效端点替换一个代理。等待检查。您将收到一个DOWN块的通知。将代理恢复到有效状态,您将看到恢复通知。 在远程服务器上部署机器人,通过systemd配置自启,提供日志记录和简单的更新。 建议:如果您使用uvloop,请确保它已安装在虚拟环境中。在Windows上不使用uvloop,在Linux上改进性能。 ✅ 检查:在您的Telegram中,当服务启动时,机器人应该对/start进行响应。在日志中可以看到启动调度程序并且web仪表板正在监听8080端口。 ✅ 检查:如果所有检查清单项目都通过且测试场景正常工作,您的机器人已成功部署并正确执行监控任务。 在app/web/server.py中启动aiohttp.web服务器。在主页上显示摘要:代理数量,上线、下线、降级数量,平均延迟,24小时正常运行时间。制作一个端点/export?from=ts&to=ts,用于CSV导出。最小的HTML可以是简单的列表和带状态的表格。视觉图表可以稍后通过简单的SVG添加。 建议:在仪表板中添加“切换间隔”的按钮,以便直接调用端点并更改settings中的值。这将加速管理,而无需通过Telegram命令。 数据库备份脚本:tar -czf backup-$(date +%F).tar.gz data/monitor.db。通过cron或systemd定时器安排每日备份。至少保留最近7个存档。 ⚠️ 注意:请勿在运行时编辑数据库文件。对于手动编辑,请停止服务,进行备份,然后再编辑并重新启动。 您已经完成了从构思到功能完整解决方案的全程:创建了基于aiogram 3的Telegram机器人,编写了基于aiohttp的代理检查模块,设置APScheduler调度器,记录历史到SQLite,实现了崩溃、降级、IP和地理位置信知的通知,在VPS上上线并通过systemd自启,并添加了简单的web仪表板与日志导出功能。现在您拥有一个灵活的工具,可进行扩展:根据组划分、调整阈值、接入图表和报告,随着负载增长迁移到PostgreSQL。下一步是自动化维护:日志轮换设置、每日数据库备份、监控机器人的系统资源。您可以进一步发展:添加对机器人命令的角色访问模型,实现基于组的SLA报告,接入替代地理数据源,使用任何方便的库制作Web界面。祝您好运,愿您的代理保持稳定的正常运行时间!可能的问题和解决方案
步骤4:配置SQLite数据库和日志记录历史
阶段目标
数据库结构
详细说明
SQL示例(简化为一行)
可能的问题和解决方案
步骤5:实现基于aiogram 3的Telegram机器人
阶段目标
详细说明
发送通知
可能的问题和解决方案
步骤6:APScheduler的检查调度程序
阶段目标
详细说明
建议:
可能的问题和解决方案
逐步说明:步骤7:将所有部分集成到main.py的入口点
阶段目标
详细说明
可能的问题和解决方案
步骤8:通知格式和代理分组
阶段目标
通知格式
代理分组
实现
可能的问题和解决方案
步骤9:在VPS(Ubuntu)上的部署和自启动
阶段目标
准备VPS
systemd 服务
可能的问题和解决方案
结果检查
检查清单
如何进行测试
成功执行的标准
常见错误及解决方案
附加功能
高级设置
优化
简单仪表板
备份
FAQ
总结