Local Email Agent

#!/usr/bin/env python3 """Local web interface for the macOS email agent.""" from __future__ import annotations import json import threading import time import webbrowser from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import unquote, urlparse import email_agent HOST = "127.0.0.1" PORT = 8787 BASE_DIR = Path(__file__).resolve().parent JOBS: dict[str, dict[str, object]] = {} HTML = r""" Local Email Agent

Unread Parsed

-

Источники пока не загружены

Style Profile

-

Профиль пока не загружен

Drafts Today

-

drafts/YYYY-MM-DD

Критерии важности

P5
Срочно, deadline, critical, высокий риск.

P4
Нужно ответить, согласовать, approve/confirm.

P3
Встреча, созвон, календарь, рабочее обсуждение.

P2
Низкий приоритет, полезно посмотреть позже.

P1
FYI, рассылки, уведомления без действия.

Labels
urgent, needs_reply, meeting, finance_or_contract, newsletter, fyi.

Письма

Просмотр письма

full parsed body

Выбери письмо слева.

Подготовленные ответы

Просмотр ответа

markdown draft

Выбери ответ слева.

Структура

Выбери файл слева или кнопку выше.

Job Output

Пока задач не было.

""" class AppHandler(BaseHTTPRequestHandler): server_version = "EmailAgentUI/0.1" def do_GET(self) -> None: parsed = urlparse(self.path) if parsed.path == "/": self.send_html(HTML) elif parsed.path == "/api/state": self.send_json(build_state()) elif parsed.path.startswith("/api/job/"): job_id = parsed.path.rsplit("/", 1)[-1] self.send_json(JOBS.get(job_id, {"status": "missing", "output": "Job not found"})) elif parsed.path.startswith("/api/file/"): self.handle_file(parsed) else: self.send_error(HTTPStatus.NOT_FOUND) def do_POST(self) -> None: command = self.path.strip("/").split("/")[-1] handlers = { "run": run_daily, "summary": run_summary, "learn": run_learn, "digest": run_digest, "drafts": run_drafts, } if command not in handlers: self.send_error(HTTPStatus.NOT_FOUND) return job_id = start_job(command, handlers[command]) self.send_json({"job_id": job_id}) def handle_file(self, parsed) -> None: kind = parsed.path.rsplit("/", 1)[-1] query = parse_query(parsed.query) path = resolve_file(kind, query.get("path", "")) if not path or not path.exists(): self.send_json({"content": ""}) return self.send_json({"content": path.read_text(encoding="utf-8", errors="replace")[:60000]}) def send_html(self, body: str) -> None: payload = body.encode("utf-8") self.send_response(HTTPStatus.OK) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) def send_json(self, value: object) -> None: payload = json.dumps(value, ensure_ascii=False, indent=2).encode("utf-8") self.send_response(HTTPStatus.OK) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) def log_message(self, format: str, *args: object) -> None: return def start_job(name: str, func) -> str: job_id = f"{int(time.time())}-{name}" JOBS[job_id] = {"status": "running", "output": ""} def target() -> None: try: JOBS[job_id] = {"status": "ok", "output": func()} except Exception as exc: # UI should show failures instead of crashing the server. JOBS[job_id] = {"status": "error", "output": f"{type(exc).__name__}: {exc}"} threading.Thread(target=target, daemon=True).start() return job_id def run_daily() -> str: sent_messages, sent_errors = email_agent.fetch_messages("all", "sent", 100, email_agent.DEFAULT_CONTENT_LIMIT) sent_messages = email_agent.filter_by_identity(sent_messages, email_agent.DEFAULT_IDENTITY) profile = email_agent.learn_style_profile(sent_messages) feedback_profile = email_agent.learn_feedback_profile(BASE_DIR / email_agent.DEFAULT_FEEDBACK_DIR) profile = email_agent.merge_profiles(profile, feedback_profile) profile_path = BASE_DIR / email_agent.DEFAULT_PROFILE_PATH email_agent.save_style_profile(profile, profile_path) messages, inbox_errors = email_agent.fetch_messages("all", "unread-mailboxes", 200, email_agent.DEFAULT_CONTENT_LIMIT) messages = email_agent.dedupe_messages(messages) digest_path = BASE_DIR / email_agent.default_digest_path() summary_path = BASE_DIR / email_agent.default_summary_path() messages_path = BASE_DIR / email_agent.default_messages_path() contacts_path = BASE_DIR / email_agent.DEFAULT_CONTACTS_PATH calendar_path = BASE_DIR / email_agent.default_calendar_path() report_path = BASE_DIR / email_agent.default_report_path() drafts_dir = BASE_DIR / email_agent.default_drafts_dir() email_agent.clear_markdown_dir(drafts_dir) digest_path.parent.mkdir(parents=True, exist_ok=True) digest_path.write_text(email_agent.build_digest(messages), encoding="utf-8") summary_path.parent.mkdir(parents=True, exist_ok=True) summary_path.write_text(json.dumps(email_agent.summarize(messages), ensure_ascii=False, indent=2) + "\n", encoding="utf-8") email_agent.write_messages_jsonl(messages, messages_path) email_agent.update_contacts_memory(messages, contacts_path) meeting_messages = email_agent.write_calendar_suggestions(messages, calendar_path) drafts = email_agent.write_drafts(email_agent.prioritize_messages(messages)[:10], drafts_dir, "auto", "learned", profile) errors = sent_errors + inbox_errors report_path.parent.mkdir(parents=True, exist_ok=True) report_path.write_text(email_agent.render_daily_report(messages, digest_path, drafts, profile, errors), encoding="utf-8") return "\n".join( [ f"Profile: {profile_path}", f"Identity: {email_agent.DEFAULT_IDENTITY}", f"Approved feedback samples: {feedback_profile.samples}", f"Summary: {summary_path}", f"Messages: {messages_path}", f"Contacts: {contacts_path}", f"Calendar: {calendar_path} ({len(meeting_messages)} meeting emails)", f"Digest: {digest_path}", f"Drafts: {drafts_dir} ({len(drafts)} files)", f"Report: {report_path}", *[f"warning: {error}" for error in errors], ] ) def run_summary() -> str: messages, errors = email_agent.fetch_messages("all", "unread-mailboxes", 200, email_agent.DEFAULT_CONTENT_LIMIT) messages = email_agent.dedupe_messages(messages) summary = email_agent.summarize(messages) path = BASE_DIR / email_agent.default_summary_path() messages_path = BASE_DIR / email_agent.default_messages_path() path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") email_agent.write_messages_jsonl(messages, messages_path) return json.dumps(summary, ensure_ascii=False, indent=2) + format_errors(errors) def run_learn() -> str: messages, errors = email_agent.fetch_messages("all", "sent", 100, email_agent.DEFAULT_CONTENT_LIMIT) messages = email_agent.filter_by_identity(messages, email_agent.DEFAULT_IDENTITY) profile = email_agent.learn_style_profile(messages) path = BASE_DIR / email_agent.DEFAULT_PROFILE_PATH email_agent.save_style_profile(profile, path) return json.dumps(email_agent.asdict(profile), ensure_ascii=False, indent=2) + format_errors(errors) def run_digest() -> str: messages, errors = email_agent.fetch_messages("all", "unread-mailboxes", 200, email_agent.DEFAULT_CONTENT_LIMIT) messages = email_agent.dedupe_messages(messages) email_agent.write_messages_jsonl(messages, BASE_DIR / email_agent.default_messages_path()) digest = email_agent.build_digest(messages) path = BASE_DIR / email_agent.default_digest_path() path.parent.mkdir(parents=True, exist_ok=True) path.write_text(digest, encoding="utf-8") return f"Digest saved to {path}\n\n{digest}" + format_errors(errors) def run_drafts() -> str: messages, errors = email_agent.fetch_messages("all", "unread-mailboxes", 200, email_agent.DEFAULT_CONTENT_LIMIT) messages = email_agent.dedupe_messages(messages) email_agent.write_messages_jsonl(messages, BASE_DIR / email_agent.default_messages_path()) profile = email_agent.load_style_profile(BASE_DIR / email_agent.DEFAULT_PROFILE_PATH) drafts_dir = BASE_DIR / email_agent.default_drafts_dir() email_agent.clear_markdown_dir(drafts_dir) drafts = email_agent.write_drafts(email_agent.prioritize_messages(messages)[:10], drafts_dir, "auto", "learned", profile) return "\n".join(str(path) for path in drafts) + format_errors(errors) def format_errors(errors: list[str]) -> str: if not errors: return "" return "\n" + "\n".join(f"warning: {error}" for error in errors) def build_state() -> dict[str, object]: summary = read_json(BASE_DIR / email_agent.default_summary_path()) or read_json(BASE_DIR / "data/reports/2026-06-16-summary.json") profile = read_json(BASE_DIR / email_agent.DEFAULT_PROFILE_PATH) or read_json(BASE_DIR / "data/style_profile.json") drafts_dir = BASE_DIR / email_agent.default_drafts_dir() drafts = [] if drafts_dir.exists(): drafts = [ {"name": path.name, "path": str(path.relative_to(BASE_DIR))} for path in sorted(drafts_dir.glob("*.md")) ] messages = read_messages(BASE_DIR / email_agent.default_messages_path()) return { "summary": summary, "profile": profile, "drafts": drafts, "replies": read_reply_summaries(drafts_dir), "messages": messages, "structure": build_structure(), "paths": { "drafts_dir": str(drafts_dir.relative_to(BASE_DIR)), "digest": email_agent.default_digest_path(), "report": email_agent.default_report_path(), "messages": email_agent.default_messages_path(), }, "updated_at": time.strftime("%Y-%m-%d %H:%M:%S"), } def read_json(path: Path) -> object | None: if not path.exists(): return None def read_messages(path: Path) -> list[dict[str, object]]: if not path.exists(): return [] messages: list[dict[str, object]] = [] for index, line in enumerate(path.read_text(encoding="utf-8", errors="replace").splitlines(), start=1): if not line.strip(): continue try: row = json.loads(line) except json.JSONDecodeError: continue row["path"] = f"{email_agent.default_messages_path()}#{index}" messages.append(row) return messages def read_reply_summaries(drafts_dir: Path) -> list[dict[str, object]]: if not drafts_dir.exists(): return [] replies: list[dict[str, object]] = [] for path in sorted(drafts_dir.glob("*.md")): content = path.read_text(encoding="utf-8", errors="replace") replies.append( { "name": path.name, "path": str(path.relative_to(BASE_DIR)), "subject": extract_field(content, "Original subject"), "priority": extract_field(content, "Priority").removeprefix("P"), "labels": extract_field(content, "Labels"), "reply": extract_section(content, "## Suggested Reply", "## Original Message Preview")[:900], } ) return replies def extract_field(content: str, field: str) -> str: pattern = rf"^\s*{re_escape(field)}:\s*(.+?)\s*$" import re match = re.search(pattern, content, flags=re.MULTILINE) return match.group(1).strip() if match else "" def extract_section(content: str, start: str, end: str) -> str: start_index = content.find(start) if start_index == -1: return "" start_index += len(start) end_index = content.find(end, start_index) if end_index == -1: end_index = len(content) return content[start_index:end_index].strip() def re_escape(value: str) -> str: import re return re.escape(value) def build_structure() -> list[dict[str, str]]: roots = [BASE_DIR / "data", BASE_DIR / "drafts"] items: list[dict[str, str]] = [] for root in roots: if not root.exists(): continue for path in sorted(root.rglob("*"))[:120]: rel = str(path.relative_to(BASE_DIR)) items.append({"name": path.name + ("/" if path.is_dir() else ""), "path": rel, "kind": "dir" if path.is_dir() else "file"}) return items try: return json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None def resolve_file(kind: str, request_path: str) -> Path | None: if kind == "digest": return BASE_DIR / email_agent.default_digest_path() if kind == "report": return BASE_DIR / email_agent.default_report_path() if kind == "profile": return BASE_DIR / email_agent.DEFAULT_PROFILE_PATH if kind == "draft" and request_path: candidate = (BASE_DIR / request_path).resolve() if BASE_DIR in candidate.parents and candidate.suffix == ".md": return candidate if kind == "message" and request_path: path_part, _, line_no = request_path.partition("#") candidate = (BASE_DIR / path_part).resolve() if BASE_DIR in candidate.parents and candidate.suffix == ".jsonl" and candidate.exists(): try: index = int(line_no) line = candidate.read_text(encoding="utf-8", errors="replace").splitlines()[index - 1] row = json.loads(line) tmp = BASE_DIR / ".message-preview.md" tmp.write_text(render_message_preview(row), encoding="utf-8") return tmp except (ValueError, IndexError, json.JSONDecodeError): return None if kind == "any" and request_path: candidate = (BASE_DIR / request_path).resolve() if BASE_DIR in candidate.parents and candidate.is_file(): return candidate return None def render_message_preview(row: dict[str, object]) -> str: labels = ", ".join(row.get("labels", [])) if isinstance(row.get("labels"), list) else "" return "\n".join( [ "# Parsed Email", "", f"Source: {row.get('source', '')}", f"Sender: {row.get('sender', '')}", f"Subject: {row.get('subject', '')}", f"Received: {row.get('received_at', '')}", f"Priority: P{row.get('priority', '')}", f"Labels: {labels}", "", "## Body", "", str(row.get("body", ""))[:8000], ] ) def parse_query(query: str) -> dict[str, str]: result: dict[str, str] = {} for part in query.split("&"): if not part: continue key, _, value = part.partition("=") result[unquote(key)] = unquote(value) return result def main() -> int: server = ThreadingHTTPServer((HOST, PORT), AppHandler) url = f"http://{HOST}:{PORT}" print(f"Email Agent UI: {url}") webbrowser.open(url) try: server.serve_forever() except KeyboardInterrupt: print("\nStopping UI") finally: server.server_close() return 0 if __name__ == "__main__": raise SystemExit(main())