Exporting Local WeChat Voice Messages on macOS

~/Library/Containers/com.tencent.xinWeChat/Data/Documents/xwechat_files/
  <account>/db_storage/
    contact/contact.db
    message/message_0.db
    message/media_0.db
    ...

import hashlib

table = "Msg_" + hashlib.md5(username.encode("utf-8")).hexdigest()
message_sql = f"""
SELECT m.local_id, m.server_id, m.create_time,
       n.user_name AS sender
FROM {table} AS m
LEFT JOIN Name2Id AS n ON n.rowid = m.real_sender_id
WHERE (m.local_type & 4294967295) = 34
"""

SELECT v.rowid AS media_rowid, v.svr_id,
       length(v.voice_data) AS bytes
FROM VoiceInfo AS v
JOIN Name2Id AS n ON n.rowid = v.chat_name_id
WHERE n.user_name = :username;

from collections import defaultdict


def match_audio(messages, media_rows):
    # media_rows already belong to this conversation, across all media shards.
    candidates = defaultdict(list)
    for row in media_rows:
        sid = str(row["svr_id"] or "0")
        if sid != "0" and int(row["bytes"] or 0) > 0:
            candidates[sid].append(row)

    for message in messages:
        matches = candidates.get(str(message["server_id"] or "0"), [])
        status = ("available" if len(matches) == 1 else
                  "not_local" if not matches else "ambiguous_audio")
        yield {**message, "status": status,
               "audio": matches[0] if len(matches) == 1 else None}

{
  "keys": {
    "contact/contact.db": {"enc_key": "<64 hexadecimal characters>"},
    "message/message_0.db": {"enc_key": "<64 hexadecimal characters>"},
    "message/media_0.db": {"enc_key": "<64 hexadecimal characters>"}
  }
}

python3 -m venv .venv
.venv/bin/python -m pip --require-virtualenv install 'pilk==0.2.4'
brew install sqlcipher ffmpeg
chmod 600 /absolute/private/keys.json

.venv/bin/python voice_export.py \
  --db-root '/absolute/account/db_storage' \
  --key-file '/absolute/private/keys.json' \
  --contact 'exact username' \
  --since 2026-08-01 --until 2026-10-13

import hashlib
import json
import subprocess
import tempfile
from pathlib import Path

import pilk


def convert_voice(data, destination):
    if not data.startswith((b"#!SILK_V3", b"\x02#!SILK_V3")):
        raise ValueError("Not recognised SILK audio")
    destination = Path(destination)
    if destination.exists():
        raise FileExistsError(destination)

    with tempfile.TemporaryDirectory(prefix="voice-convert-") as folder:
        silk = Path(folder) / "voice.silk"
        pcm = Path(folder) / "voice.pcm"
        silk.write_bytes(data)
        pilk.decode(str(silk), str(pcm), pcm_rate=24000)
        size = pcm.stat().st_size
        if size == 0 or size % 2:
            raise ValueError("Empty or incomplete 16-bit PCM")
        decoded_seconds = size / (24000 * 1 * 2)
        subprocess.run([
            "ffmpeg", "-v", "error", "-nostdin", "-n",
            "-f", "s16le", "-ar", "24000", "-ac", "1",
            "-i", str(pcm), "-c:a", "libmp3lame", "-q:a", "2",
            str(destination),
        ], check=True)

    subprocess.run([
        "ffmpeg", "-v", "error", "-nostdin", "-i", str(destination),
        "-f", "null", "-",
    ], check=True)
    probe = json.loads(subprocess.check_output([
        "ffprobe", "-v", "error", "-show_entries",
        "format=duration:stream=codec_name", "-of", "json",
        str(destination),
    ], text=True))
    if not probe["streams"] or probe["streams"][0]["codec_name"] != "mp3":
        raise ValueError("Output is not MP3")
    seconds = float(probe["format"]["duration"])
    if abs(seconds - decoded_seconds) > 0.25:
        raise ValueError("Unexpected duration change")
    return {
        "decoded_seconds": decoded_seconds,
        "mp3_seconds": seconds,
        "silk_sha256": hashlib.sha256(data).hexdigest(),
        "mp3_sha256": hashlib.sha256(destination.read_bytes()).hexdigest(),
    }

#!/usr/bin/env python3
"""Export locally stored WeChat 4.x voice messages on macOS.

Requires existing verified keys, SQLCipher, pilk and FFmpeg/ffprobe.
"""
# SQLCipher reader adapted from acmerfight:
# https://gist.github.com/acmerfight/0a01249ef72970d07a2603dbb629e80f
# The following notice covers the adapted reader.
# MIT License
#
# Copyright (c) 2026 Contributors
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import ctypes as C
import ctypes.util
import json
import os
from pathlib import Path
import re


def library():
    candidates = [os.environ.get('SQLCIPHER_LIBRARY'),
                  '/opt/homebrew/opt/sqlcipher/lib/libsqlcipher.dylib',
                  '/usr/local/opt/sqlcipher/lib/libsqlcipher.dylib',
                  ctypes.util.find_library('sqlcipher')]
    name = next((x for x in candidates if x and (not x.startswith('/') or Path(x).exists())), None)
    if not name:
        raise RuntimeError('SQLCipher library missing; install with brew install sqlcipher')
    lib = C.CDLL(name)
    lib.sqlite3_open_v2.argtypes = [C.c_char_p, C.POINTER(C.c_void_p), C.c_int, C.c_char_p]
    lib.sqlite3_open_v2.restype = C.c_int
    lib.sqlite3_close.argtypes = [C.c_void_p]
    lib.sqlite3_key.argtypes = [C.c_void_p, C.c_void_p, C.c_int]
    lib.sqlite3_key.restype = C.c_int
    lib.sqlite3_exec.argtypes = [C.c_void_p, C.c_char_p, C.c_void_p, C.c_void_p, C.c_void_p]
    lib.sqlite3_exec.restype = C.c_int
    lib.sqlite3_busy_timeout.argtypes = [C.c_void_p, C.c_int]
    return lib


class DB:
    def __init__(self, path, key):
        if not re.fullmatch(r'[0-9a-fA-F]{64}', key):
            raise ValueError('expected 64 hex characters for raw SQLCipher key')
        self.lib = library()
        self.ptr = C.c_void_p()
        flags = 1  # SQLITE_OPEN_READONLY
        rc = self.lib.sqlite3_open_v2(os.fsencode(path), C.byref(self.ptr), flags, None)
        if rc:
            self.close()
            raise RuntimeError(f'SQLCipher open failed (code {rc})')
        try:
            key_arg = ("x'" + key + "'").encode('ascii')
            rc = self.lib.sqlite3_key(self.ptr, key_arg, len(key_arg))
            if rc:
                raise RuntimeError(f'SQLCipher key setup failed (code {rc})')
            self.query('PRAGMA cipher_compatibility=4;')
            self.lib.sqlite3_busy_timeout(self.ptr, 2000)
            self.query('PRAGMA query_only=ON;')
            self.query('SELECT count(*) FROM sqlite_master;')
        except BaseException:
            self.close()
            raise

    def query(self, sql):
        rows = []
        callback_type = C.CFUNCTYPE(C.c_int, C.c_void_p, C.c_int,
                                   C.POINTER(C.c_char_p), C.POINTER(C.c_char_p))
        def collect(_, n, values, names):
            rows.append({names[i].decode(): values[i].decode('utf-8', 'replace')
                         if values[i] is not None else None for i in range(n)})
            return 0
        callback = callback_type(collect)
        rc = self.lib.sqlite3_exec(self.ptr, sql.encode(), callback, None, None)
        if rc:
            # No raw SQL or native error strings, since those may reveal secrets/content.
            raise RuntimeError(f'SQLCipher query failed (code {rc})')
        return rows

    def close(self):
        if self.ptr:
            self.lib.sqlite3_close(self.ptr)
            self.ptr = C.c_void_p()

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.close()


def load_key(path, db_path):
    p = Path(path)
    if p.stat().st_mode & 0o077:
        raise RuntimeError('key file must be private (chmod 600)')
    data = json.loads(p.read_text())
    if isinstance(data, dict) and isinstance(data.get('keys'), dict):
        data = data['keys']
    if not isinstance(data, dict):
        raise RuntimeError('key manifest must be an object')
    target = Path(db_path).resolve()
    with target.open('rb') as f:
        salt = f.read(16).hex()
    entries = []
    for name, value in data.items():
        normalized = name.replace('\\', '/')
        absolute_match = normalized.startswith('/') and Path(normalized).resolve() == target
        if normalized == salt or str(target).endswith('/' + normalized) or absolute_match:
            if isinstance(value, dict):
                value = value.get('enc_key', value.get('key'))
            if isinstance(value, str) and re.fullmatch('[0-9a-fA-F]{64}', value):
                entries.append(value.lower())
    if len(set(entries)) != 1:
        raise RuntimeError('missing or ambiguous key for this database')
    return entries[0]


import argparse
import collections
import csv
import hashlib
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
from datetime import date, datetime, time, timedelta
from pathlib import Path
from zoneinfo import ZoneInfo


def quote(value):
    return "'" + str(value).replace("'", "''") + "'"


def required_databases(root):
    return sorted([*(root / "contact").glob("contact.db"),
                   *[p for p in (root / "message").glob("message_*.db")
                     if re.fullmatch(r"message_\d+\.db", p.name)],
                   *[p for p in (root / "message").glob("media_*.db")
                     if re.fullmatch(r"media_\d+\.db", p.name)]])


def private_write(path, value):
    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    fd, tmp = tempfile.mkstemp(prefix=".write-", dir=path.parent)
    try:
        with os.fdopen(fd, "w") as f:
            json.dump(value, f, ensure_ascii=False, indent=2)
        os.replace(tmp, path)
    finally:
        if os.path.exists(tmp):
            os.unlink(tmp)


class Reader:
    def __init__(self, root, keyfile):
        self.root, self.keyfile = Path(root), Path(keyfile)

    def open(self, path):
        try:
            key = load_key(self.keyfile, path)
            return DB(path, key)
        except (RuntimeError, OSError, ValueError) as e:
            raise ValueError("Cannot read " + str(path.relative_to(self.root)) +
                             ". Supply a verified key manifest for this database; original error: " + str(e)) from e

    def contacts(self, query):
        with self.open(self.root / "contact/contact.db") as db:
            rows = db.query("SELECT username,alias,remark,nick_name FROM contact")
        needle = query.casefold()
        return [r for r in rows if not (r["username"] or "").endswith("@chatroom")
                and any(needle in (r[k] or "").casefold()
                        for k in ("username", "alias", "remark", "nick_name"))]

    def resolve(self, query):
        matches = self.contacts(query)
        exact = [r for r in matches if any((r[k] or "").casefold() == query.casefold()
                 for k in ("username", "alias", "remark", "nick_name"))]
        if len(exact) != 1:
            raise ValueError("Contact is absent or ambiguous. Run contacts and use the exact username.")
        return exact[0]

    def voices(self, contact, direction="received", since=None, until=None, timezone="Asia/Shanghai"):
        user = contact["username"]
        tz = ZoneInfo(timezone)
        lo = int(datetime.combine(date.fromisoformat(since), time.min, tz).timestamp()) if since else None
        hi = int(datetime.combine(date.fromisoformat(until) + timedelta(days=1), time.min, tz).timestamp()) if until else None
        if lo is not None and hi is not None and lo >= hi:
            raise ValueError("--since must not be later than --until")
        table = "Msg_" + hashlib.md5(user.encode()).hexdigest()
        messages = {}
        shards = [p for p in required_databases(self.root) if re.fullmatch(r"message_\d+\.db", p.name)]
        if not shards:
            raise ValueError("No local message shards found")
        for path in shards:
            with self.open(path) as db:
                if not db.query("SELECT name FROM sqlite_master WHERE name=" + quote(table)):
                    continue
                rows = db.query("SELECT m.local_id,m.server_id,m.create_time,n.user_name AS sender "
                                "FROM " + table + " m LEFT JOIN Name2Id n ON n.rowid=m.real_sender_id "
                                "WHERE (m.local_type & 4294967295)=34")
            for row in rows:
                received = row["sender"] == user
                if direction == "received" and not received:
                    continue
                if direction == "sent" and (received or row["sender"] is None):
                    continue
                timestamp = int(row["create_time"])
                if (lo is not None and timestamp < lo) or (hi is not None and timestamp >= hi):
                    continue
                sid = str(row["server_id"] or "0")
                ident = sid if sid != "0" else path.name + ":" + row["local_id"]
                record = dict(message_id=sid, local_id=row["local_id"], timestamp=timestamp,
                              sent_at=datetime.fromtimestamp(timestamp, tz).isoformat(),
                              direction="received" if received else ("sent" if row["sender"] else "unknown"),
                              message_database=path.name)
                if ident in messages and any(messages[ident][k] != record[k]
                                             for k in ("timestamp", "direction")):
                    raise ValueError("Conflicting message records for server ID " + sid)
                messages.setdefault(ident, record)
        media = collections.defaultdict(list)
        for path in required_databases(self.root):
            if not re.fullmatch(r"media_\d+\.db", path.name):
                continue
            with self.open(path) as db:
                rows = db.query("SELECT v.rowid AS media_rowid,v.svr_id,length(v.voice_data) AS bytes "
                                "FROM VoiceInfo v JOIN Name2Id n ON n.rowid=v.chat_name_id "
                                "WHERE n.user_name=" + quote(user))
            for row in rows:
                if row["svr_id"] and row["svr_id"] != "0" and int(row["bytes"] or "0") > 0:
                    media[row["svr_id"]].append((path.name, row["media_rowid"], int(row["bytes"])))
        result = []
        for record in messages.values():
            matches = media.get(record["message_id"], [])
            record["available"] = len(matches) == 1
            record["status"] = "available" if len(matches) == 1 else ("not_local" if not matches else "ambiguous_audio")
            record["audio_bytes"] = matches[0][2] if len(matches) == 1 else None
            record["_audio"] = matches[0] if len(matches) == 1 else None
            result.append(record)
        return sorted(result, key=lambda r: (r["timestamp"], r["message_id"]))

    def audio(self, record, user):
        shard, rowid, size = record["_audio"]
        with self.open(self.root / "message" / shard) as db:
            rows = db.query("SELECT hex(v.voice_data) AS body FROM VoiceInfo v "
                            "JOIN Name2Id n ON n.rowid=v.chat_name_id WHERE v.rowid=" + str(int(rowid)) +
                            " AND v.svr_id=" + str(int(record["message_id"])) +
                            " AND n.user_name=" + quote(user))
        if len(rows) != 1:
            raise ValueError("Audio changed since listing; rerun the command")
        data = bytes.fromhex(rows[0]["body"])
        if len(data) != size or not data.startswith((b"#!SILK_V3", b"\x02#!SILK_V3")):
            raise ValueError("Audio is incomplete or not recognised SILK")
        return data


def public(record):
    return {k: v for k, v in record.items() if not k.startswith("_")}


def export(reader, contact, rows, output, timezone, keep_silk=False):
    import pilk
    ffmpeg, ffprobe = shutil.which("ffmpeg"), shutil.which("ffprobe")
    if not ffmpeg or not ffprobe:
        raise ValueError("Install ffmpeg and ffprobe (brew install ffmpeg)")
    if not any(r["available"] for r in rows):
        raise ValueError("No unambiguous local audio to export")
    output = Path(output).expanduser().resolve()
    output.mkdir(parents=True, exist_ok=False, mode=0o700)
    counts = collections.Counter(r["timestamp"] for r in rows if r["available"])
    manifest = {"contact": contact["username"], "timezone": timezone,
                "source": "Local WeChat chat/media databases; no favourites or playback recovery",
                "records": []}
    private_write(output / "manifest.json", manifest)
    for row in rows:
        item = public(row)
        if row["available"]:
            data = reader.audio(row, contact["username"])
            stamp = datetime.fromtimestamp(row["timestamp"], ZoneInfo(timezone)).strftime("%Y-%m-%d_%H-%M-%S")
            if counts[row["timestamp"]] > 1:
                stamp += "_" + row["message_id"]
            mp3 = output / (stamp + ".mp3")
            with tempfile.TemporaryDirectory(prefix="wechat-voice-") as tmp:
                silk, pcm = Path(tmp) / "voice.silk", Path(tmp) / "voice.pcm"
                silk.write_bytes(data)
                pilk.decode(str(silk), str(pcm), pcm_rate=24000)
                if pcm.stat().st_size == 0 or pcm.stat().st_size % 2:
                    raise ValueError("Decoder produced empty or invalid PCM")
                decoded = pcm.stat().st_size / 48000
                subprocess.run([ffmpeg, "-v", "error", "-nostdin", "-n", "-f", "s16le",
                                "-ar", "24000", "-ac", "1", "-i", str(pcm),
                                "-c:a", "libmp3lame", "-q:a", "2", str(mp3)], check=True)
            subprocess.run([ffmpeg, "-v", "error", "-nostdin", "-i", str(mp3),
                            "-f", "null", "-"], check=True)
            probe = json.loads(subprocess.check_output([ffprobe, "-v", "error", "-show_entries",
                                "format=duration:stream=codec_name", "-of", "json", str(mp3)], text=True))
            if not probe["streams"] or probe["streams"][0]["codec_name"] != "mp3":
                raise ValueError("Output is not MP3")
            duration = float(probe["format"]["duration"])
            if abs(duration - decoded) > 0.25:
                raise ValueError("MP3 duration differs unexpectedly from decoded audio")
            if keep_silk:
                (output / (stamp + ".silk")).write_bytes(data)
            item.update(filename=mp3.name, decoded_seconds=decoded, mp3_seconds=duration,
                        silk_sha256=hashlib.sha256(data).hexdigest(),
                        mp3_sha256=hashlib.sha256(mp3.read_bytes()).hexdigest(),
                        status="exported")
        manifest["records"].append(item)
        private_write(output / "manifest.json", manifest)
    fields = sorted(set().union(*(r.keys() for r in manifest["records"])))
    with (output / "manifest.csv").open("w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fields)
        writer.writeheader()
        writer.writerows(manifest["records"])
    return {"output": str(output),
            "exported": sum(r["status"] == "exported" for r in manifest["records"]),
            "skipped": sum(r["status"] != "exported" for r in manifest["records"])}


def main():
    os.umask(0o077)
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--db-root", type=Path, required=True)
    parser.add_argument("--key-file", type=Path, required=True)
    parser.add_argument("--contact", required=True, help="Exact username, nickname, remark or alias")
    parser.add_argument("--since")
    parser.add_argument("--until")
    parser.add_argument("--timezone", default="Asia/Shanghai")
    parser.add_argument("--direction", choices=("received", "sent", "both"), default="received")
    parser.add_argument("--output", help="Export into this NEW directory; omit to list")
    parser.add_argument("--keep-silk", action="store_true")
    args = parser.parse_args()
    root = args.db_root.expanduser().resolve()
    reader = Reader(root, args.key_file.expanduser().resolve())
    try:
        contact = reader.resolve(args.contact)
        rows = reader.voices(contact, args.direction, args.since, args.until, args.timezone)
        result = (export(reader, contact, rows, args.output, args.timezone, args.keep_silk)
                  if args.output else {"contact": contact, "records": [public(r) for r in rows]})
        print(json.dumps(result, ensure_ascii=False, indent=2))
    except (ValueError, RuntimeError, OSError, subprocess.SubprocessError, ImportError) as error:
        parser.exit(1, "Error: " + str(error) + "\n")


if __name__ == "__main__":
    main()