tipbot/mail.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL2.0
# Copyright Thomas Gleixner <tglx@linutronix.de>
#

from datetime import timedelta, datetime, timezone
from email.message import EmailMessage
from email.utils import make_msgid
from email.utils import formatdate
from email.header import Header, decode_header
from email.mime.text import MIMEText
from email import message_from_string
import email.policy
import mailbox
import smtplib
import pygit2
import email
import time
import re
import os

def build_raw(addr):
    addr = addr.split('>')[0]
    try:
        return addr.split('<')[1]
    except:
        return addr

re_fromchars = re.compile('[^a-zA-Z0-9 ]')
re_compress_space = re.compile('\s+')

def clean_header(default, fallback):
    default = re_compress_space.sub(' ', default).strip()
    try:
        return default.encode('ascii').decode()
    except:
        try:
            return default.encode('UTF-8').decode()
        except:
            if fallback:
                return re_compress_space.sub(' ', fallback).strip()
            else:
                return default

def quote_name(name):
    name = name.strip()
    if re_fromchars.search(name):
        name = '"%s"' %name.replace('"', '')
    return name

def clean_from(tip, default, fallback):
    default = tip + ' ' + default
    if fallback:
        fallback = tip + ' ' + fallback
    res = clean_header(default, fallback)
    return quote_name(res)

def clean_cc(default, fallback, utf8=False):
    if default.find('>') > 0:
        default = default.split('>')[0] + '>'
    try:
        name, addr = default.split('<', 1)
        name = quote_name(name.strip())
        try:
            name = name.encode('ascii').decode()
        except:
            if not utf8:
                return fallback
            name = name.encode('UTF-8').decode()
        return name + ' <' + addr
    except:
        return fallback

class mailer(object):

    cctags = [
        "Reported-and-tested-by",
        "Reported-by",
        "Suggested-by",
        "Originally-from",
        "Originally-by",
        "Signed-off-by",
        "Tested-by",
        "Reviewed-by",
        "Acked-by",
        "Cc",
    ]

    def __init__(self, args, logger):
        self.args = args
        self.log = logger
        if args.mbox:
            self.mbox = os.path.abspath(args.mbox)

        if args.test:
            self.forceccs = []
            self.optccs = {}
        else:
            self.forceccs = vars(args).get('forceccs', [])
            self.optccs = vars(args).get('optccs', {})

    def send_mail(self, repo, branch, sha1):

        commit = repo.repo[sha1]
        subj = commit.message.split('\n')[0].strip()

        ccs = {}
        refid = None
        for l in commit.message.split('\n'):
            try:
                tag, rest = l.strip().split(':', 1)

                tag = tag.strip()
                rest = rest.strip()

                if tag in self.cctags and not self.args.test:
                    if rest.find('@') < 0:
                        continue
                    if rest.find('>') >= 0  and rest.find('<') >= 0:
                        mail = rest.rsplit('>', 1)[0] + '>'
                    else:
                        mail = rest

                    raw = build_raw(mail)
                    # Don't try the UTF8 header mess
                    # google mail is unhappy with that
                    addr = clean_cc(mail, raw)

                    if raw not in ccs:
                        ccs[raw] = addr

                elif tag == 'Link':
                    try:
                        mid = rest.rsplit('/', 1)[1]
                        if mid.find('@') > 0 and not refid:
                            refid = mid
                    except:
                        pass
            except:
                pass

        for cc in self.forceccs:
            raw = build_raw(cc)
            if raw not in ccs:
                ccs[raw] = cc

        for cc, branches in self.optccs.items():
            if branch in branches:
                raw = build_raw(cc)
                if raw not in ccs:
                    ccs[raw] = cc

        body  = ''
        if self.args.test:
            body += '\n-------------------- TEST ---------------------\n\n'

        body += 'The following commit has been merged into the %s branch of tip:\n\n' %branch

        body += 'Commit-ID:     %s\n' %sha1
        body += 'Gitweb:        https://git.kernel.org/tip/%s\n' %sha1
        body += 'Author:        %s <%s>\n' %(commit.author.name, commit.author.email)

        td = timedelta(minutes = commit.author.offset)
        tz = timezone(td)
        dt = datetime.fromtimestamp(commit.author.time, tz)
        tf = dt.strftime('%a, %d %b %Y %H:%M:%S %Z').replace('UTC', '')

        body += 'AuthorDate:    %s\n' %tf
        body += 'Committer:     %s <%s>\n' %(commit.committer.name, commit.committer.email)

        td = timedelta(minutes = commit.committer.offset)
        tz = timezone(td)
        dt = datetime.fromtimestamp(commit.committer.time, tz)
        tf = dt.strftime('%a, %d %b %Y %H:%M:%S %Z').replace('UTC', '')

        body += 'CommitterDate: %s\n' %tf

        body += '\n'
        body += commit.message
        body += '---\n'

        tree = commit.tree
        ptre = commit.parents[0].tree

        diff = ptre.diff_to_tree(tree)
        body += diff.stats.format(format=pygit2.GIT_DIFF_STATS_FULL |
                                  pygit2.GIT_DIFF_STATS_INCLUDE_SUMMARY,
                                  width=70)
        body += '\n'
        body += diff.patch

        body = body.encode('UTF-8').decode()

        msg = EmailMessage()

        msg['Return-path'] ='tip-bot2@linutronix.de'
        msg['Date'] = '%s' %formatdate()

        name = clean_from('tip-bot2 for', commit.author.name, commit.author.email.split('@')[0])
        mfrom = '%s <tip-bot2@linutronix.de>' %name
        msg['From'] = mfrom

        msg['Sender'] = 'tip-bot2@linutronix.de'
        msg.set_unixfrom('From tip-bot2 ' + time.ctime(time.time()))

        if not self.args.test:
            msg['Reply-to'] = 'linux-kernel@vger.kernel.org'
            msg['To'] = 'linux-tip-commits@vger.kernel.org'
        else:
            msg['Reply-to'] = 'tglx@linutronix.de'
            msg['To'] = 'Thomas Gleixner <tglx@linutronix.de>'

        subj = clean_header(subj, None)
        subj = '[tip: %s] %s' %(branch, subj)
        msg['Subject'] = subj

        if len(ccs) > 0:
            rcpt = ''
            for k, addr in ccs.items():
                rcpt += '%s, ' %addr
            msg['Cc'] = rcpt.rstrip(', ')

        if refid:
            msg['In-Reply-To'] = '<%s>' %refid
            msg['References'] ='<%s>' %refid

        if not msg.get('MIME-Version'):
            msg['MIME-Version'] = '1.0'
        msg['Message-ID'] = '%s' %make_msgid('tip-bot2')

        msg['X-Mailer'] = 'tip-git-log-daemon'
        msg['Robot-ID'] = '<tip-bot2.linutronix.de>'
        msg['Robot-Unsubscribe'] = 'Contact <mailto:tglx@linutronix.de> to get blacklisted from these emails'

        msg['Content-Type'] = 'text/plain'
        msg.set_param('charset', 'utf-8', header='Content-Type')
        msg['Content-Transfer-Encoding'] = '8bit'
        msg['Content-Disposition'] = 'inline'

        msg['Precedence'] = 'bulk'

        msg.set_content(body)

        if self.args.mbox:
            mbox = mailbox.mbox(self.mbox, create=True)
            try:
                mbox.add(msg)
            except:
                pol = email.policy.EmailPolicy(utf8=True)
                mbmsg = EmailMessage(pol)
                for k in msg:
                    if k not in mbmsg:
                        mbmsg[k] = msg[k]
                    mbmsg.set_content(msg.get_content())
                mbox.add(mbmsg)
            mbox.close()
        elif not self.args.smtp:
            print(msg.as_string())

        if self.args.smtp:
            to = msg['To']

            server = smtplib.SMTP('localhost')
            server.ehlo()
            server.send_message(msg)
            server.quit()