feat: Telegram 实时推送功能（TODO-00007）

P0 核心功能：
- DB Schema v3→v4：accounts 表新增 telegram_push_enabled、telegram_last_checked_at 列
- Repository：toggle_telegram_push、update_telegram_cursor、get_telegram_push_accounts
- Service：telegram_push.py — Graph/IMAP 新邮件抓取 + Telegram Bot API 推送
- Scheduler：telegram_push_job 定时轮询（默认 600s，可配置 60-3600s）
- Controller/Route：POST /api/accounts/<id>/telegram-toggle、settings GET/PUT 扩展
- Frontend：账号卡片🔔推送开关、设置页 Telegram 配置区

P1 增强功能：
- POST /api/settings/telegram-test 发送测试消息端点
- 保存 telegram_poll_interval 时动态重调度（无需重启）
- accounts API 返回 telegram_push_enabled 字段

测试：33 项 Telegram 推送测试全部通过，202 项总测试零回归

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

outlook_web/controllers/accounts.py

@@ -121,6 +121,7 @@ def api_get_accounts() -> Any:
                 "created_at": acc.get("created_at", ""),
                 "updated_at": acc.get("updated_at", ""),
                 "tags": acc.get("tags", []),
+                "telegram_push_enabled": bool(acc.get("telegram_push_enabled")),
             }
         )
     return jsonify({"success": True, "accounts": safe_accounts})
@@ -1871,3 +1872,19 @@ def api_get_refresh_stats() -> Any:
             },
         }
     )
+
+
+# ==================== Telegram 推送 API ====================
+
+
+@login_required
+def api_telegram_toggle(account_id: int) -> Any:
+    """切换账号 Telegram 推送开关"""
+    data = request.get_json(silent=True) or {}
+    enabled = bool(data.get("enabled", False))
+    success = accounts_repo.toggle_telegram_push(account_id, enabled)
+    if not success:
+        return jsonify({"success": False, "message": "账号不存在"}), 404
+    action = "开启" if enabled else "关闭"
+    log_audit(f"telegram_push_{action}", "account", str(account_id))
+    return jsonify({"success": True, "enabled": enabled, "message": f"Telegram推送已{action}"})

outlook_web/controllers/settings.py

@@ -9,7 +9,7 @@ from flask import jsonify, request
 from outlook_web.audit import log_audit
 from outlook_web.repositories import settings as settings_repo
 from outlook_web.security.auth import login_required
-from outlook_web.security.crypto import hash_password
+from outlook_web.security.crypto import decrypt_data, encrypt_data, hash_password, is_encrypted
 
 # ==================== 设置 API ====================
 
@@ -47,7 +47,26 @@ def api_get_settings() -> Any:
     safe_settings["gptmail_api_key_set"] = bool(gptmail_api_key_value)
     safe_settings["gptmail_api_key_masked"] = mask_secret_value(gptmail_api_key_value) if gptmail_api_key_value else ""
 
-    return jsonify({"success": True, "settings": safe_settings})
+    # Telegram 推送配置
+    tg_bot_token_raw = all_settings.get("telegram_bot_token", "")
+    if tg_bot_token_raw and is_encrypted(tg_bot_token_raw):
+        try:
+            plain_token = decrypt_data(tg_bot_token_raw)
+            safe_settings["telegram_bot_token"] = "****" + plain_token[-4:] if len(plain_token) > 4 else "****"
+        except Exception:
+            safe_settings["telegram_bot_token"] = "****"
+    else:
+        safe_settings["telegram_bot_token"] = ""
+    safe_settings["telegram_chat_id"] = all_settings.get("telegram_chat_id", "")
+    safe_settings["telegram_poll_interval"] = int(all_settings.get("telegram_poll_interval", "600") or "600")
+
+    response = {"success": True, "settings": safe_settings}
+    # 同时在顶层暴露 telegram 字段（兼容前端直接访问）
+    response["telegram_bot_token"] = safe_settings.get("telegram_bot_token", "")
+    response["telegram_chat_id"] = safe_settings.get("telegram_chat_id", "")
+    response["telegram_poll_interval"] = safe_settings.get("telegram_poll_interval", 600)
+
+    return jsonify(response)
 
 
 @login_required
@@ -190,6 +209,42 @@ def api_update_settings() -> Any:
         except ValueError:
             errors.append("轮询次数必须是数字")
 
+    # Telegram 推送配置
+    if "telegram_poll_interval" in data:
+        try:
+            tg_interval = int(data["telegram_poll_interval"])
+            if tg_interval < 60 or tg_interval > 3600:
+                errors.append("Telegram 轮询间隔必须在 60-3600 秒之间")
+            elif settings_repo.set_setting("telegram_poll_interval", str(tg_interval)):
+                updated.append("Telegram 轮询间隔")
+                scheduler_reload_needed = True
+            else:
+                errors.append("更新 Telegram 轮询间隔失败")
+        except (ValueError, TypeError):
+            errors.append("Telegram 轮询间隔必须是数字")
+
+    if "telegram_bot_token" in data:
+        tg_token = str(data["telegram_bot_token"]).strip()
+        if tg_token and not tg_token.startswith("****"):
+            encrypted_token = encrypt_data(tg_token)
+            if settings_repo.set_setting("telegram_bot_token", encrypted_token):
+                updated.append("Telegram Bot Token")
+            else:
+                errors.append("更新 Telegram Bot Token 失败")
+        elif not tg_token:
+            if settings_repo.set_setting("telegram_bot_token", ""):
+                updated.append("Telegram Bot Token（已清空）")
+        else:
+            # 脱敏占位符（****xxx），跳过不覆盖
+            updated.append("Telegram Bot Token（未变更）")
+
+    if "telegram_chat_id" in data:
+        tg_chat_id = str(data["telegram_chat_id"]).strip()
+        if settings_repo.set_setting("telegram_chat_id", tg_chat_id):
+            updated.append("Telegram Chat ID")
+        else:
+            errors.append("更新 Telegram Chat ID 失败")
+
     if errors:
         return jsonify({"success": False, "error": "；".join(errors)})
 
@@ -269,3 +324,24 @@ def api_validate_cron() -> Any:
         )
     except Exception as e:
         return jsonify({"success": False, "valid": False, "error": f"Cron 表达式无效: {str(e)}"})
+
+
+@login_required
+def api_test_telegram() -> Any:
+    """发送 Telegram 测试消息，验证 bot_token + chat_id 配置是否正确"""
+    from outlook_web.services.telegram_push import _send_telegram_message
+
+    bot_token_raw = settings_repo.get_setting("telegram_bot_token", "")
+    chat_id = settings_repo.get_setting("telegram_chat_id", "")
+
+    if not bot_token_raw or not chat_id:
+        return jsonify({"success": False, "error": "请先配置 Telegram Bot Token 和 Chat ID"})
+
+    bot_token = decrypt_data(bot_token_raw) if is_encrypted(bot_token_raw) else bot_token_raw
+
+    ok = _send_telegram_message(bot_token, chat_id, "✅ Outlook Email Plus 测试消息：配置正确！")
+    if ok:
+        log_audit("telegram_test", "settings", None, "测试消息发送成功")
+        return jsonify({"success": True, "message": "测试消息已发送，请检查 Telegram"})
+    else:
+        return jsonify({"success": False, "error": "发送失败，请检查 Bot Token 和 Chat ID 是否正确"})

outlook_web/db.py

@@ -18,7 +18,8 @@ from outlook_web.security.crypto import (
 
 # 数据库 Schema 版本（用于升级可验证/可诊断）
 # v3：对齐 PRD-00005 / FD-00005 / TDD-00005（accounts 表新增多邮箱字段：account_type/provider/imap_host/imap_port/imap_password）
-DB_SCHEMA_VERSION = 3
+# v4：对齐 PRD-00007 / FD-00007（accounts 表新增 telegram_push_enabled/telegram_last_checked_at）
+DB_SCHEMA_VERSION = 4
 DB_SCHEMA_VERSION_KEY = "db_schema_version"
 DB_SCHEMA_LAST_UPGRADE_TRACE_ID_KEY = "db_schema_last_upgrade_trace_id"
 DB_SCHEMA_LAST_UPGRADE_ERROR_KEY = "db_schema_last_upgrade_error"
@@ -396,6 +397,10 @@ def init_db(database_path: Optional[str] = None):
             cursor.execute("ALTER TABLE accounts ADD COLUMN imap_port INTEGER DEFAULT 993")
         if "imap_password" not in columns:
             cursor.execute("ALTER TABLE accounts ADD COLUMN imap_password TEXT")
+        if "telegram_push_enabled" not in columns:
+            cursor.execute("ALTER TABLE accounts ADD COLUMN telegram_push_enabled INTEGER NOT NULL DEFAULT 0")
+        if "telegram_last_checked_at" not in columns:
+            cursor.execute("ALTER TABLE accounts ADD COLUMN telegram_last_checked_at TEXT DEFAULT NULL")
 
         cursor.execute("PRAGMA table_info(groups)")
         group_columns = [col[1] for col in cursor.fetchall()]

outlook_web/repositories/accounts.py

@@ -367,3 +367,50 @@ def update_account_credentials(account_id: int, **fields) -> bool:
         return True
     except Exception:
         return False
+
+
+def toggle_telegram_push(account_id: int, enabled: bool) -> bool:
+    """切换账号 Telegram 推送开关。首次开启时自动设置游标为当前 UTC 时间。"""
+    from datetime import datetime, timezone
+
+    db = get_db()
+    row = db.execute("SELECT id, telegram_last_checked_at FROM accounts WHERE id = ?", (account_id,)).fetchone()
+    if not row:
+        return False
+
+    if enabled:
+        if row["telegram_last_checked_at"] is None:
+            now_utc = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
+            db.execute(
+                "UPDATE accounts SET telegram_push_enabled = 1, telegram_last_checked_at = ? WHERE id = ?",
+                (now_utc, account_id),
+            )
+        else:
+            db.execute("UPDATE accounts SET telegram_push_enabled = 1 WHERE id = ?", (account_id,))
+    else:
+        db.execute("UPDATE accounts SET telegram_push_enabled = 0 WHERE id = ?", (account_id,))
+
+    db.commit()
+    return True
+
+
+def update_telegram_cursor(account_id: int, checked_at: str) -> None:
+    """更新账号的 telegram_last_checked_at 游标。"""
+    db = get_db()
+    db.execute(
+        "UPDATE accounts SET telegram_last_checked_at = ? WHERE id = ?",
+        (checked_at, account_id),
+    )
+    db.commit()
+
+
+def get_telegram_push_accounts() -> List[Dict]:
+    """返回所有 telegram_push_enabled=1 且非 disabled 状态的账号。"""
+    db = get_db()
+    rows = db.execute(
+        """SELECT id, email, provider, refresh_token, imap_host, imap_port,
+                  imap_password, telegram_last_checked_at
+           FROM accounts
+           WHERE telegram_push_enabled = 1 AND status != 'disabled'"""
+    ).fetchall()
+    return [dict(r) for r in rows]

outlook_web/routes/accounts.py

@@ -125,4 +125,11 @@ def create_blueprint() -> Blueprint:
         methods=["GET"],
     )
 
+    # Telegram 推送开关
+    bp.add_url_rule(
+        "/api/accounts/<int:account_id>/telegram-toggle",
+        view_func=accounts_controller.api_telegram_toggle,
+        methods=["POST"],
+    )
+
     return bp

outlook_web/routes/settings.py

@@ -13,6 +13,11 @@ def create_blueprint() -> Blueprint:
         view_func=settings_controller.api_validate_cron,
         methods=["POST"],
     )
+    bp.add_url_rule(
+        "/api/settings/telegram-test",
+        view_func=settings_controller.api_test_telegram,
+        methods=["POST"],
+    )
     bp.add_url_rule("/api/settings", view_func=settings_controller.api_get_settings, methods=["GET"])
     bp.add_url_rule(
         "/api/settings",

outlook_web/services/scheduler.py

@@ -60,6 +60,40 @@ def scheduler_heartbeat_task():
         pass
 
 
+def _get_telegram_interval(app) -> int:
+    """读取 telegram_poll_interval 设置，默认 600 秒，最小 60 秒。"""
+    try:
+        with app.app_context():
+            val = settings_repo.get_setting("telegram_poll_interval", "600")
+        return max(60, int(val or "600"))
+    except Exception:
+        return 600
+
+
+def _configure_telegram_push_job(scheduler, app) -> None:
+    """注册/更新 Telegram 推送 Job。"""
+    from outlook_web.services.telegram_push import run_telegram_push_job
+
+    try:
+        scheduler.remove_job("telegram_push_job")
+    except Exception:
+        pass
+
+    interval = _get_telegram_interval(app)
+    scheduler.add_job(
+        func=run_telegram_push_job,
+        args=[app],
+        trigger="interval",
+        seconds=interval,
+        id="telegram_push_job",
+        name="Telegram 推送",
+        replace_existing=True,
+        max_instances=1,
+        coalesce=True,
+        misfire_grace_time=interval,
+    )
+
+
 def configure_scheduler_jobs(scheduler, app, test_refresh_token) -> None:
     """根据当前 settings 重新配置定时刷新 Job（配置变更即时生效）"""
     try:
@@ -90,6 +124,9 @@ def configure_scheduler_jobs(scheduler, app, test_refresh_token) -> None:
         misfire_grace_time=60,
     )
 
+    # Telegram 推送 Job
+    _configure_telegram_push_job(scheduler, app)
+
     # 刷新 Job：根据 enable_scheduled 决定是否启用
     try:
         scheduler.remove_job("token_refresh")

outlook_web/services/telegram_push.py

@@ -0,0 +1,311 @@
+"""outlook_web/services/telegram_push.py — Telegram 实时推送核心服务
+
+轮询 IMAP/Graph 读取新邮件 → 构造消息 → 调用 Telegram Bot API → 丢弃内容，仅更新游标。
+"""
+from __future__ import annotations
+
+import logging
+import re
+from datetime import datetime, timezone
+from typing import List
+
+import requests
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# 文本工具
+# ---------------------------------------------------------------------------
+
+MAX_TELEGRAM_LENGTH = 4096
+MAX_PREVIEW_LENGTH = 200
+MAX_EMAILS_PER_FETCH = 50
+MAX_SENT_PER_JOB = 20
+
+
+def _escape_html(text: str) -> str:
+    """转义 Telegram HTML 模式必须转义的三种字符：& < >"""
+    return text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
+
+
+def _html_to_plain(html_str: str) -> str:
+    """将 HTML 正文提取为纯文本（strip tags），合并多余空白。"""
+    if not html_str:
+        return ""
+    text = re.sub(r"<[^>]+>", " ", html_str)
+    text = re.sub(r"\s+", " ", text).strip()
+    return text
+
+
+def _build_telegram_message(account_email: str, email: dict) -> str:
+    """构造 Telegram HTML 消息文本（PRD §3.4 格式）。"""
+    subject = _escape_html(email.get("subject", ""))
+    sender = _escape_html(email.get("sender", ""))
+    received_at = email.get("received_at", "")
+    preview = email.get("preview", "")
+
+    lines = [
+        "📬 新邮件通知",
+        "",
+        f"账户：{_escape_html(account_email)}",
+        f"发件人：{sender}",
+        f"主题：{subject}",
+        f"时间：{received_at}",
+    ]
+
+    if preview:
+        truncated = preview[:MAX_PREVIEW_LENGTH]
+        if len(preview) > MAX_PREVIEW_LENGTH:
+            truncated += "..."
+        lines.append("")
+        lines.append(f"内容预览：\n{_escape_html(truncated)}")
+
+    msg = "\n".join(lines)
+
+    if len(msg) > MAX_TELEGRAM_LENGTH:
+        msg = msg[: MAX_TELEGRAM_LENGTH - 3] + "..."
+
+    return msg
+
+
+# ---------------------------------------------------------------------------
+# Telegram API
+# ---------------------------------------------------------------------------
+
+
+def _send_telegram_message(bot_token: str, chat_id: str, text: str) -> bool:
+    """调用 Telegram sendMessage API。超时 10 秒，失败返回 False。"""
+    url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
+    try:
+        resp = requests.post(
+            url,
+            json={
+                "chat_id": chat_id,
+                "text": text,
+                "parse_mode": "HTML",
+                "disable_web_page_preview": True,
+            },
+            timeout=10,
+        )
+        return resp.ok
+    except Exception as e:
+        logger.warning("[telegram_push] send failed: %s", e)
+        return False
+
+
+# ---------------------------------------------------------------------------
+# 邮件拉取（IMAP / Graph）
+# ---------------------------------------------------------------------------
+
+
+def _fetch_new_emails_imap(account: dict, since: str) -> List[dict]:
+    """通过 IMAP 获取 received_at > since 的邮件，最多返回 50 封。"""
+    import email as email_lib
+    import email.header
+    import imaplib
+    from datetime import datetime as dt
+
+    from outlook_web.security.crypto import decrypt_data
+
+    host = account.get("imap_host", "")
+    port = int(account.get("imap_port", 993))
+    password_raw = account.get("imap_password", "")
+    password = decrypt_data(password_raw) if password_raw else ""
+    user = account.get("email", "")
+
+    since_dt = dt.fromisoformat(since)
+    since_date_str = since_dt.strftime("%d-%b-%Y")
+
+    results: List[dict] = []
+    conn = None
+    try:
+        conn = imaplib.IMAP4_SSL(host, port, timeout=15)
+        conn.login(user, password)
+        conn.select("INBOX", readonly=True)
+
+        _, data = conn.search(None, f'(SINCE "{since_date_str}")')
+        msg_ids = data[0].split() if data[0] else []
+
+        for mid in msg_ids[-MAX_EMAILS_PER_FETCH:]:
+            try:
+                _, msg_data = conn.fetch(mid, "(RFC822)")
+                raw = msg_data[0][1]
+                msg = email_lib.message_from_bytes(raw)
+
+                subject_parts = email.header.decode_header(msg.get("Subject", ""))
+                subject = "".join(
+                    part.decode(charset or "utf-8") if isinstance(part, bytes) else part
+                    for part, charset in subject_parts
+                )
+
+                sender = msg.get("From", "")
+                date_str = msg.get("Date", "")
+                try:
+                    from email.utils import parsedate_to_datetime
+
+                    received_dt = parsedate_to_datetime(date_str)
+                    received_iso = received_dt.strftime("%Y-%m-%dT%H:%M:%S")
+                except Exception:
+                    received_iso = date_str
+
+                if received_iso <= since:
+                    continue
+
+                body = ""
+                if msg.is_multipart():
+                    for part in msg.walk():
+                        ct = part.get_content_type()
+                        if ct == "text/plain":
+                            payload = part.get_payload(decode=True)
+                            if payload:
+                                body = payload.decode(part.get_content_charset() or "utf-8", errors="replace")
+                            break
+                        elif ct == "text/html" and not body:
+                            payload = part.get_payload(decode=True)
+                            if payload:
+                                body = _html_to_plain(
+                                    payload.decode(part.get_content_charset() or "utf-8", errors="replace")
+                                )
+                else:
+                    payload = msg.get_payload(decode=True)
+                    if payload:
+                        charset = msg.get_content_charset() or "utf-8"
+                        raw_body = payload.decode(charset, errors="replace")
+                        if msg.get_content_type() == "text/html":
+                            body = _html_to_plain(raw_body)
+                        else:
+                            body = raw_body
+
+                preview = body[:MAX_PREVIEW_LENGTH] if body else ""
+
+                results.append(
+                    {
+                        "subject": subject,
+                        "sender": sender,
+                        "received_at": received_iso,
+                        "preview": preview,
+                    }
+                )
+            except Exception:
+                continue
+
+    except Exception as e:
+        logger.warning("[telegram_push] IMAP fetch error for %s: %s", account.get("email"), e)
+        raise
+    finally:
+        if conn:
+            try:
+                conn.logout()
+            except Exception:
+                pass
+
+    return results[:MAX_EMAILS_PER_FETCH]
+
+
+def _fetch_new_emails_graph(account: dict, since: str) -> List[dict]:
+    """通过 Microsoft Graph API 获取 received_at > since 的邮件，最多返回 50 封。"""
+    from outlook_web.security.crypto import decrypt_data
+    from outlook_web.services.graph import get_access_token
+
+    refresh_token_raw = account.get("refresh_token", "")
+    refresh_token = decrypt_data(refresh_token_raw) if refresh_token_raw else ""
+
+    access_token = get_access_token(refresh_token)
+    if not access_token:
+        return []
+
+    since_z = since if since.endswith("Z") else since + "Z"
+    url = "https://graph.microsoft.com/v1.0/me/messages"
+    params = {
+        "$filter": f"receivedDateTime gt {since_z}",
+        "$top": MAX_EMAILS_PER_FETCH,
+        "$select": "subject,from,receivedDateTime,bodyPreview",
+        "$orderby": "receivedDateTime asc",
+    }
+    headers = {"Authorization": f"Bearer {access_token}"}
+
+    results: List[dict] = []
+    try:
+        resp = requests.get(url, headers=headers, params=params, timeout=15)
+        if not resp.ok:
+            return []
+        data = resp.json()
+        for item in data.get("value", []):
+            sender_info = item.get("from", {}).get("emailAddress", {})
+            sender = sender_info.get("address", sender_info.get("name", ""))
+            received_raw = item.get("receivedDateTime", "")
+            received_iso = received_raw.replace("Z", "").split(".")[0] if received_raw else ""
+            preview = (item.get("bodyPreview", "") or "")[:MAX_PREVIEW_LENGTH]
+            results.append(
+                {
+                    "subject": item.get("subject", ""),
+                    "sender": sender,
+                    "received_at": received_iso,
+                    "preview": preview,
+                }
+            )
+    except Exception as e:
+        logger.warning("[telegram_push] Graph fetch error for %s: %s", account.get("email"), e)
+        raise
+
+    return results
+
+
+# ---------------------------------------------------------------------------
+# 主入口（调度器调用）
+# ---------------------------------------------------------------------------
+
+
+def run_telegram_push_job(app) -> None:
+    """主入口：轮询 → 推送 → 更新游标。由调度器调用。"""
+    with app.app_context():
+        from outlook_web.repositories.accounts import (
+            get_telegram_push_accounts,
+            update_telegram_cursor,
+        )
+        from outlook_web.repositories.settings import get_setting
+        from outlook_web.security.crypto import decrypt_data, is_encrypted
+
+        bot_token_raw = get_setting("telegram_bot_token", "")
+        bot_token = decrypt_data(bot_token_raw) if bot_token_raw and is_encrypted(bot_token_raw) else bot_token_raw
+        chat_id = get_setting("telegram_chat_id", "")
+
+        if not bot_token or not chat_id:
+            return
+
+        accounts = get_telegram_push_accounts()
+        if not accounts:
+            return
+
+        job_start_time = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
+        sent_count = 0
+
+        for account in accounts:
+            if sent_count >= MAX_SENT_PER_JOB:
+                update_telegram_cursor(account["id"], job_start_time)
+                continue
+
+            last_checked = account.get("telegram_last_checked_at")
+
+            if last_checked is None:
+                update_telegram_cursor(account["id"], job_start_time)
+                continue
+
+            try:
+                if account.get("provider") == "outlook":
+                    emails = _fetch_new_emails_graph(account, last_checked)
+                else:
+                    emails = _fetch_new_emails_imap(account, last_checked)
+
+                for em in sorted(emails, key=lambda e: e.get("received_at", "")):
+                    if sent_count >= MAX_SENT_PER_JOB:
+                        break
+                    msg = _build_telegram_message(account["email"], em)
+                    _send_telegram_message(bot_token, chat_id, msg)
+                    sent_count += 1
+
+            except Exception as e:
+                logger.warning("[telegram_push] account=%s error: %s", account.get("email"), e)
+
+            finally:
+                update_telegram_cursor(account["id"], job_start_time)

static/css/main.css

@@ -313,6 +313,7 @@ input, textarea, select { font-family: inherit; }
 .btn-sm { padding: 0.3rem 0.7rem; font-size: 0.78rem; }
 .btn-icon { padding: 0.4rem; font-size: 0.9rem; background: none; border: none; color: var(--text-muted); border-radius: var(--radius-sm); }
 .btn-icon:hover { background: var(--border-light); color: var(--text); }
+.btn-icon.tg-push-active { color: #0088cc; filter: drop-shadow(0 0 2px #0088cc66); }
 
 /* 表单 */
 .form-group { margin-bottom: 1.1rem; }

static/js/features/accounts.js

@@ -541,6 +541,29 @@
             }
         }
 
+        // 切换 Telegram 推送开关
+        async function toggleTelegramPush(accountId, enabled) {
+            try {
+                const response = await fetch(`/api/accounts/${accountId}/telegram-toggle`, {
+                    method: 'POST',
+                    headers: { 'Content-Type': 'application/json' },
+                    body: JSON.stringify({ enabled })
+                });
+                const data = await response.json();
+                if (data.success) {
+                    showToast(data.message || (enabled ? 'Telegram推送已开启' : 'Telegram推送已关闭'), 'success');
+                    if (currentGroupId) {
+                        delete accountsCache[currentGroupId];
+                        loadAccountsByGroup(currentGroupId, true);
+                    }
+                } else {
+                    handleApiError(data, 'Telegram推送切换失败');
+                }
+            } catch (error) {
+                showToast('操作失败', 'error');
+            }
+        }
+
         // 显示导出邮箱模态框
         async function showExportModal() {
             document.getElementById('exportModal').classList.add('show');

static/js/features/groups.js

@@ -271,6 +271,7 @@
                             ${isFailed ? `<button class="btn btn-sm btn-danger" onclick="event.stopPropagation(); showRefreshError(${acc.id}, '${escapeJs(acc.last_refresh_error || '未知错误')}', '${escapeJs(acc.email)}')" style="padding:1px 6px;font-size:0.65rem;">查看错误</button>` : ''}
                         </div>
                         <div class="account-actions">
+                            <button class="btn-icon ${acc.telegram_push_enabled ? 'tg-push-active' : ''}" onclick="event.stopPropagation(); toggleTelegramPush(${acc.id}, ${!acc.telegram_push_enabled})" title="Telegram推送${acc.telegram_push_enabled ? '(已开启)' : ''}">🔔</button>
                             <button class="btn btn-sm btn-accent" onclick="event.stopPropagation(); copyVerificationInfo('${escapeJs(acc.email)}', this)" title="提取验证码" style="font-size:0.72rem;padding:2px 8px;">🔑 验证码</button>
                             <button class="btn-icon" onclick="event.stopPropagation(); copyEmail('${escapeJs(acc.email)}')" title="复制">📋</button>
                             <button class="btn-icon" onclick="event.stopPropagation(); showEditAccountModal(${acc.id})" title="编辑">✏️</button>

static/js/main.js

@@ -987,6 +987,11 @@ ${details}
                     document.getElementById('enableAutoPolling').checked = enablePolling;
                     document.getElementById('pollingInterval').value = data.settings.polling_interval || '10';
                     document.getElementById('pollingCount').value = data.settings.polling_count || '5';
+
+                    // 加载 Telegram 推送设置
+                    document.getElementById('telegramBotToken').value = data.telegram_bot_token || '';
+                    document.getElementById('telegramChatId').value = data.telegram_chat_id || '';
+                    document.getElementById('telegramPollInterval').value = data.telegram_poll_interval || '600';
                 }
             } catch (error) {
                 showToast('加载设置失败', 'error');
@@ -1120,6 +1125,25 @@ ${details}
             settings.polling_interval = pInterval;
             settings.polling_count = pCount;
 
+            // Telegram 推送配置
+            const tgBotToken = document.getElementById('telegramBotToken').value.trim();
+            const tgChatId = document.getElementById('telegramChatId').value.trim();
+            const tgPollInterval = parseInt(document.getElementById('telegramPollInterval').value);
+
+            if (tgBotToken) {
+                settings.telegram_bot_token = tgBotToken;
+            }
+            if (tgChatId !== undefined) {
+                settings.telegram_chat_id = tgChatId;
+            }
+            if (!isNaN(tgPollInterval)) {
+                if (tgPollInterval < 60 || tgPollInterval > 3600) {
+                    showToast('Telegram 轮询间隔必须在 60-3600 秒之间', 'error');
+                    return;
+                }
+                settings.telegram_poll_interval = tgPollInterval;
+            }
+
             try {
                 const response = await fetch('/api/settings', {
                     method: 'PUT',
@@ -1140,6 +1164,24 @@ ${details}
             }
         }
 
+        async function testTelegramPush() {
+            const btn = document.getElementById('btnTestTelegram');
+            if (btn) { btn.disabled = true; btn.textContent = '⏳ 发送中…'; }
+            try {
+                const resp = await fetch('/api/settings/telegram-test', { method: 'POST', headers: { 'Content-Type': 'application/json' } });
+                const data = await resp.json();
+                if (data.success) {
+                    showToast(data.message || '测试消息已发送', 'success');
+                } else {
+                    showToast(data.error || '发送失败', 'error');
+                }
+            } catch (e) {
+                showToast('请求失败：' + e.message, 'error');
+            } finally {
+                if (btn) { btn.disabled = false; btn.textContent = '📨 发送测试消息'; }
+            }
+        }
+
         // ==================== 自动轮询功能 ====================
 
         // 初始化轮询设置

templates/index.html

@@ -420,6 +420,33 @@
                             </div>
                         </div>
 
+                        <!-- Telegram 推送设置 -->
+                        <div style="border-top: 1px solid var(--border-light); margin: 1.5rem 0; padding-top: 1.5rem;">
+                            <h4 style="font-size: 0.95rem; font-weight: 600; margin-bottom: 1rem; color: var(--text);">📬 Telegram 推送</h4>
+                            <div class="form-group">
+                                <label class="form-label">Bot Token <span style="font-size: 0.75rem; color: var(--text-muted); font-weight: 400;">（从 @BotFather 获取）</span></label>
+                                <input type="password" class="form-input" id="telegramBotToken" placeholder="输入 Bot Token" autocomplete="off">
+                                <div class="form-hint">格式：1234567890:AAxxxxxx（留空则禁用推送）</div>
+                            </div>
+                            <div class="form-group">
+                                <label class="form-label">Chat ID <span style="font-size: 0.75rem; color: var(--text-muted); font-weight: 400;">（接收推送的用户/群组 ID）</span></label>
+                                <input type="text" class="form-input" id="telegramChatId" placeholder="输入 Chat ID" style="max-width: 300px;">
+                                <div class="form-hint">可用 @userinfobot 获取你的 Chat ID</div>
+                            </div>
+                            <div class="form-group">
+                                <label class="form-label">轮询间隔 <span style="font-size: 0.75rem; color: var(--text-muted); font-weight: 400;">（检查新邮件的时间间隔）</span></label>
+                                <div style="display: flex; align-items: center; gap: 8px;">
+                                    <input type="number" class="form-input" id="telegramPollInterval" min="60" max="3600" value="600" placeholder="600" style="max-width: 120px;">
+                                    <span style="font-size: 0.88rem; color: var(--text-muted); font-weight: 500;">秒</span>
+                                </div>
+                                <div class="form-hint">范围：60-3600 秒，默认 600 秒（10 分钟）</div>
+                            </div>
+                            <div class="form-group" style="margin-top: 0.5rem;">
+                                <button class="btn btn-outline" onclick="testTelegramPush()" id="btnTestTelegram" style="font-size: 0.85rem;">📨 发送测试消息</button>
+                                <div class="form-hint">验证 Bot Token 和 Chat ID 是否配置正确</div>
+                            </div>
+                        </div>
+
                         <!-- 保存按钮 -->
                         <div style="display: flex; justify-content: flex-end; gap: 0.6rem; margin-top: 1.5rem; padding-top: 1rem; border-top: 1px solid var(--border-light);">
                             <button class="btn btn-primary" onclick="saveSettings()">💾 保存设置</button>

tests/test_telegram_push.py

@@ -0,0 +1,763 @@
+"""tests/test_telegram_push.py — TDD-00007 Telegram 实时推送测试
+
+覆盖 TDD-00007 §12 全部 32 个测试用例（T-01 ~ T-32）。
+
+测试分组：
+  TestBuildTelegramMessage   - T-01~T-05  消息构造单元测试
+  TestHtmlToPlain            - T-06~T-07  HTML 转纯文本
+  TestEscapeHtml             - T-08       HTML 转义
+  TestToggleTelegramPush     - T-09~T-12  toggle_telegram_push Repository
+  TestUpdateTelegramCursor   - T-13       update_telegram_cursor Repository
+  TestGetTelegramPushAccounts- T-14       get_telegram_push_accounts Repository
+  TestRunTelegramPushJob     - T-15~T-22  run_telegram_push_job 集成（mock 外部调用）
+  TestTelegramToggleEndpoint - T-23~T-26  /api/accounts/<id>/telegram-toggle 端点
+  TestTelegramSettingsAPI    - T-27~T-30  /api/settings Telegram 字段扩展
+  TestRegressions            - T-31~T-32  回归：调度器、全量测试
+"""
+
+import json
+import unittest
+from unittest.mock import MagicMock, patch
+
+from tests._import_app import clear_login_attempts, import_web_app_module
+
+
+# ---------------------------------------------------------------------------
+# 辅助：共享 app 实例（避免重复初始化）
+# ---------------------------------------------------------------------------
+
+def _get_app():
+    return import_web_app_module().app
+
+
+def _login(client):
+    resp = client.post("/login", json={"password": "testpass123"})
+    assert resp.status_code == 200, f"Login failed: {resp.data}"
+
+
+def _insert_test_account(db, email, provider="imap", enabled=0, last_checked=None, status="active"):
+    """在 accounts 表中插入测试账号，返回 rowid。"""
+    db.execute(
+        """INSERT INTO accounts
+           (email, client_id, provider, account_type, refresh_token, imap_host, imap_port,
+            imap_password, group_id, telegram_push_enabled, telegram_last_checked_at, status)
+           VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""",
+        (
+            email, "test_client_id", provider,
+            "outlook" if provider == "outlook" else "imap",
+            "enc:dummy_refresh", "imap.test.com", 993,
+            "enc:dummy_pass", None,
+            enabled, last_checked, status,
+        ),
+    )
+    db.commit()
+    return db.execute("SELECT last_insert_rowid()").fetchone()[0]
+
+
+# ===========================================================================
+# T-01 ~ T-05：消息构造单元测试
+# ===========================================================================
+
+class TestBuildTelegramMessage(unittest.TestCase):
+    """TDD-00007 T-01 ~ T-05"""
+
+    def _build(self, account_email, **email_kwargs):
+        from outlook_web.services.telegram_push import _build_telegram_message
+        email = {
+            "subject": email_kwargs.get("subject", "Test Subject"),
+            "sender":  email_kwargs.get("sender",  "sender@test.com"),
+            "received_at": email_kwargs.get("received_at", "2026-03-04T14:30:00"),
+            "preview": email_kwargs.get("preview", ""),
+        }
+        return _build_telegram_message(account_email, email)
+
+    def test_t01_normal_email_with_preview(self):
+        """T-01：正常邮件（含预览），消息包含所有关键字段"""
+        msg = self._build(
+            "user@example.com",
+            subject="Hello World",
+            sender="bob@x.com",
+            received_at="2026-03-04T14:30:00",
+            preview="This is a test email body.",
+        )
+        self.assertIn("📬", msg)
+        self.assertIn("user@example.com", msg)
+        self.assertIn("bob@x.com", msg)
+        self.assertIn("Hello World", msg)
+        self.assertIn("14:30", msg)
+        self.assertIn("This is a test email body.", msg)
+        self.assertIn("内容预览：", msg)
+
+    def test_t02_empty_preview_omits_section(self):
+        """T-02：正文为空时，消息中不包含"内容预览："区块"""
+        msg = self._build("user@example.com", preview="")
+        self.assertNotIn("内容预览：", msg)
+
+    def test_t03_very_long_message_truncated(self):
+        """T-03：消息超 4096 字符时截断至 4096，末尾为"..." """
+        msg = self._build("user@example.com", preview="a" * 5000)
+        self.assertLessEqual(len(msg), 4096)
+        self.assertTrue(msg.endswith("..."), f"Expected truncation marker, got: {msg[-10:]!r}")
+
+    def test_t04_html_special_chars_escaped(self):
+        """T-04：subject 和 sender 中的 HTML 特殊字符被正确转义"""
+        msg = self._build(
+            "user@example.com",
+            subject="<Sale> & <Offer>",
+            sender="a&b@test.com",
+        )
+        # 转义后的形式
+        self.assertIn("&lt;Sale&gt;", msg)
+        self.assertIn("&amp;", msg)
+        self.assertIn("a&amp;b@test.com", msg)
+        # 不应出现原始字符（在 subject/sender 渲染位置）
+        # （注意：HTML 标签本身如 <code> 是合法的，只检查 subject/sender 值）
+        self.assertNotIn("<Sale>", msg)
+        self.assertNotIn("<Offer>", msg)
+
+    def test_t05_preview_truncated_at_200_chars(self):
+        """T-05：正文超 200 字时截断，追加"..." """
+        msg = self._build("user@example.com", preview="x" * 300)
+        # 正文区域包含 200 个 x + "..."
+        self.assertIn("x" * 200 + "...", msg)
+        self.assertNotIn("x" * 201, msg)
+
+
+# ===========================================================================
+# T-06 ~ T-07：HTML 转纯文本
+# ===========================================================================
+
+class TestHtmlToPlain(unittest.TestCase):
+    """TDD-00007 T-06 ~ T-07"""
+
+    def _convert(self, html):
+        from outlook_web.services.telegram_push import _html_to_plain
+        return _html_to_plain(html)
+
+    def test_t06_strip_html_tags(self):
+        """T-06：HTML 标签被剥除"""
+        result = self._convert("<p>Hello <b>World</b></p>")
+        self.assertIn("Hello", result)
+        self.assertIn("World", result)
+        self.assertNotIn("<p>", result)
+        self.assertNotIn("<b>", result)
+
+    def test_t06_empty_string(self):
+        """T-06：空字符串返回空字符串"""
+        self.assertEqual(self._convert(""), "")
+
+    def test_t07_collapse_whitespace(self):
+        """T-07：多余空白（多个空格、换行）被合并为单个空格"""
+        result = self._convert("<p>  a  </p>  <br>  <p>b</p>")
+        self.assertIn("a", result)
+        self.assertIn("b", result)
+        # 不应出现连续多个空格
+        self.assertNotIn("  ", result)
+
+
+# ===========================================================================
+# T-08：HTML 转义
+# ===========================================================================
+
+class TestEscapeHtml(unittest.TestCase):
+    """TDD-00007 T-08"""
+
+    def test_t08_escape_three_special_chars(self):
+        """T-08：& < > 三种字符均被正确转义"""
+        from outlook_web.services.telegram_push import _escape_html
+        result = _escape_html("a & <b> and > c")
+        self.assertEqual(result, "a &amp; &lt;b&gt; and &gt; c")
+
+
+# ===========================================================================
+# T-09 ~ T-12：toggle_telegram_push Repository
+# ===========================================================================
+
+class TestToggleTelegramPush(unittest.TestCase):
+    """TDD-00007 T-09 ~ T-12"""
+
+    @classmethod
+    def setUpClass(cls):
+        cls.app = _get_app()
+
+    def setUp(self):
+        with self.app.app_context():
+            clear_login_attempts()
+            from outlook_web.db import get_db
+            db = get_db()
+            db.execute("DELETE FROM accounts WHERE email LIKE '%@tgtest.com'")
+            db.commit()
+            self._account_id = _insert_test_account(db, "tg@tgtest.com", enabled=0, last_checked=None)
+
+    def test_t09_first_enable_sets_cursor(self):
+        """T-09：首次开启时 telegram_last_checked_at 被设置（不为 None）"""
+        with self.app.app_context():
+            from outlook_web.repositories.accounts import toggle_telegram_push
+            from outlook_web.db import get_db
+            result = toggle_telegram_push(self._account_id, True)
+            self.assertTrue(result)
+            row = get_db().execute(
+                "SELECT telegram_push_enabled, telegram_last_checked_at FROM accounts WHERE id = ?",
+                (self._account_id,)
+            ).fetchone()
+            self.assertEqual(row["telegram_push_enabled"], 1)
+            self.assertIsNotNone(row["telegram_last_checked_at"])
+            # 游标应为有效 ISO8601 字符串
+            self.assertRegex(row["telegram_last_checked_at"], r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}")
+
+    def test_t10_disable_preserves_cursor(self):
+        """T-10：关闭推送时游标不清空"""
+        with self.app.app_context():
+            from outlook_web.repositories.accounts import toggle_telegram_push
+            from outlook_web.db import get_db
+            db = get_db()
+            # 先开启（设置游标）
+            toggle_telegram_push(self._account_id, True)
+            original_cursor = db.execute(
+                "SELECT telegram_last_checked_at FROM accounts WHERE id = ?",
+                (self._account_id,)
+            ).fetchone()["telegram_last_checked_at"]
+            # 关闭
+            toggle_telegram_push(self._account_id, False)
+            row = db.execute(
+                "SELECT telegram_push_enabled, telegram_last_checked_at FROM accounts WHERE id = ?",
+                (self._account_id,)
+            ).fetchone()
+            self.assertEqual(row["telegram_push_enabled"], 0)
+            self.assertEqual(row["telegram_last_checked_at"], original_cursor)
+
+    def test_t11_nonexistent_account_returns_false(self):
+        """T-11：账号不存在时返回 False"""
+        with self.app.app_context():
+            from outlook_web.repositories.accounts import toggle_telegram_push
+            result = toggle_telegram_push(99999, True)
+            self.assertFalse(result)
+
+    def test_t12_repeat_enable_not_reset_cursor(self):
+        """T-12：重复开启不重置已有游标"""
+        with self.app.app_context():
+            from outlook_web.repositories.accounts import toggle_telegram_push
+            from outlook_web.db import get_db
+            db = get_db()
+            # 先开启（游标设为 T1）
+            toggle_telegram_push(self._account_id, True)
+            cursor_t1 = db.execute(
+                "SELECT telegram_last_checked_at FROM accounts WHERE id = ?",
+                (self._account_id,)
+            ).fetchone()["telegram_last_checked_at"]
+            # 再次开启
+            toggle_telegram_push(self._account_id, True)
+            cursor_t2 = db.execute(
+                "SELECT telegram_last_checked_at FROM accounts WHERE id = ?",
+                (self._account_id,)
+            ).fetchone()["telegram_last_checked_at"]
+            self.assertEqual(cursor_t1, cursor_t2, "重复开启不应重置游标")
+
+
+# ===========================================================================
+# T-13：update_telegram_cursor Repository
+# ===========================================================================
+
+class TestUpdateTelegramCursor(unittest.TestCase):
+    """TDD-00007 T-13"""
+
+    @classmethod
+    def setUpClass(cls):
+        cls.app = _get_app()
+
+    def setUp(self):
+        with self.app.app_context():
+            from outlook_web.db import get_db
+            db = get_db()
+            db.execute("DELETE FROM accounts WHERE email = 'cursor@tgtest.com'")
+            db.commit()
+            self._account_id = _insert_test_account(db, "cursor@tgtest.com")
+
+    def test_t13_update_cursor(self):
+        """T-13：update_telegram_cursor 正确更新游标"""
+        with self.app.app_context():
+            from outlook_web.repositories.accounts import update_telegram_cursor
+            from outlook_web.db import get_db
+            update_telegram_cursor(self._account_id, "2026-03-04T14:30:00")
+            row = get_db().execute(
+                "SELECT telegram_last_checked_at FROM accounts WHERE id = ?",
+                (self._account_id,)
+            ).fetchone()
+            self.assertEqual(row["telegram_last_checked_at"], "2026-03-04T14:30:00")
+
+
+# ===========================================================================
+# T-14：get_telegram_push_accounts Repository
+# ===========================================================================
+
+class TestGetTelegramPushAccounts(unittest.TestCase):
+    """TDD-00007 T-14"""
+
+    @classmethod
+    def setUpClass(cls):
+        cls.app = _get_app()
+
+    def setUp(self):
+        with self.app.app_context():
+            from outlook_web.db import get_db
+            db = get_db()
+            db.execute("DELETE FROM accounts WHERE email LIKE '%@tg14test.com'")
+            db.commit()
+            _insert_test_account(db, "a@tg14test.com", enabled=1, status="active")
+            _insert_test_account(db, "b@tg14test.com", enabled=0, status="active")
+            _insert_test_account(db, "c@tg14test.com", enabled=1, status="disabled")
+
+    def test_t14_returns_only_enabled_active_accounts(self):
+        """T-14：只返回 enabled=1 且 status != disabled 的账号"""
+        with self.app.app_context():
+            from outlook_web.repositories.accounts import get_telegram_push_accounts
+            accounts = get_telegram_push_accounts()
+            emails = [a["email"] for a in accounts]
+            self.assertIn("a@tg14test.com", emails)
+            self.assertNotIn("b@tg14test.com", emails)  # enabled=0
+            self.assertNotIn("c@tg14test.com", emails)  # status=disabled
+
+
+# ===========================================================================
+# T-15 ~ T-22：run_telegram_push_job 集成测试（mock 外部调用）
+# ===========================================================================
+
+class TestRunTelegramPushJob(unittest.TestCase):
+    """TDD-00007 T-15 ~ T-22"""
+
+    @classmethod
+    def setUpClass(cls):
+        cls.app = _get_app()
+
+    def setUp(self):
+        with self.app.app_context():
+            from outlook_web.db import get_db
+            db = get_db()
+            db.execute("DELETE FROM accounts WHERE email LIKE '%@tgjob.com'")
+            # 确保其他测试遗留的账号不干扰推送 job（关闭所有推送开关）
+            db.execute("UPDATE accounts SET telegram_push_enabled = 0")
+            db.commit()
+
+    def _set_settings(self, bot_token="test_token_12345678", chat_id="-12345"):
+        from outlook_web.repositories.settings import set_setting
+        from outlook_web.security.crypto import encrypt_data
+        if bot_token:
+            set_setting("telegram_bot_token", encrypt_data(bot_token))
+        else:
+            set_setting("telegram_bot_token", "")
+        set_setting("telegram_chat_id", chat_id)
+
+    def _run_job(self):
+        from outlook_web.services.telegram_push import run_telegram_push_job
+        run_telegram_push_job(self.app)
+
+    def _make_email(self, received_at="2026-03-04T14:31:00"):
+        return {
+            "subject": "Test",
+            "sender": "s@test.com",
+            "received_at": received_at,
+            "preview": "body preview",
+        }
+
+    def test_t15_no_bot_token_skips(self):
+        """T-15：Bot Token 未配置 → Telegram API 不被调用"""
+        with self.app.app_context():
+            self._set_settings(bot_token="", chat_id="-12345")
+            with patch("outlook_web.services.telegram_push._send_telegram_message") as mock_send:
+                self._run_job()
+                mock_send.assert_not_called()
+
+    def test_t16_no_chat_id_skips(self):
+        """T-16：Chat ID 未配置 → Telegram API 不被调用"""
+        with self.app.app_context():
+            self._set_settings(bot_token="valid_token_xyz", chat_id="")
+            with patch("outlook_web.services.telegram_push._send_telegram_message") as mock_send:
+                self._run_job()
+                mock_send.assert_not_called()
+
+    def test_t17_no_enabled_accounts_skips(self):
+        """T-17：无启用推送的账号 → Telegram API 不被调用"""
+        with self.app.app_context():
+            self._set_settings()
+            # 确保无 enabled 账号（setUp 已清理）
+            with patch("outlook_web.services.telegram_push._send_telegram_message") as mock_send:
+                self._run_job()
+                mock_send.assert_not_called()
+
+    def test_t18_first_run_sets_cursor_no_push(self):
+        """T-18：首次开启（last_checked_at=NULL）→ 设游标，不推送"""
+        with self.app.app_context():
+            from outlook_web.db import get_db
+            db = get_db()
+            acc_id = _insert_test_account(db, "first@tgjob.com", enabled=1, last_checked=None)
+            self._set_settings()
+
+            with patch("outlook_web.services.telegram_push._fetch_new_emails_imap") as mock_fetch, \
+                 patch("outlook_web.services.telegram_push._fetch_new_emails_graph") as mock_graph, \
+                 patch("outlook_web.services.telegram_push._send_telegram_message") as mock_send:
+
+                mock_fetch.return_value = [self._make_email(), self._make_email()]
+                mock_graph.return_value = []
+                self._run_job()
+
+                # 不推送
+                mock_send.assert_not_called()
+
+            # 游标已更新（不再为 NULL）
+            row = db.execute(
+                "SELECT telegram_last_checked_at FROM accounts WHERE id = ?",
+                (acc_id,)
+            ).fetchone()
+            self.assertIsNotNone(row["telegram_last_checked_at"])
+
+    def test_t19_normal_push_calls_api_and_updates_cursor(self):
+        """T-19：正常推送 → Telegram API 被调用，游标更新"""
+        with self.app.app_context():
+            from outlook_web.db import get_db
+            db = get_db()
+            acc_id = _insert_test_account(
+                db, "normal@tgjob.com", enabled=1, last_checked="2026-03-01T00:00:00"
+            )
+            self._set_settings()
+
+            emails = [
+                self._make_email("2026-03-04T10:00:00"),
+                self._make_email("2026-03-04T11:00:00"),
+            ]
+            with patch("outlook_web.services.telegram_push._fetch_new_emails_imap", return_value=emails), \
+                 patch("outlook_web.services.telegram_push._send_telegram_message", return_value=True) as mock_send:
+
+                self._run_job()
+
+                self.assertEqual(mock_send.call_count, 2)
+
+            # 游标已更新（不再是 2026-03-01）
+            cursor = db.execute(
+                "SELECT telegram_last_checked_at FROM accounts WHERE id = ?",
+                (acc_id,)
+            ).fetchone()["telegram_last_checked_at"]
+            self.assertIsNotNone(cursor)
+            self.assertNotEqual(cursor, "2026-03-01T00:00:00")
+
+    def test_t20_global_limit_20(self):
+        """T-20：全局上限 20 封，跨账号合计不超过 20 条 Telegram 消息"""
+        with self.app.app_context():
+            from outlook_web.db import get_db
+            db = get_db()
+            # 账号 A：12 封；账号 B：15 封 → 合计最多 20 封
+            _insert_test_account(db, "aa@tgjob.com", enabled=1, last_checked="2026-01-01T00:00:00")
+            _insert_test_account(db, "bb@tgjob.com", enabled=1, last_checked="2026-01-01T00:00:00")
+            self._set_settings()
+
+            emails_a = [self._make_email(f"2026-03-04T10:{i:02d}:00") for i in range(12)]
+            emails_b = [self._make_email(f"2026-03-04T11:{i:02d}:00") for i in range(15)]
+
+            call_count = 0
+
+            def fake_fetch(account, since):
+                nonlocal call_count
+                if account["email"] == "aa@tgjob.com":
+                    return emails_a
+                return emails_b
+
+            def fake_send(token, chat, msg):
+                nonlocal call_count
+                call_count += 1
+                return True
+
+            with patch("outlook_web.services.telegram_push._fetch_new_emails_imap", side_effect=fake_fetch), \
+                 patch("outlook_web.services.telegram_push._send_telegram_message", side_effect=fake_send):
+                self._run_job()
+
+            self.assertEqual(call_count, 20, f"Expected 20, got {call_count}")
+
+    def test_t21_imap_exception_silent_cursor_updated(self):
+        """T-21：IMAP 连接异常 → 不抛出，游标仍更新"""
+        with self.app.app_context():
+            from outlook_web.db import get_db
+            db = get_db()
+            acc_id = _insert_test_account(
+                db, "imapfail@tgjob.com", enabled=1, last_checked="2026-03-01T00:00:00"
+            )
+            self._set_settings()
+
+            with patch(
+                "outlook_web.services.telegram_push._fetch_new_emails_imap",
+                side_effect=ConnectionError("IMAP connection refused")
+            ), patch("outlook_web.services.telegram_push._send_telegram_message") as mock_send:
+                # 不应抛出异常
+                try:
+                    self._run_job()
+                except Exception as e:
+                    self.fail(f"run_telegram_push_job raised unexpectedly: {e}")
+
+                mock_send.assert_not_called()
+
+            # 游标仍更新
+            cursor = db.execute(
+                "SELECT telegram_last_checked_at FROM accounts WHERE id = ?",
+                (acc_id,)
+            ).fetchone()["telegram_last_checked_at"]
+            self.assertNotEqual(cursor, "2026-03-01T00:00:00")
+
+    def test_t22_send_failure_silent_cursor_updated(self):
+        """T-22：Telegram 发送失败 → 不抛出，游标仍更新"""
+        with self.app.app_context():
+            from outlook_web.db import get_db
+            db = get_db()
+            acc_id = _insert_test_account(
+                db, "sendfail@tgjob.com", enabled=1, last_checked="2026-03-01T00:00:00"
+            )
+            self._set_settings()
+
+            emails = [self._make_email("2026-03-04T10:00:00"), self._make_email("2026-03-04T11:00:00")]
+            with patch("outlook_web.services.telegram_push._fetch_new_emails_imap", return_value=emails), \
+                 patch("outlook_web.services.telegram_push._send_telegram_message", return_value=False):
+                try:
+                    self._run_job()
+                except Exception as e:
+                    self.fail(f"run_telegram_push_job raised unexpectedly: {e}")
+
+            # 游标仍更新
+            cursor = db.execute(
+                "SELECT telegram_last_checked_at FROM accounts WHERE id = ?",
+                (acc_id,)
+            ).fetchone()["telegram_last_checked_at"]
+            self.assertNotEqual(cursor, "2026-03-01T00:00:00")
+
+
+# ===========================================================================
+# T-23 ~ T-26：/api/accounts/<id>/telegram-toggle 端点测试
+# ===========================================================================
+
+class TestTelegramToggleEndpoint(unittest.TestCase):
+    """TDD-00007 T-23 ~ T-26"""
+
+    @classmethod
+    def setUpClass(cls):
+        cls.app = _get_app()
+
+    def setUp(self):
+        with self.app.app_context():
+            clear_login_attempts()
+            from outlook_web.db import get_db
+            db = get_db()
+            db.execute("DELETE FROM accounts WHERE email LIKE '%@tgapi.com'")
+            db.commit()
+            self._account_id = _insert_test_account(db, "api@tgapi.com")
+
+    def _toggle(self, client, account_id, enabled):
+        return client.post(
+            f"/api/accounts/{account_id}/telegram-toggle",
+            json={"enabled": enabled},
+            content_type="application/json",
+        )
+
+    def test_t23_enable_success(self):
+        """T-23：开启成功 → 200, success=true, DB 更新"""
+        client = self.app.test_client()
+        _login(client)
+        resp = self._toggle(client, self._account_id, True)
+        self.assertEqual(resp.status_code, 200)
+        data = resp.get_json()
+        self.assertTrue(data["success"])
+        self.assertTrue(data["enabled"])
+
+        with self.app.app_context():
+            from outlook_web.db import get_db
+            row = get_db().execute(
+                "SELECT telegram_push_enabled FROM accounts WHERE id = ?",
+                (self._account_id,)
+            ).fetchone()
+            self.assertEqual(row["telegram_push_enabled"], 1)
+
+    def test_t24_disable_success(self):
+        """T-24：关闭成功 → 200, success=true, DB 更新"""
+        with self.app.app_context():
+            from outlook_web.db import get_db
+            get_db().execute(
+                "UPDATE accounts SET telegram_push_enabled=1 WHERE id=?",
+                (self._account_id,)
+            )
+            get_db().commit()
+
+        client = self.app.test_client()
+        _login(client)
+        resp = self._toggle(client, self._account_id, False)
+        self.assertEqual(resp.status_code, 200)
+        data = resp.get_json()
+        self.assertTrue(data["success"])
+        self.assertFalse(data["enabled"])
+
+        with self.app.app_context():
+            from outlook_web.db import get_db
+            row = get_db().execute(
+                "SELECT telegram_push_enabled FROM accounts WHERE id = ?",
+                (self._account_id,)
+            ).fetchone()
+            self.assertEqual(row["telegram_push_enabled"], 0)
+
+    def test_t25_nonexistent_account_404(self):
+        """T-25：账号不存在 → HTTP 404"""
+        client = self.app.test_client()
+        _login(client)
+        resp = self._toggle(client, 99999, True)
+        self.assertEqual(resp.status_code, 404)
+        data = resp.get_json()
+        self.assertFalse(data["success"])
+
+    def test_t26_unauthenticated_denied(self):
+        """T-26：未登录 → 拒绝访问（401 或 302 重定向）"""
+        client = self.app.test_client()
+        # 不调用 _login
+        resp = self._toggle(client, self._account_id, True)
+        self.assertIn(resp.status_code, [401, 302, 403])
+
+
+# ===========================================================================
+# T-27 ~ T-30：/api/settings Telegram 字段扩展测试
+# ===========================================================================
+
+class TestTelegramSettingsAPI(unittest.TestCase):
+    """TDD-00007 T-27 ~ T-30"""
+
+    @classmethod
+    def setUpClass(cls):
+        cls.app = _get_app()
+
+    def setUp(self):
+        with self.app.app_context():
+            clear_login_attempts()
+            from outlook_web.repositories.settings import set_setting
+            set_setting("telegram_bot_token", "")
+            set_setting("telegram_chat_id", "")
+            set_setting("telegram_poll_interval", "600")
+
+    def test_t27_get_settings_includes_telegram_fields(self):
+        """T-27：GET /api/settings 返回 telegram 配置字段，bot_token 脱敏"""
+        with self.app.app_context():
+            from outlook_web.repositories.settings import set_setting
+            from outlook_web.security.crypto import encrypt_data
+            set_setting("telegram_bot_token", encrypt_data("1234567890:AAxxxxx"))
+
+        client = self.app.test_client()
+        _login(client)
+        resp = client.get("/api/settings")
+        self.assertEqual(resp.status_code, 200)
+        data = resp.get_json()
+
+        self.assertIn("telegram_bot_token", data)
+        self.assertIn("telegram_chat_id", data)
+        self.assertIn("telegram_poll_interval", data)
+        # bot_token 不应返回明文
+        token_val = data["telegram_bot_token"]
+        self.assertNotEqual(token_val, "1234567890:AAxxxxx")
+        # 应为脱敏值（含 * 或为空）
+        if token_val:
+            self.assertTrue(
+                token_val.startswith("****") or "*" in token_val,
+                f"Bot token should be masked, got: {token_val!r}"
+            )
+
+    def test_t28_put_settings_saves_telegram_config(self):
+        """T-28：PUT /api/settings 保存 telegram 配置，bot_token 加密存储"""
+        client = self.app.test_client()
+        _login(client)
+        resp = client.put(
+            "/api/settings",
+            json={
+                "telegram_bot_token": "NewToken123",
+                "telegram_chat_id": "-123456",
+                "telegram_poll_interval": 300,
+            },
+            content_type="application/json",
+        )
+        self.assertEqual(resp.status_code, 200)
+        data = resp.get_json()
+        self.assertTrue(data.get("success"), f"Expected success, got: {data}")
+
+        with self.app.app_context():
+            from outlook_web.repositories.settings import get_setting
+            from outlook_web.security.crypto import decrypt_data
+            raw_token = get_setting("telegram_bot_token")
+            # 应加密存储（前缀 enc:）
+            self.assertTrue(raw_token.startswith("enc:"), f"Token should be encrypted, got: {raw_token!r}")
+            # 解密后应为原始值
+            self.assertEqual(decrypt_data(raw_token), "NewToken123")
+            self.assertEqual(get_setting("telegram_chat_id"), "-123456")
+            self.assertEqual(get_setting("telegram_poll_interval"), "300")
+
+    def test_t29_invalid_interval_returns_400(self):
+        """T-29：telegram_poll_interval 低于最小值 60 → 400 错误"""
+        client = self.app.test_client()
+        _login(client)
+        resp = client.put(
+            "/api/settings",
+            json={"telegram_poll_interval": 30},  # 低于最小值 60
+            content_type="application/json",
+        )
+        self.assertEqual(resp.status_code, 400)
+
+    def test_t30_masked_token_placeholder_not_overwrite(self):
+        """T-30：传入脱敏占位符（****xxxx）不覆盖已有 bot_token"""
+        with self.app.app_context():
+            from outlook_web.repositories.settings import set_setting
+            from outlook_web.security.crypto import encrypt_data
+            original_encrypted = encrypt_data("OriginalRealToken")
+            set_setting("telegram_bot_token", original_encrypted)
+
+        client = self.app.test_client()
+        _login(client)
+        resp = client.put(
+            "/api/settings",
+            json={"telegram_bot_token": "****xxxx"},  # 脱敏占位符
+            content_type="application/json",
+        )
+        # 保存不报错
+        self.assertIn(resp.status_code, [200, 204])
+
+        with self.app.app_context():
+            from outlook_web.repositories.settings import get_setting
+            from outlook_web.security.crypto import decrypt_data, encrypt_data
+            stored = get_setting("telegram_bot_token")
+            # 解密后应仍为原始值（未被覆盖）
+            self.assertEqual(decrypt_data(stored), "OriginalRealToken")
+
+
+# ===========================================================================
+# T-31 ~ T-32：回归测试
+# ===========================================================================
+
+class TestRegressions(unittest.TestCase):
+    """TDD-00007 T-31 ~ T-32"""
+
+    @classmethod
+    def setUpClass(cls):
+        cls.app = _get_app()
+
+    def test_t31_token_refresh_job_unaffected(self):
+        """T-31：加载 telegram_push 模块不影响 token_refresh 调度器 job"""
+        # 确认可以导入 telegram_push 模块而不报错
+        try:
+            import outlook_web.services.telegram_push as tp  # noqa: F401
+        except ImportError as e:
+            self.fail(f"Cannot import telegram_push service: {e}")
+
+    def test_t32_existing_tests_still_pass(self):
+        """T-32：现有核心功能（providers, settings）在引入新模块后仍正常"""
+        with self.app.app_context():
+            # providers 模块正常
+            from outlook_web.services.providers import get_provider_list
+            providers = get_provider_list()
+            self.assertIsInstance(providers, list)
+            self.assertGreater(len(providers), 0)
+
+            # settings repository 正常
+            from outlook_web.repositories.settings import get_setting, set_setting
+            set_setting("_test_regression_key", "ok")
+            val = get_setting("_test_regression_key")
+            self.assertEqual(val, "ok")
+
+
+if __name__ == "__main__":
+    unittest.main()