Telegram及Telegram-Bot使用指南

1. Telegram简介

1.1 Telegram是什么

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

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

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

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

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

1.2 Telegram客户端界面

1.3 Telegram注意事项

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

2. Telegram基本使用

2.1 下载安装、配置代理

1）PC端：Telegram官网下载 (opens new window) 安装并完成信息注册后，需要开启代理才能使用。

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

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

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

2.2 客户端汉化

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

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

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

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

2.3 频道检索与推荐

[1] Telegram频道推荐的博客：Telegram群组/频道/机器人推荐 (opens new window)

[2] Telegram频道搜索机器人：https://t.me/soqun (opens new window)

[3] 精选优质Telegram群组、频道和机器人：telegram-groups (opens new window)

[4] Telegram频道检索引擎：https://tgscan.xyz/ (opens new window)（代码开源，项目地址为https://github.com/tgscan-dev/tgscan (opens new window)）

2.4 创建聊天分组

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

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

2.5 解除+86号码私聊限制

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

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

解决办法：向官方机器人 @SpamBot (opens new window) 发送消息申请（跟着提示操作就行），半小时左右即可解除限制。

• 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 (opens new window) 获取虚拟的美国电话号码注册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 (opens new window) 的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 推送文本消息的命令为：

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

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

3.2.2 以脚本的方式推送消息

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

#!/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 定时任务去实现定时的消息推送。

$ crontab -e

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

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

crontab 相关命令：

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

crontab 定时任务语法：

# 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：每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库：

• 官网说明：https://core.telegram.org/api (opens new window)
• 官方文档：https://python-telegram-bot.readthedocs.io/en/latest/index.html (opens new window)
• Wiki：https://github.com/python-telegram-bot/python-telegram-bot/wiki (opens new window)
• 示例项目：https://github.com/python-telegram-bot/python-telegram-bot/tree/master/examples (opens new window)

第三方API库：

• Java：https://github.com/rubenlagus/TelegramBots (opens new window)
• Python：https://github.com/eternnoir/pyTelegramBotAPI (opens new window)
• PHP：https://github.com/Eleirbag89/TelegramBotPHP (opens new window)

3.3.2 使用 Python 开发 Telegram Bot 程序

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

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

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

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

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 (opens new window)

3.4.1 项目结构及代码

项目结构如下：

.
├── Dockerfile
├── README.md
├── build.sh
├── config.ini
├── log.py
├── requirements.txt
├── sql
│   └── push_content_structure.sql
└── yoyo-push.py

requirements.txt

PyMySQL==1.0.2
telepot==12.7
schedule==1.1.0
borax==4.0.0

yoyo-push.py

# -*- coding: utf-8 -*-

import datetime
from configparser import ConfigParser
import pymysql
import time
import schedule
import telepot
from telepot.loop import MessageLoop
from log import logger
from borax.calendars.lunardate import LunarDate


def read_config(config_path):
    cfg = ConfigParser()
    cfg.read(config_path, encoding='utf-8')
    config_dict = {k: v for section in cfg.sections() for k, v in cfg.items(section)}
    return config_dict


def query_push_content(config):
    with pymysql.connect(**config, cursorclass=pymysql.cursors.DictCursor) as db:
        with db.cursor() as cursor:
            cursor.execute("SELECT * FROM push_content WHERE is_deleted = 0 ORDER BY update_time ASC")
            results = cursor.fetchall()
    content_list = [
        {
            'id': row['id'],
            'type': row['type'],
            'person': row['person'],
            'day': row['day'],
            'memorial': row['day'].strftime('%m月%d日'),
            'desc': row['desc']
        }
        for row in results
    ]
    return content_list


def classify_push_content(content_list):
    return [
        [x for x in content_list if x['type'] == type_]
        for type_ in ['纪念日', '生日']
    ]


def calculate_days_until_next_lunar_birthday(month, day, today):
    current_year = today.year
    try:
        this_year_lunar_birthday = LunarDate(current_year, month, day).to_solar_date()
        if this_year_lunar_birthday >= today:
            return (this_year_lunar_birthday - today).days
    except ValueError:
        pass
    for next_year in range(current_year + 1, current_year + 3):
        try:
            next_year_lunar_birthday = LunarDate(next_year, month, day).to_solar_date()
            return (next_year_lunar_birthday - today).days
        except ValueError:
            continue
    return None


def generate_push_content(memorial_day_list, birthday_list):
    now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    week = datetime.datetime.now().strftime("%A")
    curr_date = datetime.datetime.now().date()

    time_frames = [(">>> 今日 ", 0, 0), (">>> 7日内 ", 1, 7), (">>> 30日内 ", 8, 30)]

    sections = [
        '〓' * 18,
        '⏳ yoyo-push',
        f"🔅 {now} {week}",
        '〓' * 18,
    ]

    sections += ['########## 👻 纪念日 ##########']
    for item in memorial_day_list:
        days_diff = (datetime.datetime.now() - item['day']).days + 1
        desc = item.get('desc', '')
        sections.append(f"与{item['person']}{desc}距今已经{days_diff}天，纪念日为{item['memorial']}")

    if not memorial_day_list:
        sections.append("👀 暂无纪念日")

    sections += ['########### 👻 生日 ###########']
    birthday_data = []
    for item in birthday_list:
        try:
            solar_date = item['day'].date()
            lunar_date = LunarDate.from_solar_date(solar_date.year, solar_date.month, solar_date.day)
            lunar_month, lunar_day = lunar_date.month, lunar_date.day
            lunar_day_count = calculate_days_until_next_lunar_birthday(lunar_month, lunar_day, curr_date)
            day_count = (solar_date.replace(year=curr_date.year) - curr_date).days
            if day_count < 0:
                day_count = (solar_date.replace(year=curr_date.year + 1) - curr_date).days

            lunar_birthday_str = f"{lunar_month}月{lunar_day}日"
            birthday_data.append((item, day_count, lunar_day_count, lunar_birthday_str))
        except Exception as e:
            logger.error(f"Error converting date for {item['person']}: {e}")

    for title, min_days, max_days in time_frames:
        frame_items = [(data[0], data[1], data[2], data[3])
                       for data in birthday_data
                       if max_days >= data[1] >= min_days or max_days >= data[2] >= min_days]
        sections.append(title)
        if frame_items:
            for idx, (item, day_count, lunar_day_count, lunar_birthday_str) in enumerate(frame_items, start=1):
                sections.append(
                    f"{idx}. {item['person']}，阳历生日为{item['memorial']}（距今{day_count}天），农历生日为{lunar_birthday_str}（距今{lunar_day_count}天）")
        else:
            sections.append(f"👀 暂无{title.strip().replace('>>> ', '')}过生日的朋友")

    sections.append('〓' * 18)
    return '\n\n'.join(sections) + '\n\n'


def push_list():
    config = read_config('./config.ini')
    content_list = query_push_content({
        'host': config['host'],
        'user': config['user'],
        'password': config['password'],
        'port': int(config['port']),
        'db': config['db'],
        'charset': 'utf8'
    })
    memorial_day_list, birthday_list = classify_push_content(content_list)
    push_str = generate_push_content(memorial_day_list, birthday_list)
    bot.sendMessage(str(config['chat_id']), push_str)
    logger.info(push_str)


def handle(msg):
    if msg['text'] == '/list':
        push_list()
    elif msg['text'] == '/data':
        content_list = query_push_content({
            'host': config['host'],
            'user': config['user'],
            'password': config['password'],
            'port': int(config['port']),
            'db': config['db'],
            'charset': 'utf8'
        })
        push_str = '\n\n'.join(map(str, content_list))
        bot.sendMessage(str(config['chat_id']), push_str)


if __name__ == '__main__':
    config = read_config('./config.ini')
    if config['env'] in ["dev"]:
        telepot.api.set_proxy(config['proxy'])
    bot = telepot.Bot(config['token'])
    MessageLoop(bot, handle).run_as_thread()
    push_list()
    if config['env'] in ["prod"]:
        logger.info('Listening ...')
        schedule.every().day.at(config['timed_task']).do(push_list)
        while True:
            schedule.run_pending()
            time.sleep(1)

log.py

# -*- 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

[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

# 基于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

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：安装项目依赖

$ pip install -r requirements.txt

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

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

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

Step3：修改配置文件

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

Step4：运行程序进行测试

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

Step5：服务器部署

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

4. 导出Telegram聊天记录

4.1 使用Telegram客户端进行导出

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

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

注意事项：如果是Mac端的话要下载那个Telegram Lite（Telegram没有该功能）

4.2 使用TelegramExporter进行导出

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

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

4.2.1 申请 Telegram 应用程序 API 权限

申请流程：

Step1：打开 https://my.telegram.org/auth?to=apps (opens new window) 网址，用于申请 Telegram 应用程序 API 权限

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

Step3：填写Create new application的表单（填写App title、Short name、Platform即可），再点击 Create application创建即可。

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

注意事项：

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

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

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

4.2.2 使用TelegramExporter导出聊天记录

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

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

$ 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：启动脚本导出数据

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

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

 _______   _                                  ______                       _          
|__   __| | |                                |  ____|                     | |           
   | | ___| | ___  __ _ _ __ __ _ _ __ ___   | |__  __  ___ __   ___  _ __| |_ ___ _ __ 
   | |/ _ \ |/ _ \/ _` | '__/ _` | '_ ` _ \  |  __| \ \/ / '_ \ / _ \| '__| __/ _ \ '__|
   | |  __/ |  __/ (_| | | | (_| | | | | | | | |____ >  <| |_) | (_) | |  | ||  __/ |   
   |_|\___|_|\___|\__, |_|  \__,_|_| |_| |_| |______/_/\_\ .__/ \___/|_|   \__\___|_|   
                   __/ |                                 | |                            
                  |___/                                  |_|                -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压缩包。

.
└── 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 导入不进去，然后我就写了个脚本进行导入，项目结构如下：

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

telegram_chat.sql

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

[Mysql]
host = 127.0.0.1
user = root
password = 123456
port = 3306
db = testdb
table = telegram_chat

log.py

# -*- 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

# -*- 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')
    section_list = cfg.sections()
    config_dict = {}
    for section in section_list:
        section_item = cfg.items(section)
        for item in section_item:
            config_dict[item[0]] = item[1]
    return config_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 知乎 (opens new window)

[2] Telegram——真正定义即时通讯 from 知乎 (opens new window)

[3] Telegram群组/频道/机器人推荐 from Newlearnerの小站 (opens new window)

[4] 导出Telegram贴纸为QQ表情包 from 小马哥哥 (opens new window)

[5] 解除 Telegram +86号码私聊限制 from VPS收割者 (opens new window)

[6] Telegram 聊天分组功能：添加聊天文件夹并对群组频道进行分组 from 老唐笔记 (opens new window)

[7] telegram-bot推送通知 from New Life (opens new window)

[8] Telegram Bot 使用文档 from cnblogs (opens new window)

[9] Telegram APIs from Telegram官方文档 (opens new window)

[10] Telegram 机器人程序开发 from 风行's Blog (opens new window)

[11] Creating your Telegram Application from Telegram官方文档 (opens new window)

[12] Create new application telegram API from stackoverflow (opens new window)

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

[14] 通过socks5代理使用telepot（telegram bot python包）from cnpython (opens new window)

[15] Telegram Bot 简明教程 II - 收指令与指令键盘 from I'm cloxnu (opens new window)

[16] Telegram 机器人的简单内联日历 from Github (opens new window)

[17] 用 Python 和 Pyrogram 编写的 Telegram Ai 聊天机器人 from Github (opens new window)

[18] 基于 Telegram 位置的天气预报信息 from Github (opens new window)

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

[20] 基于和风 API 的 Telegram 预报机器人 from Github (opens new window)