diff --git a/outlook_web/db.py b/outlook_web/db.py index a09fb5d..77d2024 100644 --- a/outlook_web/db.py +++ b/outlook_web/db.py @@ -18,8 +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） -# v4：对齐 PRD-00007 / FD-00007（accounts 表新增 telegram_push_enabled/telegram_last_checked_at） -DB_SCHEMA_VERSION = 4 +# v5：BUG-00011 P2 — Message-ID 去重防止重复推送 +DB_SCHEMA_VERSION = 5 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" @@ -556,6 +556,31 @@ def init_db(database_path: Optional[str] = None): """ ) + # v5: Telegram 推送去重日志（BUG-00011 P2） + cursor.execute( + """ + CREATE TABLE IF NOT EXISTS telegram_push_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + account_id INTEGER NOT NULL, + message_id TEXT NOT NULL, + pushed_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')), + UNIQUE(account_id, message_id) + ) + """ + ) + cursor.execute( + """ + CREATE INDEX IF NOT EXISTS idx_telegram_push_log_account_id + ON telegram_push_log(account_id) + """ + ) + cursor.execute( + """ + CREATE INDEX IF NOT EXISTS idx_telegram_push_log_pushed_at + ON telegram_push_log(pushed_at) + """ + ) + # 迁移现有明文数据为加密数据 migrate_sensitive_data(conn) diff --git a/outlook_web/services/telegram_push.py b/outlook_web/services/telegram_push.py index ce0aba0..eab0db0 100644 --- a/outlook_web/services/telegram_push.py +++ b/outlook_web/services/telegram_push.py @@ -183,6 +183,10 @@ def _fetch_new_emails_imap(account: dict, since: str) -> List[dict]: ) sender = msg.get("From", "") + # 提取 Message-ID（BUG-00011 P2 去重用） + raw_message_id = msg.get("Message-ID", "") or msg.get("Message-Id", "") + if not raw_message_id: + raw_message_id = f"imap:{account.get('email', '')}:{mid.decode() if isinstance(mid, bytes) else mid}" date_str = msg.get("Date", "") try: from email.utils import parsedate_to_datetime @@ -226,6 +230,7 @@ def _fetch_new_emails_imap(account: dict, since: str) -> List[dict]: results.append( { + "message_id": raw_message_id.strip(), "subject": subject, "sender": sender, "received_at": received_iso, @@ -267,7 +272,7 @@ def _fetch_new_emails_graph(account: dict, since: str) -> List[dict]: params = { "$filter": f"receivedDateTime gt {since_z}", "$top": MAX_EMAILS_PER_FETCH, - "$select": "subject,from,receivedDateTime,bodyPreview", + "$select": "id,subject,from,receivedDateTime,bodyPreview,internetMessageId", "$orderby": "receivedDateTime asc", } headers = {"Authorization": f"Bearer {access_token}"} @@ -285,8 +290,11 @@ def _fetch_new_emails_graph(account: dict, since: str) -> List[dict]: 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] + # BUG-00011 P2: 提取 Message-ID（优先 internetMessageId，回退 Graph id） + msg_id = item.get("internetMessageId", "") or item.get("id", "") results.append( { + "message_id": msg_id, "subject": item.get("subject", ""), "sender": sender, "received_at": received_iso, @@ -300,6 +308,45 @@ def _fetch_new_emails_graph(account: dict, since: str) -> List[dict]: return results +# --------------------------------------------------------------------------- +# Message-ID 去重（BUG-00011 P2） +# --------------------------------------------------------------------------- + +PUSH_LOG_RETENTION_DAYS = 7 # 去重记录保留天数 + + +def _is_message_pushed(db, account_id: int, message_id: str) -> bool: + """检查该邮件是否已推送过。""" + row = db.execute( + "SELECT 1 FROM telegram_push_log WHERE account_id = ? AND message_id = ?", + (account_id, message_id), + ).fetchone() + return row is not None + + +def _record_pushed_message(db, account_id: int, message_id: str) -> None: + """记录已推送的邮件 Message-ID。""" + try: + db.execute( + "INSERT OR IGNORE INTO telegram_push_log (account_id, message_id, pushed_at) VALUES (?, ?, ?)", + (account_id, message_id, datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")), + ) + db.commit() + except Exception: + pass # 去重失败不影响推送流程 + + +def _cleanup_push_log(db) -> None: + """清理超过 PUSH_LOG_RETENTION_DAYS 天的去重记录。""" + try: + from datetime import timedelta + cutoff = (datetime.now(timezone.utc) - timedelta(days=PUSH_LOG_RETENTION_DAYS)).strftime("%Y-%m-%dT%H:%M:%S") + db.execute("DELETE FROM telegram_push_log WHERE pushed_at < ?", (cutoff,)) + db.commit() + except Exception: + pass + + # --------------------------------------------------------------------------- # 主入口（调度器调用） # --------------------------------------------------------------------------- @@ -328,7 +375,7 @@ def _fetch_account_emails(account: dict) -> tuple: def run_telegram_push_job(app) -> None: - """主入口：并行轮询 → 推送 → 更新游标。由调度器调用。""" + """主入口：并行轮询 → 去重 → 推送 → 更新游标。由调度器调用。""" import time from concurrent.futures import ThreadPoolExecutor, as_completed @@ -336,6 +383,7 @@ def run_telegram_push_job(app) -> None: logger.info("[telegram_push] job started") with app.app_context(): + from outlook_web.db import get_db from outlook_web.repositories.accounts import ( get_telegram_push_accounts, update_telegram_cursor, @@ -361,6 +409,9 @@ def run_telegram_push_job(app) -> None: from datetime import timedelta recency_cutoff = (datetime.now(timezone.utc) - timedelta(hours=PUSH_RECENCY_HOURS)).strftime("%Y-%m-%dT%H:%M:%S") sent_count = 0 + dedup_skipped = 0 + + db = get_db() # 并行获取所有账号邮件 fetch_results = [] @@ -381,7 +432,6 @@ def run_telegram_push_job(app) -> None: continue # BUG-00011: 游标取 max(job_start_time, 最大 received_at) - # 防止邮件服务器时钟偏差 / fetch 窗口期间到达的邮件被重复推送 max_received = max( (em.get("received_at", "") for em in emails), default="", @@ -392,20 +442,33 @@ def run_telegram_push_job(app) -> None: update_telegram_cursor(account["id"], new_cursor) continue + account_id = account["id"] for em in sorted(emails, key=lambda e: e.get("received_at", "")): if sent_count >= MAX_SENT_PER_JOB: break # 跳过超过 PUSH_RECENCY_HOURS 的旧邮件 if em.get("received_at", "") < recency_cutoff: continue + # BUG-00011 P2: Message-ID 去重 + msg_id = em.get("message_id", "") + if msg_id and _is_message_pushed(db, account_id, msg_id): + dedup_skipped += 1 + continue + msg = _build_telegram_message(account["email"], em) if _send_telegram_message(bot_token, chat_id, msg): sent_count += 1 + if msg_id: + _record_pushed_message(db, account_id, msg_id) # 消息间延迟，防止 Telegram API 限流 if TELEGRAM_PUSH_DELAY_SEC > 0: time.sleep(TELEGRAM_PUSH_DELAY_SEC) update_telegram_cursor(account["id"], new_cursor) + # 定期清理过期去重记录（>7 天） + _cleanup_push_log(db) + elapsed = time.monotonic() - t0 - logger.info("[telegram_push] job finished: sent=%d elapsed=%.1fs", sent_count, elapsed) + logger.info("[telegram_push] job finished: sent=%d dedup_skipped=%d elapsed=%.1fs", + sent_count, dedup_skipped, elapsed) diff --git a/tests/test_telegram_push.py b/tests/test_telegram_push.py index 33153e2..6181125 100644 --- a/tests/test_telegram_push.py +++ b/tests/test_telegram_push.py @@ -361,6 +361,7 @@ class TestRunTelegramPushJob(unittest.TestCase): db.execute("DELETE FROM accounts WHERE email LIKE '%@tgjob.com'") # 确保其他测试遗留的账号不干扰推送 job（关闭所有推送开关） db.execute("UPDATE accounts SET telegram_push_enabled = 0") + db.execute("DELETE FROM telegram_push_log") db.commit() def _set_settings(self, bot_token="test_token_12345678", chat_id="-12345"): @@ -377,7 +378,9 @@ class TestRunTelegramPushJob(unittest.TestCase): run_telegram_push_job(self.app) def _make_email(self, received_at="2026-03-04T14:31:00"): + import uuid return { + "message_id": f"<{uuid.uuid4().hex}@test.com>", "subject": "Test", "sender": "s@test.com", "received_at": received_at, @@ -963,6 +966,7 @@ class TestParallelJobBehavior(unittest.TestCase): db = get_db() db.execute("DELETE FROM accounts WHERE email LIKE '%@parallel.com'") db.execute("UPDATE accounts SET telegram_push_enabled = 0") + db.execute("DELETE FROM telegram_push_log") db.commit() def _set_settings(self, bot_token="test_token_12345678", chat_id="-12345"): @@ -979,7 +983,9 @@ class TestParallelJobBehavior(unittest.TestCase): run_telegram_push_job(self.app) def _make_email(self, received_at="2026-03-04T14:31:00"): + import uuid return { + "message_id": f"<{uuid.uuid4().hex}@test.com>", "subject": "Test", "sender": "s@test.com", "received_at": received_at, @@ -1147,6 +1153,7 @@ class TestCursorClockSkew(unittest.TestCase): db = get_db() db.execute("DELETE FROM accounts WHERE email LIKE '%@clockskew.com'") db.execute("UPDATE accounts SET telegram_push_enabled = 0") + db.execute("DELETE FROM telegram_push_log") db.commit() def _set_settings(self): @@ -1172,6 +1179,7 @@ class TestCursorClockSkew(unittest.TestCase): from datetime import timedelta future_time = (datetime.now(timezone.utc) + timedelta(seconds=5)).strftime("%Y-%m-%dT%H:%M:%S") emails = [{ + "message_id": "", "subject": "Clock skew email", "sender": "s@test.com", "received_at": future_time, @@ -1204,6 +1212,7 @@ class TestCursorClockSkew(unittest.TestCase): from datetime import timedelta past_time = (datetime.now(timezone.utc) - timedelta(minutes=5)).strftime("%Y-%m-%dT%H:%M:%S") emails = [{ + "message_id": "", "subject": "Normal email", "sender": "s@test.com", "received_at": past_time, @@ -1238,6 +1247,7 @@ class TestCursorClockSkew(unittest.TestCase): from datetime import timedelta future_time = (datetime.now(timezone.utc) + timedelta(seconds=3)).strftime("%Y-%m-%dT%H:%M:%S") emails = [{ + "message_id": "", "subject": "Skew E2E", "sender": "s@test.com", "received_at": future_time, @@ -1269,5 +1279,158 @@ class TestCursorClockSkew(unittest.TestCase): mock_send2.assert_not_called() +class TestMessageIdDedup(unittest.TestCase): + """BUG-00011 P2: Message-ID 去重测试""" + + @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 '%@dedup.com'") + db.execute("UPDATE accounts SET telegram_push_enabled = 0") + db.execute("DELETE FROM telegram_push_log") + db.commit() + + def _set_settings(self): + from outlook_web.repositories.settings import set_setting + from outlook_web.security.crypto import encrypt_data + set_setting("telegram_bot_token", encrypt_data("test_token_12345678")) + set_setting("telegram_chat_id", "-12345") + + def _run_job(self): + from outlook_web.services.telegram_push import run_telegram_push_job + run_telegram_push_job(self.app) + + def test_same_message_id_not_pushed_twice(self): + """同一 message_id 的邮件在两轮 Job 中只推送 1 次""" + with self.app.app_context(): + from outlook_web.db import get_db + db = get_db() + _insert_test_account(db, "dup@dedup.com", enabled=1, + last_checked="2026-03-01T00:00:00") + self._set_settings() + + from datetime import timedelta + recent = (datetime.now(timezone.utc) - timedelta(minutes=2)).strftime("%Y-%m-%dT%H:%M:%S") + emails = [{ + "message_id": "", + "subject": "Test dedup", + "sender": "s@test.com", + "received_at": recent, + "preview": "body", + }] + + # 第一轮：推送成功 + 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 send1, \ + patch("outlook_web.services.telegram_push.TELEGRAM_PUSH_DELAY_SEC", 0): + self._run_job() + self.assertEqual(send1.call_count, 1) + + # 验证 push_log 记录 + row = db.execute( + "SELECT * FROM telegram_push_log WHERE message_id = ?", ("",) + ).fetchone() + self.assertIsNotNone(row) + + # 第二轮：同一封邮件再次被 fetch（模拟游标未正确推进的极端场景） + 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 send2, \ + patch("outlook_web.services.telegram_push.TELEGRAM_PUSH_DELAY_SEC", 0): + self._run_job() + # Message-ID 去重：不应再推送 + send2.assert_not_called() + + def test_different_message_ids_both_pushed(self): + """不同 message_id 的邮件各推送 1 次""" + with self.app.app_context(): + from outlook_web.db import get_db + db = get_db() + _insert_test_account(db, "multi@dedup.com", enabled=1, + last_checked="2026-03-01T00:00:00") + self._set_settings() + + from datetime import timedelta + recent = (datetime.now(timezone.utc) - timedelta(minutes=2)).strftime("%Y-%m-%dT%H:%M:%S") + emails = [ + {"message_id": "", "subject": "A", "sender": "s@t.com", + "received_at": recent, "preview": "a"}, + {"message_id": "", "subject": "B", "sender": "s@t.com", + "received_at": recent, "preview": "b"}, + ] + + 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, \ + patch("outlook_web.services.telegram_push.TELEGRAM_PUSH_DELAY_SEC", 0): + self._run_job() + self.assertEqual(mock_send.call_count, 2) + + def test_push_log_cleanup(self): + """过期记录（>7 天）应被清理""" + with self.app.app_context(): + from outlook_web.db import get_db + db = get_db() + _insert_test_account(db, "clean@dedup.com", enabled=1, + last_checked="2026-03-01T00:00:00") + self._set_settings() + + # 插入一条旧记录和一条新记录 + db.execute( + "INSERT INTO telegram_push_log (account_id, message_id, pushed_at) VALUES (?, ?, ?)", + (999, "", "2026-01-01T00:00:00"), + ) + db.execute( + "INSERT INTO telegram_push_log (account_id, message_id, pushed_at) VALUES (?, ?, ?)", + (999, "", datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")), + ) + db.commit() + + # 运行 job 触发 cleanup + with patch("outlook_web.services.telegram_push._fetch_new_emails_imap", return_value=[]), \ + patch("outlook_web.services.telegram_push._send_telegram_message", return_value=True), \ + patch("outlook_web.services.telegram_push.TELEGRAM_PUSH_DELAY_SEC", 0): + self._run_job() + + # 旧记录应被清理 + old = db.execute( + "SELECT 1 FROM telegram_push_log WHERE message_id = ?", ("",) + ).fetchone() + self.assertIsNone(old, "超过 7 天的记录应被清理") + + # 新记录应保留 + new = db.execute( + "SELECT 1 FROM telegram_push_log WHERE message_id = ?", ("",) + ).fetchone() + self.assertIsNotNone(new, "7 天内的记录应保留") + + def test_email_without_message_id_still_pushed(self): + """缺少 message_id 的邮件仍然正常推送（降级到仅游标防重）""" + with self.app.app_context(): + from outlook_web.db import get_db + db = get_db() + _insert_test_account(db, "nomsgid@dedup.com", enabled=1, + last_checked="2026-03-01T00:00:00") + self._set_settings() + + from datetime import timedelta + recent = (datetime.now(timezone.utc) - timedelta(minutes=2)).strftime("%Y-%m-%dT%H:%M:%S") + emails = [{ + "subject": "No Message-ID", + "sender": "s@test.com", + "received_at": recent, + "preview": "body", + }] # 注意：没有 message_id 字段 + + 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, \ + patch("outlook_web.services.telegram_push.TELEGRAM_PUSH_DELAY_SEC", 0): + self._run_job() + mock_send.assert_called_once() + + if __name__ == "__main__": unittest.main()