将*单个* IMAP 消息标记为未读
Mark a *single* IMAP message as unread
在获取所有 "UNREAD" 邮件然后遍历并获取它们之后，我试图在单个邮件上操作 IMAP 标志以将其标记为未读。
我不完全确定如何在单个消息的基础上标记消息 unread/unseen。我得到的只是消息编号，我不确定如何正确存储 UID 以仅影响单个消息。
answer in a similar question 似乎没有起作用，因为它将 错误的 消息设置为 'unread'。我如何重新设置我已提取为 'unread' 的 单个 邮件消息？
我被要求提供更多信息。在此处删除 'secret' 的细节时，这是我尝试实现的现有 运行time，这样它会尝试根据代码规则处理消息，并存储消息编号等，并在将消息的 ID 和主题存储在泡菜文件中后尝试将每条消息设置为 "UNREAD"，因为在 运行 期间 'seen' 的任何内容都将被标记为'read' 在服务器上自动设置为 'unread' status:
def main():
conn = imaplib.IMAP4('SERVER')
conn.login('username', 'passphrase')
conn.select('inbox')
(status, nums) = conn.search(None, '(UNSEEN)')
msgnums = map(int, nums[0].split())
for i in msgnums:
try:
raw_msg = conn.fetch(i, '(RFC822)')
raw_msg = conn.fetch(i, '(RFC822)')
msg = email.message_from_string(raw_msg[1][0][1])
body = "Date: %s\r\nSender: %s\r\nSubject: %s\r\n\r\n" % (msg['Date'], msg['From'], msg['Subject'])
msg_date = re.sub('/', '-', msg['Date']).replace(":", ".")
fdate = re.sub('\s+', '_', msg_date).replace(",", "")
print "Checking message: %s" % msg['Subject']
if not msg['Subject']:
continue # fname = "unknown_msg%d_%s" % (i,fdate)
elif msg['Subject'].lower().rfind('foobar') != -1:
print "Subject match 'foobar', processing: %s" % msg['Subject']
# We should have from the pickle an "observed" set of data, both subjects and message numbers.
if msg['Subject'] in PICKLED_MESSAGES['observed']['subjects']:
print "Already handled this message, moving on to next item."
# Since this was already observed we let it be removed so things don't rerun it later.
# noinspection PyBroadException
try:
PICKLED_MESSAGES['observed']['subjects'].remove(msg['Subject'])
PICKLED_MESSAGES['observed']['msgnums'].remove(i)
except:
pass
continue
else:
continue
# Do stuff with the message to store it in a special way on the filesystem
# Note that we've now looked at the message, so next-run we can see
# what was handled on the last run.
PICKLED_MESSAGES['observed']['msgnums'].append(i)
PICKLED_MESSAGES['observed']['subjects'].append(msg['Subject'])
print "PICKLED:\n%s" % PICKLED_MESSAGES['observed']
conn.uid('STORE', str(i), '-FLAGS', '(\Seen)')
except Exception:
conn.uid('STORE', str(i), '-FLAGS', '(\Seen)')
PICKLED_MESSAGES['observed']['msgnums'].remove(i)
PICKLED_MESSAGES['observed']['subjects'].remove(msg['Subject'])
print "PICKLED:\n%s\n" % PICKLED_MESSAGES
finally:
# Store the pickle file so we can use it next run.
cPickle.dump(PICKLED_MESSAGES, open('observed_msgs.pkl', 'wb'))
if __name__ == "__main__":
# pre-runtime checks - is IMAP up, etc. run first, then this:
# Initialize the PICKLED_MESSAGES data with pickle data or an empty
# structure for the pickle.
# noinspection PyBroadException
try:
PICKLED_MESSAGES = cPickle.load(open('observed_msgs.pkl', 'rb'))
except Exception as e:
PICKLED_MESSAGES = {
'observed': {
'msgnums': [],
'subjects': [],
},
}
# If all checks satisfied, continue and process the main() directives.
try:
main()
except Exception as e:
print("CRITICAL An unhandled error has occurred: %s" % str(e))
exit()
但是，不是将正确的消息设置为“未读；使用我在系统上看到的建议方法时。所以，我不完全确定我是否没有正确获取消息的 UID，或者我是否在这里遗漏了其他事情。
嗯，我今天觉得很傻。
显然正在迭代的消息编号和 conn.uid(...) 期望的消息的 UID 不一定是相同的编号。我发现必须获取 UID 并执行一些 post-fetch-processing 以获取要传递的 UID。
原始方法
在上面的 for 循环中，我能够通过以下方法获得 UID：
for i in msgnums:
# ...
msg_uid = conn.fetch(i, 'UID')[1][0].split()[2].strip('()')
# ...
这给了我 conn.uid 期望的消息的 UID，而不是普通的消息编号。我觉得有点愚蠢，因为没有意识到这一点，但这似乎解决了这个问题。
更新方法 #1（感谢评论中的 @Max）
我用 UID 等价物替换了所有 search/fetch/store 命令。
conn.search(None, '(UNSEEN)') 变为 conn.uid('SEARCH', None, '(UNSEEN)')conn.fetch(i, '(RFC822)') 变为 conn.uid('FETCH', i, '(RFC822)')conn.store(i, '-FLAGS', '(\Seen)') 变为 conn.uid('STORE', i, '-FLAGS', '(\Seen)')
更新方法#2（受#1 启发，但更进一步）
我基本上厌倦了写出 UID 命令，但还需要在另一个使用类似 IMAP 接口和命令的程序中应用类似的基于 UID 的功能。鉴于此，我决定编写一个 imaplib_extension.py 模块，'extends' imaplib 的 IMAP4 和 IMAP4_SSL 函数，并覆盖 "search"，"fetch" 和 "store" 命令带有 uid 变体，但在其他方面保持 "search"、"fetch" 和 "store" 命令原样来自 imaplib 但 returns 一个基于 UID 函数的不同结果集。
这是我的 imaplib_extension.py 文件中的内容，我只是从该模块而不是直接从 imaplib 导入 IMAP4 或 IMAP4_SSL，并替换任何 imaplib.IMAP4 和 imaplib.IMAP4_SSL 调用，稍后仅调用 IMAP4 或 IMAP4_SSL。因此，不需要import imaplib，只需from imaplib import IMAP4（或IMAP4_SSL）：
import imaplib
class IMAP4(imaplib.IMAP4):
def search(self, charset, *criteria):
# conn.uid('SEARCH', charset, criteria)
return self.uid('SEARCH', charset, " ".join(criteria))
def fetch(self, message_set, message_parts):
# conn.uid('FETCH', msgset, parts)
return self.uid('FETCH', message_set, message_parts)
def store(self, message_set, command, flags):
# conn.uid('STORE', msg_uid, '-FLAGS', '(\Seen)')
return self.uid('STORE', message_set, command, flags)
# noinspection PyPep8Naming
class IMAP4_SSL(imaplib.IMAP4_SSL):
def search(self, charset, *criteria):
# conn.uid('SEARCH', charset, criteria)
return self.uid('SEARCH', charset, " ".join(criteria))
def fetch(self, message_set, message_parts):
# conn.uid('FETCH', msgset, parts)
return self.uid('FETCH', message_set, message_parts)
def store(self, message_set, command, flags):
# conn.uid('STORE', msg_uid, '-FLAGS', '(\Seen)')
return self.uid('STORE', message_set, command, flags)
我更喜欢使用 imaplib 的这个扩展，因为命令结构与现有命令保持 相同，但可以正确使用 UID 而不是 'message numbers' 可能不是 UID。
更新方法 #3
在意识到我在 other Python 应用程序中需要这个之后，我开始努力并发布了 imaplibext on PyPI，这基本上是一个改进并充实了上面方法#2 的版本。但是，它确实具有更好的错误处理能力，并且能够为 IMAP 连接套接字实际指定超时。这是一项改进，因为您不能直接为 imaplib.IMAP4 或 imaplib.IMAP4_SSL 执行此操作，除此之外，它本质上是对 imaplib 的直接替代（尽管它的核心仍然使用imaplib).
此代码也存在于 GitHub 中，用于一般用途和改进建议以及问题报告。
在获取所有 "UNREAD" 邮件然后遍历并获取它们之后，我试图在单个邮件上操作 IMAP 标志以将其标记为未读。
我不完全确定如何在单个消息的基础上标记消息 unread/unseen。我得到的只是消息编号，我不确定如何正确存储 UID 以仅影响单个消息。
answer in a similar question 似乎没有起作用，因为它将 错误的 消息设置为 'unread'。我如何重新设置我已提取为 'unread' 的 单个 邮件消息？
我被要求提供更多信息。在此处删除 'secret' 的细节时，这是我尝试实现的现有 运行time，这样它会尝试根据代码规则处理消息，并存储消息编号等，并在将消息的 ID 和主题存储在泡菜文件中后尝试将每条消息设置为 "UNREAD"，因为在 运行 期间 'seen' 的任何内容都将被标记为'read' 在服务器上自动设置为 'unread' status:
def main():
conn = imaplib.IMAP4('SERVER')
conn.login('username', 'passphrase')
conn.select('inbox')
(status, nums) = conn.search(None, '(UNSEEN)')
msgnums = map(int, nums[0].split())
for i in msgnums:
try:
raw_msg = conn.fetch(i, '(RFC822)')
raw_msg = conn.fetch(i, '(RFC822)')
msg = email.message_from_string(raw_msg[1][0][1])
body = "Date: %s\r\nSender: %s\r\nSubject: %s\r\n\r\n" % (msg['Date'], msg['From'], msg['Subject'])
msg_date = re.sub('/', '-', msg['Date']).replace(":", ".")
fdate = re.sub('\s+', '_', msg_date).replace(",", "")
print "Checking message: %s" % msg['Subject']
if not msg['Subject']:
continue # fname = "unknown_msg%d_%s" % (i,fdate)
elif msg['Subject'].lower().rfind('foobar') != -1:
print "Subject match 'foobar', processing: %s" % msg['Subject']
# We should have from the pickle an "observed" set of data, both subjects and message numbers.
if msg['Subject'] in PICKLED_MESSAGES['observed']['subjects']:
print "Already handled this message, moving on to next item."
# Since this was already observed we let it be removed so things don't rerun it later.
# noinspection PyBroadException
try:
PICKLED_MESSAGES['observed']['subjects'].remove(msg['Subject'])
PICKLED_MESSAGES['observed']['msgnums'].remove(i)
except:
pass
continue
else:
continue
# Do stuff with the message to store it in a special way on the filesystem
# Note that we've now looked at the message, so next-run we can see
# what was handled on the last run.
PICKLED_MESSAGES['observed']['msgnums'].append(i)
PICKLED_MESSAGES['observed']['subjects'].append(msg['Subject'])
print "PICKLED:\n%s" % PICKLED_MESSAGES['observed']
conn.uid('STORE', str(i), '-FLAGS', '(\Seen)')
except Exception:
conn.uid('STORE', str(i), '-FLAGS', '(\Seen)')
PICKLED_MESSAGES['observed']['msgnums'].remove(i)
PICKLED_MESSAGES['observed']['subjects'].remove(msg['Subject'])
print "PICKLED:\n%s\n" % PICKLED_MESSAGES
finally:
# Store the pickle file so we can use it next run.
cPickle.dump(PICKLED_MESSAGES, open('observed_msgs.pkl', 'wb'))
if __name__ == "__main__":
# pre-runtime checks - is IMAP up, etc. run first, then this:
# Initialize the PICKLED_MESSAGES data with pickle data or an empty
# structure for the pickle.
# noinspection PyBroadException
try:
PICKLED_MESSAGES = cPickle.load(open('observed_msgs.pkl', 'rb'))
except Exception as e:
PICKLED_MESSAGES = {
'observed': {
'msgnums': [],
'subjects': [],
},
}
# If all checks satisfied, continue and process the main() directives.
try:
main()
except Exception as e:
print("CRITICAL An unhandled error has occurred: %s" % str(e))
exit()
但是，不是将正确的消息设置为“未读；使用我在系统上看到的建议方法时。所以，我不完全确定我是否没有正确获取消息的 UID，或者我是否在这里遗漏了其他事情。
嗯，我今天觉得很傻。
显然正在迭代的消息编号和 conn.uid(...) 期望的消息的 UID 不一定是相同的编号。我发现必须获取 UID 并执行一些 post-fetch-processing 以获取要传递的 UID。
原始方法
在上面的 for 循环中，我能够通过以下方法获得 UID：
for i in msgnums:
# ...
msg_uid = conn.fetch(i, 'UID')[1][0].split()[2].strip('()')
# ...
这给了我 conn.uid 期望的消息的 UID，而不是普通的消息编号。我觉得有点愚蠢，因为没有意识到这一点，但这似乎解决了这个问题。
更新方法 #1（感谢评论中的 @Max）
我用 UID 等价物替换了所有 search/fetch/store 命令。
conn.search(None, '(UNSEEN)')变为conn.uid('SEARCH', None, '(UNSEEN)')conn.fetch(i, '(RFC822)')变为conn.uid('FETCH', i, '(RFC822)')conn.store(i, '-FLAGS', '(\Seen)')变为conn.uid('STORE', i, '-FLAGS', '(\Seen)')
更新方法#2（受#1 启发，但更进一步）
我基本上厌倦了写出 UID 命令，但还需要在另一个使用类似 IMAP 接口和命令的程序中应用类似的基于 UID 的功能。鉴于此，我决定编写一个 imaplib_extension.py 模块，'extends' imaplib 的 IMAP4 和 IMAP4_SSL 函数，并覆盖 "search"，"fetch" 和 "store" 命令带有 uid 变体，但在其他方面保持 "search"、"fetch" 和 "store" 命令原样来自 imaplib 但 returns 一个基于 UID 函数的不同结果集。
这是我的 imaplib_extension.py 文件中的内容，我只是从该模块而不是直接从 imaplib 导入 IMAP4 或 IMAP4_SSL，并替换任何 imaplib.IMAP4 和 imaplib.IMAP4_SSL 调用，稍后仅调用 IMAP4 或 IMAP4_SSL。因此，不需要import imaplib，只需from imaplib import IMAP4（或IMAP4_SSL）：
import imaplib
class IMAP4(imaplib.IMAP4):
def search(self, charset, *criteria):
# conn.uid('SEARCH', charset, criteria)
return self.uid('SEARCH', charset, " ".join(criteria))
def fetch(self, message_set, message_parts):
# conn.uid('FETCH', msgset, parts)
return self.uid('FETCH', message_set, message_parts)
def store(self, message_set, command, flags):
# conn.uid('STORE', msg_uid, '-FLAGS', '(\Seen)')
return self.uid('STORE', message_set, command, flags)
# noinspection PyPep8Naming
class IMAP4_SSL(imaplib.IMAP4_SSL):
def search(self, charset, *criteria):
# conn.uid('SEARCH', charset, criteria)
return self.uid('SEARCH', charset, " ".join(criteria))
def fetch(self, message_set, message_parts):
# conn.uid('FETCH', msgset, parts)
return self.uid('FETCH', message_set, message_parts)
def store(self, message_set, command, flags):
# conn.uid('STORE', msg_uid, '-FLAGS', '(\Seen)')
return self.uid('STORE', message_set, command, flags)
我更喜欢使用 imaplib 的这个扩展，因为命令结构与现有命令保持 相同，但可以正确使用 UID 而不是 'message numbers' 可能不是 UID。
更新方法 #3
在意识到我在 other Python 应用程序中需要这个之后，我开始努力并发布了 imaplibext on PyPI，这基本上是一个改进并充实了上面方法#2 的版本。但是，它确实具有更好的错误处理能力，并且能够为 IMAP 连接套接字实际指定超时。这是一项改进，因为您不能直接为 imaplib.IMAP4 或 imaplib.IMAP4_SSL 执行此操作，除此之外，它本质上是对 imaplib 的直接替代（尽管它的核心仍然使用imaplib).
此代码也存在于 GitHub 中，用于一般用途和改进建议以及问题报告。