From 46b127b1f359f554b31375c4f618e81894081fe8 Mon Sep 17 00:00:00 2001 From: pewdiepie-archdaemon Date: Mon, 29 Jun 2026 14:59:25 +0000 Subject: [PATCH] Implement email auto translate cache task --- routes/email_helpers.py | 37 +++++- routes/email_routes.py | 62 +++++++++- src/builtin_actions.py | 249 ++++++++++++++++++++++++++++++++++++++-- 3 files changed, 335 insertions(+), 13 deletions(-) diff --git a/routes/email_helpers.py b/routes/email_helpers.py index a3bd99d3c..3796a6d85 100644 --- a/routes/email_helpers.py +++ b/routes/email_helpers.py @@ -424,12 +424,19 @@ SCHEDULED_DB = Path(SCHEDULED_EMAILS_DB) OWNER_SCOPED_EMAIL_CACHE_TABLES = { "email_summaries", "email_ai_replies", + "email_translations", "email_calendar_extractions", "email_urgency_alerts", "sender_signatures", } +def email_translation_body_hash(body: str) -> str: + import hashlib as _hashlib + normalized = (body or "").strip() + return _hashlib.sha256(normalized.encode("utf-8", errors="ignore")).hexdigest() + + def _email_cache_owner_clause(owner: str = "") -> tuple[str, tuple[str, ...]]: owner = (owner or "").strip() if owner: @@ -437,8 +444,15 @@ def _email_cache_owner_clause(owner: str = "") -> tuple[str, tuple[str, ...]]: return "(owner = '' OR owner IS NULL)", () -def _ensure_owner_scoped_email_cache_table(conn, table: str, create_sql: str, columns: list[str]): +def _ensure_owner_scoped_email_cache_table( + conn, + table: str, + create_sql: str, + columns: list[str], + pk_columns: list[str] | None = None, +): """Rebuild legacy Message-ID-only cache tables with owner in the PK.""" + desired_pk_cols = pk_columns or ["message_id", "owner"] conn.execute(create_sql) try: info = conn.execute(f"PRAGMA table_info({table})").fetchall() @@ -457,7 +471,7 @@ def _ensure_owner_scoped_email_cache_table(conn, table: str, create_sql: str, co else: conn.execute(f"ALTER TABLE {table} ADD COLUMN {col} TEXT") cols.append(col) - if "owner" in cols and pk_cols == ["message_id", "owner"]: + if "owner" in cols and pk_cols == desired_pk_cols: return conn.execute(f"ALTER TABLE {table} RENAME TO {table}__old") @@ -590,6 +604,25 @@ def _init_scheduled_db(): PRIMARY KEY (message_id, owner) ) """, ["message_id", "owner", "uid", "folder", "reply", "model_used", "created_at"]) + _ensure_owner_scoped_email_cache_table(conn, "email_translations", """ + CREATE TABLE IF NOT EXISTS email_translations ( + body_hash TEXT, + owner TEXT DEFAULT '', + target_language TEXT DEFAULT 'English', + uid TEXT, + folder TEXT, + subject TEXT, + sender TEXT, + translation TEXT, + same_language INTEGER DEFAULT 0, + model_used TEXT, + created_at TEXT NOT NULL, + PRIMARY KEY (body_hash, owner, target_language) + ) + """, [ + "body_hash", "owner", "target_language", "uid", "folder", "subject", "sender", + "translation", "same_language", "model_used", "created_at", + ], ["body_hash", "owner", "target_language"]) # Email tags / spam classification cache. SECURITY: keyed by # (message_id, owner) because Message-IDs are GLOBAL (a newsletter goes # to many users with the same Message-ID). Without owner-scoping, a diff --git a/routes/email_routes.py b/routes/email_routes.py index 23776313c..cdbdfe032 100644 --- a/routes/email_routes.py +++ b/routes/email_routes.py @@ -55,7 +55,7 @@ from routes.email_helpers import ( _friendly_email_auth_error, SendEmailRequest, ExtractStyleRequest, ATTACHMENTS_DIR, COMPOSE_UPLOADS_DIR, SCHEDULED_DB, - attachment_extract_dir, _email_cache_owner_clause, + attachment_extract_dir, _email_cache_owner_clause, email_translation_body_hash, ) from routes.email_pollers import _start_poller @@ -3971,6 +3971,36 @@ def setup_email_routes(): if not body: return {"success": False, "error": "No body provided"} + body_hash = email_translation_body_hash(body) + try: + _c = _sql3.connect(SCHEDULED_DB) + owner_clause, owner_params = _email_cache_owner_clause(owner) + row = _c.execute( + f"SELECT translation, same_language, model_used FROM email_translations " + f"WHERE body_hash = ? AND target_language = ? AND {owner_clause}", + (body_hash, target_language, *owner_params), + ).fetchone() + _c.close() + if row: + if int(row[1] or 0): + return { + "success": True, + "same_language": True, + "language": target_language, + "model_used": row[2] or "cached", + "cached": True, + } + if row[0]: + return { + "success": True, + "translation": row[0], + "language": target_language, + "model_used": row[2] or "cached", + "cached": True, + } + except Exception as e: + logger.warning(f"Failed to read email translation cache: {e}") + candidates = [] seen = set() @@ -4028,6 +4058,21 @@ def setup_email_routes(): content = (content or "").strip() content = _extract_reply(content) if "<<>>" in content: + try: + _c = _sql3.connect(SCHEDULED_DB) + _c.execute(""" + INSERT OR REPLACE INTO email_translations + (body_hash, owner, target_language, uid, folder, subject, sender, + translation, same_language, model_used, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, ( + body_hash, owner, target_language, data.get("uid", ""), data.get("folder", ""), + subject, sender, "", 1, model, datetime.utcnow().isoformat(), + )) + _c.commit() + _c.close() + except Exception as e: + logger.warning(f"Failed to cache same-language email translation: {e}") return {"success": True, "same_language": True, "language": target_language, "model_used": model} marker = re.search(r"<<>>\s*(.*?)\s*<<>>", content, re.S | re.I) if marker: @@ -4037,6 +4082,21 @@ def setup_email_routes(): content = re.sub(r"\s*<<>>\s*$", "", content, flags=re.I).strip() if not content: return {"success": False, "error": "Empty response from model"} + try: + _c = _sql3.connect(SCHEDULED_DB) + _c.execute(""" + INSERT OR REPLACE INTO email_translations + (body_hash, owner, target_language, uid, folder, subject, sender, + translation, same_language, model_used, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, ( + body_hash, owner, target_language, data.get("uid", ""), data.get("folder", ""), + subject, sender, content, 0, model, datetime.utcnow().isoformat(), + )) + _c.commit() + _c.close() + except Exception as e: + logger.warning(f"Failed to cache email translation: {e}") return {"success": True, "translation": content, "language": target_language, "model_used": model} except Exception as e: logger.error(f"Failed to translate email: {e}") diff --git a/src/builtin_actions.py b/src/builtin_actions.py index 5caa0a691..727d2699d 100644 --- a/src/builtin_actions.py +++ b/src/builtin_actions.py @@ -572,18 +572,247 @@ async def action_draft_email_replies(owner: str, **kwargs) -> Tuple[str, bool]: async def action_email_auto_translate(owner: str, **kwargs) -> Tuple[str, bool]: - """Placeholder for the opt-in email translation enrichment task. + """Detect recent foreign-language emails and cache translated text. - The task is seeded paused so it is visible in Tasks while the detector, - cache, and on-demand translation path are built. Returning False makes a - manual run show a clear message instead of pretending work happened. + The reader still shows the original body; it simply checks this cache + before calling the LLM on demand. Keep the scheduled pass deliberately + small so translation never turns into a mailbox-wide background crawl. """ - return ( - "Email auto-translate is registered but not implemented yet. " - "Next step: detect foreign-language emails, cache translations by body hash, " - "and stream translated text in the email reader without replacing the original.", - False, - ) + try: + import email as _email_mod + import json as _json + import re as _re + import sqlite3 as _sql3 + from datetime import datetime as _dt, timedelta as _td + + from core.database import EmailAccount as _EA, SessionLocal as _SL + from routes.email_helpers import ( + SCHEDULED_DB, + _decode_header, + _email_cache_owner_clause, + _extract_reply, + _extract_text, + _imap_connect, + email_translation_body_hash, + ) + from src.settings import load_settings + from src.task_endpoint import task_llm_call_async + + settings = load_settings() + if not settings.get("email_auto_translate", False): + raise TaskNoop("email auto-translate is disabled") + + target_language = (settings.get("email_translate_language") or "English").strip() or "English" + account_id = _email_task_account_id(kwargs) + days_back = 7 + max_process = 5 + try: + data = _json.loads((kwargs.get("prompt") or "").strip() or "{}") + if isinstance(data, dict): + days_back = max(1, min(30, int(data.get("days_back") or days_back))) + max_process = max(1, min(20, int(data.get("max_process") or max_process))) + except Exception: + pass + + db = _SL() + try: + from sqlalchemy import and_ as _and, or_ as _or + q = db.query(_EA).filter(_EA.enabled == True) # noqa: E712 + if owner: + unowned = _or(_EA.owner == None, _EA.owner == "") # noqa: E711 + same_mailbox = _or(_EA.imap_user == owner, _EA.from_address == owner) + q = q.filter(_or(_EA.owner == owner, _and(unowned, same_mailbox))) + if account_id: + q = q.filter(_EA.id == account_id) + accounts = q.all() + finally: + db.close() + if not accounts: + raise TaskNoop("no email accounts configured") + + def _cached(body_hash: str) -> bool: + c = _sql3.connect(SCHEDULED_DB) + try: + owner_clause, owner_params = _email_cache_owner_clause(owner) + row = c.execute( + f"SELECT 1 FROM email_translations " + f"WHERE body_hash = ? AND target_language = ? AND {owner_clause} LIMIT 1", + (body_hash, target_language, *owner_params), + ).fetchone() + return bool(row) + finally: + c.close() + + def _store( + body_hash: str, + *, + uid: str, + folder: str, + subject: str, + sender: str, + translation: str, + same_language: bool, + model_used: str, + ) -> None: + c = _sql3.connect(SCHEDULED_DB) + try: + c.execute(""" + INSERT OR REPLACE INTO email_translations + (body_hash, owner, target_language, uid, folder, subject, sender, + translation, same_language, model_used, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, ( + body_hash, owner, target_language, uid, folder, subject, sender, + translation, 1 if same_language else 0, model_used, _dt.utcnow().isoformat(), + )) + c.commit() + finally: + c.close() + + async def _translate(body: str, subject: str, sender: str) -> tuple[str, bool]: + content = await task_llm_call_async( + [ + { + "role": "system", + "content": ( + "You translate emails faithfully. Preserve meaning, names, dates, money, addresses, " + "bullet structure, and tone. Do not summarize or answer the email. " + "Output only the translation between <<>> and <<>>. " + "If the email is already primarily in the target language, output exactly " + "<<>>." + ), + }, + { + "role": "user", + "content": ( + f"Target language: {target_language}\n\n" + f"From: {sender}\nSubject: {subject}\n\n{body[:16000]}\n\n" + "Translate the email unless it is already primarily in the target language.\n" + "Return only:\n<<>>\ntranslated text\n<<>>" + ), + }, + ], + owner=owner, + temperature=0.2, + max_tokens=8192, + timeout=180, + ) + content = (content or "").strip() + content = _extract_reply(content) + if "<<>>" in content: + return "", True + marker = _re.search(r"<<>>\s*(.*?)\s*<<>>", content, _re.S | _re.I) + if marker: + content = marker.group(1).strip() + else: + content = _re.sub(r"^\s*<<>>\s*", "", content, flags=_re.I).strip() + content = _re.sub(r"\s*<<>>\s*$", "", content, flags=_re.I).strip() + return content, False + + since = (_dt.utcnow() - _td(days=days_back)).strftime("%d-%b-%Y") + examined = 0 + cached = 0 + translated = 0 + same_language = 0 + skipped = 0 + failures = 0 + processed = 0 + + for acct in accounts: + if processed >= max_process: + break + imap = None + try: + imap = _imap_connect(acct.id, owner=owner) + imap.select("INBOX", readonly=True) + status, data = imap.uid("SEARCH", None, f'(SINCE {since})') + if status != "OK" or not data or not data[0]: + continue + uids = list(reversed(data[0].split()))[:50] + for uid_b in uids: + if processed >= max_process: + break + uid = uid_b.decode("utf-8", errors="ignore") if isinstance(uid_b, bytes) else str(uid_b) + status, msg_data = imap.uid("FETCH", uid, "(RFC822)") + if status != "OK" or not msg_data: + continue + raw = None + for part in msg_data: + if isinstance(part, tuple) and len(part) > 1: + raw = part[1] + break + if not raw: + continue + msg = _email_mod.message_from_bytes(raw) + subject = _decode_header(msg.get("Subject", "")) + sender = _decode_header(msg.get("From", "")) + body = (_extract_text(msg) or "").strip() + examined += 1 + if len(body) < 80: + skipped += 1 + continue + body_hash = email_translation_body_hash(body) + if _cached(body_hash): + cached += 1 + continue + translation, is_same_language = await _translate(body, subject, sender) + if is_same_language: + _store( + body_hash, + uid=uid, + folder="INBOX", + subject=subject, + sender=sender, + translation="", + same_language=True, + model_used="background-task", + ) + same_language += 1 + processed += 1 + continue + if not translation: + failures += 1 + continue + _store( + body_hash, + uid=uid, + folder="INBOX", + subject=subject, + sender=sender, + translation=translation, + same_language=False, + model_used="background-task", + ) + translated += 1 + processed += 1 + except Exception as acct_e: + failures += 1 + logger.warning(f"email_auto_translate account scan failed for {getattr(acct, 'id', '?')}: {acct_e}") + finally: + if imap: + try: + imap.logout() + except Exception: + pass + + if translated == 0 and same_language == 0: + result = ( + f"no uncached foreign-language emails found " + f"(examined {examined}, cached {cached}, skipped {skipped}, failures {failures})" + ) + if failures: + return f"Email Auto Translate failed: {result}", False + raise TaskNoop(result) + return ( + f"Email Auto Translate cached {translated} translation(s), marked {same_language} same-language " + f"(examined {examined}, already cached {cached}, skipped {skipped}, failures {failures})", + True, + ) + except TaskNoop: + raise + except Exception as e: + logger.error(f"email_auto_translate action failed: {e}") + return str(e), False _TYPE_COLORS = {