Google Chat + Gmail monitor TUI for your terminal. Shows DMs, @-mentions, emoji reactions, and unread email count. Designed for a tmux pane.

chat-monitor.py

#!/usr/bin/env python3
"""
chat-monitor.py — A terminal TUI for Google Chat and Gmail.

Shows recent DMs, @-mentions, and emoji reactions from Google Chat,
plus an unread email count. Refreshes every 30 seconds. Press q to quit.

Designed to live in a small tmux pane while you work.

What shows up:
  (no prefix)  DM from someone else (last 60 min)
  @ (red)      You were @-mentioned in a group chat or space
  * (magenta)  Your message received emoji reactions

Setup:
  1. Install the Google Workspace MCP server (provides the Python dependencies):
     https://github.com/taylorwilsdon/google_workspace_mcp
  2. Set environment variables:
       GOOGLE_OAUTH_CLIENT_ID   — your Google Cloud OAuth 2.0 client ID
       GOOGLE_OAUTH_CLIENT_SECRET — your Google Cloud OAuth 2.0 client secret
       MY_NAME                  — your full name, e.g. "Phil Peralez" (default: Drew Gillson)
       MY_EMAIL                 — your Google email (default: drew@gravity.foundation)
  3. Run: python3 chat-monitor.py
  4. First run opens a browser to authorize. Credentials are cached after that.

Requirements:
  - Python 3.10+
  - google-api-python-client, google-auth-oauthlib (via workspace-mcp venv)
  - A Google Cloud OAuth client ID with these scopes enabled:
      chat.spaces.readonly, chat.messages.readonly, chat.memberships.readonly,
      directory.readonly, gmail.readonly
"""

import curses
import json
import os
import sys
import time
from datetime import datetime, timedelta, timezone

# ---------------------------------------------------------------------------
# Workspace MCP venv — adjust this path if your install location differs
# ---------------------------------------------------------------------------
VENV_SITE = os.path.expanduser(
    "~/google_workspace_mcp/.venv/lib/python3.13/site-packages"
)
if os.path.isdir(VENV_SITE):
    sys.path.insert(0, VENV_SITE)

from google.oauth2.credentials import Credentials  # noqa: E402
from google_auth_oauthlib.flow import InstalledAppFlow  # noqa: E402
from google.auth.transport.requests import Request  # noqa: E402
from googleapiclient.discovery import build  # noqa: E402

# ---------------------------------------------------------------------------
# USER CONFIGURATION — override via environment variables
# ---------------------------------------------------------------------------
MY_NAME = os.environ.get("MY_NAME", "Drew Gillson")
MY_EMAIL = os.environ.get("MY_EMAIL", "drew@gravity.foundation")

# ---------------------------------------------------------------------------
# OAUTH CONFIGURATION — set these env vars or pass them inline
# ---------------------------------------------------------------------------
CLIENT_ID = os.environ.get("GOOGLE_OAUTH_CLIENT_ID", "")
CLIENT_SECRET = os.environ.get("GOOGLE_OAUTH_CLIENT_SECRET", "")
if not CLIENT_ID or not CLIENT_SECRET:
    print("Error: GOOGLE_OAUTH_CLIENT_ID and GOOGLE_OAUTH_CLIENT_SECRET must be set.")
    sys.exit(1)
SCOPES = [
    "https://www.googleapis.com/auth/chat.spaces.readonly",
    "https://www.googleapis.com/auth/chat.messages.readonly",
    "https://www.googleapis.com/auth/chat.memberships.readonly",
    "https://www.googleapis.com/auth/directory.readonly",
    "https://www.googleapis.com/auth/gmail.readonly",
]
TOKEN_PATH = os.path.expanduser("~/.google_workspace_mcp/chat_monitor_token.json")

# ---------------------------------------------------------------------------
# DISPLAY SETTINGS
# ---------------------------------------------------------------------------
REFRESH_SECONDS = 30
MAX_CHAT_AGE_MINUTES = 60
GMAIL_QUERY = "is:unread in:inbox -from:github.com -from:noreply@github.com"

# ---------------------------------------------------------------------------
# Runtime caches (persist within a single run)
# ---------------------------------------------------------------------------
_name_cache: dict[str, str] = {}
_unread_email_ids: set[str] = set()


# ===========================================================================
# Auth
# ===========================================================================

def _get_credentials():
    """Load cached credentials or run the browser OAuth flow."""
    creds = None
    if os.path.exists(TOKEN_PATH):
        with open(TOKEN_PATH) as f:
            data = json.load(f)
        creds = Credentials(
            token=data.get("token"),
            refresh_token=data.get("refresh_token"),
            token_uri=data.get("token_uri"),
            client_id=data.get("client_id"),
            client_secret=data.get("client_secret"),
            scopes=data.get("scopes"),
        )

    # Re-auth if scopes have changed since the token was issued
    if creds and creds.scopes and set(SCOPES) - set(creds.scopes):
        creds = None

    if creds and creds.expired and creds.refresh_token:
        creds.refresh(Request())
        _save_credentials(creds)
    elif not creds or not creds.valid:
        flow = InstalledAppFlow.from_client_config(
            {
                "installed": {
                    "client_id": CLIENT_ID,
                    "client_secret": CLIENT_SECRET,
                    "auth_uri": "https://accounts.google.com/o/oauth2/auth",
                    "token_uri": "https://oauth2.googleapis.com/token",
                    "redirect_uris": ["http://localhost"],
                }
            },
            scopes=SCOPES,
        )
        creds = flow.run_local_server(port=0)
        _save_credentials(creds)

    return creds


def _save_credentials(creds):
    os.makedirs(os.path.dirname(TOKEN_PATH), exist_ok=True)
    with open(TOKEN_PATH, "w") as f:
        json.dump(
            {
                "token": creds.token,
                "refresh_token": creds.refresh_token,
                "token_uri": creds.token_uri,
                "client_id": creds.client_id,
                "client_secret": creds.client_secret,
                "scopes": list(creds.scopes) if creds.scopes else SCOPES,
            },
            f,
        )


# ===========================================================================
# Identity helpers
# ===========================================================================

def _is_me(name: str) -> bool:
    """Check if a display name belongs to the current user."""
    lower = name.lower()
    return all(part in lower for part in MY_NAME.lower().split())


def _resolve_name(people_svc, user_id: str) -> str:
    """Resolve a Chat API user ID (users/123) to a display name via People API."""
    if user_id in _name_cache:
        return _name_cache[user_id]

    people_id = user_id.replace("users/", "people/", 1)
    try:
        person = (
            people_svc.people()
            .get(resourceName=people_id, personFields="names,emailAddresses")
            .execute()
        )
        for field, key in [("names", "displayName"), ("emailAddresses", "value")]:
            entries = person.get(field, [])
            if entries:
                value = entries[0].get(key, user_id)
                if field == "emailAddresses":
                    value = value.split("@")[0]
                _name_cache[user_id] = value
                return value
    except Exception:
        pass

    _name_cache[user_id] = user_id.replace("users/", "")
    return _name_cache[user_id]


# ===========================================================================
# Chat message helpers
# ===========================================================================

def _get_sender(msg: dict, people_svc) -> str:
    """Extract the sender's display name from a Chat message."""
    sender = msg.get("sender", {})
    name = sender.get("displayName", "")
    if not name:
        raw = sender.get("name", "")
        name = _resolve_name(people_svc, raw) if raw else "?"
    return name


def _get_dm_label(chat_svc, people_svc, space_id: str) -> str:
    """Get the other person's name in a DM space."""
    try:
        resp = chat_svc.spaces().members().list(parent=space_id, pageSize=10).execute()
        for m in resp.get("memberships", []):
            member = m.get("member", {})
            display = member.get("displayName", "")
            if display and not _is_me(display):
                return display
            member_name = member.get("name", "")
            if member_name:
                resolved = _resolve_name(people_svc, member_name)
                if not _is_me(resolved):
                    return resolved
    except Exception:
        pass
    return "DM"


def _mentions_me(msg: dict) -> bool:
    """Check if a Chat message @-mentions the current user."""
    for ann in msg.get("annotations", []):
        if ann.get("type") == "USER_MENTION":
            user = ann.get("userMention", {}).get("user", {})
            if user.get("type") == "HUMAN" and _is_me(user.get("displayName", "")):
                return True
    if f"@{MY_NAME.split()[0].lower()}" in msg.get("text", "").lower():
        return True
    return False


def _msg_text(msg: dict) -> str:
    """Extract a single-line display string from a Chat message."""
    text = msg.get("text", "").replace("\n", " ").strip()
    if text:
        return text
    if msg.get("attachment"):
        return "[attachment]"
    if msg.get("cardsV2") or msg.get("cards"):
        return "[card]"
    return "[empty]"


def _format_reactions(msg: dict) -> str:
    """Format emoji reaction summaries (e.g. "checkmark thumbsup x3")."""
    reactions = msg.get("emojiReactionSummaries", [])
    if not reactions:
        return ""
    parts = []
    for r in reactions:
        emoji = r.get("emoji", {})
        symbol = emoji.get("unicode", "")
        if not symbol:
            symbol = f":{emoji.get('customEmoji', {}).get('uid', '?')}:"
        count = r.get("reactionCount", 0)
        parts.append(f"{symbol}{f'x{count}' if count > 1 else ''}")
    return " ".join(parts)


def _time_ago(iso_str: str) -> str:
    """Convert an ISO timestamp to a compact relative time (e.g. '5m', '2h')."""
    try:
        dt = datetime.fromisoformat(iso_str.replace("Z", "+00:00"))
        secs = int((datetime.now(timezone.utc) - dt).total_seconds())
        if secs < 60:
            return "now"
        if secs < 3600:
            return f"{secs // 60}m"
        if secs < 86400:
            return f"{secs // 3600}h"
        return f"{secs // 86400}d"
    except Exception:
        return "?"


def _parse_ts(iso_str: str) -> datetime:
    """Parse an ISO timestamp for sorting. Returns epoch-zero on failure."""
    try:
        return datetime.fromisoformat(iso_str.replace("Z", "+00:00"))
    except Exception:
        return datetime.min.replace(tzinfo=timezone.utc)


def _fetch_messages(chat_svc, space_id: str, count: int = 10) -> list[dict]:
    """Fetch recent messages from a Chat space, newest first."""
    try:
        resp = (
            chat_svc.spaces()
            .messages()
            .list(parent=space_id, pageSize=count, orderBy="createTime desc")
            .execute()
        )
        return resp.get("messages", [])
    except Exception:
        return []


# ===========================================================================
# Chat data fetchers — each returns a list of display entries
# ===========================================================================

def _make_entry(label, sender, text, create_time, kind):
    return {
        "label": label,
        "sender": sender,
        "text": text,
        "ago": _time_ago(create_time),
        "ts": _parse_ts(create_time),
        "kind": kind,
    }


def _fetch_dm_entries(chat_svc, people_svc, dm_spaces):
    """Most recent non-me message from each DM space."""
    entries = []
    for space in dm_spaces:
        space_id = space["name"]
        for m in _fetch_messages(chat_svc, space_id, count=5):
            sender = _get_sender(m, people_svc)
            if not _is_me(sender):
                label = _get_dm_label(chat_svc, people_svc, space_id)
                entries.append(_make_entry(
                    label, sender, _msg_text(m), m.get("createTime", ""), "dm",
                ))
                break
    return entries


def _fetch_mention_entries(chat_svc, people_svc, group_spaces):
    """Messages that @-mention me in group chats and named spaces."""
    entries = []
    for space in group_spaces:
        space_name = space.get("displayName") or "Group"
        for m in _fetch_messages(chat_svc, space["name"], count=10):
            sender = _get_sender(m, people_svc)
            if _is_me(sender):
                continue
            if _mentions_me(m):
                entries.append(_make_entry(
                    space_name, sender, _msg_text(m), m.get("createTime", ""), "mention",
                ))
    return entries


def _fetch_reacted_entries(chat_svc, people_svc, spaces, cutoff):
    """My messages that received emoji reactions (within the time cutoff)."""
    entries = []
    for space in spaces:
        space_id = space["name"]
        space_name = space.get("displayName") or "DM"
        space_type = space.get("spaceType", "")
        for m in _fetch_messages(chat_svc, space_id, count=10):
            ts = _parse_ts(m.get("createTime", ""))
            if ts < cutoff:
                break
            sender = _get_sender(m, people_svc)
            if not _is_me(sender):
                continue
            reactions = _format_reactions(m)
            if not reactions:
                continue
            label = (
                _get_dm_label(chat_svc, people_svc, space_id)
                if space_type == "DIRECT_MESSAGE"
                else space_name
            )
            entries.append(_make_entry(
                label, sender, f"{reactions}  {_msg_text(m)}",
                m.get("createTime", ""), "reacted",
            ))
    return entries


def fetch_chat(chat_svc, people_svc):
    """Fetch all chat entries: DMs, @-mentions, and reacted messages."""
    cutoff = datetime.now(timezone.utc) - timedelta(minutes=MAX_CHAT_AGE_MINUTES)

    try:
        resp = chat_svc.spaces().list(pageSize=100).execute()
    except Exception as e:
        return [_make_entry("ERROR", "", str(e), "", "dm")]

    spaces = resp.get("spaces", [])
    dm_spaces = [s for s in spaces if s.get("spaceType") == "DIRECT_MESSAGE"]
    group_spaces = [s for s in spaces if s.get("spaceType") in ("GROUP_CHAT", "SPACE")]

    entries = []
    entries.extend(_fetch_dm_entries(chat_svc, people_svc, dm_spaces))
    entries.extend(_fetch_mention_entries(chat_svc, people_svc, group_spaces))
    entries.extend(_fetch_reacted_entries(chat_svc, people_svc, dm_spaces + group_spaces, cutoff))

    # Deduplicate by (label, timestamp)
    seen = set()
    unique = []
    for e in entries:
        key = (e["label"], e["ts"])
        if key not in seen:
            seen.add(key)
            unique.append(e)

    unique.sort(key=lambda e: e["ts"], reverse=True)
    return [e for e in unique if e["ts"] > cutoff][:25]


# ===========================================================================
# Gmail — unread count
# ===========================================================================

def fetch_unread_email_count(gmail_svc):
    """Return the number of unread inbox emails (excluding GitHub notifications)."""
    global _unread_email_ids
    try:
        resp = (
            gmail_svc.users()
            .messages()
            .list(userId="me", q=GMAIL_QUERY, maxResults=50)
            .execute()
        )
        messages = resp.get("messages") or []
        _unread_email_ids = {m["id"] for m in messages}
    except Exception:
        pass  # keep previous count on error
    return len(_unread_email_ids)


# ===========================================================================
# TUI
# ===========================================================================

def _draw(stdscr, entries, status=""):
    stdscr.erase()
    height, width = stdscr.getmaxyx()

    # Color pairs
    curses.init_pair(1, curses.COLOR_CYAN, -1)      # header / rules
    curses.init_pair(2, curses.COLOR_YELLOW, -1)     # labels
    curses.init_pair(3, curses.COLOR_WHITE, -1)      # body text
    curses.init_pair(4, curses.COLOR_RED, -1)        # @ mention marker
    curses.init_pair(5, curses.COLOR_MAGENTA, -1)    # * reaction marker

    # ── Google Chat ──────────────────────────────── HH:MM:SS
    now_str = datetime.now().strftime("%H:%M:%S")
    header = " Google Chat "
    rule_len = max(width - 2 - len(header) - len(now_str) - 2, 0)
    row = 0
    try:
        stdscr.addstr(row, 0, "\u2500\u2500", curses.color_pair(1) | curses.A_BOLD)
        stdscr.addstr(header, curses.color_pair(1) | curses.A_BOLD)
        stdscr.addstr("\u2500" * rule_len, curses.color_pair(1) | curses.A_BOLD)
        stdscr.addstr(f" {now_str}", curses.color_pair(1))
    except curses.error:
        pass
    row += 1

    # Chat entries
    max_row = height - 3
    if not entries:
        try:
            stdscr.addstr(row, 2, "(no recent messages)", curses.color_pair(3) | curses.A_DIM)
        except curses.error:
            pass
    else:
        for e in entries:
            if row > max_row:
                break
            label = e["label"]
            if len(label) > 18:
                label = label[:17] + "\u2026"

            kind = e.get("kind", "dm")
            col = 0
            try:
                if kind == "mention":
                    stdscr.addstr(row, 0, "@", curses.color_pair(4) | curses.A_BOLD)
                    col = 1
                elif kind == "reacted":
                    stdscr.addstr(row, 0, "*", curses.color_pair(5) | curses.A_BOLD)
                    col = 1

                stdscr.addstr(row, col, f"{label:<18}", curses.color_pair(2) | curses.A_BOLD)
                stdscr.addstr(row, col + 18, f" {e['ago']:>3}", curses.color_pair(3) | curses.A_DIM)
                stdscr.addstr(row, col + 22, " \u2502 ", curses.color_pair(3) | curses.A_DIM)

                text = e["text"]
                max_text = max(width - col - 25, 10)
                if len(text) > max_text:
                    text = text[: max_text - 1] + "\u2026"
                stdscr.addstr(text, curses.color_pair(3))
            except curses.error:
                pass
            row += 1

    # Bottom rule + status
    try:
        stdscr.addstr(height - 2, 0, "\u2500" * width, curses.color_pair(1) | curses.A_DIM)
    except curses.error:
        pass
    if status:
        try:
            stdscr.addstr(height - 1, 0, status[: width - 1], curses.color_pair(3) | curses.A_DIM)
        except curses.error:
            pass

    stdscr.refresh()


# ===========================================================================
# Main loop
# ===========================================================================

def _tui_main(stdscr):
    curses.use_default_colors()
    curses.curs_set(0)
    stdscr.timeout(1000)

    creds = _get_credentials()
    chat_svc = build("chat", "v1", credentials=creds)
    people_svc = build("people", "v1", credentials=creds)
    gmail_svc = build("gmail", "v1", credentials=creds)

    entries = []
    n_emails = 0
    last_fetch = 0.0

    while True:
        now = time.time()

        # Refresh data on schedule
        if now - last_fetch >= REFRESH_SECONDS:
            if creds.expired and creds.refresh_token:
                creds.refresh(Request())
                _save_credentials(creds)
                chat_svc = build("chat", "v1", credentials=creds)
                people_svc = build("people", "v1", credentials=creds)
                gmail_svc = build("gmail", "v1", credentials=creds)

            _draw(stdscr, entries, status="Fetching...")
            entries = fetch_chat(chat_svc, people_svc)
            n_emails = fetch_unread_email_count(gmail_svc)
            last_fetch = time.time()

        # Status bar
        next_in = max(REFRESH_SECONDS - int(time.time() - last_fetch), 0)
        email_str = f"{n_emails} unread email{'s' if n_emails != 1 else ''}"
        _draw(stdscr, entries, status=f"Next refresh in {next_in}s  |  {email_str}  |  q to quit")

        if stdscr.getch() in (ord("q"), ord("Q")):
            break


def main():
    print("Authenticating...")
    creds = _get_credentials()
    _save_credentials(creds)
    print("Connected. Launching TUI...")
    curses.wrapper(_tui_main)


if __name__ == "__main__":
    main()