PostfixMailQueue

#!/usr/bin/python

"""
#define REC_TYPE_SIZE   'C'     /* first record, created by cleanup */
#define REC_TYPE_TIME   'T'     /* time stamp, required */
#define REC_TYPE_FULL   'F'     /* full name, optional */
#define REC_TYPE_INSP   'I'     /* inspector transport */
#define REC_TYPE_FILT   'L'     /* loop filter transport */
#define REC_TYPE_FROM   'S'     /* sender, required */
#define REC_TYPE_DONE   'D'     /* delivered recipient, optional */
#define REC_TYPE_RCPT   'R'     /* todo recipient, optional */
#define REC_TYPE_ORCP   'O'     /* original recipient, optional */
#define REC_TYPE_WARN   'W'     /* warning message time */
#define REC_TYPE_ATTR   'A'     /* named attribute for extensions */

#define REC_TYPE_RDR    '>'     /* redirect target */
#define REC_TYPE_FLGS   'f'     /* cleanup processing flags */

#define REC_TYPE_MESG   'M'     /* start message records */

#define REC_TYPE_CONT   'L'     /* long data record */
#define REC_TYPE_NORM   'N'     /* normal data record */

#define REC_TYPE_XTRA   'X'     /* start extracted records */

#define REC_TYPE_RRTO   'r'     /* return-receipt, from headers */
#define REC_TYPE_ERTO   'e'     /* errors-to, from headers */
#define REC_TYPE_PRIO   'P'     /* priority */
#define REC_TYPE_VERP   'V'     /* VERP delimiters */

#define REC_TYPE_END    'E'     /* terminator, required */
"""

import os, sys, string, re, time

VRecordError = "VRecordError"

class VRecord :
    def __init__(self,file=None) :
    if not hasattr(self, "_allowed_types") :
        self._allowed_types = "CTFILSDROWA>fMLNXrePVE"
    if file == None :
        self._type = chr(255)
        self._length = 0
        self._data = ""
        return

    self._type = file.read(1)
    if len(self._type) == 0 :
        raise VRecordError, "Unable to read type"
    if self._type not in self._allowed_types :
        file.seek(file.tell()-1)
        raise VRecordError, "Invalid record type '%s'" % self._type

    try :
        len_byte = ord(file.read(1))
    except TypeError :
        file.seek(file.tell()-1)
        raise VRecordError, "Unable to read first byte of length"

    length = len_byte & 0177
    i = 1
    while len_byte & 0200 :
        try :
        len_byte = ord(file.read(1))
        except TypeError :
        file.seek(file.tell()-1)
        raise VRecordError, "Unable to read byte %d of length" % i+1
        length = length + (len_byte & 0177) * 128 ** i
        i = i+1

    self._length = length

    buffer = ""
    while length >= 1024 :      # Should be able to read this much
        buffer = buffer + file.read(1024)
        length = length - 1024
    self._data = buffer + file.read(length)

    def __str__(self) :
    if self._data :
        return '<VRecord(%s:%d) "%s">' % \
           (self._type, self._length, self._data[:30])
    else :
        return '<VRecord(%s:%d) "">' % (self._type, self._length)

class EnvelopeRecord(VRecord) :
    def __init__(self, file=None) :
    self._allowed_types = "MCTFILSDROWVA"
    VRecord.__init__(self, file)

class MailRecord(VRecord) :
    def __init__(self, file=None) :
    self._allowed_types = "N"
    VRecord.__init__(self, file)

class QueueFile :
    def __init__(self, filename) :
    self._file = filename
    self._envelope = []
    self._mail = []
    self._client = None
    self._sender = None
    self._recipients = None
    self._mark = ""
        try :
            file = open(filename)
        except :
        raise ValueError, "Unable to read queue file %s" % self._file


    record = VRecord(file)
    while record._type in "MCTFILSDROWVA" :
        self._envelope.append(record)
        record = VRecord(file)
    while record._type in "XNAe" :
        self._mail.append(record)
        record = VRecord(file)
    if record._type != "E" :
        print record
        raise ValueError, "Unable to read queue file %s" % self._file

    def message(self, lf="\n", lines=-1) :
    print type(self._mail[0]._data), type(lf)
    buffer = self._mail[0]._data + lf
    for record in self._mail[1:] :
        buffer = buffer + record._data + lf
        lines = lines - 1
        if lines == 0 : break
    return buffer

    def node(self) :
    return string.split(self._file, os.sep)[-1]

    def client(self) :
    if self._client == None :
        for record in self._envelope :
        if record._type == "A" and \
           record._data[:20] == "reverse_client_name=" :
            self._client = record._data[20:]
            return self._client
        self._client = ""
    return self._client

    def sender(self) :
    if self._sender == None :
        self._sender = ""
        for record in self._envelope :
        if record._type == "S" :
            self._sender = record._data
            return self._sender
    return self._sender

    def recipients(self) :
    if self._recipients == None :
        self._recipients = []
        for record in self._envelope :
        if record._type == "R" :
            self._recipients.append(record._data)
    return self._recipients

    def mark(self, mark=None) :
    if mark != None :
        self._mark = mark
    return self._mark

class MonitorQueue :
    """
    This takes a top directory, and adds all files (in all sub-directories)
    that are considered a Postfix mail queue files. It stores the clients,
    the recipients and the senders from the envelopes.
    """
    def __init__(self, top) :
        self._top = []
        self._queue = {}                # The current set of evelope data
        self.addDirectory(top)

    def addDirectory(self, top) :
        if top[-1] == os.sep :
            top = top[:-1]
        if os.path.isdir(top) and \
           not top in self._top :
            self._top.append(top)
        self.update(top)

    def update(self, top=None) :
        if top == None :
            count = 0
            rd = {}
            nd = {}
            for top in self._top :
                c,r,n = self.update(top)
                count = count+c
                for x in r :
                    rd[x] = None
                for x in n :
                    nd[x] = None
            return (count, rd.keys(), nd.keys())

        files = self._queue.keys()
        new_files = []

        DIRS = [ top ]
        count = 0
        while DIRS :
            DIR, DIRS = DIRS[0], DIRS[1:]
            for name in os.listdir(DIR) :
                filename = DIR + os.sep + name
                if self._queue.has_key(filename) :
                    count = count + 1
                    continue          # No need to this more than once
                elif os.path.islink(filename) :
                    continue          # You might or might not want this
                elif os.path.isdir(filename) :
                    DIRS.append(filename)
                elif os.path.isfile(filename) :
                    count = count+1
                    try :
                        qfile = QueueFile(filename)
                    except ValueError :
                        continue
                    new_files.append(filename)
                    client = qfile.client()
                    sender = qfile.sender()
                    recipients = qfile.recipients()
                    self._queue[filename] = (client, sender, recipients)
                else :
                    # print filename, "ignored"
                    pass
        removed = []
        for filename in self._queue.keys() :
            if not os.path.isfile(filename) :
                removed.append(filename)
                del self._queue[filename]
        return (count, removed, new_files)

    def display(self) :
        files = self._queue.keys()
        files.sort()
        for filename in files :
            client = self._queue[filename][0]
            sender = self._queue[filename][1]
            recipients = ", ".join(self._queue[filename][2])
            print "%s:%s:%s:%s" % (filename,client,sender,recipients)

    def status(self, amount=30) :
        "Gives out the most common clients, sender domains, and recipient domains"
        files = self._queue.keys()
        files.sort()
        clients = {}
        senders = {}
        recipients = {}
        rx = re.compile(r".*@(.*)")
        for filename in files :
            client = self._queue[filename][0]
            if not client :
                client = "<Local>"
            try :
                clients[client] = clients[client] + 1
            except KeyError :
                clients[client] = 1

            sender = self._queue[filename][1]
            m = rx.match(sender)
            if m :
                domain = m.group(1)
                try :
                    senders[domain] = senders[domain] + 1
                except KeyError :
                    senders[domain] = 1
            elif sender :
                # print 'Unable to deal with sender "%s" for %s' % (sender, filename)
                pass

            for recipient in self._queue[filename][2] :
                m = rx.match(recipient)
                domain = m.group(1)
                try :
                    recipients[domain] = recipients[domain] + 1
                except KeyError :
                    recipients[domain] = 1

        keys = clients.keys()
        keys.sort(lambda x,y,d=clients : cmp(d[y], d[x]))
        rclient = map(lambda x,d=clients : (d[x], x), keys[:amount])

        keys = senders.keys()
        keys.sort(lambda x,y,d=senders : cmp(d[y], d[x]))
        rsender = map(lambda x,d=senders : (d[x], x), keys[:amount])

        keys = recipients.keys()
        keys.sort(lambda x,y,d=recipients : cmp(d[y], d[x]))
        rrecipient = map(lambda x,d=recipients : (d[x], x), keys[:amount])

        return (rclient, rsender, rrecipient)

    def show_status(self, old_status=None) :
        rclient, rsender, rrecipient = status = self.status()

        # print old_status
        # print status

        clients = {}
        corder = map(lambda x :x[1], rclient)
        for client in rclient :
            clients[client[1]] = client[0]
        # print clients
        for value, client in old_status[0] :
            if clients.has_key(client) :
                clients[client] = (clients[client], clients[client]-value)

        senders = {}
        sorder = map(lambda x :x[1], rsender)
        for sender in rsender :
            senders[sender[1]] = sender[0]
        for value, sender in old_status[1] :
            if senders.has_key(sender) :
                senders[sender] = (senders[sender], senders[sender]-value)

        recipients = {}
        corder = map(lambda x :x[1], rrecipient)
        for recipient in rrecipient :
            recipients[recipient[1]] = recipient[0]
        for value, recipient in old_status[2] :
            if recipients.has_key(recipient) :
                recipients[recipient] = (recipients[recipient], recipients[recipient]-value)
        return status

    def monitor(self) :

        old_status = ([],[],[])
        monitor = Monitor()
        encounted = []
        while 2+2 == 4 :
            start = time.time()
            x = self.update()
            count, removed, new_files = x
            encounted.append(count)
            if len(encounted) > 50 :
                encounted = encounted[-50:]
            updated = time.time()
            # print "Removed %d and found %d in %3.1f seconds" % (len(removed), len(new_files), updated-start)
            # Main part here
            rclient, rsender, rrecipient = status = self.status()

            monitor.wmailq(0, "Mail Queue", encounted)
            monitor.wdisplay(1, "Client", rclient, old_status[0])
            monitor.wdisplay(2, "Sender", rsender, old_status[1])
            monitor.wdisplay(3, "Recipient", rrecipient, old_status[2])

            old_status = status
            # Wrap up for next lap
            time.sleep(30.0 + start - updated)

import curses
class Monitor :
    def __init__(self) :
        self._screen = curses.initscr()
        h, w = self._screen.getmaxyx()
        h2 = h/2-1
        w2 = w/2-1
        self._windows = []
        self._windows.append(curses.newwin(h/2, w/2, 0, 0))
        self._windows.append(curses.newwin(h/2, w/2, 0, w/2))
        self._windows.append(curses.newwin(h/2, w/2, h/2, 0))
        self._windows.append(curses.newwin(h/2, w/2, h/2, w/2))
        for window in self._windows :
            window.border()
            window.refresh()

    def wtitle(self, n, title) :
        window = self._windows[n]
        window.clear()
        window.border()
        h, w = window.getmaxyx()
        offset = (w - len(title) - 2)/2
        window.addstr(0, offset, " %s " % title)

    def wline(self, n, i, text, value, old=None) :
        window = self._windows[n]
        h, w = window.getmaxyx()
        if i < h-2 :
            window.addnstr(i+1, 1, text, w-15)
            window.addnstr(i+1, w-13, "%-6d" % value, 6)
            if old :
                window.addnstr(i+1, w-7, "[%4d]" % (value-old), 6)

    def wmailq(self, n, title, queue) :
        self.wtitle(n, title)
        window = self._windows[n]
        h, w = window.getmaxyx()
        queue = queue[-(h-2):]
        m = max(queue)
        size = 1000
        while size <= m :
            size = size * 2
        window.addstr(0, w-14, " Scale %d " % size)
        i = 0
        while i < len(queue) :
            c = (size / (h-9))
            window.addstr(i+1, 1, "*" * ((queue[i] * (w-9))/size))
            window.addstr(i+1, w-7, "%-5d" % queue[i])
            i = i + 1
        window.refresh()

    def wdisplay(self, n, title, current, old) :
        self.wtitle(n, title)
        data = {}
        for value, name in old :
            data[name] = value
        i = 0
        while i < len(current) :
            value, name = current[i]
            try :
                self.wline(n, i , name, value, data[name])
            except KeyError :
                self.wline(n, i , name, value)
            i = i+1
        self._windows[n].refresh()

if __name__ == '__main__' :
    queue = MonitorQueue("/postfix/var/spool/postfix/deferred/")
    queue.addDirectory("/postfix/var/spool/postfix/active/")
    queue.monitor()