Telegram及Telegram Bot使用指南

  1. 1. Telegram简介
    1. 1.1 Telegram是什么
    2. 1.2 Telegram客户端界面
    3. 1.3 Telegram注意事项
  2. 2. Telegram基本使用
    1. 2.1 下载安装、配置代理
    2. 2.2 客户端汉化
    3. 2.3 频道推荐
    4. 2.4 创建聊天分组
    5. 2.5 解除+86号码私聊限制
    6. 2.6 导出Telegram贴纸
    7. 2.7 Telegram语音转文字
  3. 3. Telegram Bot入门指南
    1. 3.1 申请自己的 Telegram Bot
      1. 3.1.1 申请 Telegram Bot
      2. 3.1.2 修改 Telegram Bot 头像
    2. 3.2 向 Telegram Bot 推送消息
      1. 3.2.1 以命令的方式推送消息
      2. 3.2.2 以脚本的方式推送消息
      3. 3.2.3 借助 Crontab 实现定时推送
    3. 3.3 开发 Telegram Bot 程序
      1. 3.3.1 开发 Telegram Bot 程序的API库
      2. 3.3.2 使用 Python 开发 Telegram Bot 程序
    4. 3.4 开发一个推送生日及纪念日的Telegram Bot
      1. 3.4.1 项目结构及代码
      2. 3.4.2 本地运行及服务器部署
  4. 4. 导出Telegram聊天记录
    1. 4.1 使用Telegram客户端进行导出
    2. 4.2 使用TelegramExporter进行导出
      1. 4.2.1 申请 Telegram 应用程序 API 权限
      2. 4.2.2 使用TelegramExporter导出聊天记录
      3. 4.2.3 解析CSV聊天记录存入MySQL数据库
  5. 5. 参考资料

1. Telegram简介

1.1 Telegram是什么

Telegram(简称TG)是跨平台的即时通信软件,其客户端是自由及开放源代码软件,但服务器是专有软件。用户可以相互交换加密与自毁消息,发送照片、影片所有类型文件。官方提供手机版、桌面版和网页版等多种平台客户端,同时官方开放API,因此拥有许多第三方的客户端可供选择,其中多款内置中文。

  • 软件定位:加密通讯工具。

  • 软件用途:点对点加密通讯,了解IT行业趋势,交流技术问题,获取生活娱乐资源。

  • 软件优势:加密通讯、全平台同步、历史记录永不自动删除、随时删除或修改消息、传输文件无限制、链接摘要预览、已阅显示、10W人上限的超大群组、收藏贴纸方便、UI好看、轻量、永久无广告。

  • 软件劣势:软件需翻墙才能使用,导致国内社交圈子几乎没有人用;无管制所以什么样的人都有,有些不健康的东西不利于祖国花骨朵的成长;有些技术群组是外文的,没有内置的翻译功能。

1.2 Telegram客户端界面

Telegram客户端界面

1.3 Telegram注意事项

1)该软件需要翻墙才能使用;2)里面所有个人信息都不要填真实的;3)隐私全部设置为不允许任何人或者仅联系人可见;4)默认6个月不登录自动删除账号,最长可更改为1年;5)黑产、色情、邪教的频道不要关注,不要发表政治言论。

2. Telegram基本使用

2.1 下载安装、配置代理

1)PC端:Telegram官网下载 安装并完成信息注册后,需要开启代理才能使用。

  • 设置——高级——网络与代理——使用自定义代理——添加代理——协议:SOCKS5、IP:127.0.0.1、端口:1080(以实际为准)

  • 另注:也可在“设置——高级——网络与代理”处直接使用“系统代理设置”,就不用配置自定义代理了。

2)iPad端和iPhone端:直接在App Store下载(国区就有,无需美区账号),开了代理工具就能用,无需额外配置代理。

2.2 客户端汉化

默认语言选项里没有中文,需要自行汉化,各客户端均是如此。

汉化方法:打开 https://telegram.me/Tele_zh_CN 加入“电报简体中文化”频道,找一个看的顺眼的语言包点一下就可以汉化了。

  • 例如:我找的是2019.4.16的消息记录,点击“简体中文(@zh_CN 版):tg://setlanguage?lang=classic-zh-cn”语言包一键汉化。

说明:Telegram 官方只开放了语言包翻译接口,官方没有提供中文语言包,目前所有的中文语言包都是非官方用户翻译的。

2.3 频道推荐

[1] Telegram频道推荐的博客:Telegram群组/频道/机器人推荐 from Newlearnerの小站

[2] Telegram频道搜索机器人:https://t.me/soqun

注:需要啥频道,可以直接在Telegram的搜索框里搜。

2.4 创建聊天分组

由于Telegram只允许设置5个置顶消息,因此有必要创建聊天分组对消息进行分类。

设置——文件夹——设置新的文件夹

2.5 解除+86号码私聊限制

1)问题描述:

Telegram发送私聊消息时遇到如下Sorry,you can only send messages to mutual contacts at the momet.提示,对方收不到消息。

2)原因:

有段时间币圈利用 Telegram 的便利,大量发广告和拉人进群。因此 Telegram 限制了+86 大陆手机号绑定的账号的私聊(+86 的账号不能给非 +86 的账号主动发起私聊,但是 +86 的账号能给 +86 主动发起私聊),刚注册的+86 账号也不能发起私聊的,过段时间就OK了。

3)解决办法:

向官方机器人@SpamBot 发送消息申请(跟着提示操作就行),半小时左右即可解除限制。

  • Step1:发送内容 /start
  • Step2:点击按钮 But I can’t message non-contacts!
  • Step3:点击按钮 NO,I’ll never do any of this!
  • Step4:发送内容 accident

注:为了更好的保护个人隐私,建议使用 Temporary SMS and Disposable Numbers 获取虚拟的美国电话号码注册Telegram

2.6 导出Telegram贴纸

[1] 在Telegram中搜索@StickerSetBot 机器人

[2] 输入 /newpack 开启机器人,会提示 OK now send me stickers or sticker set links (200 max). 意思是现在可以开始发送要导出的贴纸或贴纸设置链接,一次有200张的限制

[3] 选择需要导出的贴纸包,点“分享贴纸”按钮,即可生成贴纸分享链接,将此链接发送给机器人

[4] 发送完成后输入 /finish 结束,等待下载完成即可(以压缩包的形式下载到TG的下载目录中)

2.7 Telegram语音转文字

Telegram客户端没有内置的语音转文字功能,但可以通过一个名为 Voicy 的Telegram Bot来实现。它支持.ogg, .flac, .wav, .mp3四种语音格式(Telegram自带的语音是.ogg格式),支持识别Arabic、Burmese、Chinese、English、French、Bengali、Catalan、Dutch、Finnish、German这10种语言。

使用方式:

  • Step1:关注 Voicy Telegram Bot,发送 /language 命令设置语言。

  • Step2:将朋友发送的语音消息转发给 Voicy Telegram Bot,等待一会儿即可显示识别结果。

注意事项:

  • 初次使用前必须先发送 /language 命令设置语言,不然无法识别,会返回Please, speak clearly, I couldn't recognize that
  • Chinese识别的语音文字是中文繁体。

3. Telegram Bot入门指南

3.1 申请自己的 Telegram Bot

3.1.1 申请 Telegram Bot

找 BotFather 官方机器人申请自己的 Telegram Bot,需要记录下:BotName、TOKEN、CHATID等信息。

  • Step1:在Telegram中添加BotFather这个账号,然后依次发送/start/newbot,按照提示即可创建一个新的机器人。记下来给你生成的token。
  • Step2:搜索刚刚创建的机器人的名字,并给它发送一条消息。(注意:需要先与机器人之间创建会话,机器人才能下发消息,否则机器人无法主动发送消息)
  • Step3:在Telegram中搜索userinfobot,并给它发送一条消息,它会返回给你chatid,将它也记下来。

3.1.2 修改 Telegram Bot 头像

还是找 BotFather 官方机器人,先发送/setuserpic,它会让你选择为哪个Bot修改,选择完之后发送头像图片给它即可。

注:头像以图片的形式发送,不要以文件的形式发送(发送时点上那个压缩图片即为图片的形式发送)

3.2 向 Telegram Bot 推送消息

3.2.1 以命令的方式推送消息

向 Telegram Bot 推送文本消息的命令为:

1
$ curl -s -X POST "URL" -d chat_id="CHAT_ID" -d text="MESSAGE"

用VPS发送该命令后,Telegram Bot里就收到了我们发送的测试消息了。

3.2.2 以脚本的方式推送消息

在VPS上创建一个testbot.sh脚本,内容如下:

1
2
3
4
5
6
7
8
#!/bin/bash

TOKEN=<TOKEN>
CHAT_ID=<CHAT_ID>
MESSAGE="Hello World"
URL="https://api.telegram.org/bot$TOKEN/sendMessage"

curl -s -X POST $URL -d chat_id=$CHAT_ID -d text="$MESSAGE"

然后使用chmod u+x testbot.sh赋予其可执行权限,再使用./testbot.sh调用,这时Telegram Bot里就收到了我们发送的测试消息了。

3.2.3 借助 Crontab 实现定时推送

我们可以借助 Crontab 定时任务去实现定时的消息推送。

1
$ crontab -e

然后添加类似如下代码保存即可:

1
59 23 * * * 脚本路径/testbot.sh

crontab 相关命令:

1
2
3
4
crontab [-u username]    // 省略用户表表示操作当前用户的crontab
-e (修改 crontab 文件,如果文件不存在会自动创建)
-l (显示 crontab 文件)
-r (删除 crontab 文件)

crontab 定时任务语法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# For details see man 4 crontabs
# Example of job definition:
# .---------------- minute (0 - 59)
# | .------------- hour (0 - 23)
# | | .---------- day of month (1 - 31)
# | | | .------- month (1 - 12) OR jan,feb,mar,apr ...
# | | | | .---- day of week (0 - 6) (Sunday=0 or 7) OR sun,mon,tue,wed,thu,fri,sat
# | | | | |
# * * * * * user-name command to be executed
定时任务的每段为:分,时,日,月,周,用户,命令
第1列表示分钟1~59 每分钟用*或者 */1表示
第2列表示小时1~23(0表示0点)
第3列表示日期1~31
第4列表示月份1~12
第5列标识号星期0~6(0表示星期天)
第6列要运行的命令

*:表示任意时间都,实际上就是“每”的意思。可以代表00-23小时或者00-12每月或者00-59分
-:表示区间,是一个范围,00 17-19 * * * cmd,就是每天17,18,19点的整点执行命令
,:是分割时段,30 3,19,21 * * * cmd,就是每天凌晨3和晚上19,21点的半点时刻执行命令
/n:表示分割,可以看成除法,*/5 * * * * cmd,每隔五分钟执行一次

crontab定时任务实例:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
实例1:每1分钟执行一次myCommand
* * * * * myCommand
实例2:每小时的第3和第15分钟执行
3,15 * * * * myCommand
实例3:在上午8点到11点的第3和第15分钟执行
3,15 8-11 * * * myCommand
实例4:每隔两天的上午8点到11点的第3和第15分钟执行
3,15 8-11 */2 * * myCommand
实例5:每周一上午8点到11点的第3和第15分钟执行
3,15 8-11 * * 1 myCommand
实例6:每晚的21:30重启smb
30 21 * * * /etc/init.d/smb restart
实例7:每月1、10、22日的4 : 45重启smb
45 4 1,10,22 * * /etc/init.d/smb restart
实例8:每周六、周日的1 : 10重启smb
10 1 * * 6,0 /etc/init.d/smb restart
实例9:每天18 : 00至23 : 00之间每隔30分钟重启smb
0,30 18-23 * * * /etc/init.d/smb restart
实例10:每星期六的晚上11 : 00 pm重启smb
0 23 * * 6 /etc/init.d/smb restart
实例11:每一小时重启smb
* */1 * * * /etc/init.d/smb restart
实例12:晚上11点到早上7点之间,每隔一小时重启smb
* 23-7/1 * * * /etc/init.d/smb restart

3.3 开发 Telegram Bot 程序

Telegram 官方为了让开发者能够参与到机器人的开发,还免费提供了功能齐全的 API,我们可以用它创建自己的机器人,下面介绍下如何创建和部署机器人程序。

3.3.1 开发 Telegram Bot 程序的API库

Github 上有很多用于开发 Telegram Bot 的API库,可以选择自己擅长的语言进行开发。

官方API库:

第三方API库:

3.3.2 使用 Python 开发 Telegram Bot 程序

以下使用 pyTelegramBotAPI 开发自己的 Telegram Bot 程序,建议使用 Pycharm 进行开发。

在开始开发之前,我们首先需要安装以下两个依赖包:

1
2
$ pip install pyTelegramBotAPI       # telebot     
$ pip install python-telegram-bot # telegram.ext

下面给出一个简单的 Telegram Bot 例子,需要注意的是,如果是在国内,需要将翻墙代理的设置写到代码里。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from telebot import logger   
from telegram.ext import Updater, CommandHandler, MessageHandler, Filters

def start(update, context):
update.message.reply_text('Hi!')

def help(update, context):
update.message.reply_text('Help!')

def echo(update, context):
update.message.reply_text(update.message.text)

def error(update, context):
logger.warning('Update "%s" caused error "%s"', update, context.error)

if __name__ == '__main__':
updater = Updater('<token>', use_context=True, request_kwargs={
'proxy_url': 'HTTPS://127.0.0.1:1080/'
})

dispatcher = updater.dispatcher

dispatcher.add_handler(CommandHandler("start", start))
dispatcher.add_handler(CommandHandler("help", help))

dispatcher.add_handler(MessageHandler(Filters.text, echo))

dispatcher.add_error_handler(error)

updater.start_polling()
updater.idle()

运行之后,便可打开你的 Telegram Bot 进行测试了,输入/start/help命令后,它会自动给你消息回应。

3.4 开发一个推送生日及纪念日的Telegram Bot

基于Python使用API开发了一个推送生日及纪念日的Telegram Bot,项目地址:https://github.com/Logistic98/yoyo-push

3.4.1 项目结构及代码

项目结构如下:

1
2
3
4
5
6
7
8
9
10
.
├── Dockerfile
├── README.md
├── build.sh
├── config.ini
├── log.py
├── requirements.txt
├── sql
│   └── push_content_structure.sql
└── yoyo-push.py

requirements.txt

1
2
3
PyMySQL==1.0.2
telepot==12.7
schedule==1.1.0

yoyo-push.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
# -*- coding: utf-8 -*-

import calendar
import datetime
from datetime import date
from configparser import ConfigParser
import pymysql
import time

import schedule
import telepot
from telepot.loop import MessageLoop
from log import logger


# 读取配置
def read_config(config_path):
cfg = ConfigParser()
cfg.read(config_path, encoding='utf-8')
env = cfg.get('Environment', 'env')
proxy = cfg.get('Environment', 'proxy')
timed_task = cfg.get('Environment', 'timed_task')
host = cfg.get('MySQL', 'host')
user = cfg.get('MySQL', 'user')
password = cfg.get('MySQL', 'password')
port = cfg.get('MySQL', 'port')
db = cfg.get('MySQL', 'db')
bot_name = cfg.get('TelegramBot', 'bot_name')
token = cfg.get('TelegramBot', 'token')
chat_id = cfg.get('TelegramBot', 'chat_id')
config_dict = {}
config_dict['env'] = env
config_dict['proxy'] = proxy
config_dict['timed_task'] = timed_task
config_dict['host'] = host
config_dict['user'] = user
config_dict['password'] = password
config_dict['port'] = port
config_dict['db'] = db
config_dict['bot_name'] = bot_name
config_dict['token'] = token
config_dict['chat_id'] = chat_id
return config_dict


# 查询推送内容
def query_push_content(config_dict):
db = pymysql.connect(host=str(config_dict['host']), user=str(config_dict['user']),
password=str(config_dict['password']), port=int(config_dict['port']),
db=str(config_dict['db']), charset='utf8')
cursor = db.cursor()

content_list = []
try:
sql = "select * from push_content where is_deleted = 0 order by update_time asc"
cursor.execute(sql)
results = cursor.fetchall()
for result in results:
item = {}
item['id'] = result[0]
item['type'] = result[1]
item['persion'] = result[2]
item['day'] = result[3]
item['memorial'] = result[4]
item['desc'] = result[5]
content_list.append(item)
except Exception as e:
logger.error(e)
cursor.close()
return content_list


# 分类推送内容
def classify_push_content(content_list):
birthday_list = []
memorial_day_list = []
birthday_id = 0
memorial_day_id = 0
for content in content_list:
if content['type'] == "生日":
birthday_id = birthday_id + 1
content['id'] = birthday_id
birthday_list.append(content)
elif content['type'] == "纪念日":
memorial_day_id = memorial_day_id + 1
content['id'] = memorial_day_id
memorial_day_list.append(content)
return memorial_day_list, birthday_list


# 生成推送内容
def gene_push_content(memorial_day_list, birthday_list):

now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
curr_date = date.today()
week = calendar.day_name[curr_date.weekday()]

push_list = []
push_list.append('〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓')
push_list.append('⏳ yoyo-push')
push_list.append("🔅 " + now + " " + week)
push_list.append('〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓')
push_list.append('############## 👻 纪念日 ##############')
for memorial_day_item in memorial_day_list:
id = memorial_day_item['id']
persion = memorial_day_item['persion']
desc = memorial_day_item['desc']
day_count = (datetime.datetime.now() - memorial_day_item['day']).days + 1
memorial = memorial_day_item['memorial']
push_list.append("{}. 与{}{}距今已经{}天,纪念日为{}".format(id, persion, desc, day_count, memorial))
if len(memorial_day_list) == 0:
push_list.append("👀 暂无纪念日")

push_list.append('############### 👻 生日 ###############')
push_list.append('>>> 今日 ')
today_num = 0
for birthday_item in birthday_list:
day_count = (birthday_item['day'].replace(year=curr_date.year) - datetime.datetime.now()).days + 1
if day_count < 0:
day_count = (birthday_item['day'].replace(year=curr_date.year + 1) - datetime.datetime.now()).days + 1
if day_count == 0:
today_num = today_num + 1
persion = birthday_item['persion']
memorial = birthday_item['memorial']
push_list.append("{}. {},生日为{}(距今{}天)".format(today_num, persion, memorial, day_count))
birthday_list.remove(birthday_item)
if today_num == 0:
push_list.append("👀 暂无今日过生日的朋友")
push_list.append('>>> 7日内 ')
week_num = 0
for birthday_item in birthday_list:
day_count = (birthday_item['day'].replace(year=curr_date.year) - datetime.datetime.now()).days + 1
if day_count < 0:
day_count = (birthday_item['day'].replace(year=curr_date.year + 1) - datetime.datetime.now()).days + 1
if 7 > day_count >= 0:
week_num = week_num + 1
persion = birthday_item['persion']
memorial = birthday_item['memorial']
push_list.append("{}. {},生日为{}(距今{}天)".format(week_num, persion, memorial, day_count))
birthday_list.remove(birthday_item)
if week_num == 0:
push_list.append("👀 暂无7日内过生日的朋友")
push_list.append('>>> 30日内 ')
mouth_num = 0
for birthday_item in birthday_list:
day_count = (birthday_item['day'].replace(year=curr_date.year) - datetime.datetime.now()).days + 1
if day_count < 0:
day_count = (birthday_item['day'].replace(year=curr_date.year + 1) - datetime.datetime.now()).days + 1
if 30 > day_count >= 0:
mouth_num = mouth_num + 1
persion = birthday_item['persion']
memorial = birthday_item['memorial']
push_list.append("{}. {},生日为{}(距今{}天)".format(mouth_num, persion, memorial, day_count))
birthday_list.remove(birthday_item)
if mouth_num == 0:
push_list.append("👀 暂无30日内过生日的朋友")
push_list.append('〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓〓')

push_str = ""
for push_item in push_list:
push_str = push_str + str(push_item) + "\n\n"
return push_str


# 消息推送的结果
def push_list():
content_list = query_push_content(config_dict)
memorial_day_list, birthday_list = classify_push_content(content_list)
push_str = gene_push_content(memorial_day_list, birthday_list)
bot.sendMessage(str(config_dict['chat_id']), push_str)
logger.info(push_str)


# 处理 Telegram Bot 消息
def handle(msg):
if msg['text'] == '/list':
push_list()
if msg['text'] == '/data':
content_list = query_push_content(config_dict)
push_str = ""
for push_item in content_list:
push_str = push_str + str(push_item) + "\n\n"
bot.sendMessage(str(config_dict['chat_id']), push_str)


if __name__ == '__main__':
config_path = './config.ini'
config_dict = read_config(config_path)
if config_dict['env'] == "dev":
telepot.api.set_proxy(str(config_dict['proxy']))
bot = telepot.Bot(str(config_dict['token']))
MessageLoop(bot, handle).run_as_thread()
logger.info('Listening ...')
# 定时任务:每日指定时间,自动发送一次推送
schedule.every().day.at(str(config_dict['timed_task'])).do(push_list)
while True:
schedule.run_pending() # 运行所有可以运行的定时任务
time.sleep(1)

log.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# -*- coding: utf-8 -*-

import logging

logger = logging.getLogger(__name__)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

# 输出到控制台
console = logging.StreamHandler()
console.setLevel(logging.INFO)
console.setFormatter(formatter)
logger.addHandler(console)

# 输出到文件
logger.setLevel(level=logging.INFO)
handler = logging.FileHandler("./yoyo-push.log")
handler.setLevel(logging.INFO)
handler.setFormatter(formatter)
logger.addHandler(handler)

config.ini

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
[Environment]
env = dev
proxy = http://127.0.0.1:1080
timed_task = 00:00

[MySQL]
host = 127.0.0.1
user = root
password = 123456
port = 3306
db = yoyo_push

[TelegramBot]
bot_name = your_bot_name
token = your_token
chat_id = your_chat_id

Dockerfile

1
2
3
4
5
6
7
8
9
10
11
12
# 基于python3.7镜像创建新镜像
FROM python:3.7
# 创建容器内部目录
RUN mkdir /code
# 将项目复制到内部目录
ADD . /code/
# 切换到工作目录
WORKDIR /code
# 安装项目依赖
RUN pip install -r requirements.txt
# 启动项目
ENTRYPOINT ["nohup","python","yoyo-push.py","&"]

build.sh

1
2
3
docker build -t yoyo-push-image .
docker run -d --name yoyo-push -e TZ="Asia/Shanghai" yoyo-push-image:latest
docker update yoyo-push --restart=always

3.4.2 本地运行及服务器部署

Step1:安装项目依赖

1
$ pip install -r requirements.txt

Step2:创建mysql数据库并导入数据

创建数据库,然后使用 push_content_structure.sql 创建数据表,自行插入数据。

1
$ CREATE DATABASE yoyo_push DEFAULT CHARSET=utf8 DEFAULT COLLATE utf8_unicode_ci;

Step3:修改配置文件

修改 config.ini 中的配置项,本地开发测试需要设置代理。

Step4:运行程序进行测试

运行 yoyo-push.py 程序,在你的Telegram Bot中输入 /list 、/data 命令查看结果。

yoyo-push效果图

Step5:服务器部署

准备docker环境,将项目整个上传到服务器,赋予可执行权限并运行build.sh脚本。

4. 导出Telegram聊天记录

4.1 使用Telegram客户端进行导出

Telegram客户端自带导出聊天记录的功能,导出格式支持 html/json,但需要24h后才可以下载

设置——高级——导出Telegram数据——授予权限(会给Telegram发送一个通知,需要Allow才可以导出,而且需要等待24h)

使用Telegram客户端进行导出-1

注意事项:如果是Mac端的话要下载那个Telegram Lite(Telegram没有该功能)

使用Telegram客户端进行导出-2

4.2 使用TelegramExporter进行导出

TelegramExporter:支持导出单个、多个、全部用户及群组的聊天记录,导出格式为CSV,可包含附件。

  • 项目地址:https://github.com/KeL3vRa/TelegramExporter
  • 项目简介:Telegram Exporter 是一种取证工具,用于提取给定 Telegram 帐户的所有聊天记录。
  • 实施过程:申请 Telegram 应用程序 API 权限——使用TelegramExporter导出聊天记录——将CSV聊天记录存入数据库进行分析

4.2.1 申请 Telegram 应用程序 API 权限

申请流程:

Step1:打开 https://my.telegram.org/auth?to=apps 网址,用于申请 Telegram 应用程序 API 权限

Step2:输入你的手机号码,之后你将在 Telegram 上收到一个代码以输入进行身份验证。

申请Telegram应用程序API权限-1

Step3:填写Create new application的表单(填写App title、Short name、Platform即可),再点击 Create application创建即可。

Step4:Telegram 将负责生成“App api_id”和“App api_hash”值,将它保存记录下来。

申请Telegram应用程序API权限-2

注意事项:

[1] 手机号码必须是带国际标识符的,比如中国大陆是+86,不然你收不到身份验证的代码,格式如下:

1
+86-你的TG注册手机号      // 以中国大陆的为例(如果用国外手机号则自行修改国际标识符位) 

[2] 关掉代理(IP地址要与手机号所属国家相符)、关掉广告屏蔽插件(比如AdGuard),否则保存表单是会莫名其妙报错。

4.2.2 使用TelegramExporter导出聊天记录

环境说明:本文是在境外Linux服务器上执行的(Win端也是支持的但我没试),Python环境为3.9(确定3.9版本没问题,我另外一个服务器用3.7版本虽然不报错,但导出结果为空,版本不对的话建议conda或者docker部署),国内的话需要配置代理。

Step1:安装依赖并修改配置文件

1
2
3
4
5
$ git clone https://github.com/KeL3vRa/TelegramExporter.git
$ cd TelegramExporter
$ pip install -r requirements
$ vim config.ini // 填写“App api_id”和“App api_hash”值(必须修改)
$ vim configuration.json // 设置为0是只保存文本,设置为1是保存文本和附件,默认值为1(可不修改)

Step2:启动脚本导出数据

1
2
$ chmod u+x run.sh
$ ./run.sh

第一次启动时,您将被要求输入与您要从中执行提取的帐户关联的电话号码(包括区号,如+86手机号),输入并按“Enter”后,系统会要求您输入数字 PIN;此 PIN 将由“Telegram”用户通过消息发送。如果启用了双因素身份验证,您还将被提示输入密码以访问用户的云。输入密码后,首次启动将完成,应用程序将准备好执行提取。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
 _______   _                                  ______                       _          
|__ __| | | | ____| | |
| | ___| | ___ __ _ _ __ __ _ _ __ ___ | |__ __ ___ __ ___ _ __| |_ ___ _ __
| |/ _ \ |/ _ \/ _` | '__/ _` | '_ ` _ \ | __| \ \/ / '_ \ / _ \| '__| __/ _ \ '__|
| | __/ | __/ (_| | | | (_| | | | | | | | |____ > <| |_) | (_) | | | || __/ |
|_|\___|_|\___|\__, |_| \__,_|_| |_| |_| |______/_/\_\ .__/ \___/|_| \__\___|_|
__/ | | |
|___/ |_| -By DMD

Pyrogram v1.0.7, Copyright (C) 2017-2020 Dan <https://github.com/delivrance>
Licensed under the terms of the GNU Lesser General Public License v3 or later (LGPLv3+)

Enter phone number or bot token: 【+86手机号】
Is "+86手机号" correct? (y/N): 【y】
The confirmation code has been sent via Telegram app
Enter confirmation code: 【验证码】
Two-step verification password required
Password hint: None
Enter password (empty to recover): 【双重验证密码】
Enter:
[1] to extract the chats for a single user
[2] to extract the chats for multiple users
[3] to extract all chats
[-1] to quit
Please enter your choice: 【1】
[create_extraction_folders] Creating extraction folders
[create_extraction_folders] Extraction folders created successfully
You can enter one of the following information:
- Phone Book name
- Telegram username
- Channel name
- Group name
- Phone number (in this case remember to indicate also the phone prefix):
- Or press enter if you want to see a list of the chats
Please enter your decision: 【Enter键】
···
[menu_get_contact] There are multiple matching chats. Which one do you want to choose?
[*] 0 Username: UnsupportedUser64Bot First Name: Unsupported Use
[*] 1 Telegram 中文社群 🅥 (supergroup)
···
【0】
···
[get_chat_ids_by_dialogs] Retrieved chat with username: xxx
[write_all_chats_logs_file] Processing chat with xxx;
[get_chat_logs_by_identifier] Downloading attached media...
···
[compress_and_hash_extraction] Creating extraction zip archive...
[compress_and_hash_extraction] Extraction zip archive created successfully
[compress_and_hash_extraction] Creating zip hashes...
[compress_and_hash_extraction] Zip hashes created successfully

之后在/extraction目录即可查看导出的数据,它会另外自动帮我们打一个zip压缩包。

1
2
3
4
5
6
7
8
9
10
11
.
└── Extraction_26-06-2022 15-13-22
├── chats
│   └── xxx_private.csv
├── extraction_archive_hash.txt
├── extraction.zip
├── media
│   └── xxx_private
│   └──photo_2021-11-16_21-04-54_xxx.jpg
└── members
└── xxx_private.csv

4.2.3 解析CSV聊天记录存入MySQL数据库

由 TelegramExporter 导出的 csv 文件我直接用 Navicat 导入不进去,然后我就写了个脚本进行导入,项目结构如下:

1
2
3
4
5
6
7
.
├── config.ini // mysql连接配置
├── input // 放置若干csv
│   └── xxx_private.csv
├── log.py // 日志记录配置
├── save-mysql.py // 导入主程序
└── telegram_chat.sql // 建表sql

telegram_chat.sql

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;

CREATE TABLE `telegram_chat` (
`ID` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键',
`USERNAME` varchar(255) DEFAULT NULL COMMENT '用户名',
`NAME` varchar(255) DEFAULT NULL COMMENT '用户昵称',
`PHONE_NUMBER` varchar(255) DEFAULT NULL COMMENT '电话号码',
`TIMESTAMP` datetime DEFAULT NULL COMMENT '发送时间',
`MESSAGE` mediumtext COMMENT '消息内容',
`DETAILS` mediumtext COMMENT '消息详情',
PRIMARY KEY (`ID`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='Telegram聊天记录表';

SET FOREIGN_KEY_CHECKS = 1;

config.ini

1
2
3
4
5
6
7
[Mysql]
host = 127.0.0.1
user = root
password = 123456
port = 3306
db = testdb
table = telegram_chat

log.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# -*- coding: utf-8 -*-

import logging

logger = logging.getLogger(__name__)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

# 输出到控制台
console = logging.StreamHandler()
console.setLevel(logging.INFO)
console.setFormatter(formatter)
logger.addHandler(console)

# 输出到文件
logger.setLevel(level=logging.INFO)
handler = logging.FileHandler("./save-mysql.log")
handler.setLevel(logging.INFO)
handler.setFormatter(formatter)
logger.addHandler(handler)

save-mysql.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
# -*- coding: utf-8 -*-

import csv
import glob
import os
import time
from configparser import ConfigParser
import chardet
import pymysql

from log import logger


# 检查文件编码
def check_charset(file_path):
with open(file_path, "rb") as f:
data = f.read(4)
charset = chardet.detect(data)['encoding']
return charset


# 实现replaceAll的功能
def replaceAll(input, toReplace, replaceWith):
while (input.find(toReplace) > -1):
input = input.replace(toReplace, replaceWith)
return input


# CSV转字典列表
def csv_to_dict_list(path, encoding):
get_data = []
with open(path, 'r', encoding=encoding) as f:
reader = csv.reader(f)
fieldnames = next(reader)
csv_reader = csv.DictReader(f, fieldnames=fieldnames)
for row in csv_reader:
key = list(row)[0]
value = list(row.values())[0]
keyList = replaceAll(key, "DETAILS (OPTIONAL)", "DETAILS").split("§")
valueList = replaceAll(replaceAll(replaceAll(replaceAll(value, '\r', ' '), '\n', ' '), '\u3000', ' '), '\x01', ' ').split("§")
rowData = dict(zip(keyList, valueList))
get_data.append(rowData)
return get_data


# 读取配置
def read_config(config_path):
cfg = ConfigParser()
cfg.read(config_path, encoding='utf-8')
host = cfg.get('Mysql', 'host')
user = cfg.get('Mysql', 'user')
password = cfg.get('Mysql', 'password')
port = cfg.get('Mysql', 'port')
db = cfg.get('Mysql', 'db')
table = cfg.get('Mysql', 'table')
mysql_dict = {}
mysql_dict['host'] = host
mysql_dict['user'] = user
mysql_dict['password'] = password
mysql_dict['port'] = port
mysql_dict['db'] = db
mysql_dict['table'] = table
return mysql_dict


# 保存到mysql数据库
def save_data(mysql_dict, csv_data_list):
if csv_data_list == []:
return None
mysql = pymysql.connect(host=str(mysql_dict['host']), user=str(mysql_dict['user']),
password=str(mysql_dict['password']), port=int(mysql_dict['port']),
db=str(mysql_dict['db']), charset='utf8mb4')
for i in csv_data_list:
cursor = mysql.cursor()
qmarks = ', '.join(['%s'] * len(i))
columns = ', '.join(i.keys())
try:
qry = "Insert Into " + str(mysql_dict['table']) + " (%s) Values (%s);" % (columns, qmarks)
values_list = []
for j in i.values():
values_list.append(j)
cursor.execute(qry, values_list)
mysql.commit()
except Exception as e:
logger.error(e)
cursor.close()
mysql.close()


if __name__ == '__main__':

logger.info("===程序启动===")

# 设置csv文件的输入目录
path = './input'
if not os.path.exists(path):
os.makedirs(path)
csv_path_list = glob.glob(path + '/*.csv')
logger.info("检测到的csv文件列表:" + str(csv_path_list))

# 读取配置文件
config_path = './config.ini'
mysql_dict = read_config(config_path)
logger.info("读取ini配置文件成功:" + str(csv_path_list))

# 遍历csv文件保存入库
for csv_path in csv_path_list:
logger.info("开始写入csv文件:" + str(csv_path))
encoding = check_charset(csv_path)
csv_data_list = csv_to_dict_list(csv_path, encoding)
save_data(mysql_dict, csv_data_list)
now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
insert_mysql_info = str(mysql_dict['table']) + "表插入了" + str(len(csv_data_list)) + "条数据,时间是" + str(now_time)
logger.info(insert_mysql_info)
logger.info("结束写入csv文件:" + str(csv_path))

logger.info("===程序结束===")

5. 参考资料

[1] 自由与失控:Telegram创始人和他的“黑客帝国” from 知乎

[2] Telegram——真正定义即时通讯 from 知乎

[3] Telegram群组/频道/机器人推荐 from Newlearnerの小站

[4] 导出Telegram贴纸为QQ表情包 from 小马哥哥

[5] 解除 Telegram +86号码私聊限制 from VPS收割者

[6] Telegram 聊天分组功能:添加聊天文件夹并对群组频道进行分组 from 老唐笔记

[7] telegram-bot推送通知 from New Life

[8] Telegram Bot 使用文档 from cnblogs

[9] Telegram APIs from Telegram官方文档

[10] Telegram 机器人程序开发 from 风行’s Blog

[11] Creating your Telegram Application from Telegram官方文档

[12] Create new application telegram API from stackoverflow

[13] 可能是全网最基础的创建telegram bot的教程 from 知乎

[14] 通过socks5代理使用telepot(telegram bot python包)from cnpython

[15] Telegram Bot 简明教程 II - 收指令与指令键盘 from I’m cloxnu

[16] Telegram 机器人的简单内联日历 from Github

[17] 用 Python 和 Pyrogram 编写的 Telegram Ai 聊天机器人 from Github

[18] 基于 Telegram 位置的天气预报信息 from Github

[19] Telegram Bot 验证加入群组的用户是否是人类 from Github

[20] 基于和风 API 的 Telegram 预报机器人 from Github