src/bugspray/git2bz.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# SPDX-License-Identifier: GPL-2.0-or-later
# Copyright (C) 2023 by the Linux Foundation

import argparse
import bugspray
import bugspray.parse
import b4
import email.message
import re

logger = bugspray.logger
b4.logger = logger
# force b4 to use EmailMessage factory
b4.emlpolicy = bugspray.emlpolicy


def update_component(product: str, component: str, dry_run: bool = False):
    logger.info('Running git2bz for %s/%s, dry_run=%s', product, component, dry_run)
    cconf = bugspray.get_component_config(product, component)
    cres = cconf.get('git_log_closes_regexes', list())
    mres = cconf.get('git_log_mentions_regexes', list())
    since = cconf.get('git_log_since', '1.week')
    tpt = bugspray.get_template_by_product_component('new_commit_notify', product, component)
    for entry in cconf.get('git_repos', list()):
        repo = entry[0]
        try:
            branch = entry[1]
        except IndexError:
            branch = 'master'
        try:
            commit_mask = entry[2]
        except IndexError:
            commit_mask = None

        logger.debug('Looking at %s:%s', repo, branch)
        for query in cconf.get('git_log_queries', list()):
            logger.debug(' query=%s', query)
            gitargs = ['log', '--grep', query, '-F', '--pretty=oneline', f'--since={since}', branch]
            lines = b4.git_get_command_lines(repo, gitargs)
            if not lines:
                logger.debug('No matches for %s', query)
                continue
            logger.debug(' query returned %s commits', len(lines))
            for line in lines:
                csha = line.split(maxsplit=1)[0]
                known_bids = bugspray.db_get_bugs_for_commit(csha)
                if len(known_bids):
                    logger.debug('This commit already processed, bugs: %s', known_bids)
                    continue

                bid = None
                gitargs = ['show', '-s', '--format=%an---%B', csha]
                ecode, cshow = b4.git_run_command(repo, gitargs)
                if ecode > 0:
                    logger.debug('Could not get commit body from %s', csha)
                    continue
                cauthor, cmsg = cshow.split('---', maxsplit=1)

                # Check closes regexes
                closes = False
                for cre in cres:
                    matches = re.search(cre, cmsg, flags=re.I | re.M)
                    if matches:
                        bid = int(matches.groups()[0])
                        logger.debug(' closes bug %s', bid)
                        closes = True
                        break

                if not closes:
                    found = False
                    # Check mentions regexes
                    for mre in mres:
                        matches = re.search(mre, cmsg, flags=re.I | re.M)
                        if matches:
                            bid = int(matches.groups()[0])
                            logger.debug(' mentions bug %s', bid)
                            found = True
                            break
                    if not found:
                        logger.debug('No regexes matched')
                        continue

                if not bid:
                    logger.debug('Could not get bug_id from comment %s', csha)
                    continue

                vals = {
                    'commit_author': cauthor.strip(),
                    'commit_id': csha,
                    'commit_text': cmsg.strip(),
                    'commit_url': '',
                }
                if commit_mask:
                    vals['commit_url'] = commit_mask % csha

                desc = tpt.safe_substitute(vals)
                msg = email.message.EmailMessage()
                body = bugspray.add_bot_signature(desc)
                msg.set_payload(body, charset='utf-8')
                if not dry_run:
                    cid = bugspray.bz_add_new_comment(bid, desc)
                    msgid = bugspray.notify_bug(bid, cid, msg, dry_run=dry_run)
                    if msgid:
                        bugspray.db_store_msgid_bid_cid(msgid, bid, cid)
                    bugspray.db_store_bug_for_commit(csha, bid)
                    if closes:
                        status, resolution = cconf.get('git_closes_with', ['RESOLVED', 'FIXED'])
                        logger.info('Commit %s closes bug %s with %s/%s', csha, bid, status, resolution)
                        bugspray.bz_set_bug_status_resolution(bid, status, resolution)
                    else:
                        logger.info('Commit %s mentions bug %s', csha, bid)
                else:
                    bugspray.notify_bug(bid, None, msg, dry_run=dry_run)


def main(cmdargs: argparse.Namespace):
    config = bugspray.get_config()
    # Iterate all components
    for bz_product, bz_components in config['components'].items():
        for bz_component in bz_components.keys():
            if config['components'][bz_product][bz_component].get('git_repos') is None:
                continue
            update_component(bz_product, bz_component, dry_run=cmdargs.dry_run)