打造telegram个人机器人（一）

这几天使用某游戏工具（ohhh，舟！），发现有一个通知功能，可以通过telegram上的机器人来通知，就对telegram机器人有了一些兴趣，记录一下对telegram机器人学习从0-1的一个简单过程。

尝试用Python开发一个个人专属的Telegram机器人，用来实现自动化任务或其他功能。

利用python-telegram-bot这个库，能轻松地创建各种功能的Telegram机器人，无论是简单的消息回复，还是复杂的群组管理。

1、创建第一个机器人

打开 Telegram，搜索 @BotFather。
点击进入 BotFather，并发送 /start 命令。
然后输入命令 /newbot 创建一个新机器人。

创建成功后就需要为这个机器人命令以及给他一个用户名。
名称：机器人的显示名称（例如：XXXXX_Bot）。
用户名：必须以 bot 结尾，例如 xxxxx_bot。

创建成功后，BotFather 会生成一个唯一的 Token（如 123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11）。妥善保存这个 Token，因为它是与机器人 API 交互的关键。1

设置机器人描述：发送 /setdescription 给 BotFather。按提示选择你的机器人并输入描述信息。

设置机器人头像：发送 /setuserpic，然后上传一张图片作为头像。

2、与机器人进行交互

打开你的机器人（在 Telegram 中搜索你刚创建的用户名）。

点击 Start 按钮，你的机器人现在可以响应基本命令(这里配置后机器人才会做基本的响应内容，不然没有响应)

3、设置机器人功能

要让机器人执行特定功能，通常需要结合开发工具或平台进行编程。以下是两种常见方式：

1. 使用第三方 Bot 管理平台

以下第三方平台可以协助配置机器人，完成基础功能实现，如自定义回复，创建菜单等。

ManyBot
Chatfuel
Botsify

在第一次接触后，试用了一下manybot

在telegram中搜索manybot，关注后根据指示操作就行，也是非常便捷的（具体使用这里并未过多研究，有需要者自行根据文档学习。）

但我这里最开始是为了获取机器的token和chatID来连接我的游戏工具，然后就去研究如何去获取chatID。

通过 getUpdates API 获取 Chat ID

1. 确保与机器人交互：打开 Telegram，找到你的机器人，点击 Start 或者直接发送一条消息给机器人（如 "Hello"）。
2. 调用 getUpdates 在浏览器中访问以下 URL（替换<YOUR_BOT_TOKEN>为你的机器人 Token）：

https://api.telegram.org/bot<YOUR_BOT_TOKEN>/getUpdates

3. 查看返回结果 API 会返回一个 JSON 数据，其中包含 Chat ID。示例如下：

{
    "ok": true,
    "result": [
        {
            "update_id": 123456789,
            "message": {
                "message_id": 1,
                "from": {
                    "id": 987654321,
                    "is_bot": false,
                    "first_name": "John",
                    "username": "johndoe",
                    "language_code": "en"
                },
                "chat": {
                    "id": 987654321,
                    "first_name": "John",
                    "username": "johndoe",
                    "type": "private"
                },
                "date": 1692849830,
                "text": "Hello"
            }
        }
    ]
}

chat.id 字段的值就是你的 Chat ID，例如：987654321。

获取chatID踩坑

另外，在这个地方也踩了个坑，获取chatID时碰到了下述错误

{
"ok": false,
"error_code": 409,
"description": "Conflict: can't use getUpdates method while webhook is active; use deleteWebhook to delete the webhook first"
}

这是因为我先使用了manybot这个平台来连接我的bot，启用了 Webhook，所以无法使用 getUpdates 方法。Telegram 机器人不能同时使用 Webhook 和轮询（Polling）模式。要解决这个问题，需要先删除 Webhook。

解决方法：删除 Webhook

可以通过以下步骤删除 Webhook，从而切换到使用 getUpdates 的模式。

1. 调用 deleteWebhook API 在浏览器中访问以下 URL（将 <YOUR_BOT_TOKEN> 替换为你的机器人Token）：

https://api.telegram.org/bot<YOUR_BOT_TOKEN>/deleteWebhook

如果成功，会返回以下响应：

{
    "ok": true,
    "result": true,
    "description": "Webhook was deleted"
}

2. 再次调用 getUpdates 删除 Webhook 后，你可以再次尝试访问以下 URL 来获取 Chat ID。

这里我也是获取到的我的chatID，通过机器人发送了一条测试通知

2. 编程 Telegram 机器人

结束完上面的工具连接，我就在想，可不可以去做一个属于我自己的机器人，做一些日常工作的自动化工作。决定使用python来尝试一下

首先就是需要一个开发环境

下面是需要的一些环境配置：

确保已安装 Python 3.7 或更高版本
安装 python-telegram-bot 库
文本编辑器或 IDE：推荐 VSCode 或 PyCharm。
Postman（可选）：方便测试 Telegram API。

接下来就是进行编写基础机器人代码。

此处附上我目前做的一些简单功能模块，其他内容后续再打包放在个人git库中。

欢迎界面

from common import Update, CommandHandler, MessageHandler, filters, ContextTypes

# 定义 /start 命令的功能
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text('你好，Milktea！我是你的专属机器人，很高兴为您服务！')

# 定义普通消息处理功能
async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(f'你说了：{update.message.text}')

# 提供注册命令和消息处理器的方法
def register_handlers(application):
    # 注册 /start 命令处理器
    application.add_handler(CommandHandler("start", start))
    # 注册普通消息处理器
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))

根据IP实现实时天气查询

import httpx

# 高德 API Keys
GAODE_API_KEY = "XXXXXXXXXXXXXXXXXXX"

# 高德 API URLs
GAODE_IP_URL = "https://restapi.amap.com/v3/ip"
GAODE_WEATHER_URL = "https://restapi.amap.com/v3/weather/weatherInfo"


# 获取 IP 定位信息
async def get_location_by_ip():
    """
    使用高德 API 自动通过 IP 定位获取城市编码（adcode）和城市信息。
    """
    try:
        params = {
            "key": GAODE_API_KEY,
        }
        async with httpx.AsyncClient() as client:
            response = await client.get(GAODE_IP_URL, params=params)
            data = response.json()

        if data.get("status") == "1":
            adcode = data.get("adcode")
            city = data.get("city")
            return adcode, city
        else:
            return None, "无法确定当前位置"
    except Exception as e:
        return None, f"IP 定位失败: {e}"


# 获取天气信息
async def get_weather_by_adcode(adcode):
    """
    使用高德 API 根据城市编码（adcode）获取实时天气信息。
    """
    try:
        params = {
            "key": GAODE_API_KEY,
            "city": adcode,
            "extensions": "base",  # base: 实时天气, all: 预报天气
        }
        async with httpx.AsyncClient() as client:
            response = await client.get(GAODE_WEATHER_URL, params=params)
            data = response.json()

        if data.get("status") == "1" and "lives" in data:
            weather_data = data["lives"][0]
            city = weather_data["city"]
            weather = weather_data["weather"]
            temperature = weather_data["temperature"]
            wind_direction = weather_data["winddirection"]
            wind_power = weather_data["windpower"]
            humidity = weather_data["humidity"]
            report_time = weather_data["reporttime"]

            return (
                f"城市: {city}\n"
                f"天气: {weather}\n"
                f"温度: {temperature}°C\n"
                f"风向: {wind_direction}\n"
                f"风力: {wind_power}级\n"
                f"湿度: {humidity}%\n"
                f"更新时间: {report_time}"
            )
        else:
            return "无法获取天气信息，请稍后再试。"
    except Exception as e:
        return f"查询天气失败: {e}"


# 定义 /weather 命令功能
async def weather(update, context):
    """
    处理 Telegram /weather 命令，自动获取用户 IP 定位并推送天气信息。
    """
    try:
        # 获取用户的 IP 地址对应的城市编码和信息
        adcode, city_info = await get_location_by_ip()

        if adcode:
            # 根据城市编码查询天气
            weather_info = await get_weather_by_adcode(adcode)
            await update.message.reply_text(weather_info)
        else:
            await update.message.reply_text(city_info)

    except Exception as e:
        await update.message.reply_text(f"查询天气失败: {e}")


# 注册命令处理器
def register_handlers(application):
    """
    将 /weather 命令注册到 Telegram Bot。
    """
    from telegram.ext import CommandHandler
    application.add_handler(CommandHandler("weather", weather))

显示当前时间

from common import Update, CommandHandler, ContextTypes
from datetime import datetime

# 定义 /time 命令的功能
async def time(update: Update, context: ContextTypes.DEFAULT_TYPE):
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    await update.message.reply_text(f'当前的时间是: {now}')

# 注册 /time 命令的处理器
def register_handlers(application):
    application.add_handler(CommandHandler("time", time))

主函数

from common import Application
import logging
import importlib
import os

# 替换成你的API Token
TOKEN = "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"

# 设置日志记录
logging.basicConfig(
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.INFO
)

# 错误处理器
async def error_handler(update, context):
    logging.error(msg="Exception while handling an update:", exc_info=context.error)
    if update and update.effective_message:
        await update.effective_message.reply_text("发生错误，请稍后再试。")

# 自动加载模块
def load_modules(application):
    for filename in os.listdir(os.path.dirname(__file__)):
        if filename.endswith(".py") and filename not in ("bot.py", "common.py"):
            module_name = filename[:-3]
            module = importlib.import_module(module_name)
            if hasattr(module, "register_handlers"):
                module.register_handlers(application)

def main():
    application = Application.builder().token(TOKEN).build()

    # 动态加载模块并注册处理器
    load_modules(application)

    # 启动机器人
    application.run_polling()

if __name__ == '__main__':
    main()