《利用Telegram官方Bot构建自动化下载通知系统：代码实现与部署》

import logging
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes

# 配置日志
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)

# 用您从@BotFather获得的Token替换此处
BOT_TOKEN = "YOUR_BOT_TOKEN_HERE"

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """用户发送/start时调用，将其加入订阅列表"""
    user = update.effective_user
    user_id = user.id
    # 此处应调用函数将user_id存入数据库的订阅表
    # add_subscriber(user_id)
    await update.message.reply_html(
        rf"Hi {user.mention_html()}! 您已成功订阅Telegram官方版本更新通知。当检测到新版安装包时，我会第一时间通知您！"
    )

async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """发送帮助信息"""
    await update.message.reply_text("可用命令:\n/start - 订阅更新通知\n/unsubscribe - 取消订阅\n/status - 查看当前监控状态")

def main():
    """启动Bot"""
    application = Application.builder().token(BOT_TOKEN).build()
    application.add_handler(CommandHandler("start", start))
    application.add_handler(CommandHandler("help", help_command))
    # 启动Bot，使用Webhook或轮询（Polling）。开发阶段常用轮询。
    application.run_polling(allowed_updates=Update.ALL_TYPES)

if __name__ == '__main__':
    main()

import requests
import hashlib
from datetime import datetime

def fetch_latest_desktop_release():
    """从GitHub API获取Telegram Desktop最新发布信息"""
    url = "https://api.github.com/repos/telegramdesktop/tdesktop/releases/latest"
    headers = {'Accept': 'application/vnd.github.v3+json'}
    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
        data = response.json()
        release_info = {
            'tag_name': data.get('tag_name'), # 版本号，如 "v4.0.0"
            'name': data.get('name'),
            'published_at': data.get('published_at'),
            'body': data.get('body', '')[:500], # 更新日志摘要
            'assets': []
        }
        for asset in data.get('assets', []):
            if asset['name'].endswith('.exe') or asset['name'].endswith('.dmg') or asset['name'].endswith('.AppImage'):
                release_info['assets'].append({
                    'name': asset['name'],
                    'download_url': asset['browser_download_url'],
                    'size': asset['size']
                })
        return release_info
    except requests.RequestException as e:
        logger.error(f"获取GitHub Release失败: {e}")
        return None

def check_for_update(stored_version, latest_release_info):
    """比对版本，检测是否有更新"""
    if not latest_release_info:
        return False, None
    latest_version = latest_release_info['tag_name']
    # 简单的字符串比较，对于语义化版本（SemVer）可能需要更复杂的解析库
    if latest_version != stored_version:
        logger.info(f"检测到新版本: {stored_version} -> {latest_version}")
        return True, latest_release_info
    return False, None

import sqlite3
import json

DB_PATH = "telegram_monitor.db"

def init_database():
    """初始化数据库，创建表"""
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    # 存储版本信息的表
    c.execute('''CREATE TABLE IF NOT EXISTS version_history
                 (platform TEXT PRIMARY KEY, current_version TEXT, release_info TEXT, last_checked TIMESTAMP)''')
    # 存储订阅用户的表
    c.execute('''CREATE TABLE IF NOT EXISTS subscribers
                 (user_id INTEGER PRIMARY KEY, subscribed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)''')
    conn.commit()
    conn.close()

def update_version_in_db(platform, version, release_info):
    """更新或插入平台版本信息"""
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    release_info_json = json.dumps(release_info)
    c.execute('''INSERT OR REPLACE INTO version_history (platform, current_version, release_info, last_checked)
                 VALUES (?, ?, ?, ?)''', (platform, version, release_info_json, datetime.utcnow().isoformat()))
    conn.commit()
    conn.close()

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.interval import IntervalTrigger

async def scheduled_check(context: ContextTypes.DEFAULT_TYPE):
    """定时执行的任务：检查所有平台的更新"""
    logger.info("开始执行定时检查...")
    platform = "telegram_desktop"
    stored_info = get_stored_version(platform) # 从数据库获取当前存储的信息
    latest_info = fetch_latest_desktop_release()

    is_update, new_release_info = check_for_update(stored_info.get('current_version') if stored_info else None, latest_info)

    if is_update:
        # 1. 更新数据库
        update_version_in_db(platform, latest_info['tag_name'], latest_info)
        # 2. 发送通知给所有订阅者
        await broadcast_update(context.bot, platform, new_release_info)

async def broadcast_update(bot, platform, release_info):
    """向所有订阅者广播更新消息"""
    subscribers = get_all_subscribers() # 从数据库获取所有订阅者ID
    message = format_notification_message(platform, release_info)
    for user_id in subscribers:
        try:
            await bot.send_message(chat_id=user_id, text=message, parse_mode='HTML', disable_web_page_preview=False)
        except Exception as e:
            logger.error(f"向用户 {user_id} 发送消息失败: {e}")
            # 可选：如果用户已封锁Bot，将其从订阅列表中移除
            # remove_subscriber(user_id)

def format_notification_message(platform, release_info):
    """格式化美观的通知消息"""
    assets_links = "\n".join([f"<a href='{a['download_url']}'>{a['name']}</a> ({a['size'] // (1024*1024)} MB)" for a in release_info.get('assets', [])[:3]])
    return f"""
🚀 <b>Telegram 桌面版发现新版本！</b>

<b>版本:</b> {release_info['tag_name']}
<b>发布日期:</b> {release_info['published_at'][:10]}
<b>更新摘要:</b>
{release_info['body'][:300]}...

<b>官方下载链接：</b>
{assets_links}

<em>提示：请始终从官方或可信渠道下载以确保安全。</em>
"""

def main():
    application = Application.builder().token(BOT_TOKEN).build()
    # ... 添加命令处理器 ...

    # 初始化数据库
    init_database()

    # 创建并启动定时任务调度器
    scheduler = AsyncIOScheduler()
    # 每30分钟执行一次检查
    scheduler.add_job(scheduled_check, IntervalTrigger(minutes=30), args=[application])
    scheduler.start()

    application.run_polling()