Gmail

pip install httpx geniffy

uv add httpx geniffy

import base64
import html
import re
from datetime import datetime, timezone

import httpx
from geniffy import Geniffy, NotFoundError

geniffy = Geniffy()                               # reads GENIFFY_API_KEY
LABELS = {"channel": "gmail"}
QUERY = "in:sent newer_than:1y"                   # the first run: every thread the user wrote in, this past year
QUOTED = re.compile(r"^(On .+ wrote:|-+ ?Original Message ?-+|_{20,})$")    # where a reply starts quoting


def gmail(token: str) -> httpx.Client:
    """One user's mailbox, with the access token from their Google sign-in."""
    return httpx.Client(base_url="https://gmail.googleapis.com/gmail/v1/users/me", timeout=60,
                        headers={"Authorization": f"Bearer {token}"})


def plain(part: dict) -> str:
    """A message's plain text, from whichever of its parts holds it."""
    data = part.get("body", {}).get("data")
    if part.get("mimeType") == "text/plain" and data and not part.get("filename"):
        return base64.urlsafe_b64decode(data + "=" * (-len(data) % 4)).decode("utf-8", "replace")
    for sub in part.get("parts", []):
        if text := plain(sub):
            return text
    return ""


def headers(message: dict) -> dict:
    return {h["name"].lower(): h["value"] for h in message["payload"].get("headers", [])}


def own_words(text: str) -> str:
    """What a message adds, without the earlier messages it quotes."""
    kept = []
    for line in text.splitlines():
        if QUOTED.match(line.strip()):
            break
        if not line.startswith(">"):
            kept.append(line)
    return "\n".join(kept).strip()


def conversation(thread: dict) -> tuple[str, str, datetime] | None:
    """A thread as one note: who wrote each message, when, and its own words; with its subject and the time
    of its latest message. None when the user never wrote in it."""
    messages = [m for m in thread.get("messages", []) if not {"TRASH", "SPAM", "DRAFT"} & set(m.get("labelIds", []))]
    if not any("SENT" in m.get("labelIds", []) for m in messages):
        return None
    parts, when = [], None
    for m in messages:
        words = own_words(plain(m["payload"]) or html.unescape(m.get("snippet", "")))
        when = datetime.fromtimestamp(int(m["internalDate"]) / 1000, timezone.utc)
        if words:
            parts.append(f"From {headers(m).get('from', 'someone')}, {when:%d %b %Y}:\n{words}")
    if not parts:
        return None
    return headers(messages[0]).get("subject") or "(no subject)", "\n\n".join(parts), when


def keep(mem, api: httpx.Client, thread_id: str) -> bool:
    """Add one thread under its Gmail id. False when it is gone, or the user never wrote in it."""
    got = api.get(f"/threads/{thread_id}", params={"format": "full"})
    found = None if got.status_code == 404 else conversation(got.raise_for_status().json())
    if found is None:
        return False
    subject, text, when = found
    mem.memories.add(text, title=subject, said_at=when, external_id=f"gmail:{thread_id}", labels=LABELS)
    return True


def forget(mem, thread_id: str) -> None:
    try:
        mem.sources.delete(external_id=f"gmail:{thread_id}")
    except NotFoundError:                         # never added: no one answered it
        pass


def read_all(mem, api: httpx.Client) -> None:
    """Every thread QUERY finds; and what memory holds from Gmail that is no longer there goes."""
    seen, params = set(), {"q": QUERY, "maxResults": 100}
    while True:
        out = api.get("/threads", params=params).raise_for_status().json()
        seen |= {f"gmail:{t['id']}" for t in out.get("threads", []) if keep(mem, api, t["id"])}
        if "nextPageToken" not in out:
            break
        params["pageToken"] = out["nextPageToken"]
    mem.sources.delete_labelled(LABELS, keep=seen)


def changed(api: httpx.Client, history_id: str) -> tuple[set, str] | None:
    """The threads that changed since history_id, and the id to pass next time. None when Gmail no
    longer keeps its history that far back."""
    threads, params = set(), {"startHistoryId": history_id, "maxResults": 500}
    while True:
        got = api.get("/history", params=params)
        if got.status_code == 404:
            return None
        out = got.raise_for_status().json()
        for record in out.get("history", []):
            for item in record.get("messagesAdded", []) + record.get("messagesDeleted", []):
                threads.add(item["message"]["threadId"])
            for item in record.get("labelsAdded", []) + record.get("labelsRemoved", []):
                if {"TRASH", "SPAM"} & set(item.get("labelIds", [])):    # binned or marked as spam, or back
                    threads.add(item["message"]["threadId"])
        if "nextPageToken" not in out:
            return threads, out["historyId"]
        params["pageToken"] = out["nextPageToken"]


def sync(user_id: str, token: str, history_id: str | None = None) -> str:
    """Bring one user's email conversations into their memory. Returns the id to pass next time: pass None
    the first time and QUERY is read; after that, only threads that changed since the last run are fetched."""
    mem = geniffy.space(f"user_{user_id}")
    with gmail(token) as api:
        found = changed(api, history_id) if history_id else None
        if found is None:                         # the first run, or history too old to ask for
            start = api.get("/profile").raise_for_status().json()["historyId"]
            read_all(mem, api)
            return start
        threads, latest = found
        for thread_id in threads:
            if not keep(mem, api, thread_id):
                forget(mem, thread_id)
        return latest


def disconnect(user_id: str) -> int:
    """The user disconnected Gmail: everything that came from it goes. Returns how many threads."""
    return geniffy.space(f"user_{user_id}").sources.delete_labelled(LABELS)