"""Outlook voice-command handlers — Python parity for the Rust `outlook` pack. Talks to Outlook via the `win32com.client` COM bridge (pywin32). When pywin32 isn't installed or Outlook isn't running we degrade gracefully: every handler reports the failure through the speak callback instead of crashing. Exposes four handlers, mirroring the Rust pack: do_outlook_unread_count(action, voice) do_outlook_latest(action, voice) do_outlook_send_clipboard(action, voice) do_outlook_summarize_inbox(action, voice) """ import re import subprocess from typing import Optional # Lazy import — pywin32 is Windows-only and may not be installed on dev/CI. try: import win32com.client as _w32com # type: ignore _W32_OK = True except Exception: # ImportError, OSError, etc. _w32com = None _W32_OK = False _speak_fn = print def set_speak(fn): """Register the voice output callback (called by extensions.set_speak).""" global _speak_fn _speak_fn = fn def _speak(text: str) -> None: try: _speak_fn(text) except Exception as exc: print(f"[outlook] speak: {exc}") # ── Outlook session helpers ──────────────────────────────────────────────── def _outlook_session(): """Return (app, namespace) or (None, None) on any failure.""" if not _W32_OK: return None, None try: app = _w32com.Dispatch("Outlook.Application") ns = app.GetNamespace("MAPI") return app, ns except Exception as exc: print(f"[outlook] session: {exc}") return None, None def _unavailable_msg() -> str: return "Outlook недоступен — запусти Outlook сначала." def _strip_html(s: str) -> str: if not s: return "" s = re.sub(r"<[^>]+>", " ", s) s = (s.replace(" ", " ").replace("&", "&") .replace("<", "<").replace(">", ">") .replace(""", '"').replace("'", "'")) return re.sub(r"\s+", " ", s).strip() def _get_clipboard() -> str: """Fetch clipboard via PowerShell. Returns "" on failure.""" try: res = subprocess.run( ["powershell", "-NoProfile", "-Command", "Get-Clipboard -Raw"], capture_output=True, text=True, encoding="utf-8", timeout=10, ) return (res.stdout or "").strip() except Exception as exc: print(f"[outlook] clipboard: {exc}") return "" # ── handlers ─────────────────────────────────────────────────────────────── def do_outlook_unread_count(action, voice): _, ns = _outlook_session() if ns is None: _speak(_unavailable_msg()) return try: inbox = ns.GetDefaultFolder(6) # olFolderInbox # Prefer the cheap property; fall back to scanning items. try: count = int(inbox.UnReadItemCount) except Exception: count = sum(1 for it in inbox.Items if getattr(it, "UnRead", False)) except Exception as exc: print(f"[outlook] unread_count: {exc}") _speak(_unavailable_msg()) return if count == 0: _speak("Непрочитанных писем нет.") elif count == 1: _speak("Одно непрочитанное письмо.") else: _speak(f"Непрочитанных писем: {count}.") def do_outlook_latest(action, voice): _, ns = _outlook_session() if ns is None: _speak(_unavailable_msg()) return try: inbox = ns.GetDefaultFolder(6) items = inbox.Items items.Sort("[ReceivedTime]", True) latest = None for it in items: try: if it.Class == 43: # olMail latest = it break except Exception: continue if latest is None: _speak("Входящие пусты.") return from_name = ( getattr(latest, "SenderName", "") or getattr(latest, "SenderEmailAddress", "") or "неизвестного отправителя" ) subject = getattr(latest, "Subject", "") or "без темы" body = getattr(latest, "Body", "") or "" if not body: body = _strip_html(getattr(latest, "HTMLBody", "") or "") body = re.sub(r"\s+", " ", body).strip() if len(body) > 200: body = body[:200] + "..." except Exception as exc: print(f"[outlook] latest: {exc}") _speak(_unavailable_msg()) return _speak(f"Последнее письмо от {from_name}. Тема: {subject}. {body}") _SEND_TRIGGERS = ( "отправь письмо ", "отправь почту ", "напиши письмо ", "отправь это ", "email this to ", "send email to ", "send this email to ", "mail this to ", ) def _extract_recipient(voice: str) -> str: """Pull the recipient name off the voice command by stripping the trigger.""" if not voice: return "" low = voice.lower() for trig in _SEND_TRIGGERS: if trig in low: idx = low.find(trig) return voice[idx + len(trig):].strip() return voice.strip() def _resolve_recipient(ns, query: str) -> Optional[dict]: if not query or ns is None: return None try: r = ns.CreateRecipient(query) r.Resolve() if not r.Resolved: return None addr = None try: addr = r.AddressEntry.GetExchangeUser().PrimarySmtpAddress except Exception: pass if not addr: try: addr = r.AddressEntry.Address except Exception: pass if not addr: addr = r.Name return {"name": r.Name, "address": addr} except Exception as exc: print(f"[outlook] resolve: {exc}") return None def do_outlook_send_clipboard(action, voice): recipient = _extract_recipient(voice or "") if not recipient: _speak("Кому отправить? Назови имя.") return app, ns = _outlook_session() if ns is None or app is None: _speak(_unavailable_msg()) return resolved = _resolve_recipient(ns, recipient) if not resolved: _speak(f"Не нашёл контакт «{recipient}».") return body = _get_clipboard() first_line = (body.split("\n", 1)[0] if body else "").strip() if len(first_line) > 60: subject = first_line[:60].strip() else: subject = first_line if not subject: subject = "From J.A.R.V.I.S." try: mail = app.CreateItem(0) # olMailItem mail.To = resolved["address"] mail.Subject = subject mail.Body = body mail.Send() except Exception as exc: print(f"[outlook] send: {exc}") _speak("Не получилось отправить — Outlook отказал.") return _speak(f"Отправил {resolved['name']}. Тема: {subject}.") def _llm_summarize(emails) -> Optional[str]: """Call the configured LLM to summarise the email list. Returns None on any failure, callers should provide a fallback.""" try: import llm_backend except Exception: return None if not emails: return None listing = "\n".join( f"{i+1}. From: {m['from']} | Subject: {m['subject']}" for i, m in enumerate(emails) ) sys_prompt = ( "Ты помощник по почте. Кратко опиши, что лежит в инбоксе, ОДНИМ абзацем " "(2-4 предложения). Сгруппируй по темам/отправителям если возможно. " "Без преамбулы, без перечисления списком." ) user_prompt = "Непрочитанные письма:\n" + listing try: reply = llm_backend.complete( messages=[ {"role": "system", "content": sys_prompt}, {"role": "user", "content": user_prompt}, ], max_tokens=250, temperature=0.4, ) if reply and isinstance(reply, str) and reply.strip(): return reply.strip() except Exception as exc: print(f"[outlook] llm: {exc}") return None def do_outlook_summarize_inbox(action, voice, limit: int = 5): _, ns = _outlook_session() if ns is None: _speak(_unavailable_msg()) return try: inbox = ns.GetDefaultFolder(6) items = inbox.Items items.Sort("[ReceivedTime]", True) emails = [] for it in items: if len(emails) >= limit: break try: if it.Class != 43: continue if not getattr(it, "UnRead", False): continue except Exception: continue emails.append({ "from": getattr(it, "SenderName", "") or "", "subject": getattr(it, "Subject", "") or "", }) except Exception as exc: print(f"[outlook] summarize: {exc}") _speak(_unavailable_msg()) return if not emails: _speak("Непрочитанных писем нет.") return summary = _llm_summarize(emails) if not summary: # Fallback to a flat enumeration. parts = "; ".join(f"{m['from']} — {m['subject']}" for m in emails) summary = f"Непрочитанных: {len(emails)}. {parts}." _speak(summary)