peebz/pi2bz.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 peebz
import peebz.parse
import b4
import urllib.parse
import email.message
import gzip
import datetime

from typing import List, Set

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


def get_query_results(query_url: str) -> List:
    loc = urllib.parse.urlparse(query_url)
    logger.debug('query=%s', query_url)
    logger.debug('grabbing search results from %s', loc.netloc)
    session = peebz.get_requests_session()
    # For the query to retrieve a mbox file, we need to send a POST request
    resp = session.post(query_url, data='')
    if resp.status_code == 404:
        raise LookupError('Nothing matching query=%s', query_url)
    if resp.status_code != 200:
        raise LookupError('Server returned an error for %s: %s' % (query_url, resp.status_code))
    t_mbox = gzip.decompress(resp.content)
    resp.close()
    if not len(t_mbox):
        raise LookupError('Nothing matching query=%s', query_url)
    return b4.split_and_dedupe_pi_results(t_mbox)


def get_sorted_thread(url: str, msgid: str) -> List[email.message.EmailMessage]:
    loc = urllib.parse.urlparse(url)
    mbox_url = url.rstrip('/') + '/' + urllib.parse.quote_plus(msgid) + '/t.mbox.gz'
    logger.debug('mbox_url=%s', mbox_url)
    logger.debug('grabbing thread from %s', loc.netloc)
    session = peebz.get_requests_session()
    resp = session.get(mbox_url)
    if resp.status_code == 404:
        raise LookupError('Nothing matching mbox_url=%s', mbox_url)
    if resp.status_code != 200:
        raise LookupError('Server returned an error for %s: %s' % (mbox_url, resp.status_code))
    t_mbox = gzip.decompress(resp.content)
    resp.close()

    deduped = b4.split_and_dedupe_pi_results(t_mbox)
    if not deduped:
        raise LookupError('No messages matching mbox_url=%s' % mbox_url)
    strict = b4.get_strict_thread(deduped, msgid)
    return peebz.sort_msgs_by_received(strict)


def get_new_msgs(msgs: List[email.message.EmailMessage]) -> List[email.message.EmailMessage]:
    new_msgs = list()
    for msg in msgs:
        msgid = b4.LoreMessage.get_clean_msgid(msg)
        try:
            peebz.db_get_bid_cid_by_msgid(msgid)
            continue
        except LookupError:
            new_msgs.append(msg)

    return new_msgs


def get_tracked_bug_msgids(product: str, component: str) -> Set[str]:
    cconf = peebz.get_component_config(product, component)
    params = {
        'include_fields': 'id',
        'product': product,
        'component': component,
        'quicksearch': 'OPEN',
        'chfieldfrom': '90d',
    }
    params.update(cconf.get('bz_query_params', dict()))
    rdata = peebz.bz_rest('bug', params=params)
    msgids = set()
    for bdata in rdata.get('bugs', list()):
        bid = bdata['id']
        try:
            msgid = peebz.db_get_msgid_by_bid_cid(bid, None)
            logger.debug('bid=%s is tracked as msgid=%s', bid, msgid)
            msgids.add(msgid)
        except LookupError:
            logger.debug('Not tracking bid=%s', bid)

    return msgids


def update_component(product: str, component: str, dry_run: bool = False):
    logger.info('Running pi2bz for %s/%s, dry_run=%s', product, component, dry_run)
    cconf = peebz.get_component_config(product, component)
    tracked = get_tracked_bug_msgids(product, component)
    url = cconf.get('pi_url').rstrip('/')
    now = datetime.datetime.utcnow()

    seen_msgids = set()
    updates = list()
    if len(tracked):
        logger.info('Checking for updates in %s tracked threads', len(tracked))
        for msgid in tracked:
            try:
                tmsgs = get_sorted_thread(url, msgid)
            except LookupError:
                logger.debug('No results returned for msgid=%s', msgid)
                continue

            for tmsg in tmsgs:
                tmsgid = b4.LoreMessage.get_clean_msgid(tmsg)
                if tmsgid in seen_msgids:
                    logger.debug('Already seen %s', tmsgid)
                    continue
                seen_msgids.add(tmsgid)
                try:
                    peebz.db_get_bid_cid_by_msgid(tmsgid)
                    logger.debug('%s has already been processed', tmsgid)
                    continue
                except LookupError:
                    logger.debug('New message in tracked thread: %s', tmsgid)
                    updates.append(tmsg)

    # Now grab the latest query matches
    query = cconf.get('pi_query')
    if query:
        logger.info('Running query for %s/%s', product, component)
        try:
            last_check = peebz.db_get_query_last_check(product, component)
            query += f' AND dt:{last_check}..'
        except LookupError:
            pass
        qquery = urllib.parse.quote_plus(query)
        query_url = url.rstrip('/') + f'/?x=m&q={qquery}'
        lastdt = now.strftime('%Y%m%d%H%M%S')
        try:
            msgs = get_query_results(query_url)
            for msg in msgs:
                msgid = b4.LoreMessage.get_clean_msgid(msg)
                if msgid in seen_msgids:
                    logger.debug('Already seen %s', msgid)
                    continue

                # New thing to track!
                seen_msgids.add(msgid)
                author = peebz.msg_get_author(msg)
                fromaddr = author[1]
                if not peebz.bz_check_user_allowed(fromaddr, product, component):
                    logger.debug('skipping msg %s', msg.get('Subject'))
                    continue
                # Retrieve and queue up the entire thread
                try:
                    tmsgs = get_sorted_thread(url, msgid)
                except LookupError:
                    logger.debug('No results returned for msgid=%s', msgid)
                    continue
                for tmsg in tmsgs:
                    tmsgid = b4.LoreMessage.get_clean_msgid(tmsg)
                    seen_msgids.add(tmsgid)
                    updates.append(tmsg)

        except LookupError:
            logger.info('No new results for product=%s, component=%s', product, component)

        if not dry_run:
            peebz.db_store_query_last_check(product, component, lastdt)

    if not updates:
        logger.info('No new messages to add to bugzilla for %s/%s', product, component)
        return

    for msg in updates:
        logger.debug('Recording %s', msg.get('Subject'))
        peebz.parse.process_rfc2822(msg, product, component, dry_run=dry_run)


def main(cmdargs: argparse.Namespace):
    config = peebz.get_config()
    # Iterate all components
    for bz_product, bz_components in config['components'].items():
        for bz_component in bz_components.keys():
            update_component(bz_product, bz_component, dry_run=cmdargs.dry_run)