"""Email reader CARD — the whole thread on Ahmed's HUD, never in a browser.

Ahmed has a standing NEVER rule against Gmail-in-the-browser (enforced in code
by voice/show_gate.py). This card is the SANCTIONED way for him to actually
READ mail: he says "pull up Elie's email", Jarvis calls ``mail_show``, the
engine fetches the FULL thread (every reply, the HTML body, the attachments),
sanitises it, and emits ONE event — the HUD opens a reader card he can scroll,
open attachments from, and REPLY from. There is no browser anywhere in here,
and there must never be.

  Jarvis (voice) -> mail_show / mail_show_id  (open the card)
                    mail_close                (drop it)
  Ahmed  (HUD)   -> clicks an attachment chip / Load images / Send
                    -> HUD writes control/email_action.json
                    -> ControlWatcher._email_action applies it and, for a sent
                       reply, drops a note in the master's inbox so Jarvis says
                       it out loud.

Two mailboxes, two backends, one card:
  work      Gmail API (voice/gmail_control.py) — real thread ids
  business  ahmad@revalstudio.com over IMAP (voice/email_control.py) — the
            "thread" is reassembled from References/In-Reply-To, best effort

Engine → HUD events (voice.events.emit):
  email_thread      {thread_id, account, subject, messages[]}   (oldest first)
  email_attachment  {thread_id, msg_id, att_id, path, ok, error}
  email_reply_sent  {thread_id, ok, error}
  card_close        {draft_id: thread_id}    (reuses the compose card's close)
Set MCP_MAIL_CARD=0 to disable.
"""

from __future__ import annotations

import os
import re
import subprocess
import sys
from datetime import datetime, timedelta, timezone
from email.utils import parseaddr
from html import escape
from html.parser import HTMLParser
from pathlib import Path

from voice.events import emit

_TZ = timezone(timedelta(hours=3))                       # Riyadh, +03:00
ATT_ROOT = Path.home() / ".jarvis" / "attachments"       # downloaded chips land here

# thread_id -> {"account", "raw" (backend thread), "unblocked" (msg_ids whose
# remote images Ahmed loaded), "paths" ({(msg_id, att_id): local path})}
_THREADS: dict[str, dict] = {}
_ORDER: list[str] = []                                   # so mail_close("") works

# What a blocked remote image collapses to: a 1x1 transparent GIF. Email is
# wall-to-wall tracking pixels; nothing remote is fetched until Ahmed asks.
_PIXEL = ("data:image/gif;base64,R0lGODlhAQABAIAAAAAAAP///yH5BAEAAAAALAAAAAAB"
          "AAEAAAIBRAA7")


def enabled() -> bool:
    if os.environ.get("MCP_MAIL_CARD", "1") == "0":
        return False
    from voice import email_control, gmail_control
    return gmail_control.enabled() or email_control.enabled()


# --- HTML sanitising ------------------------------------------------------
# The HUD renders body_html in a WKWebView (JS off, non-navigating). That is
# the second line of defence; THIS is the first. Everything executable is cut
# out here, before the HTML ever leaves the engine.

_DROP_TREE = {"script", "iframe", "object", "embed", "applet", "frame",
              "frameset", "noscript", "form", "svg", "math", "template"}
_DROP_TAG = {"link", "meta", "base"}          # dropped, but keep their content
_VOID = {"area", "base", "br", "col", "embed", "hr", "img", "input", "link",
         "meta", "param", "source", "track", "wbr"}
_URL_ATTRS = {"src", "href", "background", "poster", "action", "data",
              "codebase", "cite", "longdesc"}
_KILL_ATTRS = {"srcset", "formaction", "xlink:href", "http-equiv", "ping"}

_JS_RE = re.compile(r"^(javascript|vbscript|data:text/html)", re.I)
_CSS_URL_RE = re.compile(r"url\(\s*['\"]?\s*https?://[^)]*\)", re.I)
_CSS_JS_RE = re.compile(r"(javascript|vbscript)\s*:|expression\s*\(|@import[^;]*;",
                        re.I)
_TAG_RE = re.compile(r"<[^>]+>")


class _Sanitizer(HTMLParser):
    """Rebuilds the HTML, keeping only what's safe to render.

    Drops script/iframe/object/embed/svg/form subtrees, link/meta/base tags,
    every on*= handler and javascript: URL. Inline cid: images become data:
    URIs from the message's own parts (safe — they already arrived with the
    mail). Remote http(s) images become a transparent pixel until Ahmed hits
    "Load images" (they're tracking pixels far more often than they're
    pictures)."""

    def __init__(self, inline: dict, block_remote: bool) -> None:
        super().__init__(convert_charrefs=True)
        self.inline = inline or {}
        self.block = block_remote
        self.blocked = 0          # how many remote images we swallowed
        self.out: list[str] = []
        self._skip = 0            # depth inside a dropped subtree
        self._css = False         # inside <style> (raw CSS, must not escape)

    # -- tags
    def handle_starttag(self, tag: str, attrs: list) -> None:
        if self._skip:
            if tag in _DROP_TREE:
                self._skip += 1
            return
        if tag in _DROP_TREE:
            self._skip = 1
            return
        if tag in _DROP_TAG:
            return
        if tag == "style":
            self._css = True
        self.out.append(f"<{tag}{self._attrs(tag, attrs)}>")

    def handle_startendtag(self, tag: str, attrs: list) -> None:
        if self._skip or tag in _DROP_TREE or tag in _DROP_TAG:
            return
        self.out.append(f"<{tag}{self._attrs(tag, attrs)} />")

    def handle_endtag(self, tag: str) -> None:
        if self._skip:
            if tag in _DROP_TREE:
                self._skip -= 1
            return
        if tag in _DROP_TAG or tag in _VOID:
            return
        if tag == "style":
            self._css = False
        self.out.append(f"</{tag}>")

    # -- content
    def handle_data(self, data: str) -> None:
        if self._skip:
            return
        self.out.append(self.css(data) if self._css else escape(data, quote=False))

    def handle_comment(self, data: str) -> None:
        pass  # conditional comments can smuggle markup — drop them all

    # -- attributes
    def _attrs(self, tag: str, attrs: list) -> str:
        out = []
        for raw_k, raw_v in attrs:
            k = (raw_k or "").lower()
            v = raw_v or ""
            if k.startswith("on") or k in _KILL_ATTRS:
                continue
            if k == "style":
                v = self.css(v)
            elif k in _URL_ATTRS:
                url = self.url(tag, k, v)
                if url is None:
                    continue
                v = url
            elif _JS_RE.match(v.strip()):
                continue
            out.append(f' {k}="{escape(v, quote=True)}"')
        return "".join(out)

    def url(self, tag: str, key: str, value: str) -> str | None:
        """Rewrite one URL attribute. None = drop the attribute entirely."""
        s = (value or "").strip()
        flat = re.sub(r"[\s\x00-\x1f]", "", s).lower()   # java\nscript: evasion
        if not flat or _JS_RE.match(flat):
            return None
        if flat.startswith("cid:"):
            part = self.inline.get(s.split(":", 1)[1].strip().strip("<>"))
            if not part:
                return None                              # unknown cid — no src
            return f"data:{part['mime']};base64,{part['b64']}"
        if flat.startswith("data:"):
            return s if flat.startswith("data:image/") else None
        if flat.startswith(("http://", "https://")):
            if key == "href":
                return s          # links are fine: the HUD opens them outside
            if self.block:
                self.blocked += 1
                return _PIXEL
            return s
        if key == "href" and flat.startswith(("mailto:", "tel:")):
            return s
        return None               # relative URLs are meaningless in mail

    def css(self, css: str) -> str:
        """Neutralise CSS: no javascript:/expression()/@import, and no remote
        url() (a background image tracks just as well as an <img>)."""
        out = _CSS_JS_RE.sub("", css or "")
        if self.block:
            out = _CSS_URL_RE.sub("url(about:blank)", out)
        return out


def sanitize_html(html: str, inline: dict | None = None,
                  block_remote: bool = True) -> tuple[str, int]:
    """Sanitised HTML + how many remote images were blocked."""
    if not html:
        return "", 0
    p = _Sanitizer(inline or {}, block_remote)
    try:
        p.feed(html)
        p.close()
    except Exception:  # noqa: BLE001 — malformed HTML must never kill the card
        return escape(_TAG_RE.sub(" ", html), quote=False), 0
    return "".join(p.out), p.blocked


def _plain(html: str) -> str:
    """Rough text fallback for a message that only has an HTML body."""
    txt = re.sub(r"(?is)<(script|style)[^>]*>.*?</\1>", " ", html or "")
    txt = re.sub(r"(?i)<br\s*/?>|</p>|</div>|</tr>", "\n", txt)
    txt = _TAG_RE.sub("", txt)
    from html import unescape
    return re.sub(r"\n{3,}", "\n\n", unescape(txt)).strip()


# --- attachments on disk ---------------------------------------------------

def safe_name(name: str) -> str:
    """A filename that can only ever land INSIDE the attachment dir: no path
    separators, no '..', no control characters, no leading dot."""
    base = os.path.basename((name or "").replace("\\", "/").strip())
    base = re.sub(r"[\x00-\x1f/]+", "", base).strip(". ")
    base = base.replace("..", ".")
    return base[:120] or "attachment"


def dest_path(thread_id: str, filename: str) -> Path:
    """Where an attachment of this thread gets written (dir created)."""
    d = ATT_ROOT / safe_name(thread_id or "mail")
    d.mkdir(parents=True, exist_ok=True)
    return d / safe_name(filename)


# --- accounts --------------------------------------------------------------

def _account(account: str) -> str:
    """A spoken account word -> a backend account. 'business'/'reval' = the
    revalstudio IMAP mailbox; anything else = a Gmail label (default 'work')."""
    a = (account or "").strip().lower()
    from voice import email_control, gmail_control
    if ("business" in a or "reval" in a) and email_control.enabled():
        return "business"
    if not gmail_control.enabled() and email_control.enabled():
        return "business"
    return gmail_control._resolve(a) if a else gmail_control._default_label()


def _my_addresses(account_addr: str) -> set[str]:
    """Every address that means 'Ahmed sent this' — the mailbox's own address
    plus anything in MAIL_MY_ADDRESSES."""
    mine = {a.strip().lower() for a in
            os.environ.get("MAIL_MY_ADDRESSES", "").split(",") if a.strip()}
    for a in (account_addr, os.environ.get("EMAIL_ADDRESS", "")):
        if a:
            mine.add(a.strip().lower())
    return mine


# --- thread -> the email_thread payload ------------------------------------

def _when(raw_date: str, ts: float) -> datetime:
    from email.utils import parsedate_to_datetime
    try:
        d = parsedate_to_datetime(raw_date)
        if d.tzinfo is None:
            d = d.replace(tzinfo=timezone.utc)
        return d
    except Exception:  # noqa: BLE001 — garbled Date header, fall back to ts
        return datetime.fromtimestamp(ts or 0, tz=timezone.utc)


def _now() -> datetime:
    return datetime.now(_TZ)


def _human(d: datetime) -> str:
    """Pre-formatted Riyadh time — the HUD just prints it."""
    now = _now()
    d = d.astimezone(_TZ)
    if d.date() == now.date():
        return d.strftime("%H:%M")
    if (now - d).days < 6:
        return d.strftime("%a %H:%M")
    if d.year == now.year:
        return d.strftime("%d %b %H:%M")
    return d.strftime("%d %b %Y")


def payload(thread_id: str) -> dict:
    """Build the `email_thread` event body from the cached raw thread."""
    st = _THREADS[thread_id]
    raw = st["raw"]
    mine = _my_addresses(raw.get("account_addr", ""))
    msgs = []
    for m in raw.get("messages", []):
        mid = m.get("msg_id", "")
        name, addr = parseaddr(m.get("from", ""))
        block = mid not in st["unblocked"]
        html, blocked = sanitize_html(m.get("body_html", ""),
                                      m.get("inline") or {}, block)
        text = m.get("body_text", "") or _plain(m.get("body_html", ""))
        when = _when(m.get("date", ""), m.get("ts", 0))
        atts = []
        for a in m.get("attachments", []):
            atts.append({**a, "path": st["paths"].get(
                (mid, a.get("att_id", "")), a.get("path", "") or "")})
        msgs.append({
            "msg_id": mid,
            "from_name": name or addr or "(unknown)",
            "from_addr": addr,
            "to": m.get("to", ""), "cc": m.get("cc", ""),
            "date": when.astimezone(_TZ).isoformat(),
            "date_human": _human(when),
            "outgoing": bool(m.get("outgoing")) or addr.lower() in mine,
            "body_html": html, "body_text": text,
            "images_blocked": blocked,
            "attachments": atts,
        })
    return {"thread_id": thread_id, "account": st["account"],
            "subject": raw.get("subject", "(no subject)"), "messages": msgs}


def emit_thread(thread_id: str) -> dict:
    p = payload(thread_id)
    emit("email_thread", thread_id=p["thread_id"], account=p["account"],
         subject=p["subject"], messages=p["messages"])
    return p


def _fetch(account: str, ident: str) -> dict:
    """Pull a thread from whichever backend owns this account."""
    if account == "business":
        from voice import email_control
        return email_control.thread(ident)
    from voice import gmail_control
    return gmail_control.thread(account, ident)


def open_thread(account: str, ident: str) -> dict:
    """Fetch a thread, cache it, and put it on Ahmed's HUD. Returns the
    payload that went out."""
    raw = _fetch(account, ident)
    tid = raw.get("thread_id") or ident
    _THREADS[tid] = {"account": account, "raw": raw, "unblocked": set(),
                     "paths": {}}
    if tid not in _ORDER:
        _ORDER.append(tid)
    return emit_thread(tid)


def _summary(p: dict) -> str:
    """The one line Jarvis speaks after the card lands."""
    msgs = p["messages"]
    n = len(msgs)
    atts = sum(len(m["attachments"]) for m in msgs)
    # name the CORRESPONDENT, not Ahmed — his own reply is usually the last one
    who = next((m["from_name"] for m in reversed(msgs) if not m["outgoing"]),
               msgs[-1]["from_name"] if msgs else "?")
    bits = [f"{n} message{'s' if n != 1 else ''}"]
    if atts:
        bits.append(f"{atts} attachment{'s' if atts != 1 else ''}")
    return (f"Pulled up the '{p['subject']}' thread with {who} on Ahmed's HUD "
            f"— {', '.join(bits)}. He's reading it on the card.")


def show(query: str, account: str = "") -> str:
    """Newest thread matching `query` -> card. Returns Jarvis' line."""
    acct = _account(account)
    if acct == "business":
        from voice import email_control
        ids = email_control.search_uids(query, 1)
    else:
        from voice import gmail_control
        ids = gmail_control.search_ids(acct, query, 1)
    if not ids:
        return f"No message matching '{query}' in the {acct} mailbox."
    return _summary(open_thread(acct, ids[0]))


def show_id(msg_id: str, account: str = "") -> str:
    """Card for an explicit message/thread id (chains off gmail_search)."""
    return _summary(open_thread(_account(account), msg_id))


def latest_id() -> str | None:
    return _ORDER[-1] if _ORDER else None


def drop(thread_id: str = "") -> None:
    """Forget the thread engine-side. Used when the HUD already closed the card
    itself (Ahmed clicked ✕) — no event, the card is gone already."""
    tid = thread_id or latest_id() or ""
    if not tid:
        return
    _THREADS.pop(tid, None)
    if tid in _ORDER:
        _ORDER.remove(tid)


def close(thread_id: str = "") -> None:
    """Drop the thread AND tell the HUD to dismiss the card (Jarvis closing it
    on Ahmed's word — `mail_close`)."""
    tid = thread_id or latest_id() or ""
    if not tid:
        return
    drop(tid)
    emit("card_close", draft_id=tid)


# --- HUD actions (driven by ControlWatcher._email_action) -------------------

def _acct_of(thread_id: str, account: str) -> str:
    st = _THREADS.get(thread_id)
    return st["account"] if st else _account(account)


def download(thread_id: str, msg_id: str, att_id: str, filename: str,
             account: str = "") -> str:
    """Ahmed clicked an attachment chip: fetch it, remember the path (so a
    re-emit keeps it), and tell the HUD where it landed."""
    acct = _acct_of(thread_id, account)
    try:
        if acct == "business":
            from voice import email_control
            path = email_control.attachment(msg_id, att_id, filename, thread_id)
        else:
            from voice import gmail_control
            path = gmail_control.attachment(acct, msg_id, att_id, filename,
                                            thread_id)
    except Exception as e:  # noqa: BLE001 — a bad chip must not kill the card
        emit("email_attachment", thread_id=thread_id, msg_id=msg_id,
             att_id=att_id, path="", ok=False, error=str(e)[:200])
        raise
    st = _THREADS.get(thread_id)
    if st:
        st["paths"][(msg_id, att_id)] = path
    emit("email_attachment", thread_id=thread_id, msg_id=msg_id, att_id=att_id,
         path=path, ok=True, error="")
    return path


def open_path(path: str) -> str:
    """Open a downloaded attachment in its default app (Preview, etc.)."""
    p = str(path or "").strip()
    if not p or not os.path.isfile(p):
        return ""
    try:
        if sys.platform == "darwin":
            subprocess.Popen(["open", p])
        elif os.name == "nt":
            os.startfile(p)  # type: ignore[attr-defined]  # noqa: S606
        else:
            subprocess.Popen(["xdg-open", p])
    except Exception as e:  # noqa: BLE001 — a failed open must not crash the app
        print(f"  [mail] open failed: {str(e)[:120]}")
    return p


def load_images(thread_id: str, msg_id: str) -> None:
    """Ahmed hit "Load images" on one message — re-emit the thread with that
    message's remote images restored (the rest stay blocked)."""
    st = _THREADS.get(thread_id)
    if not st:
        return
    st["unblocked"].add(msg_id)
    emit_thread(thread_id)


def reply(thread_id: str, msg_id: str, body: str, reply_all: bool = False,
          account: str = "") -> tuple[bool, str]:
    """Send Ahmed's typed reply on the thread. Threading headers are preserved
    by the backends (Gmail: threadId + In-Reply-To/References; IMAP: the same
    two headers). Returns (ok, human_summary)."""
    acct = _acct_of(thread_id, account)
    if not str(body).strip():
        return False, "the reply was empty."
    try:
        if acct == "business":
            from voice import email_control
            out = email_control.reply(msg_id, body, reply_all)
        else:
            from voice import gmail_control
            out = gmail_control._reply(acct, msg_id, body, reply_all)
        emit("email_reply_sent", thread_id=thread_id, ok=True, error="")
        return True, out
    except Exception as e:  # noqa: BLE001 — report the failure, keep the card
        emit("email_reply_sent", thread_id=thread_id, ok=False,
             error=str(e)[:200])
        return False, f"failed: {e}"


# --- MCP server -----------------------------------------------------------

def _text(msg: str) -> dict:
    return {"content": [{"type": "text", "text": msg}]}


def build_server():
    """Build the in-process ``mail`` MCP server."""
    import asyncio

    from claude_agent_sdk import tool, create_sdk_mcp_server

    @tool("mail_show",
          "Put an email THREAD on Ahmed's HUD as a readable card — use this "
          "whenever he wants to SEE/read/'pull up' an email (never open Gmail "
          "in a browser, he has a hard rule against it). `query` matches the "
          "sender's name/address or subject words (Gmail search syntax works "
          "on the work account). `account` = work (default) or business "
          "(ahmad@revalstudio.com). Shows the whole thread — every reply, the "
          "formatted body, the attachments — and he can reply from the card.",
          {"query": str, "account": str})
    async def mail_show(args: dict) -> dict:
        q = str(args.get("query", "")).strip()
        try:
            out = await asyncio.to_thread(
                show, q, str(args.get("account", "")).strip())
            return _text(out)
        except Exception as e:  # noqa: BLE001
            return _text(f"mail_show failed: {e}")

    @tool("mail_show_id",
          "Same as mail_show but by an explicit message/thread id — chain it "
          "off a gmail_search / email_search result when you already know "
          "which message he means." ,
          {"msg_id": str, "account": str})
    async def mail_show_id(args: dict) -> dict:
        mid = str(args.get("msg_id", "")).strip()
        if not mid:
            return _text("mail_show_id failed: no message id.")
        try:
            out = await asyncio.to_thread(
                show_id, mid, str(args.get("account", "")).strip())
            return _text(out)
        except Exception as e:  # noqa: BLE001
            return _text(f"mail_show_id failed: {e}")

    @tool("mail_close",
          "Close the email card on Ahmed's HUD ('close that', 'done with it').",
          {"thread_id": str})
    async def mail_close(args: dict) -> dict:
        tid = str(args.get("thread_id", "")).strip() or latest_id()
        if not tid:
            return _text("no email card is open.")
        close(tid)
        return _text("Closed the email card.")

    return create_sdk_mcp_server(
        name="mail", version="1.0.0",
        tools=[mail_show, mail_show_id, mail_close])