自己动手写一个获取当日传播预测的Telegram Bot[第二部分]

本文最后更新于 152 天前，其中的信息可能已经有所发展或是发生改变。

既然闲来无事，那么我们就来具体地讲一讲这个程序

让我们再放一次程序

import requests
from datetime import datetime, time, timedelta
from telegram import Bot
from telegram.ext import Updater, CommandHandler
from apscheduler.schedulers.background import BackgroundScheduler
from pytz import utc
 
TOKEN = 'INPUT_YOUR_TOKEN_HERE'
URL = 'INPUT_YOUR_URL_HERE'
SEND_TIME = time(hour=0, minute=0, second=0)
 
bot = Bot(token=TOKEN)
scheduler = BackgroundScheduler(timezone=utc)
 
# 缓存的图片数据和最后更新时间
cached_image_data = None
last_update_time = None
 
def get_image_url():
    response = requests.get(URL)
    if response.status_code == 200:
        return response.content
    return None
 
def fetch_image_data():
    global cached_image_data, last_update_time
    cached_image_data = get_image_url()
    last_update_time = datetime.utcnow()
 
def send_image_to_user(update, context):
    if cached_image_data:
        chat_id = update.effective_chat.id
        bot.send_photo(chat_id=chat_id, photo=cached_image_data, filename='image.jpg')
    else:
        context.bot.send_message(chat_id=update.effective_chat.id, text="Failed to get the image.")
 
def start(update, context):
    context.bot.send_message(chat_id=update.effective_chat.id, text="请通过 /getcondition 指令来获取当日传播预测图")
 
def helpcommand(update, context):
    context.bot.send_message(chat_id=update.effective_chat.id, text="Bot commands:\n/getcondition To get the image of condition\n/help To list commands of the bot\nAuthor:Mashiro Chino")
 
def main():
    fetch_image_data()  # 预先获取并缓存图片数据
 
    updater = Updater(TOKEN, use_context=True)
    dispatcher = updater.dispatcher
 
    start_handler = CommandHandler('start', start)
    get_condition_handler = CommandHandler('getcondition', send_image_to_user)
    help_handler = CommandHandler('help', helpcommand)
 
    dispatcher.add_handler(start_handler)
    dispatcher.add_handler(get_condition_handler)
    dispatcher.add_handler(help_handler)
 
    scheduler.add_job(fetch_image_data, 'cron', hour=0, minute=0, second=0)  # 每天UTC时间00:00运行任务
    scheduler.start()
 
    updater.start_polling()
 
if __name__ == '__main__':
    main()

一、引用库

在程序一开始我们引用了大量的依赖库，使得这个程序更加简单，那我们具体来看看这几个库

import requests
from datetime import datetime, time, timedelta
from telegram import Bot
from telegram.ext import Updater, CommandHandler
from apscheduler.schedulers.background import BackgroundScheduler
from pytz import utc

我们分别引用了requests，datetime和time，python-telegram-bot，apscheduler以及pytz库

其中，request库用于来发送HTTP请求，获取每日传播图片，datetime和time用于处理时间等操作，python-telegram-bot库用于实现Telegram Bot与用户的交互操作，apscheduler库用于定时任务调度，以及pytz库用于处理时区(在这里我主要使用UTC时间)

这里from telegram import Bot和from telegram.ext import Updater, Commander两句都是调用了python-telegram-bot库中的两个不同的类

二、定义全局变量和对象

这里定义全局变量能够方便程序整体运行方便

TOKEN = 'INPUT_YOUR_TOKEN_HERE'
URL = 'INPUT_YOUR_URL_HERE'
SEND_TIME = time(hour=0, minute=0, second=0)
 
bot = Bot(token=TOKEN)
scheduler = BackgroundScheduler(timezone=utc)
 
# 缓存的图片数据和最后更新时间
cached_image_data = None
last_update_time = None

在Python中，变量的定义无需声明类型，可以直接定义（不过你阅读整个程序可以看到，这几个量其实都是C/C++的const类型常量，整体程序中不会有所变动）

第一个常量TOKEN是你的Telegram Bot的访问凭证，通过TOKEN你可以将程序联系到你的Bot上

第二个常量URL是你将要获取传播预测图的地址

第三个常量SEND_TIME是原本我想写的定时发送的效果，后来没想到怎么写就耽搁在那里了，后期会补充上

第四个bot对象是通过Telegram Bot API创建的实例，它充当与Telegram服务器进行交互的机器人。通过使用bot对象，可以发送消息、发送图片等与用户进行交互的操作

第五个scheduler对象是BackgroundScheduler类的实例，它用于设置定时任务。通过调度器，可以设置定时执行的函数或任务，以特定的时间间隔自动触发函数的执行

第六个cached_image_data变量是用于缓存的太阳活动预测图片数据。在fetch_image_data()函数中，通过调用get_image_url()函数获取到的图片数据会被保存到cached_image_data变量中。这样，在后续的操作中可以使用该缓存数据，避免每次请求时都重新获取图片数据，提高效率

第七个last_update_time变量是用于记录缓存的图片数据最后更新时间的变量。在fetch_image_data()函数中，每次成功获取图片数据后，会更新last_update_time的值为当前时间。这样，可以在需要时检查数据是否已过期，或者在需要更新数据时进行比较

三、子函数部分

程序里定义了大量的子函数，这样方便程序的调用，也能使程序写起来更加方便

def get_image_url():
    response = requests.get(URL)
    if response.status_code == 200:
        return response.content
    return None
 
def fetch_image_data():
    global cached_image_data, last_update_time
    cached_image_data = get_image_url()
    last_update_time = datetime.utcnow()
 
def send_image_to_user(update, context):
    if cached_image_data:
        chat_id = update.effective_chat.id
        bot.send_photo(chat_id=chat_id, photo=cached_image_data, filename='image.jpg')
    else:
        context.bot.send_message(chat_id=update.effective_chat.id, text="Failed to get the image.")
 
def start(update, context):
    context.bot.send_message(chat_id=update.effective_chat.id, text="请通过 /getcondition 指令来获取当日传播预测图")
 
def helpcommand(update, context):
    context.bot.send_message(chat_id=update.effective_chat.id, text="Bot commands:\n/getcondition To get the image of condition\n/help To list commands of the bot\nAuthor:Mashiro Chino")

1.`get_image_url()`函数

def get_image_url():
    response = requests.get(URL)
    if response.status_code == 200:
        return response.content
    return None

调用方式：get_image_url()函数没有在代码中被显式调用，而是被fetch_image_data()函数调用
运行流程：该函数发送HTTP请求，获取太阳活动预测图片的URL，并返回图片的内容（response.content）。如果请求成功（状态码为200），则返回图片内容；否则返回None

2.`fetch_image_data()`函数

def fetch_image_data():
    global cached_image_data, last_update_time
    cached_image_data = get_image_url()
    last_update_time = datetime.utcnow()

调用方式：在main()函数的开头被调用，用于预先获取并缓存图片数据。同时，该函数还被定时任务调度器（scheduler）以每天UTC时间的00:00运行
运行流程：该函数内部首先调用get_image_url()函数获取太阳活动预测图片的内容，并将其保存在全局变量cached_image_data中。然后，更新last_update_time为当前的UTC时间

3.`send_image_to_user(update, context)`函数

def send_image_to_user(update, context):
    if cached_image_data:
        chat_id = update.effective_chat.id
        bot.send_photo(chat_id=chat_id, photo=cached_image_data, filename='image.jpg')
    else:
        context.bot.send_message(chat_id=update.effective_chat.id, text="Failed to get the image.")

调用方式：该函数被命令处理器（get_condition_handler）调用，用于将缓存的图片发送给用户。
运行流程：该函数首先检查全局变量cached_image_data是否存在图片数据。如果存在，它会使用bot对象通过Telegram Bot API将图片发送给用户；否则，会发送一条失败消息

4.`start(update, context)`函数

def start(update, context):
    context.bot.send_message(chat_id=update.effective_chat.id, text="请通过 /getcondition 指令来获取当日传播预测图")

调用方式：该函数被命令处理器（start_handler）调用，用于处理/start命令
运行流程：该函数会向用户发送一条欢迎消息

5.`helpcommand(update, context)`函数

def helpcommand(update, context):
    context.bot.send_message(chat_id=update.effective_chat.id, text="Bot commands:\n/getcondition To get the image of condition\n/help To list commands of the bot\nAuthor:Mashiro Chino")

调用方式：该函数被命令处理器（help_handler）调用，用于处理/help命令
运行流程：该函数会向用户发送一条帮助信息，列出可用的命令以及作者信息

由此，我们小结一下这部分子函数的定义：

get_image_url()函数负责发送HTTP请求获取太阳活动预测图片的URL，并返回图片的内容
fetch_image_data()函数在程序开始时预先获取并缓存图片数据，同时也被定时任务调度器调用以更新数据
send_image_to_user(update, context)函数用于将缓存的图片发送给用户
start(update, context)函数处理/start命令，发送欢迎消息给用户
helpcommand(update, context)函数处理/help命令，发送帮助信息给用户

四、主函数部分

主函数将所有的子程序调用，也就写成了一个Telegram Bot

def main():
    fetch_image_data()  # 预先获取并缓存图片数据
 
    updater = Updater(TOKEN, use_context=True)
    dispatcher = updater.dispatcher
 
    start_handler = CommandHandler('start', start)
    get_condition_handler = CommandHandler('getcondition', send_image_to_user)
    help_handler = CommandHandler('help', helpcommand)
 
    dispatcher.add_handler(start_handler)
    dispatcher.add_handler(get_condition_handler)
    dispatcher.add_handler(help_handler)
 
    scheduler.add_job(fetch_image_data, 'cron', hour=0, minute=0, second=0)  # 每天UTC时间00:00运行任务
    scheduler.start()
 
    updater.start_polling()

fetch_image_data()

首先主函数先调用了fetch_image_data()函数，来对传播情况预测图进行一个缓存

updater = Updater(TOKEN, use_context=True)
dispatcher = updater.dispatcher

创建了一个Updater对象，传入了TOKEN作为参数，用于与Telegram进行交互，将use_context参数设置为True，以启用上下文（context）的使用，并且创建了一个dispatcher对象，用于处理不同的命令和消息

start_handler = CommandHandler('start', start)
get_condition_handler = CommandHandler('getcondition', send_image_to_user)
help_handler = CommandHandler('help', helpcommand)

创建了三个命令处理器（CommandHandler）对象，分别处理/start、/getcondition和/help命令

start_handler命令处理器关联了start函数，用于处理/start命令
get_condition_handler命令处理器关联了send_image_to_user函数，用于处理/getcondition命令
help_handler命令处理器关联了helpcommand函数，用于处理/help命令

dispatcher.add_handler(start_handler)
dispatcher.add_handler(get_condition_handler)
dispatcher.add_handler(help_handler)

将上述三个命令处理器添加到dispatcher中，用于处理对应的命令

scheduler.add_job(fetch_image_data, 'cron', hour=0, minute=0, second=0)  # 每天UTC时间00:00运行任务
scheduler.start()

首先使用调度器（scheduler）设置一个定时任务，即在每天的UTC时间00:00运行fetch_image_data()函数。其中调度器使用'cron'作为触发器类型，指定了每天的00:00时刻触发任务。然后调度器开始运行，开始定时任务的调度

updater.start_polling()

这句话用于启动updater对象的轮询，开始监听来自Telegram的命令和消息，并进行相应的处理

综上所述，main()函数主要是实现了整个程序的逻辑流程。它首先调用fetch_image_data()函数预先获取并缓存图片数据，然后创建了Updater对象和命令处理器，将命令处理器添加到dispatcher中。接下来，设置了定时任务调度器，以每天的UTC时间00:00运行fetch_image_data()函数。最后，启动了updater对象的轮询，开始监听用户的命令和消息

在main()函数中，涉及的主要变量的类型和作用如下：

updater：Updater对象，用于与Telegram进行交互
dispatcher：Dispatcher对象，用于处理命令和消息
start_handler、get_condition_handler和help_handler：CommandHandler对象，处理对应的命令
scheduler：BackgroundScheduler对象，用于设置定时任务
TOKEN：字符串类型，存储Telegram Bot的令牌

最后再送大家这个Bot分别用Go和C++实现的代码（Go特点就是，不用配置运行环境，而C++则是我写程序的主要语言）

Go实现

//Go
package main

import (
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"strings"
	"time"

	tgbotapi "github.com/go-telegram-bot-api/telegram-bot-api"
)

const (
	Token = "INPUT_YOUR_TOKEN_HERE"
	URL   = "INPUT_YOUR_URL_HERE"
)

var (
	bot               *tgbotapi.BotAPI
	cachedImageData   []byte
	lastUpdateUTC     time.Time
	fetchImageTrigger <-chan time.Time
)

func main() {
	var err error
	bot, err = tgbotapi.NewBotAPI(Token)
	if err != nil {
		log.Fatal(err)
	}

	bot.Debug = true

	log.Printf("Authorized on account %s", bot.Self.UserName)

	updateConfig := tgbotapi.NewUpdate(0)
	updateConfig.Timeout = 60

	updates, err := bot.GetUpdatesChan(updateConfig)
	if err != nil {
		log.Fatal(err)
	}

	go fetchImagePeriodically()

	for update := range updates {
		if update.Message == nil { // ignore any non-Message Updates
			continue
		}

		switch update.Message.Text {
		case "/start":
			msg := tgbotapi.NewMessage(update.Message.Chat.ID, "请通过 /getcondition 指令来获取当日传播预测图")
			bot.Send(msg)
		case "/getcondition":
			sendImage(update.Message.Chat.ID)
		case "/help":
			helpCommand(update.Message.Chat.ID)
		}
	}
}

func fetchImage() ([]byte, error) {
	resp, err := http.Get(URL)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("HTTP request failed with status code %d", resp.StatusCode)
	}

	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}

	return body, nil
}

func sendImage(chatID int64) {
	if cachedImageData != nil {
		msg := tgbotapi.NewPhotoUpload(chatID, tgbotapi.FileReader{Name: "image.jpg", Reader: strings.NewReader(string(cachedImageData))})
		bot.Send(msg)
	} else {
		msg := tgbotapi.NewMessage(chatID, "无法获取图片，请稍后再试")
		bot.Send(msg)
	}
}

func helpCommand(chatID int64) {
	msg := tgbotapi.NewMessage(chatID, "Bot commands:\n/getcondition To get the image of condition\n/help To list commands of the bot\nAuthor:Mashiro Chino")
	bot.Send(msg)
}

func fetchImagePeriodically() {
	now := time.Now().UTC()
	nextFetchTime := time.Date(now.Year(), now.Month(), now.Day()+1, 0, 0, 0, 0, time.UTC)
	duration := nextFetchTime.Sub(now)

	fetchImageTrigger = time.Tick(duration)

	for {
		select {
		case <-fetchImageTrigger:
			log.Println("Fetching image...")
			imageData, err := fetchImage()
			if err != nil {
				log.Println("Failed to fetch image:", err)
				cachedImageData = nil
			} else {
				cachedImageData = imageData
			}
		}
	}
}

在使用Go来写这个Bot之前，你需要安装第三方库 “github.com/go-telegram-bot-api/telegram-bot-api”

可以通过以下命令来安装

go get github.com/go-telegram-bot-api/telegram-bot-api

C++实现

//C++
#include <iostream>
#include <string>
#include <chrono>
#include <thread>
#include <curl/curl.h>
#include <tgbot/tgbot.h>

const std::string TOKEN = "YOUR_TELEGRAM_BOT_TOKEN";
const std::string URL = "https://www.hamqsl.com/solar101vhfpic.php";

// 缓存的图片数据和最后更新时间
std::string cachedImageData;
std::chrono::system_clock::time_point lastUpdateTime;

std::string get_image_url() {
    CURL* curl = curl_easy_init();
    if (curl) {
        curl_easy_setopt(curl, CURLOPT_URL, URL.c_str());
        curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L);
        curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, [](char* ptr, size_t size, size_t nmemb, std::string* data) {
            data->append(ptr, size * nmemb);
            return size * nmemb;
        });
        std::string imageData;
        curl_easy_setopt(curl, CURLOPT_WRITEDATA, &imageData);
        CURLcode res = curl_easy_perform(curl);
        if (res == CURLE_OK) {
            return imageData;
        }
        curl_easy_cleanup(curl);
    }
    return "";
}

void fetch_image_data() {
    cachedImageData = get_image_url();
    lastUpdateTime = std::chrono::system_clock::now();
}

void send_image_to_user(TgBot::Message::Ptr message, TgBot::Bot& bot) {
    if (!cachedImageData.empty()) {
        bot.getApi().sendPhoto(message->chat->id, TgBot::InputFile::fromData(cachedImageData, "image.jpg"));
    } else {
        bot.getApi().sendMessage(message->chat->id, "Failed to get the image.");
    }
}

void start(TgBot::Message::Ptr message, TgBot::Bot& bot) {
    bot.getApi().sendMessage(message->chat->id, "请通过 /getcondition 指令来获取当日传播预测图");
}

void help_command(TgBot::Message::Ptr message, TgBot::Bot& bot) {
    bot.getApi().sendMessage(message->chat->id, "Bot commands:\n/getcondition To get the image of condition\n/help To list commands of the bot\nAuthor:Mashiro Chino");
}

int main() {
    TgBot::Bot bot(TOKEN);

    fetch_image_data(); // 预先获取并缓存图片数据

    bot.getEvents().onCommand("start", [&bot](TgBot::Message::Ptr message) {
        start(message, bot);
    });

    bot.getEvents().onCommand("getcondition", [&bot](TgBot::Message::Ptr message) {
        send_image_to_user(message, bot);
    });

    bot.getEvents().onCommand("help", [&bot](TgBot::Message::Ptr message) {
        help_command(message, bot);
    });

    // 每天UTC时间00:00运行任务
    auto current_time = std::chrono::system_clock::now();
    auto next_fetch_time = std::chrono::system_clock::time_point(std::chrono::duration_cast<std::chrono::hours>(current_time.time_since_epoch() + std::chrono::hours(24)));
    auto duration = next_fetch_time - current_time;

    std::thread scheduler_thread([&bot]() {
        while (true) {
            fetch_image_data();
            std::this_thread::sleep_for(std::chrono::hours(24));
        }
    });

    bot.getApi().deleteWebhook();

    TgBot::TgLongPoll longPoll(bot);
    while (true) {
        try {
            longPoll.start();
        } catch (const TgBot::TgException& e) {
            std::cout << "error: " << e.what() << std::endl;
        }
    }

    scheduler_thread.join();

    return 0;
}

请注意，C++的程序使用了 libcurl 和 tgbot 库来进行HTTP请求和与Telegram Bot API的交互。在编译之前，请确保已经正确安装这些库，并将编译选项配置为正确链接这些库

另外，C++的程序结构可能与Python有所不同。在C++中，我使用了线程来定期获取图片数据，因此程序需要使用 -lpthread 参数进行编译

libcurl库的安装：

Debian/Ubuntu:

sudo apt-get install libcurl4-openssl-dev

Red Hat/CentOS:

sudo yum install libcurl-devel

macOS:

brew install curl

tgbot库的安装:

Linux:

sudo apt-get install g++ make libboost-dev libboost-system-dev libssl-dev
git clone https://github.com/reo7sp/tgbot-cpp.git
cd tgbot-cpp
cmake .
make -j4
sudo make install

macOS:

brew install tgbot-cpp

Windows:

git clone https://github.com/Microsoft/vcpkg.git
cd vcpkg
./bootstrap-vcpkg.bat
./vcpkg install --triplet x64-windows tgbot-cpp

题外话：

~~应该没有人能像我这样水文章吧（逃~~

一、引用库

二、定义全局变量和对象

三、子函数部分

1.get_image_url()函数

2.fetch_image_data()函数

3.send_image_to_user(update, context)函数

4.start(update, context)函数

5.helpcommand(update, context)函数