Implement email auto translate cache task

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

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 "<<<SAME_LANGUAGE>>>" 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"<<<TRANSLATION>>>\s*(.*?)\s*<<<END>>>", content, re.S | re.I)
             if marker:
@@ -4037,6 +4082,21 @@ def setup_email_routes():
                 content = re.sub(r"\s*<<<END>>>\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}")

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 <<<TRANSLATION>>> and <<<END>>>. "
+                            "If the email is already primarily in the target language, output exactly "
+                            "<<<SAME_LANGUAGE>>>."
+                        ),
+                    },
+                    {
+                        "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<<<TRANSLATION>>>\ntranslated text\n<<<END>>>"
+                        ),
+                    },
+                ],
+                owner=owner,
+                temperature=0.2,
+                max_tokens=8192,
+                timeout=180,
+            )
+            content = (content or "").strip()
+            content = _extract_reply(content)
+            if "<<<SAME_LANGUAGE>>>" in content:
+                return "", True
+            marker = _re.search(r"<<<TRANSLATION>>>\s*(.*?)\s*<<<END>>>", content, _re.S | _re.I)
+            if marker:
+                content = marker.group(1).strip()
+            else:
+                content = _re.sub(r"^\s*<<<TRANSLATION>>>\s*", "", content, flags=_re.I).strip()
+                content = _re.sub(r"\s*<<<END>>>\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 = {