File Manager

Editing File: /usr/local/cpanel/3rdparty/mailman/cron/checkdbs

#! /usr/local/cpanel/3rdparty/bin/python2
#
# Copyright (C) 1998-2018 by the Free Software Foundation, Inc.
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301,
# USA.

"""Check for pending admin requests and mail the list owners if necessary.

Usage: %(PROGRAM)s [options]

Options:

-h/--help
        Print this message and exit.
"""

import sys
import time
import getopt
from types import UnicodeType

import paths

# Import this after paths so we get Mailman's copy of the email package
from email.Charset import Charset

from Mailman import mm_cfg
from Mailman import Utils
from Mailman import MailList
from Mailman import Message
from Mailman import i18n

# Work around known problems with some RedHat cron daemons
import signal
signal.signal(signal.SIGCHLD, signal.SIG_DFL)

NL = '\n'
PROGRAM = sys.argv[0]

_ = i18n._
i18n.set_language(mm_cfg.DEFAULT_SERVER_LANGUAGE)

now = time.time()

def usage(code, msg=''):
    if code:
        fd = sys.stderr
    else:
        fd = sys.stdout
    print >> fd, _(__doc__)
    if msg:
        print >> fd, msg
    sys.exit(code)

def main():
    try:
        opts, args = getopt.getopt(sys.argv[1:], 'h', ['help'])
    except getopt.error, msg:
        usage(1, msg)

for opt, arg in opts:
        if opt in ('-h', '--help'):
            usage(0)

if args:
        usage(1)

for name in Utils.list_names():
        # the list must be locked in order to open the requests database
        mlist = MailList.MailList(name)
        try:
            count = mlist.NumRequestsPending()
            # While we're at it, let's evict yesterday's autoresponse data
            midnightToday = Utils.midnight()
            evictions = []
            for sender in mlist.hold_and_cmd_autoresponses.keys():
                date, respcount = mlist.hold_and_cmd_autoresponses[sender]
                if Utils.midnight(date) < midnightToday:
                    evictions.append(sender)
            if evictions:
                for sender in evictions:
                    del mlist.hold_and_cmd_autoresponses[sender]
                # Only here have we changed the list's database
                mlist.Save()
            if count:
                i18n.set_language(mlist.preferred_language)
                realname = mlist.real_name
                discarded = auto_discard(mlist)
                if discarded:
                    count = count - discarded
                    text = _(
             'Notice: %(discarded)d old request(s) automatically expired.\n\n')
                else:
                    text = ''
                if count:
                    text += Utils.maketext(
                        'checkdbs.txt',
                        {'count'    : count,
                         'host_name': mlist.host_name,
                         'adminDB'  : mlist.GetScriptURL('admindb', absolute=1),
                         'real_name': realname,
                         }, mlist=mlist)
                    text += '\n' + pending_requests(mlist)
                    subject = _(
                        '%(count)d %(realname)s moderator request(s) waiting')
                else:
                    subject = _(
                        '%(realname)s moderator request check result')
                msg = Message.UserNotification(mlist.GetOwnerEmail(),
                                               mlist.GetBouncesEmail(),
                                               subject, text,
                                               mlist.preferred_language)
                msg.send(mlist, **{'tomoderators': 1})
        finally:
            mlist.Unlock()

def pending_requests(mlist):
    # Must return a byte string
    lcset = Utils.GetCharSet(mlist.preferred_language)
    pending = []
    first = 1
    for id in mlist.GetSubscriptionIds():
        if first:
            pending.append(_('Pending subscriptions:'))
            first = 0
        when, addr, fullname, passwd, digest, lang = mlist.GetRecord(id)
        if fullname:
            if isinstance(fullname, UnicodeType):
                fullname = fullname.encode(lcset, 'replace')
            fullname = ' (%s)' % fullname
        pending.append('    %s%s %s' % (addr, fullname, time.ctime(when)))
    first = 1
    for id in mlist.GetUnsubscriptionIds():
        if first:
            pending.append(_('Pending unsubscriptions:'))
            first = 0
        addr = mlist.GetRecord(id)
        pending.append('    %s' % addr)
    first = 1
    for id in mlist.GetHeldMessageIds():
        if first:
            pending.append(_('\nPending posts:'))
            first = 0
        info = mlist.GetRecord(id)
        when, sender, subject, reason, text, msgdata = mlist.GetRecord(id)
        subject = Utils.oneline(subject, lcset)
        date = time.ctime(when)
        reason = _(reason)
        pending.append(_("""\
From: %(sender)s on %(date)s
Subject: %(subject)s
Cause: %(reason)s"""))
        pending.append('')
    # Coerce all items in pending to a Unicode so we can join them
    upending = []
    charset = Utils.GetCharSet(mlist.preferred_language)
    for s in pending:
        if isinstance(s, UnicodeType):
            upending.append(s)
        else:
            upending.append(unicode(s, charset, 'replace'))
    # Make sure that the text we return from here can be encoded to a byte
    # string in the charset of the list's language.  This could fail if for
    # example, the request was pended while the list's language was French,
    # but then it was changed to English before checkdbs ran.
    text = u'\n'.join(upending)
    charset = Charset(Utils.GetCharSet(mlist.preferred_language))
    incodec = charset.input_codec or 'ascii'
    outcodec = charset.output_codec or 'ascii'
    if isinstance(text, UnicodeType):
        return text.encode(outcodec, 'replace')
    # Be sure this is a byte string encodeable in the list's charset
    utext = unicode(text, incodec, 'replace')
    return utext.encode(outcodec, 'replace')

def auto_discard(mlist):
    # Discard old held messages
    discard_count = 0
    expire = mlist.max_days_to_hold * 86400 # days
    heldmsgs = mlist.GetHeldMessageIds()
    if expire and len(heldmsgs):
        for id in heldmsgs:
            if now - mlist.GetRecord(id)[0] > expire:
                mlist.HandleRequest(id, mm_cfg.DISCARD)
                discard_count += 1
        mlist.Save()
    return discard_count

if __name__ == '__main__':
    main()

Back to Directory File Manager