#!/usr/bin/env python3
"""
Local iMessage reader (read-only)
=================================
Reads the local macOS Messages database (``~/Library/Messages/chat.db``) so the
vault agent can answer "what did X text me?" — without ever writing to, or
sending, anything. The `imessage` skill drives this; a person rarely runs it by
hand.

Read-only by construction: opens the DB in SQLite read-only URI mode (falling
back to a temp *copy* only if Messages holds a write lock). It NEVER writes to
chat.db, NEVER writes to the vault, and CANNOT send messages — sending is a
separate, deliberately-unbuilt capability (see Agent/docs/iMessage.md).

Runs on the machine where Messages is signed in — i.e. the owner's laptop, NOT an
always-on automation host (the opposite of the digest/sweep engine). Requires the
app running the assistant (Claude, or the terminal it runs in) to have macOS Full
Disk Access. Without it every query returns "authorization denied".

Optional config — read from Agent/config.json under the `imessage` key, all
optional (built-in defaults apply when absent):
    imessage.default_since     default time window          (default "24h")
    imessage.default_service   all | imessage | sms         (default "all")
    imessage.contacts_file     path to the handle->name map
                               (default <vault>/Agent/scripts/imessage/contacts.json;
                               the shared Agent/scripts/contacts/handles.json from the
                               contacts module is always read underneath it)

Usage:
    python3 imessage_read.py                        # last 24h, all conversations
    python3 imessage_read.py --since 7d             # 24h | 7d | 2w | today | yesterday
    python3 imessage_read.py --since 2026-07-01     # since a date — whole window, uncapped
    python3 imessage_read.py --from alex            # handles/names matching "alex"
    python3 imessage_read.py --list-threads         # recent conversations + last activity
    python3 imessage_read.py --thread "chat123..."  # one conversation by identifier/guid/name
    python3 imessage_read.py --service imessage     # blue only (drop forgeable SMS)
    python3 imessage_read.py --limit 200 --json     # machine-readable, capped at 200

The window is the bound, not a row count: `--limit` defaults to 0 (no cap), so a
multi-day `--since` returns the whole window. Pass a positive `--limit` only when
you deliberately want the newest N, and watch stderr — every run prints a one-line
coverage summary (window start · rows · threads · oldest message), and a loud
TRUNCATED warning if a cap swallowed older messages. Both go to stderr, so `--json`
stdout stays clean.

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,  # noqa: F401
                                             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()

SCRIPT = Path(__file__).resolve()
VAULT = resolve_vault()
DB_PATH = Path.home() / "Library" / "Messages" / "chat.db"


# ── config (Core-reads-Config seam; personalization never hardcoded here) ─────
def _load_config():
    """Read optional `imessage` tuning from the vault's Config (vault_core: the one
    locator + loader). A missing file or key is fine — built-in defaults apply. No
    identity is read here; this only tunes defaults (window, service, contacts-map
    path)."""
    cfg = _vc_load_config(resolve_vault())
    return cfg.get("imessage", {}) if isinstance(cfg, dict) else {}


IM = _load_config()
DEFAULT_SINCE = IM.get("default_since", "24h")
DEFAULT_SERVICE = IM.get("default_service", "all")
# The module-specific map (gitignored PII) lives IN THE VAULT under the module's asset
# folder — not beside this engine file, which ships inside the installed package.
CONTACTS = (Path(IM["contacts_file"]).expanduser() if IM.get("contacts_file")
            else VAULT / "Agent" / "scripts" / "imessage" / "contacts.json")
# The shared map the contacts module writes from macOS Contacts (handles.py); the
# module-specific file above wins on a clash.
SHARED_HANDLES = shared_handles_path(VAULT, _vc_load_config(VAULT))


# ── handles & contacts ───────────────────────────────────────────────────────
# norm_handle() is the shared implementation (engine/handles.py) so this reader, the
# WhatsApp reader, and the Contacts importer agree on what "the same handle" means.


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)


def display_name(handle, contacts):
    if not handle:
        return "Me"
    return contacts.get(norm_handle(handle), handle)


# ── time ─────────────────────────────────────────────────────────────────────
def cocoa_to_dt(value):
    """message.date is ns since 2001 (modern macOS) or s (very old). Detect by magnitude."""
    if value is None:
        return None
    v = int(value)
    seconds = v / 1_000_000_000 if v > 1_000_000_000_000 else v
    try:
        return dt.datetime.fromtimestamp(seconds + COCOA_EPOCH)
    except (ValueError, OverflowError, OSError):
        return None


def dt_to_cocoa_ns(when):
    return int((when.timestamp() - COCOA_EPOCH) * 1_000_000_000)




# ── attributedBody decode ────────────────────────────────────────────────────
_ARCHIVE_TOKENS = {"NSString", "NSMutableString", "NSAttributedString", "NSObject",
                   "NSDictionary", "NSNumber", "NSValue", "streamtyped", "__kIM"}


def decode_attributed_body(blob):
    """Best-effort text from message.attributedBody (a typedstream NSAttributedString).
    Modern macOS often leaves message.text NULL and stores the body only here. Primary
    parse reads the length-prefixed NSString; the fallback scrapes the longest printable
    run. Offsets validated against real chat.db data on the reference instance."""
    if not blob:
        return None
    # primary: length-prefixed NSString inside the typedstream. The string payload
    # sits after the marker b"\x84\x01+" ('+' = 0x2b string type code), then a
    # length byte (0x81 → 2-byte LE length, 0x82 → 4-byte LE length).
    try:
        i = blob.index(b"NSString")
        k = blob.find(b"\x84\x01+", i + len(b"NSString"))
        if k != -1 and k - i < 16:
            p = k + 3
            length = blob[p]
            p += 1
            if length == 0x81:                                # 2-byte little-endian length
                length = int.from_bytes(blob[p:p + 2], "little"); p += 2
            elif length == 0x82:                              # 4-byte
                length = int.from_bytes(blob[p:p + 4], "little"); p += 4
            text = blob[p:p + length].decode("utf-8", errors="replace").strip()
            if text and text not in _ARCHIVE_TOKENS:
                return text
    except Exception:
        pass
    # fallback: longest printable run that isn't an archive class name
    try:
        runs = re.findall(r"[ -~¡-]{2,}", blob.decode("utf-8", errors="ignore"))
        runs = [r.strip() for r in runs if r.strip() not in _ARCHIVE_TOKENS]
        return max(runs, key=len) if runs else None
    except Exception:
        return None


def message_text(row):
    txt = row["text"]
    if txt and txt.strip():
        return txt.strip()
    return decode_attributed_body(row["attributedBody"]) or "[no text — attachment or reaction]"


# ── db ───────────────────────────────────────────────────────────────────────
FDA_MESSAGE = (
    "chat.db: 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 ~/Library/Messages/chat.db is blocked by macOS TCC."
)


def _is_permission_error(msg):
    msg = msg.lower()
    return any(t in msg for t in ("authoriz", "denied", "not permitted", "unable to open"))


def open_db():
    """Open chat.db read-only. Prefer the live file (sees WAL rows); only on a genuine
    write lock fall back to a temp snapshot copy. Any permission failure — on the open
    OR on the copy — is the TCC / Full Disk Access gate, reported as one clear message."""
    if not DB_PATH.exists():
        raise SystemExit("chat.db not found at %s — is this the machine signed into Messages?" % DB_PATH)
    try:
        con = sqlite3.connect("file:%s?mode=ro" % DB_PATH, uri=True)
        con.row_factory = sqlite3.Row
        con.execute("SELECT 1 FROM message LIMIT 1")
        return con, None
    except sqlite3.DatabaseError as e:            # DatabaseError is the parent of OperationalError
        if _is_permission_error(str(e)):
            raise SystemExit(FDA_MESSAGE)
        try:
            tmp = Path(tempfile.mkdtemp(prefix="imsg-"))
            for suffix in ("", "-wal", "-shm"):
                src = Path(str(DB_PATH) + suffix)
                if src.exists():
                    shutil.copy2(src, tmp / src.name)
            con = sqlite3.connect("file:%s?mode=ro" % (tmp / "chat.db"), uri=True)
            con.row_factory = sqlite3.Row
            con.execute("SELECT 1 FROM message LIMIT 1")
            return con, tmp
        except (PermissionError, OSError, sqlite3.DatabaseError):
            raise SystemExit(FDA_MESSAGE)



def fetch_messages(con, since=None, service="all", limit=0, over=1, thread=None):
    """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."""
    where, params = [], []
    if since:
        where.append("m.date >= ?"); params.append(dt_to_cocoa_ns(since))
    if service == "imessage":
        where.append("m.service = 'iMessage'")
    elif service == "sms":
        where.append("(m.service IS NULL OR m.service != 'iMessage')")
    if thread:
        # filter in SQL, not in Python: --thread ignores --since, so a post-hoc filter
        # would have to pull the whole history into memory to find one conversation.
        where.append("(LOWER(COALESCE(c.chat_identifier,'')) LIKE ? ESCAPE '\\'"
                     " OR LOWER(COALESCE(c.guid,'')) LIKE ? ESCAPE '\\'"
                     " OR LOWER(COALESCE(c.display_name,'')) LIKE ? ESCAPE '\\')")
        params += [like_pattern(thread)] * 3
    sql = """
        SELECT m.ROWID AS rid, m.date, m.is_from_me, m.service, m.text, m.attributedBody,
               h.id AS handle, c.chat_identifier, c.guid AS chat_guid,
               c.display_name AS chat_name, c.ROWID AS chat_id
        FROM message m
        LEFT JOIN handle h ON m.handle_id = h.ROWID
        LEFT JOIN chat_message_join cmj ON cmj.message_id = m.ROWID
        LEFT JOIN chat c ON c.ROWID = cmj.chat_id
    """
    if where:
        sql += " WHERE " + " AND ".join(where)
    sql += " ORDER BY m.date 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):
    sql = """
        SELECT c.ROWID AS chat_id, c.chat_identifier, c.guid, c.display_name,
               MAX(m.date) AS last_date, COUNT(m.ROWID) AS n
        FROM chat c
        JOIN chat_message_join cmj ON cmj.chat_id = c.ROWID
        JOIN message m ON m.ROWID = cmj.message_id
        GROUP BY c.ROWID
        ORDER BY last_date DESC
        LIMIT ?
    """
    return con.execute(sql, (limit,)).fetchall()


# ── rendering ────────────────────────────────────────────────────────────────
def thread_key(row):
    return row["chat_name"] or row["chat_identifier"] or row["handle"] or "(unknown)"


def matches(row, query, contacts):
    if not query:
        return True
    q = query.lower()
    hay = " ".join(filter(None, [
        row["handle"] or "", row["chat_identifier"] or "", row["chat_name"] or "",
        display_name(row["handle"], contacts),
    ])).lower()
    return q in hay or (norm_handle(query) and norm_handle(query) in norm_handle(row["handle"] or ""))


def render_text(rows, contacts):
    groups = {}
    for r in rows:
        groups.setdefault(thread_key(r), []).append(r)
    order = sorted(groups, key=lambda k: max(x["date"] for x in groups[k]), reverse=True)
    out = []
    for key in order:
        msgs = sorted(groups[key], key=lambda x: x["date"])
        out.append("\n=== %s ===" % (contacts.get(norm_handle(key), key)))
        for r in msgs:
            when = cocoa_to_dt(r["date"])
            stamp = when.strftime("%b %d %H:%M") if when else "??"
            who = "Me" if r["is_from_me"] else display_name(r["handle"], contacts)
            svc = "" if r["service"] == "iMessage" else " [SMS]"
            out.append("  %s  %s%s: %s" % (stamp, who, svc, message_text(r).replace("\n", " ")))
    return "\n".join(out) if out else "(no messages in range)"


def render_json(rows, contacts):
    items = []
    for r in sorted(rows, key=lambda x: x["date"]):
        when = cocoa_to_dt(r["date"])
        items.append({
            "time": when.isoformat() if when else None,
            "thread": r["chat_name"] or r["chat_identifier"] or r["handle"],
            "from": "me" if r["is_from_me"] else display_name(r["handle"], contacts),
            "handle": r["handle"],
            "service": r["service"] or "SMS",
            "is_from_me": bool(r["is_from_me"]),
            "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("imessage_read: %s" % msg, file=sys.stderr)


def report_coverage(rows, since, limit, truncated):
    """State what the run actually covered, so a truncated window can't pass for a
    quiet week — plus a loud warning when a cap swallowed older messages."""
    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 threads · 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 threads 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 iMessage/SMS reader.")
    ap.add_argument("--since", default=DEFAULT_SINCE, help=SINCE_FORMS)
    ap.add_argument("--from", dest="who", help="filter by handle or contacts.json name (substring)")
    ap.add_argument("--thread", help="one conversation by chat_identifier / guid / display name")
    ap.add_argument("--service", choices=["all", "imessage", "sms"], default=DEFAULT_SERVICE)
    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 conversations + last activity")
    ap.add_argument("--json", action="store_true")
    args = ap.parse_args()

    contacts = load_contacts()
    since = None if (args.list_threads or args.thread) else parse_since(args.since)   # validate args before DB
    con, tmp = open_db()
    try:
        if args.list_threads:
            rows = list_threads(con, limit=args.limit or 30)
            if args.json:
                print(json.dumps([{
                    "thread": r["display_name"] or r["chat_identifier"],
                    "identifier": r["chat_identifier"], "guid": r["guid"],
                    "last": cocoa_to_dt(r["last_date"]).isoformat() 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 "??"
                    name = r["display_name"] or contacts.get(norm_handle(r["chat_identifier"] or ""), r["chat_identifier"])
                    print("%s  %-28s  (%d msgs)" % (stamp, name, r["n"]))
            return

        # --from filters in Python (contacts.json aliases don't exist in SQL), so a
        # capped fetch over-reads to leave room for the filter. --thread filters in SQL.
        over = 8 if args.who else 1
        rows, sql_cap = fetch_messages(con, since=since, service=args.service,
                                       limit=args.limit, over=over, thread=args.thread)
        truncated = bool(sql_cap) and len(rows) >= sql_cap

        if args.who:
            rows = [r for r in rows if matches(r, args.who, contacts)]
        if args.limit:
            truncated = truncated or len(rows) > args.limit
            rows = rows[:args.limit]

        print(render_json(rows, contacts) if args.json else render_text(rows, contacts))
        report_coverage(rows, since, args.limit, truncated)
    finally:
        con.close()
        if tmp:
            shutil.rmtree(tmp, ignore_errors=True)


if __name__ == "__main__":
    main()
