首页
学习
活动
专区
圈层
工具
发布
社区首页 >专栏 >5行Python代码,把你的微信聊天记录变成可搜索的数据库

5行Python代码,把你的微信聊天记录变成可搜索的数据库

作者头像
用户11081884
发布2026-07-20 20:39:42
发布2026-07-20 20:39:42
400
举报

那个让我翻了两小时聊天记录的下午

上周四,同事问我:“上次你在群里发的那个餐厅叫什么来着?就是带包间的那个。”

我说我找找。

打开微信群聊,开始往上翻。翻了十分钟,没翻到。群里消息太多了,每天几百条。我试着用微信自带的搜索,搜“餐厅”,出来几百条结果,从2026年到现在的都有,一条一条看,根本看不过来。

又试了搜“包间”。出来十几条,都不是我要的那条。

最后花了将近两个小时,手动从大概那个时间段的聊天记录里一条一条看,才找到。原来当时说的是“饭馆”不是“餐厅”,提到包间的时候说的是“有单间”。

微信自带的搜索,只支持关键词精确匹配。你想不起来当时用的什么词,就搜不到。

当天晚上我就写了这个脚本。


思路

把微信聊天记录导入到SQLite数据库里,利用SQLite的FTS5全文搜索引擎来查询。

FTS5不是简单的关键词匹配。能分词,支持模糊搜索,还能按相关度排序。最关键的是,SQLite是Python自带的,不用装任何额外服务。

整体流程:

  1. 导出微信聊天记录(CSV格式)
  2. Python脚本读取CSV,写入SQLite
  3. 建FTS5全文索引
  4. 写一个查询函数,随时搜

怎么导出微信聊天记录

微信自己没有导出功能。需要用第三方工具。

目前比较靠谱的方案:

方案一:WechatExporter(推荐)

GitHub开源项目,支持Mac和Windows。把手机备份文件解析成HTML或CSV。地址搜 BlueMatthew/WechatExporter 就能找到。

方案二:PyWxDump

也是开源的,可以直接从Windows版微信的数据库里解密提取。搜 xaoyaoo/PyWxDump

不管你用哪个工具,最终目标是拿到一个CSV文件,格式如下:

代码语言:javascript
复制
timestamp,sender,content,chat_name
2026-03-15 14:23:01,张三,周末去哪吃饭,朋友聚餐群
2026-03-15 14:23:45,我,上次去的那个湘菜馆不错,朋友聚餐群
2026-03-15 14:24:12,李四,哪个湘菜馆?,朋友聚餐群
2026-03-15 14:24:58,我,就是望京那个 好像叫湘味楼,朋友聚餐群
2026-03-15 14:25:30,张三,有包间吗,朋友聚餐群
2026-03-15 14:26:01,我,有 上次我们就是在包间吃的 有单间,朋友聚餐群

每行一条消息,四个字段:时间、发送人、消息内容、聊天名称(群名或联系人昵称)。

有了这个CSV,后面的事情就简单了。


完整代码

代码语言:javascript
复制
"""
微信聊天记录全文搜索工具
将导出的CSV聊天记录导入SQLite,支持FTS5全文检索

依赖:Python 3.8+(无需额外安装第三方库)
"""

import csv
import sqlite3
from datetime import datetime
from pathlib import Path

# ============ 配置 ============

CSV_FILE = "wechat_export.csv"      # 导出的聊天记录CSV文件
DB_FILE = "wechat_search.db"        # SQLite数据库文件名


# ============ 1. 建库建表 ============

def init_database(db_path=DB_FILE):
    """初始化SQLite数据库,创建普通表和FTS5虚拟表"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    # 创建主表,存原始数据
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS messages (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT,
            sender TEXT,
            content TEXT,
            chat_name TEXT,
            date_str TEXT
        )
    """)

    # 创建FTS5虚拟表,用于全文搜索
    # tokenize='unicode61' 对中文支持较好
    # 也可以用 'jieba' 分词,但需要额外配置
    cursor.execute("""
        CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts
        USING fts5(
            content,
            sender,
            chat_name,
            content='messages',
            content_rowid='id',
            tokenize='unicode61'
        )
    """)

    # 创建触发器,主表插入数据时自动同步到FTS表
    cursor.execute("""
        CREATE TRIGGER IF NOT EXISTS messages_ai AFTER INSERT ON messages BEGIN
            INSERT INTO messages_fts(rowid, content, sender, chat_name)
            VALUES (new.id, new.content, new.sender, new.chat_name);
        END;
    """)

    conn.commit()
    returnconn


# ============ 2. 导入CSV数据 ============

def import_csv(conn, csv_path=CSV_FILE):
    """从CSV文件导入聊天记录到数据库"""
    cursor = conn.cursor()

    # 先看看已经有多少数据,避免重复导入
    cursor.execute("SELECT COUNT(*) FROM messages")
    existing_count = cursor.fetchone()[0]

    imported = 0
    skipped = 0

    with open(csv_path, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row in reader:
            timestamp = row.get("timestamp", "").strip()
            sender = row.get("sender", "").strip()
            content = row.get("content", "").strip()
            chat_name = row.get("chat_name", "").strip()

            if not content:
                skipped += 1
                continue

            # 提取日期部分,方便按日期筛选
            date_str = timestamp[:10] if len(timestamp) >= 10 else""

            cursor.execute(
                "INSERT INTO messages (timestamp, sender, content, chat_name, date_str) VALUES (?, ?, ?, ?, ?)",
                (timestamp, sender, content, chat_name, date_str),
            )
            imported += 1

    conn.commit()

    print(f"导入完成:新增 {imported} 条,跳过空消息 {skipped} 条")
    print(f"数据库总计:{existing_count + imported} 条消息")


# ============ 3. 全文搜索 ============

def search_messages(conn, query, chat_name=None, sender=None, limit=20):
    """
    全文搜索聊天记录

    参数:
        query: 搜索关键词
        chat_name: 限定聊天对象(可选)
        sender: 限定发送人(可选)
        limit: 返回结果数量
    """
    cursor = conn.cursor()

    # 构建FTS5查询
    # FTS5 支持 AND OR NOT 等操作符
    # 默认用关键词搜索所有字段
    fts_query = query

    # 基础查询:从FTS表搜索,关联主表拿完整数据
    sql = """
        SELECT
            m.timestamp,
            m.sender,
            m.content,
            m.chat_name,
            rank
        FROM messages_fts fts
        JOIN messages m ON m.id = fts.rowid
        WHERE messages_fts MATCH ?
    """
    params = [fts_query]

    # 可选:按聊天名称过滤
    if chat_name:
        sql += " AND m.chat_name = ?"
        params.append(chat_name)

    # 可选:按发送人过滤
    if sender:
        sql += " AND m.sender = ?"
        params.append(sender)

    # 按相关度排序(rank越小越相关)
    sql += " ORDER BY rank LIMIT ?"
    params.append(limit)

    cursor.execute(sql, params)
    results = cursor.fetchall()

    return results


def search_by_date_range(conn, query, start_date, end_date, limit=20):
    """在指定日期范围内搜索"""
    cursor = conn.cursor()

    sql = """
        SELECT
            m.timestamp,
            m.sender,
            m.content,
            m.chat_name,
            rank
        FROM messages_fts fts
        JOIN messages m ON m.id = fts.rowid
        WHERE messages_fts MATCH ?
          AND m.date_str BETWEEN ? AND ?
        ORDER BY rank
        LIMIT ?
    """
    cursor.execute(sql, (query, start_date, end_date, limit))
    return cursor.fetchall()


# ============ 4. 统计功能 ============

def get_chat_stats(conn):
    """获取各聊天的消息统计"""
    cursor = conn.cursor()
    cursor.execute("""
        SELECT
            chat_name,
            COUNT(*) as msg_count,
            MIN(date_str) as earliest,
            MAX(date_str) as latest
        FROM messages
        GROUP BY chat_name
        ORDER BY msg_count DESC
    """)
    returncursor.fetchall()


# ============ 5. 交互式查询 ============

def print_results(results):
    """格式化打印搜索结果"""
    if not results:
        print("没有找到匹配的消息\n")
        return

    for i, (timestamp, sender, content, chat_name, rank) in enumerate(results, 1):
        print(f"[{i}] {timestamp}  [{chat_name}]  {sender}:")
        print(f"    {content}")
        print()


def interactive_search(conn):
    """交互式搜索循环"""
    print("\n"+"="*50)
    print("微信聊天记录搜索")
    print("="*50)
    print("命令说明:")
    print("  直接输入关键词 → 全文搜索")
    print("  /chat 群名 → 限定在某个群搜索")
    print("  /from 人名 → 限定某人发的消息")
    print("  /date 起始日期 结束日期 关键词 → 按日期范围搜索")
    print("  /stats → 查看各聊天消息统计")
    print("  /quit → 退出")
    print("="*50+"\n")

    current_chat = None
    current_sender = None

    whileTrue:
        try:
            user_input = input("搜索> ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n再见")
            break

        if not user_input:
            continue

        if user_input == "/quit":
            print("再见")
            break

        elif user_input == "/stats":
            stats = get_chat_stats(conn)
            print(f"\n{'聊天名称':<20} {'消息数':>8} {'最早日期':<12} {'最近日期':<12}")
            print("-"*56)
            forchat_name, count, earliest, latestinstats[:20]:
                print(f"{chat_name:<20} {count:>8} {earliest:<12} {latest:<12}")
            print()

        elif user_input.startswith("/chat "):
            current_chat = user_input[6:].strip()
            current_sender = None
            print(f"已限定搜索范围:{current_chat}\n")

        elif user_input.startswith("/from "):
            current_sender = user_input[6:].strip()
            print(f"已限定发送人:{current_sender}\n")

        elif user_input.startswith("/date "):
            parts = user_input[6:].strip().split(" ", 2)
            if len(parts) >= 3:
                start_date, end_date, query = parts
                results = search_by_date_range(conn, query, start_date, end_date)
                print(f"日期范围 {start_date} ~ {end_date} 的搜索结果:\n")
                print_results(results)
            else:
                print("格式:/date 2024-01-01 2024-06-30 关键词\n")

        elif user_input == "/all":
            current_chat = None
            current_sender = None
            print("已清除搜索限定,搜索全部记录\n")

        else:
            # 普通关键词搜索
            results = search_messages(
                conn,
                user_input,
                chat_name=current_chat,
                sender=current_sender,
            )
            print()
            print_results(results)


# ============ 主程序 ============

def main():
    db_path = Path(DB_FILE)
    csv_path = Path(CSV_FILE)

    # 初始化数据库
    conn = init_database(db_path)

    # 如果CSV存在且数据库是新的,就导入
    if csv_path.exists():
        import_csv(conn, str(csv_path))
    else:
        print(f"提示:未找到 {CSV_FILE},将使用已有数据库")
        cursor = conn.cursor()
        cursor.execute("SELECT COUNT(*) FROM messages")
        count = cursor.fetchone()[0]
        if count == 0:
            print("数据库为空,请先导出微信聊天记录为CSV文件")
            print(f"并将文件命名为 {CSV_FILE} 放在当前目录")
            return
        print(f"数据库中已有 {count} 条消息")

    # 进入交互式搜索
    interactive_search(conn)
    conn.close()


if __name__ == "__main__":
    main()

运行效果

先把CSV文件放到脚本同目录下,然后运行:

代码语言:javascript
复制
python wechat_search.py
代码语言:javascript
复制
导入完成:新增 158432 条,跳过空消息 2041 条
数据库总计:158432 条消息

==================================================
微信聊天记录搜索
==================================================
命令说明:
  直接输入关键词 → 全文搜索
  /chat 群名 → 限定在某个群搜索
  /from 人名 → 限定某人发的消息
  /date 起始日期 结束日期 关键词 → 按日期范围搜索
  /stats → 查看各聊天消息统计
  /quit → 退出
==================================================

搜索> 湘菜馆

[1] 2024-03-15 14:23:45  [朋友聚餐群]  我:
    上次去的那个湘菜馆不错

[2] 2024-03-15 14:24:12  [朋友聚餐群]  李四:
    哪个湘菜馆?

[3] 2024-03-15 14:24:58  [朋友聚餐群]  我:
    就是望京那个 好像叫湘味楼

搜索> /chat 朋友聚餐群
已限定搜索范围:朋友聚餐群

搜索> 包间

[1] 2024-03-15 14:26:01  [朋友聚餐群]  我:
    有 上次我们就是在包间吃的 有单间

搜索> /date 2024-03-01 2024-03-31 聚餐
日期范围 2024-03-01 ~ 2024-03-31 的搜索结果:

[1] 2024-03-15 14:23:01  [朋友聚餐群]  张三:
    周末去哪吃饭

[2] 2024-03-20 18:45:12  [朋友聚餐群]  李四:
    周五聚餐的事定了吗

搜索> /stats

聊天名称                 消息数   最早日期     最近日期
--------------------------------------------------------
朋友聚餐群               12456   2022-06-15  2024-11-20
工作项目群               34521   2023-01-08  2024-11-21
老王                     8923    2021-09-01  2024-11-19

搜“包间”的时候,即使消息原文用的是“单间”,FTS5也能通过相关度排序把它排在前面。当然这取决于具体的分词效果,中文场景下 unicode61 分词器表现还行。


关于中文分词

FTS5内置的 unicode61 分词器,对中文的处理不算完美。它是按字符级别拆的,不是按词。搜“湘菜馆”能找到结果,是因为三个字都匹配上了。

如果你想要更精确的中文分词,可以接入jieba。改造方式:

代码语言:javascript
复制
import jieba

def tokenize_chinese(text):
    """用jieba分词"""
    return " ".join(jieba.cut(text))

然后在导入CSV的时候,对content先做一遍分词,存到FTS表里。搜索时也对查询词做分词。效果会好不少,但导入速度会慢一些。

对于日常搜个关键词找聊天记录来说,unicode61 已经够用了。


数据量大了怎么办

15万条消息,搜索响应在毫秒级别。SQLite的FTS5在这方面表现很好。

如果你有好几年的聊天记录,量级到了百万条,也没问题。SQLite本身支持几十GB的数据,FTS5的索引效率也不错。

实在觉得慢了,可以加个日期范围过滤,先缩小搜索范围再做全文检索。代码里的 search_by_date_range 函数就是干这个的。


一些安全提醒

聊天记录是很隐私的数据。几点建议:

  • 生成的 wechat_search.db 文件不要传到公共位置
  • CSV导出文件用完可以删掉,数据库里已经有了
  • 如果你在公司的电脑上跑,注意别把私人聊天记录导出来
  • 不要把数据库上传到GitHub

最后

这个脚本没有任何第三方依赖。Python自带的 sqlite3csv 模块就搞定了。FTS5是SQLite 3.9.0以后内置的功能,Python 3.8+ 的SQLite版本都支持。

5行代码是个标题党。但核心逻辑确实不多——建表、导入、查询,加起来也就几十行。

如果你还想加什么功能,比如按时间线浏览、导出某人的所有消息、统计聊天活跃度,思路都在代码里了,扩展起来不难。

下一篇写什么,评论区见。

“无他,惟手熟尔”!有需要的用起来!

本文参与 腾讯云自媒体同步曝光计划,分享自微信公众号。
原始发表:2026-07-08,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 Nicholas与Pypi 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 那个让我翻了两小时聊天记录的下午
  • 思路
  • 怎么导出微信聊天记录
  • 完整代码
  • 运行效果
  • 关于中文分词
  • 数据量大了怎么办
  • 一些安全提醒
  • 最后
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档