本文將詳細介紹如何使用 Cloudflare Worker、Hono 框架和 Telegram Bot API 構建一個自動化系統，該系統能夠監控 RSS 訂閱源並將最新更新推送到 Telegram 頻道。這個解決方案完全基於 Serverless 架構，無需管理伺服器，且大部分情況下可以在 Cloudflare 的免費方案內運行，非常適合個人開發者和小型團隊使用。
技術概述
Cloudflare Worker
Cloudflare Worker 是一種 Serverless 運算平台，允許開發者在 Cloudflare 的全球邊緣網絡上運行 JavaScript 代碼。它具有低延遲、高可用性和零冷啟動時間的特點，非常適合構建輕量級的 Web 應用和自動化任務。
Cloudflare Worker 提供了免費方案，每天允許 100,000 次請求，足以滿足大多數個人項目的需求。此外，它還支持定時觸發器（Cron Triggers），可以按照預定的時間表自動執行代碼，這正是我們 RSS 訂閱推送系統所需的功能。
Hono 框架
Hono 是一個專為邊緣計算環境設計的輕量級 Web 框架，特別適合在 Cloudflare Worker 上運行。它提供了簡潔的 API 設計和路由系統，使得在 Serverless 環境中構建 Web 應用變得更加容易。
Hono 的特點包括：
- 零依賴，體積小，適合在邊緣環境運行
- 支持中間件系統，便於功能擴展
- 提供類型安全的 API 設計，對 TypeScript 友好
- 性能卓越，專為低延遲優化
Telegram Bot API
Telegram Bot API 允許開發者創建機器人，這些機器人可以執行各種操作，包括發送消息、圖片和文件等。通過創建一個 Telegram Bot 並將其添加到頻道中，我們可以實現自動將 RSS 更新推送到 Telegram 頻道的功能。
開發環境準備
先決條件
在開始之前，請確保您已經：
- 註冊了 Cloudflare 帳戶
- 安裝了 Node.js 和 npm
- 創建了 Telegram 機器人並獲取了 API Token
- 創建了 Telegram 頻道並將機器人添加為管理員
安裝 Wrangler CLI
Wrangler 是 Cloudflare 提供的命令行工具，用於開發和部署 Cloudflare Worker。我們首先需要全局安裝 Wrangler：
npm install -g wrangler安裝完成後，使用以下命令登錄到您的 Cloudflare 帳戶：
wrangler login這會打開瀏覽器窗口，要求您授權 Wrangler 訪問您的 Cloudflare 帳戶。
項目設置
創建新項目
使用 Wrangler 創建一個新的 Cloudflare Worker 項目：
wrangler init rss-to-telegram
cd rss-to-telegram
然後，安裝必要的依賴：
npm install hono rss-parser node-fetch配置項目
編輯 wrangler.toml 文件，添加必要的配置：
name = "rss-to-telegram"
main = "src/index.ts"
compatibility_flags = [ "nodejs_compat" ]
compatibility_date = "2024-09-23"
[triggers]
crons = ["0 * * * *"] # 每小時執行一次注意：不要在配置文件中直接存儲敏感信息如 Telegram Bot Token。我們將在後面使用 Cloudflare 的環境變量來安全地存儲這些值。
實現 RSS 解析功能
首先，我們需要創建一個函數來獲取和解析 RSS feed。在 src 目錄下創建一個新文件 rss.ts：
import Parser from 'rss-parser';
// 定義 Feed 項目的類型
export interface FeedItem {
title: string;
link: string;
pubDate: string;
content?: string;
contentSnippet?: string;
}
// 使用 fetch API 獲取資料
async function fetchUrl(url: string): Promise<string> {
console.log('使用 fetch 獲取：', url);
try {
const response = await fetch(url);
if (!response.ok) {
throw new Error(`請求失敗，狀態碼：${response.status}`);
}
const contentType = response.headers.get('content-type') || '';
if (!contentType.includes('xml') && !contentType.includes('rss')) {
console.warn(`警告：內容類型可能不是 XML: ${contentType}`);
}
return await response.text();
} catch (error) {
console.error('獲取 URL 時出錯：', error);
throw error;
}
}
// 獲取最新的 RSS 項目
export async function getLatestRssItems(feedUrl: string): Promise<FeedItem[]> {
console.log('獲取 RSS feed:', feedUrl);
const parser = new Parser();
try {
// 使用 fetch API 獲取 RSS 內容
const xmlContent = await fetchUrl(feedUrl);
// 使用 rss-parser 解析 XML 內容
const feed = await parser.parseString(xmlContent);
// 獲取當前時間和 24 小時前的時間點
const now = new Date();
const twentyFourHoursAgo = new Date(now);
twentyFourHoursAgo.setHours(now.getHours() - 24);
console.log(`過濾 24 小時內的項目，當前時間：${now.toISOString()}, 24 小時前：${twentyFourHoursAgo.toISOString()}`);
// 按發布日期排序，最新的在前，並只保留 24 小時內的項目
const sortedItems = feed.items
.map(item => ({
title: item.title || '無標題',
link: item.link || '',
pubDate: item.pubDate || new Date().toISOString(),
content: item.content,
contentSnippet: item.contentSnippet
}))
.filter(item => {
const pubDate = new Date(item.pubDate);
return pubDate >= twentyFourHoursAgo;
})
.sort((a, b) => new Date(b.pubDate).getTime() - new Date(a.pubDate).getTime());
console.log(`找到 ${feed.items.length} 個項目，其中 ${sortedItems.length} 個在 24 小時內`);
return sortedItems;
} catch (error) {
console.error('獲取 RSS feed 時出錯：', error);
return [];
}
}實現 Telegram 集成
接下來，我們創建一個用於發送消息到 Telegram 頻道的模塊。在 src 目錄下創建 telegram.ts：
// 發送消息到 Telegram 頻道
export async function sendToTelegram(
botToken: string,
channelId: string,
message: string
): Promise<boolean> {
const url = `https://api.telegram.org/bot${botToken}/sendMessage`;
try {
const response = await fetch(url, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
chat_id: channelId,
text: message,
parse_mode: 'HTML',
disable_web_page_preview: false,
}),
});
if (!response.ok) {
const error = await response.json();
console.error('Telegram API 錯誤:', error);
return false;
}
return true;
} catch (error) {
console.error('發送到 Telegram 時出錯:', error);
return false;
}
}
// 格式化 RSS 項目為 Telegram 消息
export function formatRssItemForTelegram(item: {
title: string;
link: string;
contentSnippet?: string;
}): string {
const snippet = item.contentSnippet
? `\n\n${item.contentSnippet.substring(0, 150)}${
item.contentSnippet.length > 150 ? '...' : ''
}`
: '';
return `${escapeHtml(item.title)}${snippet}\n\n閱讀全文`;
}
// HTML 特殊字符轉義
function escapeHtml(text: string): string {
return text.replace(/&/g, '&')
.replace(/</g, '<')
.replace(/>/g, '>')
.replace(/"/g, '"');
}實現存儲功能
為了避免重複發送同一篇文章，我們需要記錄已經發送過的文章。由於 Cloudflare Worker 是無狀態的，我們將使用 Cloudflare KV 來存儲這些信息。
首先，運行以下命令創建 KV 命名空間，並獲取 ID：
wrangler kv namespace create "RSS_STORAGE"
然後，更新 wrangler.toml 文件，添加 KV 配置：
[[kv_namespaces]]
binding = "RSS_STORAGE"
id = "<ID>"接下來，創建 src/storage.ts 文件：
export interface Storage {
getProcessedItems(): Promise<string[]>;
markItemAsProcessed(itemUrl: string): Promise<void>;
}
export class KVStorage implements Storage {
constructor(private kvNamespace: KVNamespace) {}
async getProcessedItems(): Promise<string[]> {
const items = await this.kvNamespace.get('processed_items');
return items ? JSON.parse(items) : [];
}
async markItemAsProcessed(itemUrl: string): Promise<void> {
const items = await this.getProcessedItems();
// 如果項目已存在，則不需要再添加
if (items.includes(itemUrl)) return;
// 添加新項目並保留最近的 100 個項目（防止存儲過大）
const updatedItems = [itemUrl, ...items].slice(0, 100);
await this.kvNamespace.put('processed_items', JSON.stringify(updatedItems));
}
}使用 Hono 構建應用
現在，我們可以使用 Hono 框架創建我們的主應用。編輯 src/index.ts 文件：
import { Hono } from 'hono';
import { getLatestRssItems } from './rss';
import { sendToTelegram, formatRssItemForTelegram } from './telegram';
import { KVStorage } from './storage';
// 定義環境變量的類型
interface Env {
RSS_FEED_URL: string;
TELEGRAM_BOT_TOKEN: string;
TELEGRAM_CHANNEL_ID: string;
RSS_STORAGE: KVNamespace;
}
const app = new Hono();
// 健康檢查路由
app.get('/', (c) => {
return c.text('RSS 到 Telegram 推送服務正在運行中！');
});
// 手動觸發更新的路由
app.get('/check-updates', async (c) => {
const env = c.env;
try {
const result = await checkAndSendUpdates(env);
return c.json({ success: true, sent: result.sent, error: result.error });
} catch (error) {
return c.json({ success: false, error: error.message }, 500);
}
});
// 檢查更新並發送到 Telegram 的主要邏輯
async function checkAndSendUpdates(env: Env) {
const storage = new KVStorage(env.RSS_STORAGE);
const processedItems = await storage.getProcessedItems();
// 獲取最新的 RSS 項目
const latestItems = await getLatestRssItems(env.RSS_FEED_URL);
// 過濾出尚未處理的項目
const newItems = latestItems.filter(
(item) => item.link && !processedItems.includes(item.link)
);
if (newItems.length === 0) {
return { sent: 0, error: null };
}
let sent = 0;
let error = null;
// 發送新項目到 Telegram
for (const item of newItems) {
try {
const message = formatRssItemForTelegram(item);
const success = await sendToTelegram(
env.TELEGRAM_BOT_TOKEN,
env.TELEGRAM_CHANNEL_ID,
message
);
if (success) {
await storage.markItemAsProcessed(item.link);
sent++;
}
} catch (err) {
console.error('發送項目時出錯：', err);
error = err.message;
// 繼續處理其他項目
}
}
return { sent, error };
}
// 導出 Cloudflare Worker 入口點
export default {
fetch: app.fetch,
// 定時任務處理函數
async scheduled(event: ScheduledEvent, env: Env, ctx: ExecutionContext) {
ctx.waitUntil(checkAndSendUpdates(env));
},
};部署到 Cloudflare
在部署之前，我們需要設置環境變量。這些敏感信息不應該存儲在代碼中：
wrangler secret put TELEGRAM_BOT_TOKEN
# 按照提示輸入您的 Telegram Bot Token
wrangler secret put TELEGRAM_CHANNEL_ID
# 輸入您的 Telegram 頻道 ID，格式如 @channelname 或 -100xxxxxxxxxx然後，部署您的 Worker：
wrangler deploy部署完成後，您將看到 Worker 的 URL，例如 https://rss-to-telegram.username.workers.dev。
測試和故障排除
手動觸發更新
您可以通過訪問 /check-updates 端點來手動觸發更新檢查：
https://rss-to-telegram.username.workers.dev/check-updates查看日誌
使用以下命令查看 Worker 的日誌，這對於調試非常有用：
wrangler tail常見問題
- RSS 解析錯誤：確保 RSS Feed URL 是正確的，並且返回有效的 XML。
- Telegram 發送失敗：檢查 Bot Token 是否正確，並確保機器人已被添加到頻道並具有發送消息的權限。
- 定時任務不執行：確保
wrangler.toml中的 cron 表達式格式正確。
進階優化
支持多個 RSS 訂閱
要支持多個 RSS 訂閱，我們可以修改配置和代碼：
// 在 wrangler.toml 中，將 RSS_FEED_URL 改為 JSON 格式的字符串
// [vars]
// RSS_FEEDS = "[{\"url\":\"https://example1.com/feed.xml\",\"name\":\"Blog 1\"},{\"url\":\"https://example2.com/feed.xml\",\"name\":\"Blog 2\"}]"
// 然後在代碼中解析這個 JSON
const feeds = JSON.parse(env.RSS_FEEDS);
for (const feed of feeds) {
// 為每個訂閱獲取更新
}添加自定義消息格式
您可以擴展 formatRssItemForTelegram 函數，支持更多的自定義格式：
export function formatRssItemForTelegram(
item: FeedItem,
format: 'default' | 'compact' | 'detailed' = 'default'
): string {
switch (format) {
case 'compact':
return `${escapeHtml(item.title)}\n閱讀全文`;
case 'detailed':
return `${escapeHtml(item.title)}\n\n${
item.contentSnippet || ''
}\n\n發布時間：${new Date(item.pubDate).toLocaleString()}\n閱讀全文`;
default:
// 原始格式
return `${escapeHtml(item.title)}${
item.contentSnippet ? `\n\n${item.contentSnippet.substring(0, 150)}...` : ''
}\n\n閱讀全文`;
}
}添加錯誤通知
當系統出現問題時，您可能希望收到通知。可以添加一個函數來發送錯誤報告：
async function sendErrorNotification(
botToken: string,
adminChatId: string,
error: string
): Promise {
await sendToTelegram(
botToken,
adminChatId,
`❌ RSS 推送系統出現錯誤：\n\n${error}`
);
}開源代碼
我已經將本文中介紹的 RSS 訂閱推送系統的完整代碼開源在 GitHub 上，歡迎大家訪問和使用：
https://github.com/calpa/rss-to-telegram
主要功能
- 🔄 自動檢查 RSS 訂閱源的新內容
- ⏱️ 過濾最近 24 小時內發布的內容
- 📲 將新文章發送到指定的 Telegram 頻道
- 🚫 通過追蹤已發送的項目防止重複發送
- ⏰ 通過 Cloudflare Workers cron 觸發器定時運行（默認每小時一次）
- 🔘 提供手動更新端點，可按需檢查更新
如何使用
- 克隆倉庫
- 安裝依賴：
npm install - 配置環境變量：
wrangler secret put RSS_FEED_URL wrangler secret put TELEGRAM_BOT_TOKEN wrangler secret put TELEGRAM_CHANNEL_ID - 部署到 Cloudflare Workers：
npm run deploy
完整的設置和使用說明請參考 GitHub 倉庫中的 README 文件。
如果您對這個項目有任何問題或建議，歡迎在 GitHub 上提交 issue 或 pull request。
結論
在本文中，我們完成了一個使用 Cloudflare Worker、Hono 框架和 Telegram Bot API 構建的 RSS 訂閱推送系統。這個系統能夠定期檢查 RSS 訂閱源的更新，並將新文章自動推送到 Telegram 頻道，是一個實用的內容自動化分發工具。
這個項目展示了 Serverless 架構的強大之處，我們無需管理伺服器，只需關注業務邏輯的實現。使用 Cloudflare Worker，我們能夠以幾乎零成本運行這個系統，適合個人開發者和小型項目使用。
進一步，您可以擴展這個系統，添加更多功能，如支持更多類型的內容源、實現內容過濾、添加更多的通知渠道等。由於 Hono 框架的靈活性，這些擴展都可以相對容易地實現。
希望本文對您理解如何使用現代 Web 技術構建實用的自動化工具有所幫助。
