引言

在这本逐步指南中,您将从零开始创建一个完全功能的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机器人的令牌并创建私人聊天以接收通知,这样机器人就可以在发生事件时发送消息。

逐步说明

  1. 在您的电话或计算机上打开Telegram应用。
  2. 在搜索栏中输入BotFather,打开带有蓝色勾的官方帐户。
  3. 点击“开始”或输入命令/start,然后输入/newbot。
  4. 想个名字,比如:ProxyMonitorBot。
  5. 想个独特的用户名,以bot结尾。例如:proxy_monitor_helper_bot。
  6. 复制发放的令牌。形式如字符和数字的组合,以冒号分隔。
  7. 创建一个私人聊天或群组,机器人将发送通知到该群组。将机器人添加到其中。
  8. 在聊天中发送任意消息,使聊天出现在您的账户列表中。
  9. 打开机器人对话并点击“开始”,以激活它。

重要事项:请务必保管好令牌。不要将其传递给任何人。如果令牌泄露,恶意用户将能够控制您的机器人。如果令牌被泄露,请在BotFather中选择/revoke并获取新的令牌。

建议:创建一个单独的群组用于警报,您和机器人只有2人。这样通知不会在其他消息中丢失。

✅ 检查:您已获得令牌并将机器人添加到所需的聊天中。机器人在个人对话中对/start命令做出响应。

可能的问题和解决方案

  • 问题:机器人不响应/start。原因:您未点击“开始”或令牌在代码中输入错误。解决方案:确保机器人已激活,稍后我们将正确指定.env文件中的令牌。
  • 问题:未看到聊天ID。原因:我们尚未获取。解决方案:稍后机器人中的命令将自动显示chat_id。

步骤2:准备Python环境和项目结构

阶段目标

创建项目的工作文件夹,Python 3.11+的虚拟环境,安装依赖、配置和目录结构。

逐步说明

  1. 在您的机器上打开终端或通过SSH连接到VPS。
  2. 使用命令检查Python版本:python3 --version。如果版本低于3.11,请安装最新版本。
  3. 在Ubuntu上执行:sudo apt update; sudo apt install -y python3.11 python3.11-venv python3-pip。
  4. 创建项目文件夹:mkdir proxy-monitor-bot; cd proxy-monitor-bot。
  5. 创建虚拟环境:python3.11 -m venv .venv。
  6. 激活环境:source .venv/bin/activate。
  7. 更新pip:pip install --upgrade pip。
  8. 安装依赖:pip install aiogram aiohttp aiosqlite APScheduler python-dotenv uvloop。
  9. 创建文件夹结构:mkdir app app/modules app/db app/bot app/web。
  10. 创建文件: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的限制。

详细说明

  1. 打开文件app/modules/proxy_checker.py。
  2. 将代理模型定义为字典:id、label、type(http, https, socks5)、endpoint(host:port)、username、password、region、group、expected_country、last_ip。
  3. 创建异步函数build_proxy_url,从type、host、port和可选的username:password生成格式为scheme://user:pass@host:port的字符串。
  4. 创建异步函数fetch_ip(session, proxy_url),用于通过代理获取外部IP。使用url https://api.ipify.org?format=json或类似服务,返回您在JSON中的IP。记录调用前后的时间,以获取延迟。
  5. 创建地理数据缓存:内存中的字典和SQLite中的表格,以便在重启时不丢失缓存。键为IP,值为国家、城市、组织和缓存时间的JSON。
  6. 创建函数fetch_geo(session, ip),对ip-api.com/json/{ip}?fields=status,country,countryCode,regionName,city,org,query进行GET请求。如果状态为fail,则返回错误标志;如果成功则返回地理数据。
  7. 实现函数check_proxy(proxy),逻辑为:构建proxy_url,尝试在超时内获取IP,测量延迟,然后获取地理数据(从缓存或网络),生成结果:可用性(ok)、毫秒级的延迟、外部IP、国家和城市,以及结果类型:UP、DEGRADED或DOWN。
  8. 定义DEGRADED标准:如果延迟超过您的阈值,例如1500-2000毫秒,或如果最近的检查中错误频率明显。为了简单:如果有响应,但延迟超过2秒 — DEGRADED,否则为UP。如果没有响应 — DOWN。
  9. 返回详细结果以便记录:时间戳、proxy_id、status、latency_ms或None、ip或None、geo或None、error_reason、changed_ip(如果IP与上一个不同为True或False)、changed_geo(如果国家发生变化)。
  10. 添加异常处理: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。

可能的问题和解决方案

  • 问题:ClientHttpProxyError错误。原因:代理不支持请求的类型或认证错误。解决方案:检查类型和凭证,更新端点。
  • 问题:地理位置始终为unknown。原因:ip-api请求错误或被封锁。解决方案:检查IP_API_URL和网络可用性,启用异常日志记录。

步骤4:配置SQLite数据库和日志记录历史

阶段目标

在SQLite中创建表格,以存储代理列表、检查历史、事件、地理缓存和设置。这将为您提供详细的统计数据和构建图表的能力。

数据库结构

  • proxies: id, label, type, endpoint, username, password, region, group_name, expected_country, last_ip, enabled, created_at, updated_at。
  • checks: id, proxy_id, ts, status, latency_ms, ip, country, city, org, error_reason。
  • incidents: id, proxy_id, opened_at, closed_at, kind (DOWN或DEGRADED), last_status, last_message, active。
  • geo_cache: ip, country, city, org, ts。
  • settings: key, value(例如,默认检查间隔)。

详细说明

  1. 打开app/db/models.py。描述用于创建表的SQL常量。添加proxy_id和ts的索引,以加快查询速度。
  2. 创建函数init_db(db_path),该函数在需要时创建data文件夹,打开aiosqlite.connect连接,并执行create table if not exists以创建所有表。
  3. 添加用于记录结果的函数:add_proxy,update_proxy_last_ip,add_check,open_incident,close_incident_if_needed,upsert_geo_cache,get_recent_checks,get_uptime_stats。
  4. 在app/db/migrations.py中实现迁移。第一次启动只需执行所有create table。版本更新时,该脚本将添加缺失的列。

SQL示例(简化为一行)

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的响应。

可能的问题和解决方案

  • 问题:no such table错误。原因:未初始化。解决方案:在所有操作之前调用init_db。
  • 问题:写入锁定。原因:并发事务。解决方案:尽可能将结果记录分组为批次,或使用较短的事务。

步骤5:实现基于aiogram 3的Telegram机器人

阶段目标

启动一个能够响应命令、能够向您的聊天发送通知并能够从.env文件读取设置的机器人。配置基础命令:/start,/status,/chatid,/addproxy,/list,/enable,/disable,/interval,/export。

详细说明

  1. 打开app/bot/handlers.py。创建命令路由。对于aiogram 3,使用Router、Dispatcher和异步处理程序。
  2. 实现/start:回答简短的功能描述和当前检查间隔。
  3. 实现/chatid:发送聊天ID,方便您在.env文件中保存TELEGRAM_CHAT_ID。这很方便,如果您在个人聊天或群组中启动机器人。
  4. 实现/status:查找每个代理的最新N个检查并输出简短摘要:状态、延迟、国家、更新时间。
  5. 实现/list:列出代理,包括id、label、type、region、enabled。
  6. 实现/addproxy:通过type、endpoint、username、password、region、group、expected_country传递参数。第一步可以做一个简单版本,接受格式为type=http endpoint=host:port region=EU的字符串,并解析参数。
  7. 实现/enable和/disable,带参数id,快速启用和禁用特定代理。
  8. 实现/interval N以改变检查间隔(以分钟为单位)。将值保存到settings表中。
  9. 实现/export生成CSV,基于一段时间内的最后检查记录。稍后我们还将在web仪表板中添加类似的按钮。

发送通知

在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命令可能会是空的,这是正常的。

可能的问题和解决方案

  • 问题:机器人无法启动。原因:令牌错误。解决方案:检查.env中BOT_TOKEN,空格和换行符。
  • 问题:命令无法执行。原因:Router未连接。解决方案:确保您在Dispatcher中添加了路由器并启动了轮询。

步骤6:APScheduler的检查调度程序

阶段目标

设置每5-15分钟定期检查所有启用代理,考虑到当前设置中的间隔,并且并行检查而不超载,记录历史,并在变化时发送警报。

详细说明

  1. 打开app/scheduler.py。初始化AsyncIOScheduler。
  2. 创建任务run_checks,从数据库中选择所有enabled=1的代理,将它们放入并发的异步任务池进行检查,并收集结果。
  3. 对于每一个结果,将一行写入checks。如果发生变化,请更新proxies中的last_ip。
  4. 事件逻辑:如果状态为DOWN且没有事件,则创建incidents的记录(active=1)。如果状态再次为UP,则关闭事件(active=0,closed_at=ts)。对于DEGRADED同样,但可以将DEGRADED合并为一个活动事件,或在恢复时关闭。
  5. 发送通知:在第一次转为DOWN时、DEGRADED时、恢复为UP时、IP变化或国家变化时。如果许多代理同时更改状态,请将消息分组:发送一个标题和项目列表。
  6. 创建作业scheduler.add_job(run_checks, 'interval', minutes=CHECK_INTERVAL_MINUTES, id='checks', replace_existing=True)。添加对/interval的响应,以便在不重启机器人的情况下重新启动作业,使其使用新间隔。
  7. 启用scheduler.start(),同时启动机器人和web服务器在main中。

建议:

建议:为调度添加0-60秒的随机抖动,以减少峰值负载,同时避免正好同步请求。

✅ 检查:暂时将CHECK_INTERVAL_MINUTES设置为1并启动应用。日志中您应该会看到每分钟执行的检查,并且新行被添加到表checks中。

可能的问题和解决方案

  • 问题:检查未启动。原因:调度器未启动。解决方案:确保调用了scheduler.start(),并且在创建作业时没有异常。
  • 问题:数据库写入锁定。原因:写入频率过高。解决方案:缩短间隔或在写入时使用批量处理。

逐步说明:步骤7:将所有部分集成到main.py的入口点

阶段目标

完整构建应用:配置、数据库、机器人、检查器、调度器和web仪表板。确保正确启动和停止,处理终止信号。

详细说明

  1. 打开app/main.py。导入asyncio、logging、uvloop(如果是Linux)、aiogram、init_db、路由器、调度器、Web服务器的功能。
  2. 通过python-dotenv加载.env中的配置。检查必要的键:BOT_TOKEN和DB_PATH。
  3. 初始化数据库:await init_db(DB_PATH)。
  4. 创建aiogram 3的Dispatcher和Router,添加命令处理程序。
  5. 创建APScheduler并根据间隔添加run_checks任务。
  6. 并行启动三个组件:机器人轮询、scheduler.start()、web仪表板。使用asyncio.gather并行启动。
  7. 处理SIGINT和SIGTERM信号:正确停止调度器,关闭与数据库的连接并结束轮询。

简化的启动思路为一行: 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开放,并且调度程序已开始在设置的间隔下执行检查。

可能的问题和解决方案

  • 问题:地址0.0.0.0:8080被占用。原因:端口被其他进程占用。解决方案:在.env中更改DASHBOARD_PORT或停止正在占用的进程。
  • 问题:在启动轮询时出错。原因:令牌错误或网络问题。解决方案:检查BOT_TOKEN和网络接入。

步骤8:通知格式和代理分组

阶段目标

使通知清晰、简洁且信息丰富。配置按区域和类型进行的代理分组,以使消息结构化。

通知格式

  • 标题:代理状态变化。
  • 副标题:日期和时间。
  • 按优先级划分的部分:DOWN、然后是DEGRADED、恢复、IP变化、地理变化。
  • 每个项目:[区域][组][Label]([type])状态、延迟、IP、国家、原因。
  • 图标:DOWN用❌标记,DEGRADED用⚠️标记,UP用✅标记。

代理分组

  • 按区域字段分组:EU、US、ASIA等。
  • 按group_name字段分组:逻辑分组,例如EUROPE、PAID、TESTING。
  • 按类型分组:http、https、socks5。

实现

  1. 在app/bot/notifications.py中创建group_by_region和group_by_type函数,以便在发送之前对结果进行分组。这将减少发送的消息数量并提高可读性。
  2. 为每个块生成简短字符串,包括关键数据:"EU, EUROPE, EU-1 (http): ❌ DOWN, timeout"或"US, AMERICA, US-2 (socks5): ⚠️ DEGRADED, 2300ms, country US"。
  3. 如果变更较多,请先发送一个总结标题,然后按区域发送多条消息。这比发送100条单独的消息要好。

✅ 检查:暂时用无效端点替换一个代理。等待检查。您将收到一个DOWN块的通知。将代理恢复到有效状态,您将看到恢复通知。

可能的问题和解决方案

  • 问题:消息过长。原因:细节过多。解决方案:缩短字段,详细信息可通过命令/status或/details id查看。
  • 问题:重复通知。原因:状态未变时重复发送。解决方案:仅在状态变化时发送通知,并在IP或国家发生明显变化时发送。

步骤9:在VPS(Ubuntu)上的部署和自启动

阶段目标

在远程服务器上部署机器人,通过systemd配置自启,提供日志记录和简单的更新。

准备VPS

  1. 连接到VPS:ssh user@server_ip。
  2. 更新系统:sudo apt update && sudo apt upgrade -y。
  3. 如果尚未安装Python,请安装:sudo apt install -y python3.11 python3.11-venv python3-pip。
  4. 如果需要,创建用于应用的用户:sudo adduser botuser;然后以该用户登录。
  5. 通过scp或git clone将项目复制到服务器。
  6. 建立虚拟环境并安装依赖,如步骤2所示。
  7. 检查.env。填入实际值BOT_TOKEN和TELEGRAM_CHAT_ID。

systemd 服务

  1. 创建服务文件:sudo nano /etc/systemd/system/proxy-monitor.service。
  2. 示例Unit(单行): [Unit] Description=Proxy Monitor Bot After=network.target [Service] Type=simple User=botuser WorkingDirectory=/home/botuser/proxy-monitor-bot ExecStart=/home/botuser/proxy-monitor-bot/.venv/bin/python -m app.main Environment=PYTHONUNBUFFERED=1 Restart=on-failure RestartSec=5 [Install] WantedBy=multi-user.target。
  3. 重载守护程序:sudo systemctl daemon-reload。
  4. 启用自启动功能:sudo systemctl enable proxy-monitor.service。
  5. 启动:sudo systemctl start proxy-monitor.service。
  6. 检查日志:journalctl -u proxy-monitor.service -f。

建议:如果您使用uvloop,请确保它已安装在虚拟环境中。在Windows上不使用uvloop,在Linux上改进性能。

✅ 检查:在您的Telegram中,当服务启动时,机器人应该对/start进行响应。在日志中可以看到启动调度程序并且web仪表板正在监听8080端口。

可能的问题和解决方案

  • 问题:服务立即崩溃。原因:Python或工作目录路径错误。解决方案:检查ExecStart和WorkingDirectory的路径以及用户权限。
  • 问题:外部不能访问8080端口。原因:UFW或云防火墙配置。解决方案:开启该端口:sudo ufw allow 8080或在提供商面板中配置规则。

结果检查

检查清单

  • 机器人接受命令/start、/status、/list、/chatid。
  • 调度程序在间隔内启动检查,记录出现在checks表中。
  • 当代理崩溃时,机器人向Telegram发送通知。
  • IP变化被记录并在消息中标记。
  • 地理位置得到确认并与预期国家进行比较。
  • 仪表板在指定端口上运行,显示正常运行时间和最近事件。
  • CSV导出正常工作并输出正确的文件。

如何进行测试

  1. 断开一个代理或指定无效端口。等待警报DOWN。
  2. 暂时更改端点以触发IP变化。确保机器人指示IP和国家的变化(如有变化)。
  3. 通过网络限制增加人为延迟(如果可能),或者通过沉重请求检查,以使其变为DEGRADED。
  4. 检查命令:/interval 5、/list、/enable、/disable,并确保状态展现正确。

成功执行的标准

  • 无错误的稳定检查。
  • 在出现真实问题时及时警报。
  • 事件逻辑正常工作:DOWN时开启,UP时关闭。

✅ 检查:如果所有检查清单项目都通过且测试场景正常工作,您的机器人已成功部署并正确执行监控任务。

常见错误及解决方案

  • 问题:Telegram中没有通知→ 原因:TELEGRAM_CHAT_ID错误或机器人不是群组成员→ 解决方案:在所需聊天中执行/chatid,保存ID到.env中,将机器人添加到聊天并给予其写权限。
  • 问题:所有代理DOWN,尽管它们实际可用→ 原因:代理URL格式或类型错误→ 解决方案:检查type=http、https或socks5,检查user:pass授权和endpoint host:port,单独测试一个代理。
  • 问题:地理位置未确定→ 原因:ip-api限额或网络错误→ 解决方案:启用缓存,减少对ip-api的请求频率,检查IP_API_URL。
  • 问题:数据库增长过快→ 原因:检查频率过高且代理数量庞大→ 解决方案:增加间隔,启用旧记录清理,导出归档并删除超过30-90天的记录。
  • 问题:所有代理延迟较高→ 原因:服务器过载或网络问题→ 解决方案:增加VPS资源,减少PROXY_CONCURRENCY,使用更近的区域。
  • 问题:相同事件的重复警报→ 原因:未保存事件状态→ 解决方案:使用incidents表格,仅在状态变化时发送警报。
  • 问题:机器人在异常情况时频繁崩溃→ 原因:未捕获的错误→ 解决方案:将网络操作包裹在try-except中,增加日志记录和重试。

附加功能

高级设置

  • 灵活的降级阈值:在settings表中根据组存储延迟阈值。
  • 不同组的检查间隔:重要代理每5分钟检查,次要代理每15-30分钟。
  • 扩展命令解析:使用key=value格式和严格的参数验证。

优化

  • 在Linux上启用uvloop以加速事件循环。
  • 在SQLite中使用批量写入或WAL模式以加速并发写入。
  • 为信号量参数化,以便在速度和负载之间取得平衡。

简单仪表板

在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个存档。

⚠️ 注意:请勿在运行时编辑数据库文件。对于手动编辑,请停止服务,进行备份,然后再编辑并重新启动。

FAQ

  • 如何添加带授权的代理?在命令/addproxy中指定username和password,或在endpoint中添加user:pass。示例:type=http endpoint=user:pass@host:port。
  • 如何限制夜间检查?在APScheduler中实现类似cron的时间表,添加时间窗口的触发器,或在夜间通过自动规则简单地增加间隔。
  • 如果ip-api.com无法访问该怎么办?使用缓存并临时跳过地理请求。稍后可以添加备用地理数据源。
  • 如何过滤通知?添加重要性级别和配置:仅发送DOWN,而DEGRADED则每小时发送一次。
  • 可以使用PostgreSQL吗?可以。将aiosqlite替换为asyncpg,调整SQL和连接。对于较大的负载这更方便。
  • 如何安全存储令牌?在服务器的.env中设置权限。周期性更新令牌。
  • 为什么机器人有时会超时?网络不稳定。增加PROBE_TIMEOUT,减少并发数,检查路由。
  • 可以检查IPv6代理吗?可以,如果目标服务支持IPv6。请确认endpoint返回IPv6,ip-api正常工作。
  • 如何监控特定的网络运营商?在消息中保存org字段来自ip-api,并与预期值进行比较。
  • 如何暂时禁用代理?使用/disable id命令,暂时将代理从检查中排除,但不从数据库中删除。

总结

您已经完成了从构思到功能完整解决方案的全程:创建了基于aiogram 3的Telegram机器人,编写了基于aiohttp的代理检查模块,设置APScheduler调度器,记录历史到SQLite,实现了崩溃、降级、IP和地理位置信知的通知,在VPS上上线并通过systemd自启,并添加了简单的web仪表板与日志导出功能。现在您拥有一个灵活的工具,可进行扩展:根据组划分、调整阈值、接入图表和报告,随着负载增长迁移到PostgreSQL。下一步是自动化维护:日志轮换设置、每日数据库备份、监控机器人的系统资源。您可以进一步发展:添加对机器人命令的角色访问模型,实现基于组的SLA报告,接入替代地理数据源,使用任何方便的库制作Web界面。祝您好运,愿您的代理保持稳定的正常运行时间!