长毛象新通知推送TGBot

长毛象新通知推送TGBot

2025-04-21
#工具箱 , #分享 , #编码

其实蜗牛哥之前开发过一个Chrome扩展，可以在浏览器里显示一个图标和通知数量。但是我电脑上现在Chrome一打开十几个进程，我实在不想再新增扩展了。

所以这两天用Deepseek糊了一个搭配Cloudflare推送到TGBot的Worker脚本，效果如截图。设置好长毛象和TGBot的设置后，程序会每5分钟检测一次有没有新通知，有的话才进行推送操作。

准备KV

先添加一个KV，名称 KV_STORE ，备用。

准备长毛象TOken

获取长毛象Access Token，不放心的话在权限部分可以单独只设置通知获取权限

  1. 登录你的 Mastodon 实例（如 https://mastodon.social）。
  2. 进入 "Preferences（偏好设置） > Development（开发）（或直接访问 https:///settings/applications）。
  3. 点击 "New Application（新建应用）：
    1. Application Name（应用名称）：填写你的应用名称（如 MyBot）。
    2. Website（网站）（可选）：填写你的应用网站（如果没有可留空）。
    3. Scopes（权限）：选择你需要的 API 权限（如 read、write、follow 等）。
  4. 点击 "Submit（提交），系统会生成：
    1. Client Key（客户端 ID）
    2. Client Secret（客户端密钥）
    3. Access Token（访问令牌）（可直接使用）

准备Cloudflare Worker

再添加一个Worker，代码如下，并修改代码中的 config 部分的配置为你自己的设置，其中长毛象token

// 配置部分
const config = {
  // Mastodon 配置
  mastodon: {
    instance: 'https://your-instance.social', // 替换为你的 Mastodon 实例地址
    accessToken: 'ZTDJN2ZMZMZI5MTU1MZHH',  // 替换为你的 Mastodon 访问令牌
    lastNotificationIdKey: 'last_notification_id' // KV 存储中保存最后通知ID的键名
  },
  
  // Telegram 配置
  telegram: {
    botToken: 'your-bot-token', // 替换为你的 Telegram Bot Token
    chatId: 'your-tg-chart-id'               // 替换为接收消息的聊天ID
  },
  
  // 检查间隔（分钟）
  checkInterval: 5
};

// 主处理函数
export default {
  async scheduled(event, env, ctx) {
    // 执行检查通知任务
    await checkNotifications(env);
  },
  
  async fetch(request, env) {
    // 手动触发检查
    if (new URL(request.url).pathname === '/check') {
      await checkNotifications(env);
      return new Response('Notification check triggered');
    }
    
    return new Response('Not found', { status: 404 });
  }
};

// 检查未读通知
async function checkNotifications(env) {
  try {
    // 获取上次处理的通知ID
    let lastNotificationId = await env.KV_STORE.get(config.mastodon.lastNotificationIdKey);
    lastNotificationId = lastNotificationId || '0';
    
    // 获取新通知
    const notifications = await fetchMastodonNotifications(lastNotificationId);
    
    if (notifications.length > 0) {
      // 有新通知，发送到 Telegram
      await sendToTelegram(notifications, env);
      
      // 更新最后处理的通知ID
      const latestId = notifications[0].id;
      await env.KV_STORE.put(config.mastodon.lastNotificationIdKey, latestId);
      
      console.log(`Sent ${notifications.length} new notifications to Telegram. Latest ID: ${latestId}`);
    } else {
      console.log('No new notifications.');
    }
  } catch (error) {
    console.error('Error checking notifications:', error);
  }
}

// 从 Mastodon 获取通知
async function fetchMastodonNotifications(sinceId) {
  const url = new URL(`${config.mastodon.instance}/api/v1/notifications`);
  url.searchParams.append('exclude_types[]', 'follow');
  url.searchParams.append('exclude_types[]', 'follow_request');
  url.searchParams.append('since_id', sinceId);
  
  const response = await fetch(url.toString(), {
    headers: {
      'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36',
      'Authorization': `Bearer ${config.mastodon.accessToken}`
    }
  });
  
  if (!response.ok) {
    throw new Error(`Mastodon API error: ${response.status} ${response.statusText}`);
  }
  
  return await response.json();
}

// 发送通知到 Telegram
async function sendToTelegram(notifications, env) {
  // 按时间倒序排列通知
  notifications.sort((a, b) => new Date(b.created_at) - new Date(a.created_at));
  
  let message = `📨 你有 ${notifications.length} 条新通知:\n\n`;
  
  notifications.forEach(notification => {
    const sender = notification.account; // 通知发起者
    const senderName = sender.display_name || sender.username;
    const senderHandle = `@${sender.acct}`;
    const senderUrl = sender.url;
    
    // 根据不同通知类型构建不同消息
    switch(notification.type) {
      case 'mention':
        const mentionContent = stripHTML(notification.status.content);
        message += `💬 <b>${senderName}</b> (${senderHandle}) 提到了你:\n${mentionContent}\n\n`;
        break;
        
      case 'reply':
        const replyContent = stripHTML(notification.status.content);
        message += `↩️ <b>${senderName}</b> (${senderHandle}) 回复了你:\n${replyContent}\n\n`;
        break;
        
      case 'reblog':
        const reblogContent = notification.status 
          ? stripHTML(notification.status.content) 
          : "[内容不可用]";
        message += `🔁 <b>${senderName}</b> (${senderHandle}) 转发了你的嘟文:\n${reblogContent}\n\n`;
        break;
        
      case 'favourite':
        const favContent = notification.status 
          ? stripHTML(notification.status.content) 
          : "[内容不可用]";
        message += `⭐ <b>${senderName}</b> (${senderHandle}) 喜欢了你的嘟文:\n${favContent}\n\n`;
        break;
        
      case 'poll':
        message += `📊 <b>${senderName}</b> (${senderHandle}) 的投票已结束\n`;
        break;
        
      case 'follow':
        message += `👤 <b>${senderName}</b> (${senderHandle}) 关注了你\n`;
        break;
        
      case 'follow_request':
        message += `🫂 <b>${senderName}</b> (${senderHandle}) 请求关注你\n`;
        break;
        
      default:
        message += `ℹ️ 你有一条新通知 (${notification.type}) 来自 <b>${senderName}</b>\n`;
    }
  });
  
  // 添加来源链接
  message += `\n查看所有通知: ${config.mastodon.instance}/notifications`;
  
  // 发送到 Telegram
  const url = `https://api.telegram.org/bot${config.telegram.botToken}/sendMessage`;
  
  const response = await fetch(url, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({
      chat_id: config.telegram.chatId,
      text: message,
      disable_web_page_preview: true,
      parse_mode: 'HTML'
    })
  });
  
  if (!response.ok) {
    throw new Error(`Telegram API error: ${response.status} ${response.statusText}`);
  }
}

// 去除 HTML 标签
function stripHTML(html) {
  return html.replace(/<[^>]*>?/gm, '');
}

配置KV和定时执行

去Cloudflare Worker设置页面绑定KV和设置定时执行。

除了定时执行外，你还可以用 https://wokrerurl.dev/check 手动触发
258、谷雨

加入评论