使用 Telethon 实现 Telegram 群组消息监控和转发

本文介绍如何使用 Telethon 库在自己的服务器上实现 Telegram 群组消息的监控和转发，即使你不是群组的管理员。通过配置文件动态更新群组 ID，脚本将自动加载最新配置，实现实时转发消息。

前置条件

1. 你需要有一个 Telegram 用户账户。

2. 你需要访问 my.telegram.org 以获取 `api_id` 和 `api_hash`。

操作步骤

步骤1：安装必要的软件包

首先，在你的服务器上安装必要的软件包：

sudo apt update
sudo apt install python3 python3-pip python3-venv -y

步骤 2：创建和激活虚拟环境

创建一个虚拟环境来隔离你的 Python 包：

python3 -m venv myenv

激活虚拟环境：

source myenv/bin/activate

激活后，你的命令行提示符可能会发生变化，表示你现在在虚拟环境中。

步骤 3：安装 Telethon

在虚拟环境中使用 pip 安装 Telethon：

pip install telethon

步骤 4：创建配置文件

创建一个配置文件 config.json，用于存储群组 ID 和其他配置信息：

nano config.json

将以下内容粘贴到 config.json 文件中，并替换为你的实际值：

{
    "api_id": "YOUR_API_ID",
    "api_hash": "YOUR_API_HASH",
    "phone_number": "YOUR_PHONE_NUMBER",
    "source_chat_ids": [-1001234567890, -1009876543210],
    "target_chat_id": -1001122334455
}

步骤 5：编写并运行脚本

1. 创建并编辑脚本文件

创建一个新的 Python 脚本文件 forward_user.py：

nano forward_user.py

将以下内容粘贴到 forward_user.py 文件中：

import os
import json
import asyncio
from telethon import TelegramClient, events

# 获取脚本所在目录的绝对路径
script_dir = os.path.dirname(os.path.abspath(__file__))
config_path = os.path.join(script_dir, 'config.json')

# 从配置文件中读取配置信息
def load_config():
    with open(config_path, 'r') as f:
        return json.load(f)

# 初始化客户端和配置
config = load_config()
client = TelegramClient('session_name', config['api_id'], config['api_hash'])

async def main():
    # 登录
    await client.start(config['phone_number'])

    print("Bot is running...")

    # 定义事件处理器
    @client.on(events.NewMessage(chats=config['source_chat_ids']))
    async def handler(event):
        target_chat_id = config['target_chat_id']
        # 转发消息到目标群组
        await client.forward_messages(target_chat_id, event.message)

    # 运行直到客户端断开连接
    await client.run_until_disconnected()

# 启动客户端
with client:
    client.loop.run_until_complete(main())

2. 运行脚本

确保你仍然在虚拟环境中，然后运行脚本：

python forward_user.py

首次运行脚本时，你可能需要输入你的电话号码和 Telegram 发送的验证码，以完成用户身份验证。

获取群组或频道的 ID

方法 1：使用 Telethon 获取群组或频道的 ID

创建一个新的 Python 脚本文件 get_chat_id.py：

nano get_chat_id.py

将以下内容粘贴到 get_chat_id.py 文件中：

from telethon.sync import TelegramClient

# 替换为你的实际值
api_id = 'YOUR_API_ID'
api_hash = 'YOUR_API_HASH'
phone_number = 'YOUR_PHONE_NUMBER'

# 初始化客户端
client = TelegramClient('session_name', api_id, api_hash)

async def main():
    # 登录
    await client.start(phone_number)
    print("登录成功！")

    # 获取对话列表
    dialogs = await client.get_dialogs()
    for dialog in dialogs:
        print(f"Name: {dialog.name}, ID: {dialog.id}")

with client:
    client.loop.run_until_complete(main())

运行脚本：

python get_chat_id.py

后台运行且开机启动

要让一个 Python 脚本在后台运行并且开机启动，你可以使用 systemd 来创建一个系统服务。以下是详细步骤：

步骤 1：创建一个 Python 脚本

首先，确保你的 Python 脚本 get_chat_id.py 能够独立运行并且执行预期的任务。这个脚本似乎是用来获取 Telegram 群组或频道的 ID。

步骤 2：创建一个 systemd 服务

创建服务文件在 /etc/systemd/system/ 目录下创建一个新的服务文件，例如 forward_user.service：

sudo nano /etc/systemd/system/forward_user.service

编辑服务文件将以下内容粘贴到 forward_user.service 文件中：

[Unit]
Description=Telegram Message Forwarder
After=network.target

[Service]
User=root
WorkingDirectory=/root/
ExecStart=/root/myenv/bin/python3 /root/forward_user.py
Restart=always
RestartSec=10
StandardOutput=syslog
StandardError=syslog
SyslogIdentifier=forward_user

[Install]
WantedBy=multi-user.target

解释说明：

• User: 设置为 root 用户，因为你是使用 root 用户权限进行操作。
• WorkingDirectory: 设置为存放 forward_user.py 脚本的目录路径。
• ExecStart: 指定 Python 解释器路径为 /usr/bin/python3，并指定完整的 get_chat_id.py 脚本路径。
• Restart: 如果服务意外终止，会自动重启。
• RestartSec: 设置重启间隔时间为 10 秒。
• StandardOutput 和 StandardError: 将输出和错误日志记录到系统日志中。
• SyslogIdentifier: 日志标识符，用于识别服务的日志条目。

注意事项：

• 确保将 /path/to/your/script/directory 和 /path/to/your/script/forward_user.py 替换为实际的路径。
• /usr/bin/python3 是默认的 Python 3 解释器路径，如果你的系统中的 Python 3 解释器路径不同，请相应调整。
• 保存并关闭文件后，确保通过 systemctl daemon-reload 命令重载 systemd 配置，以便使新的服务配置生效。

完成这些步骤后，你可以使用 systemctl enable forward_user.service 命令来启用并设置开机自启动，使用 systemctl start forward_user.service 命令来启动服务，并使用 systemctl status forward_user.service 命令来检查服务的运行状态。

保存并关闭文件

按 Ctrl + O 保存文件，然后按 Ctrl + X 关闭编辑器。

步骤 3：启用和启动服务

• 重载 systemd

sudo systemctl daemon-reload

• 启用服务（开机自启动）

sudo systemctl enable forward_user.service

• 启动服务

sudo systemctl start forward_user.service

• 检查服务状态

sudo systemctl status forward_user.service

• 如果一切正常，你应该看到类似以下的输出：

● get_chat_id.service - Get Telegram Chat IDs Loaded: loaded (/etc/systemd/system/get_chat_id.service; enabled; vendor preset: enabled) Active: active (running) since Mon 2024-07-16 12:00:00 UTC; 10s ago

步骤 4：验证服务是否正常运行

通过 status 命令可以验证服务是否正常运行。如果服务启动失败，可以检查日志文件 /var/log/syslog 或者使用 journalctl -xe 命令查看详细错误信息。

现在，get_chat_id.py 脚本将在后台持续运行，并且在系统启动时自动启动。