#!/usr/bin/env python3
"""
Local WhatsApp reader (read-only)
=================================
Reads the local WhatsApp message store kept by the native macOS WhatsApp app
(``~/Library/Group Containers/group.net.whatsapp.WhatsApp.shared/ChatStorage.sqlite``)
so the vault agent can answer "what did X send me on WhatsApp?" — without ever
writing to, or sending, anything. The `whatsapp` skill drives this; a person rarely
runs it by hand.

Read-only by construction: the store is snapshot-copied (db + -wal + -shm) to a temp
dir and the *copy* is queried, so the live database WhatsApp holds open is never
touched. It NEVER writes to the store, NEVER writes to the vault, and CANNOT send
messages — sending is a separate, deliberately-unbuilt capability.

Runs on the machine where the WhatsApp Mac app is signed in — the owner's laptop, NOT
an always-on automation host. Requires the app running the assistant (Claude, or the
terminal it runs in) to have macOS Full Disk Access. The old Electron "WhatsApp
Desktop" kept its data in an unreadable IndexedDB — this reader needs the native
(App Store) app, which keeps the same SQLite schema as WhatsApp for iPhone.

Optional config — read from Agent/config.json under the `whatsapp` key, all optional:
    whatsapp.default_since     default time window                  (default "24h")
    whatsapp.include_groups    read group chats too                 (default true)
    whatsapp.contacts_file     path to a handle->name map           (default Agent/scripts/whatsapp/contacts.json)
    whatsapp.db                path to ChatStorage.sqlite           (default the native app's store)

Usage:
    python -m alexandria_vault.engine.whatsapp_read --vault .                  # last 24h, all chats
    python -m alexandria_vault.engine.whatsapp_read --vault . --since 7d       # 24h | 7d | 2w | today | yesterday
    python -m alexandria_vault.engine.whatsapp_read --vault . --from alex      # a person (name or number)
    python -m alexandria_vault.engine.whatsapp_read --vault . --thread "Family" # one chat by name / JID
    python -m alexandria_vault.engine.whatsapp_read --vault . --list-threads   # recent chats + last activity
    python -m alexandria_vault.engine.whatsapp_read --vault . --direct-only    # no group chats
    python -m alexandria_vault.engine.whatsapp_read --vault . --limit 200 --json

The window is the bound, not a row count: `--limit` defaults to 0 (no cap). Every run
prints a one-line coverage summary to stderr (window start · rows · chats · oldest
message) and a loud TRUNCATED warning if a cap swallowed older messages, so `--json`
stdout stays clean. Names come from the store itself (the WhatsApp partner name from
the owner's address book, group-member names, push names) and then from the handle
maps — the shared one the contacts module writes and this module's own contacts.json.

Read-only. Never writes, never sends. Python 3.9+, standard library only.
"""
import argparse
import datetime as dt
import json
import re
import shutil
import sqlite3
import sys
import tempfile
from pathlib import Path

from alexandria_vault.engine import consume_vault_arg, resolve_vault
from alexandria_vault.engine.handles import (COCOA_EPOCH, SINCE_FORMS, shared_handles_path,
                                             like_pattern, load_handle_map, norm_handle,
                                             parse_since)
from alexandria_vault.vault_core import load_config as _vc_load_config

# `--vault` must land in VAULT_DIR before the module-level config load below.
consume_vault_arg()

VAULT = resolve_vault()
DEFAULT_DB = (Path.home() / "Library" / "Group Containers" / "group.net.whatsapp.WhatsApp.shared"
              / "ChatStorage.sqlite")

# ZWACHATSESSION.ZSESSIONTYPE
SESSION_DIRECT, SESSION_GROUP, SESSION_BROADCAST, SESSION_STATUS = 0, 1, 2, 3

# ZWAMESSAGE.ZMESSAGETYPE — the iPhone/Mac schema's well-known values. Anything else
# is labelled by number rather than guessed at.
MESSAGE_TYPES = {
    0: "text", 1: "image", 2: "video", 3: "voice note", 4: "contact card", 5: "location",
    6: "group event", 7: "link", 8: "document", 10: "system notice", 11: "gif",
    14: "deleted message", 15: "sticker",
    # 59 = the in-chat call entry (the "Voice call" / "Missed voice call" row). Confirmed
    # 2026-09-03 against the app's own CallHistory.sqlite: every incoming call, missed or
    # answered, has a type-59 row within ±2 min, and 105/112 type-59 rows in the call-history
    # window sit next to a logged call. No text, no media; a 1:1-chat row in both directions.
    59: "call",
}
# Types that carry NO text and NO media by nature — an event row, not a message — and so
# render as a bare label rather than "[unsupported message type N]". (59 graduated to
# MESSAGE_TYPES once identified; the seam stays for the next one.)
EVENT_TYPES = {}


# ── config (Core-reads-Config seam; personalization never hardcoded here) ─────
def _load_config():
    """Optional `whatsapp` tuning from the vault's Config. A missing file or key is
    fine — built-in defaults apply. No identity is read here."""
    cfg = _vc_load_config(VAULT)
    wa = cfg.get("whatsapp", {}) if isinstance(cfg, dict) else {}
    return wa if isinstance(wa, dict) else {}


WA = _load_config()
DEFAULT_SINCE = WA.get("default_since", "24h")
INCLUDE_GROUPS = bool(WA.get("include_groups", True))
DB_PATH = Path(WA["db"]).expanduser() if WA.get("db") else DEFAULT_DB
CONTACTS = (Path(WA["contacts_file"]).expanduser() if WA.get("contacts_file")
            else VAULT / "Agent" / "scripts" / "whatsapp" / "contacts.json")
SHARED_HANDLES = shared_handles_path(VAULT, _vc_load_config(VAULT))


def load_contacts():
    """handle→name, normalised: the shared map from macOS Contacts first, then this
    module's own contacts.json on top (a hand-written label beats an imported one)."""
    return load_handle_map(SHARED_HANDLES, CONTACTS)


# ── JIDs, time ───────────────────────────────────────────────────────────────
def jid_phone(jid):
    """'4915551234567@s.whatsapp.net' -> '+4915551234567'; a group / LID / broadcast
    JID has no phone in it -> ''."""
    if not jid or "@" not in jid:
        return ""
    user, domain = jid.split("@", 1)
    if domain != "s.whatsapp.net" or not user.isdigit():
        return ""
    return "+" + user


def is_group_jid(jid):
    return bool(jid) and jid.endswith("@g.us")


def cocoa_to_dt(value):
    """ZMESSAGEDATE is seconds since 2001-01-01 (a REAL)."""
    if value is None:
        return None
    try:
        return dt.datetime.fromtimestamp(float(value) + COCOA_EPOCH)
    except (ValueError, OverflowError, OSError):
        return None


def dt_to_cocoa(when):
    return when.timestamp() - COCOA_EPOCH


# ── database ─────────────────────────────────────────────────────────────────
FDA_MESSAGE = (
    "ChatStorage.sqlite: access not permitted (macOS Full Disk Access).\n"
    "Grant Full Disk Access to the app running your assistant — Claude, or the terminal it runs in:\n"
    "  System Settings -> Privacy & Security -> Full Disk Access -> enable it -> re-open the app.\n"
    "Until then every read of WhatsApp's store under ~/Library/Group Containers/ is blocked by macOS TCC."
)


def open_db(db_path=None):
    """Snapshot the store (+wal/shm) to a temp dir and open the copy read-only. WhatsApp
    keeps the live file open in WAL mode, so the copy is the only safe thing to read."""
    db_path = Path(db_path or DB_PATH)
    if not db_path.exists():
        raise SystemExit(
            "no WhatsApp store at %s — is the native WhatsApp Mac app installed and signed in on this "
            "machine? (The old Electron 'WhatsApp Desktop' keeps an unreadable IndexedDB; install the App "
            "Store app.)" % db_path)
    tmp = Path(tempfile.mkdtemp(prefix="whatsapp-"))
    try:
        for suffix in ("", "-wal", "-shm"):
            src = Path(str(db_path) + suffix)
            if src.exists():
                shutil.copy2(src, tmp / src.name)
    except (PermissionError, OSError):
        shutil.rmtree(tmp, ignore_errors=True)
        raise SystemExit(FDA_MESSAGE)
    copy = tmp / db_path.name
    con = None
    try:
        con = sqlite3.connect("file:%s?mode=ro" % copy, uri=True)
        con.execute("SELECT 1 FROM ZWAMESSAGE LIMIT 1")
    except sqlite3.DatabaseError:
        if con is not None:
            con.close()
        try:
            con = sqlite3.connect(str(copy))          # ro open can fail on WAL recovery; the copy is ours
            con.execute("SELECT 1 FROM ZWAMESSAGE LIMIT 1")
        except sqlite3.DatabaseError as e:
            if con is not None:
                con.close()
            shutil.rmtree(tmp, ignore_errors=True)
            raise SystemExit("%s is not a readable WhatsApp store (%s)" % (db_path, e))
    con.row_factory = sqlite3.Row
    return con, tmp


# Columns the reader uses that a WhatsApp build might lack (the Core Data schema drifts
# by version). Each is selected as NULL when absent so a drift degrades a label, never
# the whole read. The join keys and the core message/session columns are required.
OPTIONAL_COLUMNS = {
    "ZWAMESSAGE": ("ZPUSHNAME", "ZGROUPMEMBER", "ZMEDIAITEM", "ZFROMJID", "ZMESSAGETYPE"),
    "ZWACHATSESSION": ("ZPARTNERNAME", "ZSESSIONTYPE"),
    "ZWAGROUPMEMBER": ("ZCONTACTNAME", "ZMEMBERJID"),
    "ZWAMEDIAITEM": ("ZTITLE", "ZVCARDNAME"),
}


def _present(con):
    """{table: set(columns)} for the tables the reader touches."""
    out = {}
    for table in OPTIONAL_COLUMNS:
        try:
            out[table] = {r[1] for r in con.execute("PRAGMA table_info(%s)" % table)}
        except sqlite3.DatabaseError:
            out[table] = set()
    return out


def col(present, alias, table, column, as_name):
    """`alias.column AS as_name`, or `NULL AS as_name` when the schema lacks it."""
    if column in present.get(table, ()) or column not in OPTIONAL_COLUMNS.get(table, ()):
        return "%s.%s AS %s" % (alias, column, as_name)
    return "NULL AS %s" % as_name


def session_filter(include_groups, include_all):
    if include_all:
        return None
    return (SESSION_DIRECT, SESSION_GROUP) if include_groups else (SESSION_DIRECT,)


def fetch_messages(con, since=None, limit=0, over=1, thread=None, sessions=(SESSION_DIRECT, SESSION_GROUP)):
    """Rows newest-first. limit=0 means no cap — the --since window is the bound.
    Returns (rows, sql_cap) so the caller can tell a full window from a truncated one."""
    pr = _present(con)
    where, params = [], []
    if since:
        where.append("m.ZMESSAGEDATE >= ?"); params.append(dt_to_cocoa(since))
    if sessions and "ZSESSIONTYPE" in pr["ZWACHATSESSION"]:
        where.append("s.ZSESSIONTYPE IN (%s)" % ",".join("?" * len(sessions))); params += list(sessions)
    if thread:
        where.append("(LOWER(COALESCE(s.ZPARTNERNAME,'')) LIKE ? ESCAPE '\\'"
                     " OR LOWER(COALESCE(s.ZCONTACTJID,'')) LIKE ? ESCAPE '\\')")
        params += [like_pattern(thread)] * 2
    have_gm = "ZGROUPMEMBER" in pr["ZWAMESSAGE"] and pr["ZWAGROUPMEMBER"]
    have_mi = "ZMEDIAITEM" in pr["ZWAMESSAGE"] and pr["ZWAMEDIAITEM"]
    sql = """
        SELECT m.Z_PK AS rid, m.ZMESSAGEDATE AS date, m.ZISFROMME AS is_from_me,
               %s, m.ZTEXT AS text, %s, %s,
               s.Z_PK AS chat_id, s.ZCONTACTJID AS chat_jid, %s, %s,
               %s, %s, %s, %s
        FROM ZWAMESSAGE m
        JOIN ZWACHATSESSION s ON s.Z_PK = m.ZCHATSESSION
        %s
        %s
    """ % (col(pr, "m", "ZWAMESSAGE", "ZMESSAGETYPE", "mtype"),
           col(pr, "m", "ZWAMESSAGE", "ZFROMJID", "from_jid"),
           col(pr, "m", "ZWAMESSAGE", "ZPUSHNAME", "push_name"),
           col(pr, "s", "ZWACHATSESSION", "ZPARTNERNAME", "chat_name"),
           col(pr, "s", "ZWACHATSESSION", "ZSESSIONTYPE", "session_type"),
           col(pr, "gm", "ZWAGROUPMEMBER", "ZCONTACTNAME", "member_name") if have_gm else "NULL AS member_name",
           col(pr, "gm", "ZWAGROUPMEMBER", "ZMEMBERJID", "member_jid") if have_gm else "NULL AS member_jid",
           col(pr, "mi", "ZWAMEDIAITEM", "ZTITLE", "media_title") if have_mi else "NULL AS media_title",
           col(pr, "mi", "ZWAMEDIAITEM", "ZVCARDNAME", "media_vcard") if have_mi else "NULL AS media_vcard",
           "LEFT JOIN ZWAGROUPMEMBER gm ON gm.Z_PK = m.ZGROUPMEMBER" if have_gm else "",
           "LEFT JOIN ZWAMEDIAITEM mi ON mi.Z_PK = m.ZMEDIAITEM" if have_mi else "")
    if where:
        sql += " WHERE " + " AND ".join(where)
    sql += " ORDER BY m.ZMESSAGEDATE DESC"
    sql_cap = max(limit * over, limit) if limit else 0
    if sql_cap:
        sql += " LIMIT ?"
        params.append(sql_cap)
    return con.execute(sql, params).fetchall(), sql_cap


def list_threads(con, limit=30, sessions=(SESSION_DIRECT, SESSION_GROUP)):
    pr = _present(con)
    where, params = [], []
    if sessions and "ZSESSIONTYPE" in pr["ZWACHATSESSION"]:
        where.append("s.ZSESSIONTYPE IN (%s)" % ",".join("?" * len(sessions))); params += list(sessions)
    sql = """
        SELECT s.Z_PK AS chat_id, s.ZCONTACTJID AS chat_jid, %s, %s,
               MAX(m.ZMESSAGEDATE) AS last_date, COUNT(m.Z_PK) AS n
        FROM ZWACHATSESSION s
        JOIN ZWAMESSAGE m ON m.ZCHATSESSION = s.Z_PK
    """ % (col(pr, "s", "ZWACHATSESSION", "ZPARTNERNAME", "chat_name"),
           col(pr, "s", "ZWACHATSESSION", "ZSESSIONTYPE", "session_type"))
    if where:
        sql += " WHERE " + " AND ".join(where)
    sql += " GROUP BY s.Z_PK ORDER BY last_date DESC LIMIT ?"
    params.append(limit)
    return con.execute(sql, params).fetchall()


def push_names(con):
    """JID -> profile push name (what the other party calls themselves)."""
    try:
        return {r[0]: r[1] for r in con.execute("SELECT ZJID, ZPUSHNAME FROM ZWAPROFILEPUSHNAME")
                if r[0] and r[1]}
    except sqlite3.DatabaseError:
        return {}


# ── names & text ─────────────────────────────────────────────────────────────
def thread_name(row, contacts):
    if row["chat_name"]:
        return row["chat_name"]
    phone = jid_phone(row["chat_jid"])
    if phone and contacts.get(norm_handle(phone)):
        return contacts[norm_handle(phone)]
    return phone or row["chat_jid"] or "(unknown chat)"


def sender_name(row, contacts, pushes):
    """Who sent an inbound message: the store's own names first (partner name for a
    direct chat, the group-member name in a group), then the handle maps, then the
    push name, then the bare number."""
    if row["is_from_me"]:
        return "Me"
    if row["session_type"] == SESSION_GROUP:
        if row["member_name"]:
            return row["member_name"]
        jid = row["member_jid"] or (row["from_jid"] if not is_group_jid(row["from_jid"]) else "")
    else:
        if row["chat_name"]:
            return row["chat_name"]
        jid = row["chat_jid"] or row["from_jid"]
    phone = jid_phone(jid)
    if phone and contacts.get(norm_handle(phone)):
        return contacts[norm_handle(phone)]
    push = row["push_name"] or pushes.get(jid or "")
    if push:
        return push + ((" (%s)" % phone) if phone else "")
    if phone:
        return phone
    if jid and jid.endswith("@lid"):
        return "(unnamed, WhatsApp id %s)" % jid.split("@")[0]     # a LID, not a dialable number
    return (jid.split("@")[0] if jid else "") or "(unknown)"


def message_text(row):
    kind = MESSAGE_TYPES.get(row["mtype"])
    text = (row["text"] or "").strip()
    if row["mtype"] in (0, 7):
        return text or "[no text]"
    if kind is None:
        if row["mtype"] in EVENT_TYPES and not text:
            return "[%s (type %s)]" % (EVENT_TYPES[row["mtype"]], row["mtype"])
        return ("[type %s] %s" % (row["mtype"], text)) if text else "[unsupported message type %s]" % row["mtype"]
    detail = text or (row["media_title"] or "").strip() or (row["media_vcard"] or "").strip()
    return "[%s%s]" % (kind, (": " + detail) if detail else "")


def thread_key(row):
    return row["chat_jid"] or row["chat_name"] or "(unknown)"


def matches(row, query, contacts, pushes):
    q = query.lower()
    sender = "" if row["is_from_me"] else sender_name(row, contacts, pushes)   # "Me" is not a person to filter by
    hay = " ".join([sender, thread_name(row, contacts),
                    row["chat_jid"] or "", row["member_jid"] or "", row["from_jid"] or ""]).lower()
    if q in hay:
        return True
    nq = norm_handle(query)
    return bool(nq) and any(nq in norm_handle(jid_phone(j)) for j in (row["chat_jid"], row["member_jid"], row["from_jid"]) if j)


# ── rendering ────────────────────────────────────────────────────────────────
def render_text(rows, contacts, pushes):
    threads = {}
    for r in rows:
        threads.setdefault(thread_key(r), []).append(r)
    out = []
    for key, items in threads.items():
        first = items[0]
        label = thread_name(first, contacts) + (" [group]" if first["session_type"] == SESSION_GROUP else "")
        out.append("\n=== %s ===" % label)
        for r in sorted(items, key=lambda x: x["date"] or 0):
            when = cocoa_to_dt(r["date"])
            stamp = when.strftime("%Y-%m-%d %H:%M") if when else "??"
            out.append("[%s] %s: %s" % (stamp, sender_name(r, contacts, pushes), message_text(r)))
    return "\n".join(out).strip() or "(no messages in this window)"


def render_json(rows, contacts, pushes):
    items = []
    for r in rows:
        when = cocoa_to_dt(r["date"])
        items.append({
            "thread": thread_name(r, contacts),
            "thread_id": r["chat_jid"],
            "group": r["session_type"] == SESSION_GROUP,
            "when": when.isoformat(timespec="seconds") if when else None,
            "from": sender_name(r, contacts, pushes),
            "is_from_me": bool(r["is_from_me"]),
            "type": MESSAGE_TYPES.get(r["mtype"], "type %s" % r["mtype"]),
            "text": message_text(r),
        })
    return json.dumps(items, ensure_ascii=False, indent=2)


# ── coverage reporting (stderr, so --json stdout stays parseable) ────────────
def warn(msg):
    print("whatsapp_read: %s" % msg, file=sys.stderr)


def report_coverage(rows, since, limit, truncated):
    stamp = lambda w: w.isoformat(timespec="seconds")
    oldest = min((cocoa_to_dt(r["date"]) for r in rows if cocoa_to_dt(r["date"])), default=None)
    warn("window %s → %d msgs / %d chats · oldest %s" % (
        stamp(since) if since else "(all history)",
        len(rows), len({thread_key(r) for r in rows}),
        stamp(oldest) if oldest else "—"))
    if truncated:
        warn("TRUNCATED — hit --limit %d, so older messages in this window were dropped "
             "and whole quiet chats may be missing. Re-run with --limit 0 (no cap) to "
             "cover the window." % limit)


def _non_negative(value):
    n = int(value)
    if n < 0:
        raise argparse.ArgumentTypeError("--limit must be 0 (no cap) or a positive count")
    return n


def main():
    ap = argparse.ArgumentParser(description="Read-only local WhatsApp reader (native Mac app store).")
    ap.add_argument("--since", default=None,
                    help="%s (default %s; a --thread read covers the whole chat unless --since is given)"
                         % (SINCE_FORMS, DEFAULT_SINCE))
    ap.add_argument("--from", dest="who", help="filter by sender/chat name or number (substring)")
    ap.add_argument("--thread", help="one chat by partner/group name or JID")
    ap.add_argument("--direct-only", action="store_true", help="1:1 chats only (no groups)")
    ap.add_argument("--include-all", action="store_true",
                    help="every session type, including broadcasts and status updates")
    ap.add_argument("--limit", type=_non_negative, default=0,
                    help="max messages, newest first; 0 = no cap (default) — the --since window is "
                         "the bound. Truncation is warned about on stderr. (--list-threads shows 30 unless set.)")
    ap.add_argument("--list-threads", action="store_true", help="recent chats + last activity")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--db", default=None, help="path to ChatStorage.sqlite (default the native app's store)")
    args = ap.parse_args()

    contacts = load_contacts()
    # Validate args before touching the DB. A --thread read spans the whole chat unless the
    # caller narrows it explicitly; the default window applies to the cross-chat read only.
    if args.list_threads:
        since = None
    elif args.since is not None:
        since = parse_since(args.since)
    else:
        since = None if args.thread else parse_since(DEFAULT_SINCE)
    sessions = session_filter(INCLUDE_GROUPS and not args.direct_only, args.include_all)
    con, tmp = open_db(args.db)
    try:
        pushes = push_names(con)
        if args.list_threads:
            rows = list_threads(con, limit=args.limit or 30, sessions=sessions)
            if args.json:
                print(json.dumps([{
                    "thread": thread_name(r, contacts), "thread_id": r["chat_jid"],
                    "group": r["session_type"] == SESSION_GROUP,
                    "last": cocoa_to_dt(r["last_date"]).isoformat(timespec="seconds") if cocoa_to_dt(r["last_date"]) else None,
                    "messages": r["n"],
                } for r in rows], ensure_ascii=False, indent=2))
            else:
                for r in rows:
                    when = cocoa_to_dt(r["last_date"])
                    stamp = when.strftime("%Y-%m-%d %H:%M") if when else "??"
                    print("%s  %-28s%s  (%d msgs)" % (stamp, thread_name(r, contacts),
                                                       " [group]" if r["session_type"] == SESSION_GROUP else "",
                                                       r["n"]))
            return

        over = 8 if args.who else 1          # --from filters in Python; over-read to leave room
        rows, sql_cap = fetch_messages(con, since=since, limit=args.limit, over=over,
                                       thread=args.thread, sessions=sessions)
        truncated = bool(sql_cap) and len(rows) >= sql_cap
        if args.who:
            rows = [r for r in rows if matches(r, args.who, contacts, pushes)]
        if args.limit:
            truncated = truncated or len(rows) > args.limit
            rows = rows[:args.limit]
        print(render_json(rows, contacts, pushes) if args.json else render_text(rows, contacts, pushes))
        report_coverage(rows, since, args.limit, truncated)
    finally:
        con.close()
        shutil.rmtree(tmp, ignore_errors=True)


if __name__ == "__main__":
    main()
