scripts/whatsapp_read.py — Desktop Chat Reader

#!/usr/bin/env python3
# ============================================================
# File: whatsapp_read.py
# Created by: Zero Width Lab
# Date: 2026-09-02
# Version: 0.1.3; requires Python 3.11 or newer
# Description: Read-only CLI over the macOS WhatsApp Desktop SQLite database.
#              Standalone: Python standard library only, no third-party
#              dependencies, built for ad-hoc reading by people and AI agents.
# By Zero Width Lab (zerowidthlab.com)
# ============================================================
#
# WHY THIS EXISTS
#   WhatsApp has no local API; on macOS, WhatsApp Desktop keeps its messages
#   in ChatStorage.sqlite. This script wraps that database with the schema
#   quirks already handled for ad-hoc reading by people and AI agents.
#
# QUICK START
#   python3 scripts/whatsapp_read.py chats --days 30
#   python3 scripts/whatsapp_read.py chats --search alice
#   python3 scripts/whatsapp_read.py read "Alice" --limit 30
#   python3 scripts/whatsapp_read.py read 120000000000000001@g.us --days 7
#   python3 scripts/whatsapp_read.py search "forecast" --days 60
#   python3 scripts/whatsapp_read.py contacts --search example-co
#   ...add --json to any command for machine-readable output.
#
# SAFETY
#   Reads a private temporary DB/WAL/SHM snapshot with mode=ro. Sources
#   are never written; only the private system-temp copy is written, then
#   deleted on normal exit, errors, Ctrl-C, SIGTERM and SIGHUP. SIGKILL or
#   power loss can leave the copy in system temp. Requires up to 1 GiB free
#   space for DB/WAL/SHM. Changing snapshots are retried 3 times.
#
# SCHEMA NOTES
#   ZWAMESSAGE          one row per message
#     ZMESSAGEDATE      Core Data epoch seconds (since 2001-01-01 UTC).
#                       Add 978307200 for Unix epoch. Order by this.
#     ZTEXT             body; NULL for most media and system messages
#     ZISFROMME         1 = sent by the account owner
#     ZMESSAGETYPE      see MESSAGE_TYPES below; the range goes well past 15
#     ZGROUPMEMBER      FK to ZWAGROUPMEMBER.Z_PK — how you identify the
#                       sender of a GROUP message
#     ZPUSHNAME         DO NOT USE as a name — the large majority of rows hold
#                       a base64 protobuf blob, not text
#   ZWACHATSESSION      one or more rows per chat; group by ZCONTACTJID when
#                       presenting the chat directory
#     ZCONTACTJID       the JID. '@g.us' = group, '@s.whatsapp.net' or
#                       '@lid' = individual ('@lid' is a multi-device
#                       linked identity)
#     ZPARTNERNAME      chat display name
#   ZWAGROUPMEMBER
#     ZMEMBERJID        clean JID of a group participant; this is the reliable
#                       field
#     ZCONTACTNAME      DO NOT USE — usually empty and otherwise can contain a
#                       base64 protobuf blob. Never a name.
#     ZFIRSTNAME        DO NOT USE — same problem (base64 protobuf).
#   ZWAPROFILEPUSHNAME  JID -> display name directory. This is what turns a
#                       raw JID into a human name.
#   ZWAMEDIAITEM        media metadata; multiple rows can reference one
#                       message, so select at most one ZTITLE / ZMEDIALOCALPATH
#
#   SENDER RESOLUTION — the part naive implementations get wrong.
#   There is no usable name column on the message row. Use this chain:
#     ZISFROMME = 1                     -> "ME"
#     1:1 chat (JID not '@g.us')        -> ZWACHATSESSION.ZPARTNERNAME
#     group  -> ZGROUPMEMBER -> ZWAGROUPMEMBER.ZMEMBERJID, then look that JID
#               up in ZWAPROFILEPUSHNAME; if absent, show the raw JID
#   If the directory has no matching name, resolution falls back to the raw
#   JID. Never reach for ZPUSHNAME, ZCONTACTNAME or
#   ZFIRSTNAME as a fallback — all three can be protobuf blobs.
#
# GOTCHAS
#   - ZCONTACTJID is not unique. The chat directory groups duplicate sessions;
#     the resolver picks the session with the most messages, then lowest PK.
#   - ZTEXT is NULL for media/system rows. Never assume a body exists.
#   - There is no FTS index; `search` is a LIKE scan with literal substring
#     matching, so user-provided %, _ and backslash are escaped.
#   - Media/profile metadata is selected with scalar subqueries so duplicate
#     metadata rows cannot duplicate messages; message ordering also uses PK.
#   - If WhatsApp Desktop is mid-write you may get SQLITE_BUSY. The script
#     reports it plainly instead of returning empty results.
# ============================================================

from __future__ import annotations

import sys

if sys.version_info < (3, 11):
    print(
        f"Desktop Chat Reader needs Python 3.11 or newer; this is {sys.version_info.major}.{sys.version_info.minor}",
        file=sys.stderr,
    )
    raise SystemExit(1)

import argparse
import errno
import json
import math
import os
import re
import sqlite3
import shutil
import signal
import tempfile
import stat
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable

__author__ = "Zero Width Lab"
MAX_DATABASE_BYTES = 1024 * 1024 * 1024
MAX_VALUE_BYTES = 1024 * 1024
MAX_RESULTS = 1000
MAX_QUERY_SECONDS = 30  # Per-query wall-clock limit checked every 1000 VM steps.

CORE_DATA_EPOCH = 978307200  # 2001-01-01 00:00:00 UTC

DEFAULT_DB = Path(
    os.environ.get("WHATSAPP_DB")
    or os.path.expanduser(
        "~/Library/Group Containers/group.net.whatsapp.WhatsApp.shared/ChatStorage.sqlite"
    )
)

REQUIRED_TABLES = {
    "ZWAMESSAGE",
    "ZWACHATSESSION",
    "ZWAGROUPMEMBER",
    "ZWAPROFILEPUSHNAME",
    "ZWAMEDIAITEM",
}

# Observed message type labels. Anything not listed renders as "type:<n>"
# rather than being silently mislabeled.
MESSAGE_TYPES = {
    0: "text",
    1: "image",
    2: "video",
    3: "voice",
    4: "contact",
    5: "location",
    6: "group-event",
    7: "link",  # URL in ZTEXT, page title in ZWAMEDIAITEM.ZTITLE
    8: "document",  # filename in ZTEXT, human caption in ZTITLE
    10: "group-event",  # event code lives in ZGROUPEVENTTYPE
    11: "gif",
    12: "media",
    13: "gif",
    # Type 14 semantics are UNSETTLED. Verifiable payload fields include a
    # group JID in ZVCARDNAME and a member JID in ZVCARDSTRING, often with
    # empty ZTEXT. It is not a contact card (that is type 4). Some forensic
    # work treats 14 as the Delete-for-Everyone retract marker; other research
    # points at type 15, which can also have empty text. Left descriptive
    # rather than guessed.
    14: "group-ref",
    15: "sticker",
    27: "media",
    28: "poll",
    43: "reaction",
    46: "media",
    54: "notice",
    57: "notice",
    59: "e2e-notice",
    60: "notice",
    66: "notice",
    75: "notice",
}


# ── plumbing ────────────────────────────────────────────────


def require_supported_platform() -> None:
    if sys.platform != "darwin":
        die("macOS only: this CLI refuses database I/O on unsupported systems")


class SnapshotChanged(Exception):
    """A source changed while the private copy was being made."""


class SnapshotLifetime:
    """Own cleanup independently of connection construction and assignment."""

    def __init__(self) -> None:
        self.directory: Path | None = None
        self.previous_handlers = {}

    def start(self) -> None:
        try:
            for name in ("SIGTERM", "SIGHUP"):
                number = getattr(signal, name)
                self.previous_handlers[number] = signal.getsignal(number)
                signal.signal(number, self.interrupted)
        except BaseException:
            self.close()
            raise

    @staticmethod
    def interrupted(number, frame) -> None:
        raise SystemExit(128 + number)

    def close(self) -> None:
        try:
            if self.directory is not None:
                directory = self.directory
                self.directory = None
                shutil.rmtree(directory)
        finally:
            for number, handler in self.previous_handlers.items():
                signal.signal(number, handler)
            self.previous_handlers.clear()


class SnapshotConnection(sqlite3.Connection):
    snapshot_directory: Path | None = None
    lifetime: SnapshotLifetime | None = None

    def close(self) -> None:
        try:
            super().close()
        finally:
            if self.lifetime is not None:
                self.lifetime.close()
                self.snapshot_directory = None


def source_signature(info) -> tuple:
    return info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns


def copy_snapshot(db_path: Path, lifetime: SnapshotLifetime) -> Path:
    """Copy DB/WAL/SHM without source writes; retry changed sources three times."""
    # Resolve ancestors for macOS system temp aliases, but never the source leaf.
    db_path = db_path.expanduser()
    parent = db_path.parent.resolve(strict=True)
    for attempt in range(3):
        lifetime.directory = Path(tempfile.mkdtemp(prefix="desktop-chat-reader-"))
        directory = lifetime.directory
        sources = []
        completed = False
        try:
            directory.chmod(0o700)
            total = 0
            for suffix in ("", "-wal", "-shm"):
                path = parent / (db_path.name + suffix)
                try:
                    descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
                except FileNotFoundError:
                    if not suffix:
                        die("Database not found")
                    sources.append((path, None, None))
                    continue
                sources.append((path, descriptor, os.fstat(descriptor)))
                info = sources[-1][2]
                if not stat.S_ISREG(info.st_mode):
                    die("Database sources must be regular files; symlinks are refused")
                total += info.st_size
                if total > MAX_DATABASE_BYTES:
                    die("Database and WAL/SHM exceed combined 1073741824-byte limit")
                target = directory / ("ChatStorage.sqlite" + suffix)
                target_fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
                try:
                    copied = 0
                    while True:
                        chunk = os.read(descriptor, min(1024 * 1024, MAX_DATABASE_BYTES - total + info.st_size - copied + 1))
                        if not chunk:
                            break
                        copied += len(chunk)
                        if total - info.st_size + copied > MAX_DATABASE_BYTES:
                            die("Database and WAL/SHM exceed combined 1073741824-byte limit")
                        view = memoryview(chunk)
                        while view:
                            written = os.write(target_fd, view)
                            if written <= 0:
                                raise OSError("snapshot write made no progress")
                            view = view[written:]
                    if source_signature(info) != source_signature(os.fstat(descriptor)):
                        raise SnapshotChanged()
                finally:
                    os.close(target_fd)
            # Check the entire set again, including absent sidecars and replacements.
            for path, descriptor, info in sources:
                try:
                    current = path.lstat()
                except FileNotFoundError:
                    current = None
                if info is None:
                    if current is not None:
                        raise SnapshotChanged()
                elif current is None or source_signature(info) != source_signature(current) or source_signature(info) != source_signature(os.fstat(descriptor)):
                    raise SnapshotChanged()
            completed = True
            return directory
        except SnapshotChanged:
            if attempt == 2:
                die("Database sources changed during snapshot; refused after 3 attempts. Close WhatsApp Desktop and retry.")
        finally:
            for _, descriptor, _ in sources:
                if descriptor is not None:
                    os.close(descriptor)
            if not completed:
                shutil.rmtree(directory)
                lifetime.directory = None
    raise AssertionError("snapshot attempts exhausted")


def connect(db_path: Path, lifetime: SnapshotLifetime | None = None) -> sqlite3.Connection:
    """Open a private snapshot read-only, preserving committed WAL-only rows."""
    require_supported_platform()
    owner = lifetime if lifetime is not None else SnapshotLifetime()
    conn = None
    try:
        if lifetime is None:
            owner.start()
        directory = copy_snapshot(db_path, owner)
        copied = directory / "ChatStorage.sqlite"
        conn = sqlite3.connect(f"{copied.as_uri()}?mode=ro", uri=True, timeout=5,
                               factory=SnapshotConnection)
        conn.lifetime = owner
        conn.snapshot_directory = directory
        conn.setlimit(sqlite3.SQLITE_LIMIT_LENGTH, MAX_VALUE_BYTES)
        conn.execute("PRAGMA temp_store=MEMORY")
        tables = {r[0] for r in query(conn, "SELECT name FROM sqlite_master WHERE type='table'")}
        missing = REQUIRED_TABLES - tables
        if missing:
            die(f"Schema drift — missing tables: {sorted(missing)}. Re-verify before trusting output.")
        return conn
    except BaseException as exc:
        try:
            if conn is not None:
                conn.close()
        finally:
            owner.close()
        if isinstance(exc, FileNotFoundError):
            die(f"Database path or parent folder not found: {db_path}")
        if isinstance(exc, OSError) and exc.errno == errno.ELOOP:
            die(f"Database source is a symlink; symlinked sources are refused: {db_path}")
        if isinstance(exc, (sqlite3.Error, OSError, ValueError)):
            die(f"Cannot open database snapshot for {db_path}: {exc}")
        raise


def die(msg: str) -> None:
    print(f"error: {msg}", file=sys.stderr)
    raise SystemExit(1)


def query(conn: sqlite3.Connection, sql: str, params: Iterable = ()) -> list[tuple]:
    deadline = time.monotonic() + MAX_QUERY_SECONDS
    def stop_long_query() -> int:
        return int(time.monotonic() >= deadline)
    conn.set_progress_handler(stop_long_query, 1000)
    try:
        cursor = conn.execute(sql, tuple(params))
        rows = cursor.fetchmany(MAX_RESULTS + 1)
        if len(rows) > MAX_RESULTS:
            die("Query exceeds 1000-row limit; narrow the selection")
        for row in rows:
            for field, value in zip(cursor.description, row):
                if isinstance(value, bytes):
                    die(f"Malformed database value: BLOB in {field[0]}")
                if isinstance(value, str) and len(value.encode("utf-8")) > MAX_VALUE_BYTES:
                    die("Database value exceeds 1048576-byte limit")
                if isinstance(value, float) and not math.isfinite(value):
                    die(f"Malformed database value: non-finite {field[0]}")
                if field[0] in {"ZISFROMME", "ZMESSAGETYPE", "Z_PK", "ZMESSAGEDATE"}:
                    if value is not None and not isinstance(value, (int, float)):
                        die(f"Malformed database numeric value: {field[0]}")
                if field[0] == "ZISFROMME" and value not in (0, 1):
                    die("Malformed database boolean: ZISFROMME")
        return rows
    except sqlite3.Error as exc:
        if "locked" in str(exc).lower() or "busy" in str(exc).lower():
            die(f"Database is locked ({exc}); wait briefly and retry")
        if "interrupted" in str(exc).lower():
            die(f"Query exceeds {MAX_QUERY_SECONDS}-second wall-clock limit; narrow the selection")
        die(f"Query failed: {exc}")
    finally:
        conn.set_progress_handler(None, 0)


def non_negative_finite_float(value: str) -> float:
    try:
        parsed = float(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("must be a finite number >= 0") from exc
    if not math.isfinite(parsed) or parsed < 0:
        raise argparse.ArgumentTypeError("must be a finite number >= 0")
    return parsed


def positive_int(value: str) -> int:
    try:
        parsed = int(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("must be an integer from 1 to 1000") from exc
    if not 1 <= parsed <= MAX_RESULTS:
        raise argparse.ArgumentTypeError("must be an integer from 1 to 1000")
    return parsed


def cutoff(days: float | None) -> float | None:
    """Convert 'N days ago' into a Core Data epoch threshold."""
    if days is None:
        return None
    return time.time() - days * 86400 - CORE_DATA_EPOCH


def to_iso(core_data_seconds: Any) -> str:
    if core_data_seconds is None:
        return ""
    try:
        value = float(core_data_seconds)
    except (TypeError, ValueError):
        die("Malformed database timestamp: expected finite epoch seconds")
    if not math.isfinite(value):
        die("Malformed database timestamp: expected finite epoch seconds")
    try:
        return (
            datetime.fromtimestamp(value + CORE_DATA_EPOCH, tz=timezone.utc)
            .astimezone()
            .isoformat(timespec="seconds")
        )
    except (OverflowError, OSError, ValueError):
        die("Malformed database timestamp: outside supported date range")


def like_pattern(term: str) -> str:
    escaped = term.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
    return f"%{escaped}%"


def as_text(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, bytes):
        die("Malformed database text: unexpected BLOB")
    if not isinstance(value, str):
        die("Malformed database text: expected TEXT or NULL")
    return value


def printable(value: str) -> str:
    text = re.sub(r"[\r\n]+", " ", as_text(value))
    return "".join(
        character
        for character in text
        if not (ord(character) < 32 or 127 <= ord(character) <= 159)
    )


def sender_name(
    from_me: bool,
    is_group: bool,
    chat_name: Any,
    member_jid: Any,
    profile_push: Any,
) -> str:
    """Best available human name for a message's sender.

    Deliberately never falls back to ZWAMESSAGE.ZPUSHNAME,
    ZWAGROUPMEMBER.ZCONTACTNAME or ZFIRSTNAME: all three hold base64 protobuf
    rather than text, so using them prints garbage instead of a name.
    """
    if from_me:
        return "ME"
    if not is_group:
        # In a 1:1 chat the sender is unambiguously the chat partner.
        return as_text(chat_name).strip() or as_text(member_jid).strip() or "?"
    return as_text(profile_push).strip() or as_text(member_jid).strip() or "?"


def type_label(t: int | None) -> str:
    if t is None:
        return "unknown"
    return MESSAGE_TYPES.get(t, f"type:{t}")


def emit(rows: list[dict[str, Any]], as_json: bool, render) -> None:
    if as_json:
        print(json.dumps(rows, ensure_ascii=False, indent=2))
    else:
        render(rows)


# ── chat resolution ─────────────────────────────────────────


def resolve_chat(conn: sqlite3.Connection, term: str) -> tuple[int, str, str]:
    """Turn a JID or a name fragment into (session_pk, jid, display_name).

    ZCONTACTJID is not unique, so when several sessions match we take the one
    holding the most messages. When several *different* chats match a name
    fragment we stop and list them rather than guessing.
    """
    rows = query(
        conn,
        """
        SELECT cs.Z_PK, cs.ZCONTACTJID, COALESCE(cs.ZPARTNERNAME,''),
               (SELECT COUNT(*) FROM ZWAMESSAGE m WHERE m.ZCHATSESSION = cs.Z_PK)
          FROM ZWACHATSESSION cs
         WHERE cs.ZCONTACTJID = ?
         ORDER BY 4 DESC, cs.Z_PK ASC
        """,
        (term,),
    )
    if not rows:
        rows = query(
            conn,
            """
            SELECT cs.Z_PK, cs.ZCONTACTJID, COALESCE(cs.ZPARTNERNAME,''),
                   (SELECT COUNT(*) FROM ZWAMESSAGE m WHERE m.ZCHATSESSION = cs.Z_PK)
              FROM ZWACHATSESSION cs
             WHERE cs.ZPARTNERNAME LIKE ? ESCAPE '\\'
             ORDER BY 4 DESC, cs.Z_PK ASC
            """,
            (like_pattern(term),),
        )
    if not rows:
        die(
            f"No chat matches {term!r}. "
            "Run `whatsapp_read.py chats --search <name>` to find it."
        )

    rows = [(r[0], as_text(r[1]), as_text(r[2]), r[3]) for r in rows]
    distinct_jids = {r[1] for r in rows}
    if len(distinct_jids) > 1:
        lines = "\n".join(
            f"  {printable(r[1]):<42} {printable(r[2])[:34]:<34} "
            f"{r[3]:>6} msgs"
            for r in rows[:12]
        )
        die(
            f"{term!r} matches {len(distinct_jids)} chats — pass an exact JID:\n{lines}"
        )
    return rows[0][0], rows[0][1], rows[0][2]


# ── commands ────────────────────────────────────────────────


def cmd_chats(conn: sqlite3.Connection, args) -> None:
    where, params = ["1=1"], []
    if args.search:
        where.append("COALESCE(cs.ZPARTNERNAME,'') LIKE ? ESCAPE '\\'")
        params.append(like_pattern(args.search))
    since = cutoff(args.days)
    if since is not None:
        where.append("m.ZMESSAGEDATE > ?")
        params.append(since)
    if args.groups_only:
        where.append("cs.ZCONTACTJID LIKE '%@g.us'")
    if args.dms_only:
        where.append("cs.ZCONTACTJID NOT LIKE '%@g.us'")
    if not args.include_status:
        # ZSESSIONTYPE 2 is the broadcast container and 3 is the status bucket;
        # both are real rows with real messages but are not conversations.
        where.append("COALESCE(cs.ZSESSIONTYPE, 0) NOT IN (2, 3)")
        where.append("cs.ZCONTACTJID NOT LIKE '%.status'")

    rows = query(
        conn,
        f"""
        SELECT cs.ZCONTACTJID,
               COALESCE(MAX(NULLIF(cs.ZPARTNERNAME,'')),''),
               COUNT(m.Z_PK),
               MAX(m.ZMESSAGEDATE)
          FROM ZWACHATSESSION cs
          JOIN ZWAMESSAGE m ON m.ZCHATSESSION = cs.Z_PK
         WHERE {" AND ".join(where)}
         GROUP BY cs.ZCONTACTJID
         ORDER BY MAX(m.ZMESSAGEDATE) DESC, cs.ZCONTACTJID
         LIMIT ?
        """,
        (*params, args.limit),
    )
    out = [
        {
            "jid": as_text(r[0]),
            "name": as_text(r[1]),
            "is_group": as_text(r[0]).endswith("@g.us"),
            "messages": r[2],
            "last_message_at": to_iso(r[3]),
        }
        for r in rows
    ]

    def render(items):
        if not items:
            print("(no chats matched)")
            return
        print(f"{'LAST':<25} {'MSGS':>7}  {'NAME':<32} JID")
        print("-" * 116)
        for it in items:
            kind = "G" if it["is_group"] else " "
            print(
                f"{printable(it['last_message_at']):<25} {it['messages']:>7}  "
                f"{kind} {printable(it['name'])[:30]:<30} {printable(it['jid'])}"
            )
        print(f"\n{len(items)} chats")

    emit(out, args.json, render)


def cmd_read(conn: sqlite3.Connection, args) -> None:
    pk, jid, name = resolve_chat(conn, args.chat)
    is_group = jid.endswith("@g.us")

    where, params = ["m.ZCHATSESSION = ?"], [pk]
    since = cutoff(args.days)
    if since is not None:
        where.append("m.ZMESSAGEDATE > ?")
        params.append(since)
    if args.text_only:
        conn.create_function("has_text_body", 1, lambda value: bool(as_text(value).strip()))
        where.append("has_text_body(m.ZTEXT)")

    rows = query(
        conn,
        f"""
        SELECT m.ZMESSAGEDATE, m.ZISFROMME, m.ZMESSAGETYPE, m.ZTEXT,
               gm.ZMEMBERJID,
               (SELECT pn.ZPUSHNAME
                  FROM ZWAPROFILEPUSHNAME pn
                 WHERE pn.ZJID = gm.ZMEMBERJID
                 LIMIT 1),
               (SELECT mi.ZTITLE
                  FROM ZWAMEDIAITEM mi
                 WHERE mi.ZMESSAGE = m.Z_PK
                 LIMIT 1),
               (SELECT mi.ZMEDIALOCALPATH
                  FROM ZWAMEDIAITEM mi
                 WHERE mi.ZMESSAGE = m.Z_PK
                 LIMIT 1)
          FROM ZWAMESSAGE m
          LEFT JOIN ZWAGROUPMEMBER gm ON m.ZGROUPMEMBER = gm.Z_PK
         WHERE {" AND ".join(where)}
         ORDER BY m.ZMESSAGEDATE DESC, m.Z_PK DESC
         LIMIT ?
        """,
        (*params, args.limit),
    )
    rows.reverse()  # chronological for reading

    out = []
    for (
        zdate,
        from_me,
        mtype,
        text,
        member_jid,
        profile_push,
        title,
        path,
    ) in rows:
        sender = sender_name(bool(from_me), is_group, name, member_jid, profile_push)
        body = as_text(text).strip()
        caption = as_text(title).strip()
        if not body:
            media_path = as_text(path)
            extra = caption or (Path(media_path).name if media_path else "")
            body = f"[{type_label(mtype)}{': ' + extra if extra else ''}]"
        elif caption and caption != body:
            # Documents put the FILENAME in ZTEXT and the human caption in
            # ZTITLE; link previews put the URL in ZTEXT and the page title in
            # ZTITLE. Rendering ZTEXT alone silently drops written context.
            body = f"{body} — {caption}"
        out.append(
            {
                "at": to_iso(zdate),
                "sender": sender,
                "from_me": bool(from_me),
                "type": type_label(mtype),
                "text": body,
            }
        )

    def render(items):
        header = (
            f"{printable(name or jid)}  <{printable(jid)}>  —  {len(items)} messages"
        )
        print(header)
        print("=" * min(len(header), 100))
        for it in items:
            print(
                f"[{printable(it['at'])}] {printable(it['sender'])}: "
                f"{printable(it['text'])}"
            )

    emit(out, args.json, render)


def cmd_search(conn: sqlite3.Connection, args) -> None:
    where = ["m.ZTEXT LIKE ? ESCAPE '\\'"]
    params: list[Any] = [like_pattern(args.term)]
    since = cutoff(args.days)
    if since is not None:
        where.append("m.ZMESSAGEDATE > ?")
        params.append(since)
    if args.chat:
        pk, _, _ = resolve_chat(conn, args.chat)
        where.append("m.ZCHATSESSION = ?")
        params.append(pk)

    rows = query(
        conn,
        f"""
        SELECT m.ZMESSAGEDATE, m.ZISFROMME, cs.ZCONTACTJID,
               COALESCE(cs.ZPARTNERNAME,''), m.ZTEXT,
               gm.ZMEMBERJID,
               (SELECT pn.ZPUSHNAME
                  FROM ZWAPROFILEPUSHNAME pn
                 WHERE pn.ZJID = gm.ZMEMBERJID
                 LIMIT 1)
          FROM ZWAMESSAGE m
          JOIN ZWACHATSESSION cs ON m.ZCHATSESSION = cs.Z_PK
          LEFT JOIN ZWAGROUPMEMBER gm ON m.ZGROUPMEMBER = gm.Z_PK
         WHERE {" AND ".join(where)}
         ORDER BY m.ZMESSAGEDATE DESC, m.Z_PK DESC
         LIMIT ?
        """,
        (*params, args.limit),
    )
    out = [
        {
            "at": to_iso(r[0]),
            "chat": as_text(r[3]) or as_text(r[2]),
            "chat_jid": as_text(r[2]),
            "sender": sender_name(
                bool(r[1]),
                as_text(r[2]).endswith("@g.us"),
                r[3],
                r[5],
                r[6],
            ),
            "text": as_text(r[4]).strip(),
        }
        for r in rows
    ]

    def render(items):
        if not items:
            print("(no matches)")
            return
        for it in items:
            print(
                f"[{printable(it['at'])}] {printable(it['chat'])[:28]} / "
                f"{printable(it['sender'])}: {printable(it['text'])[:180]}"
            )
        print(f"\n{len(items)} matches")

    emit(out, args.json, render)


def cmd_contacts(conn: sqlite3.Connection, args) -> None:
    where, params = ["1=1"], []
    if args.search:
        where.append(
            "(ZPUSHNAME LIKE ? ESCAPE '\\' OR ZJID LIKE ? ESCAPE '\\')"
        )
        pattern = like_pattern(args.search)
        params += [pattern, pattern]
    rows = query(
        conn,
        f"""
        SELECT ZJID, ZPUSHNAME FROM ZWAPROFILEPUSHNAME
         WHERE {" AND ".join(where)}
         ORDER BY ZPUSHNAME
         LIMIT ?
        """,
        (*params, args.limit),
    )
    out = [{"jid": as_text(r[0]), "name": as_text(r[1])} for r in rows]

    def render(items):
        for it in items:
            print(f"{printable(it['name'])[:38]:<38} {printable(it['jid'])}")
        print(f"\n{len(items)} contacts")

    emit(out, args.json, render)


# ── entrypoint ──────────────────────────────────────────────


def main() -> None:
    p = argparse.ArgumentParser(
        prog="whatsapp_read.py",
        description="Read-only access to WhatsApp Desktop messages (macOS).",
        epilog=(
            "examples:\n"
            "  %(prog)s chats --days 30\n"
            "  %(prog)s chats --search alice\n"
            '  %(prog)s read "Bob" --limit 40\n'
            "  %(prog)s read 120000000000000001@g.us --days 7\n"
            '  %(prog)s search "forecast" --days 60\n'
            "  %(prog)s contacts --search example-co\n"
            "By Zero Width Lab (zerowidthlab.com)\n"
        ),
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    p.add_argument(
        "--db",
        type=Path,
        default=DEFAULT_DB,
        help="path to ChatStorage.sqlite (overrides WHATSAPP_DB)",
    )
    p.add_argument("--json", action="store_true", help="emit JSON instead of text")

    # Repeated on every subcommand so `read X --json` works as readily as
    # `--json read X`. Without this argparse only accepts the flag before the
    # subcommand, which is a silent footgun for callers.
    # default=SUPPRESS matters: without it the subparser would write its own
    # default back over a value already given before the subcommand.
    common = argparse.ArgumentParser(add_help=False)
    common.add_argument(
        "--db", type=Path, default=argparse.SUPPRESS, help=argparse.SUPPRESS
    )
    common.add_argument(
        "--json",
        action="store_true",
        default=argparse.SUPPRESS,
        help=argparse.SUPPRESS,
    )

    sub = p.add_subparsers(dest="cmd", required=True)

    c = sub.add_parser(
        "chats", parents=[common], help="list chats, most recently active first"
    )
    c.add_argument("--search", help="filter by chat name substring")
    c.add_argument(
        "--days",
        type=non_negative_finite_float,
        help="only chats active in the last N days",
    )
    c.add_argument("--limit", type=positive_int, default=40)
    chat_kind = c.add_mutually_exclusive_group()
    chat_kind.add_argument("--groups-only", action="store_true")
    chat_kind.add_argument("--dms-only", action="store_true")
    c.add_argument(
        "--include-status",
        action="store_true",
        help="also list status/broadcast rows (hidden by default)",
    )
    c.set_defaults(func=cmd_chats)

    r = sub.add_parser("read", parents=[common], help="read messages from one chat")
    r.add_argument("chat", help="JID, or a substring of the chat name")
    r.add_argument(
        "--limit", type=positive_int, default=50, help="most recent N (default 50)"
    )
    r.add_argument("--days", type=non_negative_finite_float, help="only the last N days")
    r.add_argument(
        "--text-only",
        action="store_true",
        help="only rows with a non-empty text body",
    )
    r.set_defaults(func=cmd_read)

    s = sub.add_parser(
        "search", parents=[common], help="substring search across message bodies"
    )
    s.add_argument("term")
    s.add_argument("--chat", help="restrict to one chat (JID or name)")
    s.add_argument("--days", type=non_negative_finite_float, help="only the last N days")
    s.add_argument("--limit", type=positive_int, default=50)
    s.set_defaults(func=cmd_search)

    k = sub.add_parser(
        "contacts",
        parents=[common],
        help="JID -> push-name directory (not your address book)",
    )
    k.add_argument("--search", help="filter by name or JID substring")
    k.add_argument("--limit", type=positive_int, default=100)
    k.set_defaults(func=cmd_contacts)

    args = p.parse_args()
    require_supported_platform()
    lifetime = SnapshotLifetime()
    conn = None
    try:
        lifetime.start()
        conn = connect(args.db, lifetime)
        args.func(conn, args)
    finally:
        try:
            if conn is not None:
                conn.close()
        finally:
            lifetime.close()


if __name__ == "__main__":
    main()

License: MIT (read it).

Back to the listing