Telegram机器人对接数据库存储聊天记录完整指南：从连接到查询

在开发Telegram机器人的过程中，我们经常需要保存聊天记录，用于数据分析、客服归档或长期留存。Telegram Bot API本身不提供消息存储能力，因此对接数据库是唯一可靠的方案。本文将带你从零开始，实现一个完整的数据库存储系统，涵盖数据库选择、表结构设计、Python代码实现以及查询示例。

为什么需要数据库存储聊天记录？

Telegram服务器只保存最近的消息，官方也没有提供完整的聊天记录导出接口。对于需要长期运营的机器人（如客服、日志机器人、社群管理），将聊天记录持久化到本地数据库，可以带来以下好处：

数据主权：完全掌控自己的数据，不受平台限制。
高级查询：利用SQL语法快速检索历史消息。
离线分析：可对聊天数据进行统计、挖掘和可视化。
备份容灾：避免因封号或数据丢失造成损失。

数据库选型：SQLite还是MySQL？

对于个人项目或中小型机器人，SQLite是首选。它无需单独部署，直接嵌入应用，支持标准SQL，适合快速开发和单机场景。如果机器人会扩展到多实例部署或需要高并发写入，建议升级到MySQL或PostgreSQL。本文以SQLite为例，但代码可以轻松迁移到MySQL。

表结构设计

为了完整保存聊天信息，我们需要设计三张表：users、chats和messages。

CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    telegram_id INTEGER UNIQUE NOT NULL,
    first_name TEXT,
    last_name TEXT,
    username TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE IF NOT EXISTS chats (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    chat_id INTEGER UNIQUE NOT NULL,
    chat_type TEXT,
    title TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE IF NOT EXISTS messages (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    message_id INTEGER NOT NULL,
    user_id INTEGER,
    chat_id INTEGER,
    text TEXT,
    date TIMESTAMP,
    FOREIGN KEY (user_id) REFERENCES users(id),
    FOREIGN KEY (chat_id) REFERENCES chats(id)
);

这样设计的好处是：用户和聊天信息冗余少，查询消息时可以通过外键关联获取详细上下文。

使用Python对接数据库

我们使用流行库python-telegram-bot和sqlite3（Python内置）。首先安装依赖：

pip install python-telegram-bot

然后创建连接和初始化表结构：

import sqlite3

conn = sqlite3.connect('bot.db', check_same_thread=False)
cursor = conn.cursor()
# 执行上述CREATE语句...
cursor.execute("INSERT OR IGNORE INTO ...")
conn.commit()

保存聊天的核心逻辑

在机器人更新处理函数中，我们需要统一拦截Message，将用户、聊天和消息写入数据库。为了避免重复，使用INSERT OR IGNORE配合唯一索引。

from telegram.ext import Application, MessageHandler, filters

def save_message(update, context):
    msg = update.message
    if not msg:
        return

    user = msg.from_user
    chat = msg.chat

    # 写入用户
    cursor.execute("INSERT OR IGNORE INTO users (telegram_id, first_name, last_name, username) VALUES (?,?,?,?)",
                   (user.id, user.first_name, user.last_name, user.username))
    cursor.execute("SELECT id FROM users WHERE telegram_id=?", (user.id,))
    user_db_id = cursor.fetchone()[0]

    # 写入聊天
    cursor.execute("INSERT OR IGNORE INTO chats (chat_id, chat_type, title) VALUES (?,?,?)",
                   (chat.id, chat.type, chat.title))
    cursor.execute("SELECT id FROM chats WHERE chat_id=?", (chat.id,))
    chat_db_id = cursor.fetchone()[0]

    # 写入消息
    cursor.execute("INSERT INTO messages (message_id, user_id, chat_id, text, date) VALUES (?,?,?,?,?)",
                   (msg.message_id, user_db_id, chat_db_id, msg.text, msg.date))
    conn.commit()

注意在main中注册这个处理器，并设置对所有消息类型生效：

application.add_handler(MessageHandler(filters.ALL, save_message))

查询聊天记录的实用示例

存储的最终目的是便于查询。以下列出几个高频需求。

1. 查看某位用户最近20条消息

SELECT text, date FROM messages 
WHERE user_id = (SELECT id FROM users WHERE telegram_id = 123456) 
ORDER BY date DESC LIMIT 20;

2. 统计某个群聊的消息总量

SELECT COUNT(*) FROM messages 
WHERE chat_id = (SELECT id FROM chats WHERE chat_id = -100123456789);

3. 按关键词搜索消息

SELECT m.text, u.username, c.title, m.date 
FROM messages m 
JOIN users u ON m.user_id = u.id 
JOIN chats c ON m.chat_id = c.id 
WHERE m.text LIKE '%关键词%' 
ORDER BY m.date DESC;

生产环境注意事项

异步问题：python-telegram-bot是异步框架，SQLite的同步写入可能阻塞事件循环。建议使用asyncio.to_thread或改用异步数据库驱动（如aiosqlite）。
连接管理：不要在每个请求中重复打开连接，应使用连接池（如SQLAlchemy）或长期保持一个连接（注意线程安全）。
字段扩展：如果需要存储媒体文件或语音消息，增加content_type和file_id字段。
隐私合规：存储聊天记录可能涉及用户隐私，建议模糊化敏感信息并告知用户。

总结

对接数据库是Telegram机器人进阶的必经之路。通过本文的学习，你已经掌握了基础的数据表设计、消息保存和查询方法。这套方案可以轻松扩展，例如加入情感分析、词云统计或自动备份功能。立刻动手，让你机器人的记忆永不丢失！