ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

基于RP2040与Telegram Bot的GitHub仓库实时监控系统实现

基于RP2040与Telegram Bot的GitHub仓库实时监控系统实现 1. 项目缘起为什么需要实时监控GitHub仓库作为一名开发者我经常遇到这样的场景团队里某个核心依赖库更新了或者自己关注的开源项目发布了新版本但总是后知后觉。等发现时可能已经过去了好几天错过了第一时间了解新特性或修复兼容性问题的时机。手动刷新GitHub页面显然不现实而GitHub自带的Watch功能通知又过于“嘈杂”它会推送Star、Fork、Issue等各种动态真正关心的代码提交Commit、发布Release反而容易被淹没。我需要一个“哨兵”一个能7x24小时紧盯特定仓库一旦有新的代码推送或版本发布就立刻通过一个我高频使用的渠道通知我的工具。这个工具最好足够轻量、部署简单并且完全由我自己掌控不依赖复杂的第三方SaaS服务毕竟有些内部仓库也不方便接入。于是“Real-time GitHub Repository Monitoring with PICO”这个想法就诞生了。它的核心目标很明确利用树莓派PICO这类微型开发板结合GitHub API实现对指定仓库的精准、实时监控并将变动消息推送到Telegram打造一个低成本、高可定制化的个人开发情报站。2. 核心架构与工具选型为什么是PICOTelegram在构思这个监控系统时我评估了几个关键组件每一个选择背后都有其考量。2.1 监控终端为何选用RP2040开发板如PICO首先需要明确这里的“PICO”是一个泛指核心是指使用RP2040微控制器芯片的开发板例如Raspberry Pi Pico、Pico W或者国内一些兼容板如Luckfox Pico。选择它而非树莓派全功能主板或云服务器主要基于以下几点极低功耗与常驻运行这个监控脚本需要长时间不间断运行7x24小时。树莓派4B即使空载也有数瓦的功耗而一块PICO开发板的运行功耗通常仅在几十到一百毫瓦级别非常适合作为长期插电运行的“网络哨兵”电费几乎可以忽略不计。成本与资源利用PICO开发板价格极其低廉通常几十元人民币用它来跑一个简单的HTTP请求和逻辑判断脚本是“杀鸡用牛刀”式的高性价比选择。将性能更强的树莓派或云服务器资源释放给更复杂的任务。网络能力必须选择带有Wi-Fi功能的型号如Raspberry Pi Pico W或Luckfox Pico。这是它能访问GitHub API和Telegram Bot API的基础。Pico W内置了Infineon CYW43439无线芯片提供了完整的TCP/IP网络栈支持。编程友好性RP2040支持MicroPython和C/C开发。对于此类网络应用MicroPython以其简洁的语法和丰富的库支持如urequests,network成为首选开发调试效率远高于嵌入式C。注意如果你手头只有无Wi-Fi的Pico则需要通过额外连接ESP-01s这类AT指令Wi-Fi模块来实现网络功能复杂度会提高。因此Pico W是更推荐的选择。2.2 通知渠道为何锁定Telegram Bot通知渠道需要满足实时、可靠、跨平台且易于集成。邮件太慢且容易被归类为垃圾邮件微信集成个人账号复杂且存在风控风险而Telegram Bot几乎是为此类场景量身定做API简单强大Telegram Bot API设计清晰通过HTTPS POST发送消息即可无需处理复杂的登录态或鉴权流程。即时到达消息推送几乎是实时的并且支持Markdown格式可以很好地格式化代码仓库的更新信息。跨平台与历史记录Telegram客户端覆盖全平台消息云端同步可以随时回溯查看历史通知。高度隔离Bot与个人账号分离即使Bot被滥用或废弃也不影响主账号安全。2.3 系统工作流全景图整个系统的工作流程可以概括为一个循环初始化PICO上电连接Wi-Fi。数据获取向GitHub API发起HTTP GET请求查询目标仓库的最新状态如最新提交的SHA哈希值、最新发布的版本号。状态比对将获取到的最新状态与本地存储的上一次状态进行比对。判断与通知如果状态发生变化如SHA不同、版本号更新则构造通知消息通过Telegram Bot API发送到指定聊天。状态更新与等待将新的状态更新到本地存储如PICO的文件系统然后进入休眠或等待一段时间例如每5分钟检查一次。循环跳回步骤2开始下一个监控周期。这个流程的核心在于“状态比对”。我们不需要每次都拉取完整的仓库信息只需获取一个能唯一标识“最新状态”的指纹如最新提交的SHA与本地指纹对比即可高效判断是否有更新。3. 环境搭建与核心代码实现接下来我们进入实操环节。你需要准备一块Pico W或类似板卡、一根Micro USB数据线以及一个能运行Thonny或类似IDE的电脑。3.1 前期准备创建Bot与获取Token创建Telegram Bot在Telegram中搜索BotFather。发送/newbot指令按提示设置机器人名称和用户名。创建成功后BotFather会提供一个HTTP API Token格式类似1234567890:ABCDEFGhijklmnOpqrstUvWxyz-abcde。妥善保存此Token它是你的Bot的钥匙。启动你的Bot并给它发送一条消息如/start。获取Chat ID在浏览器中访问以下URL将YourBotToken替换为你的Tokenhttps://api.telegram.org/botYourBotToken/getUpdates在与你Bot的聊天窗口发送任意消息。刷新浏览器页面你会在返回的JSON数据中看到一个message对象其下的chat对象里包含id字段。这个数字就是你的Chat ID。私聊时这就是你的个人ID如果将Bot拉入群组则需要获取群组的Chat ID。确定监控目标明确你要监控的GitHub仓库格式为:owner/:repo例如micropython/micropython。3.2 PICO开发环境配置刷入MicroPython固件从 Raspberry Pi 官网下载最新的 Pico W MicroPython UF2 固件文件。按住Pico W上的BOOTSEL按钮并连接电脑将其识别为U盘后将下载的UF2文件拖入完成后会自动重启。安装IDE推荐使用 Thonny IDE。安装后在Thonny的“运行” - “选择解释器”中选择“MicroPython (Raspberry Pi Pico)”并连接你的Pico W。关键库确认MicroPython固件通常内置了urequests和network库。我们主要依赖这两个库进行网络请求和连接。3.3 核心监控脚本详解下面是一个完整的、可运行的MicroPython脚本main.py。将其上传到Pico W后它会在每次上电时自动运行。import network import urequests as requests import time import json import ubinascii import os # 配置区域 # Wi-Fi 配置 WIFI_SSID 你的Wi-Fi名称 WIFI_PASSWORD 你的Wi-Fi密码 # Telegram Bot 配置 TELEGRAM_BOT_TOKEN 你的BotToken TELEGRAM_CHAT_ID 你的ChatID # 注意这里可以是字符串或数字但JSON传输时需统一类型 # GitHub 仓库配置 GITHUB_REPO_OWNER micropython GITHUB_REPO_NAME micropython GITHUB_API_URL fhttps://api.github.com/repos/{GITHUB_REPO_OWNER}/{GITHUB_REPO_NAME} # 设置GitHub API请求头使用Token可以避免速率限制可选但推荐 GITHUB_HEADERS { # ‘Authorization’: ‘token your_github_personal_access_token‘, # 如果需要取消注释并填入Token User-Agent: PicoW-GitHub-Monitor/1.0 } # 监控配置 CHECK_INTERVAL 300 # 检查间隔单位秒 (例如 300秒5分钟) STATE_FILE repo_state.json # 用于存储上次状态的本地文件名 # 函数定义 def connect_wifi(ssid, password): 连接Wi-Fi网络 wlan network.WLAN(network.STA_IF) wlan.active(True) if not wlan.isconnected(): print(f‘正在连接Wi-Fi: {ssid}...‘) wlan.connect(ssid, password) # 等待连接最多20秒 max_wait 20 while max_wait 0: if wlan.isconnected(): break max_wait - 1 print(‘等待连接...‘, max_wait) time.sleep(1) if wlan.isconnected(): print(‘Wi-Fi连接成功!‘) print(‘网络配置:‘, wlan.ifconfig()) return True else: print(‘Wi-Fi连接失败!‘) return False def send_telegram_message(bot_token, chat_id, text): 通过Telegram Bot发送消息 url fhttps://api.telegram.org/bot{bot_token}/sendMessage payload { chat_id: chat_id, text: text, parse_mode: Markdown # 使用Markdown格式让消息更易读 } headers {‘Content-Type‘: ‘application/json‘} try: # 使用json参数urequests会自动处理序列化和headers response requests.post(url, jsonpayload, headersheaders) resp_json response.json() response.close() # 重要关闭响应释放资源 if resp_json.get(‘ok‘): print(fTelegram消息发送成功。) return True else: print(fTelegram消息发送失败: {resp_json}) return False except Exception as e: print(f发送Telegram消息时发生异常: {e}) return False def get_latest_commit_sha(): 获取仓库默认分支的最新提交SHA try: # 先获取仓库信息其中包含默认分支名 repo_info requests.get(GITHUB_API_URL, headersGITHUB_HEADERS) repo_data repo_info.json() repo_info.close() default_branch repo_data.get(‘default_branch‘, ‘main‘) # 通常是main或master # 获取该分支的最新提交 commits_url f{GITHUB_API_URL}/commits/{default_branch} response requests.get(commits_url, headersGITHUB_HEADERS) # GitHub API可能会返回一个包含多个提交的数组我们取第一个最新 commits_data response.json() response.close() if isinstance(commits_data, dict) and ‘sha‘ in commits_data: # 当直接请求单个提交时返回的是对象 latest_sha commits_data[‘sha‘] elif isinstance(commits_data, list) and len(commits_data) 0: # 当请求分支提交列表时返回的是数组 latest_sha commits_data[0][‘sha‘] else: print(无法从响应中解析出提交SHA) return None # SHA很长我们只取前7位作为简短标识类似git log --oneline显示的 short_sha latest_sha[:7] return short_sha except Exception as e: print(f获取最新提交SHA时出错: {e}) return None def get_latest_release_tag(): 获取仓库最新的发布版本标签 releases_url f{GITHUB_API_URL}/releases/latest try: response requests.get(releases_url, headersGITHUB_HEADERS) # 注意如果没有任何Release这里会返回404 if response.status_code 404: print(该仓库尚未创建任何Release。) response.close() return None release_data response.json() response.close() latest_tag release_data.get(‘tag_name‘) return latest_tag except Exception as e: print(f获取最新Release时出错: {e}) return None def load_previous_state(): 从本地文件加载上一次的仓库状态 try: with open(STATE_FILE, ‘r‘) as f: state json.load(f) print(f已加载历史状态: {state}) return state except (OSError, ValueError): # 文件不存在或内容无效返回空状态 print(未找到历史状态文件或文件无效将创建新状态。) return {‘last_commit_sha‘: None, ‘last_release_tag‘: None} def save_current_state(commit_sha, release_tag): 将当前状态保存到本地文件 state { ‘last_commit_sha‘: commit_sha, ‘last_release_tag‘: release_tag } try: with open(STATE_FILE, ‘w‘) as f: json.dump(state, f) print(f状态已保存: {state}) except Exception as e: print(f保存状态文件时出错: {e}) # 主程序循环 def main(): print( GitHub仓库监控器 (PICO W) 启动 ) # 1. 连接Wi-Fi if not connect_wifi(WIFI_SSID, WIFI_PASSWORD): # 连接失败可能是配置错误或信号问题可以尝试重启或进入深度睡眠 print(Wi-Fi连接失败程序终止。) return # 2. 加载上一次的监控状态 previous_state load_previous_state() last_known_commit previous_state.get(‘last_commit_sha‘) last_known_release previous_state.get(‘last_release_tag‘) # 3. 首次启动获取当前状态并保存但不通知避免上电就发一条消息 print(首次运行初始化状态...) current_commit get_latest_commit_sha() current_release get_latest_release_tag() if current_commit: # 如果之前没有记录或者记录与当前不同理论上首次运行一定不同则更新状态 if last_known_commit ! current_commit: save_current_state(current_commit, current_release) print(f初始提交状态已记录: {current_commit}) if last_known_commit is not None: # 如果之前有记录且不同说明是程序重启后发现了新提交应该通知 message f *仓库更新通知* \n\n message f仓库: *{GITHUB_REPO_OWNER}/{GITHUB_REPO_NAME}*\n message f有新的提交\n message f最新提交SHA: {current_commit}\n message f[查看提交对比](https://github.com/{GITHUB_REPO_OWNER}/{GITHUB_REPO_NAME}/compare/{last_known_commit}...{current_commit}) send_telegram_message(TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, message) else: print(无法获取初始提交状态监控可能无法正常工作。) # 4. 进入主监控循环 print(f进入监控循环每 {CHECK_INTERVAL} 秒检查一次...) while True: try: print(f\n--- 开始新一轮检查 ({time.ticks_ms()}) ---) # 获取当前最新状态 current_commit get_latest_commit_sha() current_release get_latest_release_tag() message_parts [] # 用于组装通知消息 # 检查提交更新 if current_commit and current_commit ! last_known_commit: print(f检测到新提交! 旧: {last_known_commit}, 新: {current_commit}) msg_part f *代码提交更新*\n msg_part f仓库: {GITHUB_REPO_OWNER}/{GITHUB_REPO_NAME}\n msg_part f最新提交: {current_commit}\n if last_known_commit: msg_part f[对比更改](https://github.com/{GITHUB_REPO_OWNER}/{GITHUB_REPO_NAME}/compare/{last_known_commit}...{current_commit}) else: msg_part f[查看提交](https://github.com/{GITHUB_REPO_OWNER}/{GITHUB_REPO_NAME}/commit/{current_commit}) message_parts.append(msg_part) last_known_commit current_commit # 更新内存中的状态 # 检查发布更新 if current_release and current_release ! last_known_release: print(f检测到新发布! 旧: {last_known_release}, 新: {current_release}) msg_part f *新版本发布!*\n msg_part f仓库: {GITHUB_REPO_OWNER}/{GITHUB_REPO_NAME}\n msg_part f最新版本: *{current_release}*\n msg_part f[查看Release详情](https://github.com/{GITHUB_REPO_OWNER}/{GITHUB_REPO_NAME}/releases/tag/{current_release}) message_parts.append(msg_part) last_known_release current_release # 更新内存中的状态 # 如果有任何更新发送通知并保存状态到文件 if message_parts: full_message *GitHub仓库监控通知* \n\n \n\n.join(message_parts) if send_telegram_message(TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, full_message): # 只有通知发送成功才更新持久化状态避免网络波动导致状态丢失 save_current_state(last_known_commit, last_known_release) else: print(通知发送失败本次状态变更将不会保存下次循环会重新检测。) else: print(未检测到更新。) except Exception as e: print(f主循环发生未知错误: {e}) # 可以选择记录错误或尝试恢复 # 等待下一个检查周期 print(f等待 {CHECK_INTERVAL} 秒后再次检查...) time.sleep(CHECK_INTERVAL) # 运行主程序 if __name__ ‘__main__‘: main()4. 脚本深度解析与关键问题处理上面的脚本可以直接运行但理解其关键细节和潜在问题能让你更好地定制和排错。4.1 状态持久化策略为什么用文件而不用内存PICO在断电后RAM中的数据会丢失。因此我们必须将上一次检测到的状态提交SHA和发布标签持久化存储。这里选择了PICO内置的Flash文件系统将状态保存为一个JSON文件repo_state.json。每次启动时脚本会读取这个文件获取“上一次已知的状态”当检测到更新并成功发送通知后再将新状态写回文件。这样即使PICO意外重启也不会重复发送旧通知。4.2 GitHub API调用优化与限流处理GitHub API对未认证的请求有严格的速率限制每小时60次。我们的脚本每5分钟300秒请求一次每小时12次远低于限制通常没有问题。但为了更稳妥以及访问私有仓库我强烈建议使用Personal Access Token (PAT)。生成PAT在GitHub Settings - Developer settings - Personal access tokens - Tokens (classic) 中生成一个Token只需勾选repo访问仓库信息权限即可。使用PAT将生成的Token填入脚本GITHUB_HEADERS字典的‘Authorization‘: ‘token ghp_xxxxxx‘中取消注释并替换。这样可以将速率限制提升至每小时5000次并可以访问你有权限的私有仓库。4.3 异常处理与网络鲁棒性MicroPython运行在嵌入式环境网络不稳定是常态。脚本中使用了try...except块来包裹关键的网络请求get_latest_commit_sha,get_latest_release_tag,send_telegram_message。一旦发生超时、DNS解析失败或HTTP错误异常会被捕获并打印错误信息而不会导致整个程序崩溃。程序会继续执行等待下一个循环周期再次尝试。4.4 首次运行逻辑避免“开机轰炸”脚本在首次运行或状态文件丢失时有一个特殊逻辑它会获取当前仓库状态并保存但不会发送通知。除非它检测到状态文件中的旧记录与当前状态不同这通常发生在脚本更新后或手动删除状态文件后再次运行。这个设计避免了每次给PICO上电或重置时都向Telegram发送一条“发现新状态”的冗余通知。4.5 消息格式与用户体验Telegram消息使用了Markdown格式parse_mode: “Markdown”使消息更清晰*加粗*用于强调关键信息如仓库名、版本号。代码用于显示提交SHA。[链接文字](URL)用于嵌入可直接点击的链接如对比链接或Release页面。这大大提升了通知的实用性和可操作性。5. 部署、测试与进阶玩法5.1 部署到PICO W并实现开机自启使用Thonny将完整的main.py脚本上传到Pico W的根目录。确保文件名是main.py。MicroPython设备上电后会自动寻找并执行根目录下的main.py或boot.py文件。上传完成后可以按一下Pico W的复位键RST或者重新拔插USB线脚本就会自动运行。通过Thonny的“Shell”窗口可以看到串口输出的日志信息。5.2 功能测试与验证模拟更新最直接的测试方法是手动向你监控的仓库推送一个提交或创建一个新的Release。等待一个检查周期例如5分钟观察Telegram是否收到通知。修改状态文件你也可以通过Thonny的文件管理器直接修改Pico W上的repo_state.json文件将last_commit_sha改成一个旧的、不存在的SHA值。保存后脚本在下个周期会发现“变化”并发送通知。这是一种安全的测试方法。观察日志通过串口输出你可以清晰地看到每个循环的步骤连接状态、API请求结果、比对结果、发送消息是否成功。这是最重要的调试手段。5.3 进阶定制与扩展思路这个基础框架有很大的扩展空间监控多个仓库将GITHUB_REPO_OWNER和GITHUB_REPO_NAME改为列表在主循环中遍历每个仓库进行检查。注意调整状态文件的结构以存储多个仓库的状态。监控特定分支或PR修改get_latest_commit_sha函数将API请求的URL指向特定分支如.../commits/develop或拉取请求.../pulls。丰富通知内容从GitHub API的响应中提取更多信息如提交者、提交信息摘要、变更文件数量等并填充到Telegram消息中。降低功耗如果使用电池供电可以在time.sleep()期间将Pico W置于深度睡眠模式。但这需要额外的硬件连接如GPIO唤醒和更复杂的程序逻辑因为深度睡眠会断开Wi-Fi唤醒后需要重新连接。添加本地指示利用Pico W的板载LEDGPIO 25或外接一个LED在检测到更新时闪烁提供本地视觉反馈。接入其他通知服务除了Telegram可以类似地集成Server酱微信、BarkiOS、钉钉机器人、Slack等只需替换send_telegram_message函数为对应服务的API调用。5.4 常见问题与排查无法连接Wi-Fi检查SSID和密码是否正确确保Pico W在路由器信号覆盖范围内。可以尝试在connect_wifi函数中增加更长的等待时间或重试逻辑。Telegram消息发送失败检查Bot Token和Chat ID是否正确。确保你已经向Bot发送过/start消息。通过浏览器直接访问https://api.telegram.org/botYourBotToken/getMe可以测试Token是否有效。GitHub API返回403或404403通常是速率超限请考虑添加GitHub Token。404可能是仓库地址拼写错误或者请求的API端点不存在例如仓库确实没有Release时请求/releases/latest会返回404脚本已处理此情况。程序运行一段时间后停止可能是内存泄漏或网络异常导致程序崩溃。确保在每次urequests请求后都调用了response.close()来释放资源。考虑在顶层while循环外再包裹一个异常捕获实现“看门狗”机制在发生不可恢复错误时软重启。状态文件损坏如果文件写入过程中断电可能导致JSON文件格式错误。可以在load_previous_state函数中增加更严格的异常处理在文件损坏时删除并重建一个默认状态文件。这个基于PICO的GitHub仓库监控器虽然代码量不大但完整地实践了嵌入式设备接入互联网服务IoT的典型流程网络连接、数据获取、逻辑处理、状态持久化、消息推送。它提供了一个极具性价比的自动化解决方案让你不再错过任何重要的代码更新。你可以根据自己的需求轻松地修改和扩展它打造属于你自己的智能开发助手。
返回列表