feat: add process_unread_days setting to email handler
linuztx committed
Mar 16, 2026 at 23:50 UTC
0eb6b3b4f3c1a7e3e96346110a9e8d6dca91a13b
5 files changed
+113
-9
plugins/_email_integration/default_config.yaml
+1
@@ -12,6 +12,7 @@ handlers: []
12
# poll_mode: seconds
13
# poll_interval_seconds: 15
14
# poll_interval_cron: "*/2 * * * *"
15
+# process_unread_days: 0
16
# sender_whitelist: []
17
# project: ""
18
# dispatcher_instructions: ""
plugins/_email_integration/extensions/python/job_loop/_10_email_poll.py
+9
@@ -56,6 +56,8 @@ async def _handler_poll_loop(handler_name: str) -> None:
56
_state_lock,
57
)
58
59
+ first_poll = True
60
+
61
while True:
62
config = plugins.get_plugin_config(PLUGIN_NAME) or {}
63
handlers = config.get("handlers", [])
@@ -69,6 +71,13 @@ async def _handler_poll_loop(handler_name: str) -> None:
71
try:
72
async with _state_lock:
73
state = _load_state()
74
+ # On first poll after startup, reset state so the first-run
75
+ # path processes recent unread emails via date-based search.
76
+ # Subsequent polls use normal UID tracking.
77
+ if first_poll:
78
+ if int(handler_cfg.get("process_unread_days", 0)) > 0:
79
+ state.pop(handler_name, None)
80
+ first_poll = False
81
await _poll_single_handler(handler_cfg, state)
82
_save_state(state)
83
except Exception as e:
plugins/_email_integration/helpers/handler.py
+30
-7
@@ -24,6 +24,7 @@ from plugins._email_integration.helpers.imap_client import (
24
connect_imap,
25
disconnect_imap,
26
fetch_new,
27
+ fetch_unread_since,
28
get_highest_uid,
29
connect_exchange,
30
fetch_unread_exchange,
@@ -78,9 +79,10 @@ async def _poll_single_handler(handler_cfg: dict, state: dict):
79
account_type = handler_cfg.get("account_type", "imap")
80
whitelist = handler_cfg.get("sender_whitelist") or []
81
last_uid = state.get(name, {}).get("last_uid", 0)
82
+ process_unread_days = int(handler_cfg.get("process_unread_days", 0))
83
84
if account_type == "exchange":
83
- messages = await _fetch_exchange(handler_cfg, whitelist)
85
+ messages = await _fetch_exchange(handler_cfg, whitelist, process_unread_days)
86
if messages:
87
await _dispatch_all(handler_cfg, messages)
88
return
@@ -92,11 +94,28 @@ async def _poll_single_handler(handler_cfg: dict, state: dict):
94
password=handler_cfg.get("password", ""),
95
)
96
try:
95
- # First run: record current highest UID, don't process old emails
97
+ # First run: optionally process unread from last N days
98
if last_uid == 0:
97
- highest = await get_highest_uid(client)
98
- state[name] = {"last_uid": highest}
99
- PrintStyle.info(f"Email ({name}): initialized, tracking from UID {highest}")
99
+ if process_unread_days > 0:
100
+ messages, highest = await fetch_unread_since(
101
+ client, DOWNLOAD_FOLDER, process_unread_days, whitelist or None,
102
+ )
103
+ highest = highest or await get_highest_uid(client)
104
+ state[name] = {"last_uid": highest}
105
+ if messages:
106
+ PrintStyle.info(
107
+ f"Email ({name}): processing {len(messages)} unread"
108
+ f" from last {process_unread_days} days"
109
+ )
110
+ await _dispatch_all(handler_cfg, messages)
111
+ else:
112
+ PrintStyle.info(
113
+ f"Email ({name}): no unread in last {process_unread_days} days"
114
+ )
115
+ else:
116
+ highest = await get_highest_uid(client)
117
+ state[name] = {"last_uid": highest}
118
+ PrintStyle.info(f"Email ({name}): initialized, tracking from UID {highest}")
119
return
120
121
messages, new_uid = await fetch_new(
@@ -114,13 +133,17 @@ async def _poll_single_handler(handler_cfg: dict, state: dict):
133
await disconnect_imap(client)
134
135
117
-async def _fetch_exchange(cfg: dict, whitelist: list[str]) -> list[InboundMessage]:
136
+async def _fetch_exchange(
137
+ cfg: dict, whitelist: list[str], since_days: int = 0,
138
+) -> list[InboundMessage]:
139
account = await connect_exchange(
140
server=cfg.get("imap_server", ""),
141
username=cfg.get("username", ""),
142
password=cfg.get("password", ""),
143
)
123
- return await fetch_unread_exchange(account, DOWNLOAD_FOLDER, whitelist or None)
144
+ return await fetch_unread_exchange(
145
+ account, DOWNLOAD_FOLDER, whitelist or None, since_days=since_days,
146
+ )
147
148
149
async def _dispatch_all(handler_cfg: dict, messages: list[InboundMessage]):
plugins/_email_integration/helpers/imap_client.py
+62
-2
@@ -1,4 +1,8 @@
1
-"""IMAP/Exchange email reader. No agent/tool dependencies."""
1
+"""
2
+IMAP/Exchange email reader.
3
+
4
+No agent/tool dependencies.
5
+"""
6
7
import asyncio
8
import email
@@ -6,6 +10,7 @@ import os
10
import re
11
import uuid
12
from dataclasses import dataclass, field
13
+from datetime import datetime, timedelta
14
from email.header import decode_header
15
from email.message import Message as EmailMessage
16
from fnmatch import fnmatch
@@ -133,6 +138,56 @@ async def get_highest_uid(client: IMAPClient) -> int:
138
return await loop.run_in_executor(None, _search)
139
140
141
+async def fetch_unread_since(
142
+ client: IMAPClient,
143
+ download_folder: str,
144
+ days: int,
145
+ sender_whitelist: list[str] | None = None,
146
+ max_messages: int = 10,
147
+) -> tuple[list[InboundMessage], int]:
148
+ """Fetch unread emails from the last N days. Returns (messages, highest_uid)."""
149
+ loop = asyncio.get_event_loop()
150
+ since_date = datetime.now() - timedelta(days=days)
151
+
152
+ def _search():
153
+ client.select_folder("INBOX")
154
+ try:
155
+ return client.gmail_search(
156
+ f"category:primary is:unread after:{since_date.strftime('%Y/%m/%d')}"
157
+ )
158
+ except Exception:
159
+ return client.search(["UNSEEN", "SINCE", since_date.date()]) # type: ignore[arg-type]
160
+
161
+ msg_ids = await loop.run_in_executor(None, _search)
162
+ if not msg_ids:
163
+ return [], 0
164
+
165
+ highest_uid = max(msg_ids)
166
+
167
+ if len(msg_ids) > max_messages:
168
+ PrintStyle.standard(
169
+ f"Email: {len(msg_ids)} unread, processing latest {max_messages}"
170
+ )
171
+ msg_ids = msg_ids[-max_messages:]
172
+ else:
173
+ PrintStyle.standard(
174
+ f"Email: found {len(msg_ids)} unread messages from last {days} days"
175
+ )
176
+
177
+ results: list[InboundMessage] = []
178
+ for msg_id in msg_ids:
179
+ try:
180
+ msg = await _fetch_single(client, msg_id, download_folder, sender_whitelist)
181
+ if msg:
182
+ results.append(msg)
183
+ except Exception as e:
184
+ PrintStyle.error(
185
+ f"Email: error processing message {msg_id}: {format_error(e)}"
186
+ )
187
+
188
+ return results, highest_uid
189
+
190
+
191
async def _fetch_single(
192
client: IMAPClient,
193
msg_id: int,
@@ -208,13 +263,18 @@ async def fetch_unread_exchange(
263
account,
264
download_folder: str,
265
sender_whitelist: list[str] | None = None,
266
+ since_days: int = 0,
267
) -> list[InboundMessage]:
268
from exchangelib import Q
269
270
loop = asyncio.get_event_loop()
271
272
def _sync():
217
- return list(account.inbox.filter(Q(is_read=False)))
273
+ q = Q(is_read=False)
274
+ if since_days > 0:
275
+ since = datetime.now(tz=account.default_timezone) - timedelta(days=since_days)
276
+ q &= Q(datetime_received__gte=since)
277
+ return list(account.inbox.filter(q))
278
279
items = await loop.run_in_executor(None, _sync)
280
results: list[InboundMessage] = []
plugins/_email_integration/webui/config.html
+11
@@ -24,6 +24,7 @@
24
poll_mode: 'seconds',
25
poll_interval_seconds: 15,
26
poll_interval_cron: '*/2 * * * *',
27
+ process_unread_days: 0,
28
sender_whitelist: [],
29
project: '',
30
dispatcher_instructions: '',
@@ -194,6 +195,16 @@
195
</div>
196
</div>
197
198
+ <div class="field">
199
+ <div class="field-label">
200
+ <div class="field-title">Process Unread (days)</div>
201
+ <div class="field-description">On first run, process unread emails from the last N days. 0 = skip existing emails.</div>
202
+ </div>
203
+ <div class="field-control">
204
+ <input type="number" x-model.number="handler.process_unread_days" min="0" placeholder="0" />
205
+ </div>
206
+ </div>
207
+
208
<div class="field">
209
<div class="field-label">
210
<div class="field-title">Sender Whitelist</div>