okooooooo/outlookEmailPlus
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: BUG-00011 P2 Message-ID 去重防止重复推送

Browse Source 
- 新增 telegram_push_log 表 (schema v5)，UNIQUE(account_id, message_id)
- IMAP fetch 提取 Message-ID header; Graph fetch 提取 internetMessageId
- 推送前查 push_log 跳过已推送邮件，推送后记录
- 每次 Job 结束清理 >7 天的去重记录
- 新增 4 个 TestMessageIdDedup 测试 + 更新 _make_email 含 message_id
- 228 tests passing (游标防重 + Message-ID 去重双保险)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
胡绍康andCopilot committed 2026-03-05 22:34:16 +08:00
1 parent d6af769401
commit 71b5cc0293
3 files changed
+257 -6
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

outlook_web/db.py
+27 -2
View File
@@ -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)
outlook_web/services/telegram_push.py
+67 -4
View File
@@ -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)
tests/test_telegram_push.py
+163
View File
@@ -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": "<future@example.com>",
"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": "<past@example.com>",
"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": "<skew-e2e@example.com>",
"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": "<unique123@mail.com>",
"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 = ?", ("<unique123@mail.com>",)
).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": "<aaa@mail.com>", "subject": "A", "sender": "s@t.com",
"received_at": recent, "preview": "a"},
{"message_id": "<bbb@mail.com>", "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, "<old@mail.com>", "2026-01-01T00:00:00"),
)
db.execute(
"INSERT INTO telegram_push_log (account_id, message_id, pushed_at) VALUES (?, ?, ?)",
(999, "<new@mail.com>", 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 = ?", ("<old@mail.com>",)
).fetchone()
self.assertIsNone(old, "超过 7 天的记录应被清理")
# 新记录应保留
new = db.execute(
"SELECT 1 FROM telegram_push_log WHERE message_id = ?", ("<new@mail.com>",)
).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()