将*单个* IMAP 邮件标记为未读

Posted 2023-03-15

【中文标题】将*单个* IMAP 邮件标记为未读

【英文标题】：Mark a *single* IMAP message as unread

【发布时间】：2017-07-26 15:57:40

【问题描述】：

在获取所有“未读”消息然后遍历它们并获取它们之后，我正在尝试操纵单个消息上的 IMAP 标志以将其标记为未读。

我不完全确定如何在单个消息的基础上将消息标记为未读/未见。我得到的只是消息号，我不确定如何正确存储 UID 以仅影响单个消息。

answer in a similar question 似乎不起作用，因为它将 错误 消息设置为“未读”。如何设置我再次获取为“未读”的单个邮件消息？

---

我被要求提供更多信息。在去掉这里“秘密”的细节的同时，这是我尝试实现的现有运行时，它尝试根据代码规则处理消息，并存储消息编号等，并尝试在将消息的 id 和主题存储在 pickle 文件中后，将每条消息设置为“未读”，因为在运行期间“看到”的任何内容都将在服务器上自动标记为“已读”，而不是设置为“未读”状态：

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，或者我在这里是否缺少其他东西。

【问题讨论】：

如果它在“错误”消息上设置标志，则您必须指定错误的消息 ID。没有任何details on what you're doing and what you're getting，我们不能说别的。

@ivan_pozdeev 添加了详细信息，例如代码等，以及我根据邮箱上的 IMAP 检查正在观察发生的内容是否未读已发布.

【参考方案1】：

好吧，我今天觉得自己很愚蠢。

显然，被迭代的消息编号和conn.uid(...) 期望的消息的 UID 不一定是相同的编号。我发现必须获取 UID 并进行一些获取后处理才能让 UID 传递出去。

---

原始方法

在上面的 for 循环中，我能够使用以下内容获取 UID：

for i in msgnums:
    # ...
    msg_uid = conn.fetch(i, 'UID')[1][0].split()[2].strip('()')
    # ...

这给了我conn.uid 所期待的消息的 UID，而不是普通的消息编号。没有意识到这一点，我觉得有点愚蠢，但这似乎解决了问题。

---

更新方法 #1（感谢 cmets 中的 @Max）

我用 UID 等效项替换了所有搜索/获取/存储命令。

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 模块，它“扩展”imaplib 的IMAP4 和IMAP4_SSL 函数，并用uid 覆盖“搜索”、“获取”和“存储”命令变体，但以其他方式保留来自 imaplib 的“搜索”、“获取”和“存储”命令，但返回基于 UID 函数的不同结果集。

这是我的imaplib_extension.py 文件中的内容，我只是从该模块导入IMAP4 或IMAP4_SSL，而不是直接从imaplib 导入，并将任何imaplib.IMAP4 和imaplib.IMAP4_SSL 调用替换为仅@ 987654346@ 或 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 而不是可能不是 UID 的“消息编号”。

---

更新方法#3

在意识到我在其他 Python 应用程序中需要这个之后，我下定决心并发布了imaplibext on PyPI，它基本上是上面方法#2 的改进和充实版本。但是，它确实具有更好的错误处理能力，以及为 IMAP 连接套接字实际指定超时的能力。这是一项改进，因为您不能直接为imaplib.IMAP4 或imaplib.IMAP4_SSL 执行此操作，除此之外，它本质上是imaplib 的直接替代品（尽管它的核心仍然使用imaplib）。

此代码也存在于GitHub，用于一般使用和改进建议以及问题报告。

【讨论】：

或者总是使用 UID，包括：conn.uid('SEARCH', ...) 和 conn.uid('FETCH', ..) 和 conn.uid('STORE')。通过UID SEARCH，您将永远不会有序列号，并且必须在它们和 UID 之间进行转换。

@Max 这是否适用于“获取所有未读邮件”？这样我们就需要使用“搜索”或其他东西并获得正确的消息 UID 来获取所有内容？

conn.uid('SEARCH', ...) 的工作方式与 conn.search(...) 完全相同，只是它返回 UID 而不是消息序列号。与 uid('FETCH', ...) vs fetch(...) 和 uid('STORE') vs store() 相同

@Max 感谢您的建议，基于 UID 的函数非常棒。它还导致我扩展 imaplib 以用基于 UID 的函数替换 fetch/store/search 函数，即使我不必这样做（我宁愿能够就地放置一个“覆盖”的新模块标准功能并且仍然有效，而不是...）

谢谢，这很有帮助