#!/usr/bin/env python3 # -*- coding: utf-8 -*- # vim: set ts=4 # # Copyright 2025-present Linaro Limited # # SPDX-License-Identifier: MIT import logging import json import os import re import subprocess as sp import argparse import tempfile from datetime import datetime, timedelta logging.basicConfig(level=logging.INFO) logger = logging.getLogger() def run_git(args): cmd = ["git"] cmd = cmd + args logger.debug(f"{cmd=}") proc = sp.Popen(cmd, stdout=sp.PIPE, stderr=sp.PIPE) stdout, stderr = proc.communicate() proc.ok = proc.returncode == 0 proc.out = stdout.decode('utf-8', errors='replace') proc.err = stderr.decode('utf-8', errors='replace') return proc def search_git_tree(args): # Regex patterns message_id_pattern = r"^Message-I(D|d):\s*<([^>]+)>" in_reply_to_pattern = r"^In-Reply-To:\s*<([^>]+)>" subject_pattern = r"^Subject:\s*(.*)" date_pattern = r"^Date:\s*(.*)" from_user_emails = [ "anders.roxell@linaro.org", "daniel.diaz@linaro.org", "naresh.kamboju@linaro.org", "theodore.grey@linaro.org", ] user_emails = [ "lkft@linaro.org", ] by = ["Tested-by:", "Reported-by:"] from_user_emails_regex = "|".join(from_user_emails) from_pattern = f"From: .*<((?:{from_user_emails_regex})).*" user_emails_regex = "|".join(user_emails) by_regex = "|".join(by) by_pattern = f"\s*((?:{by_regex}) .*<((?:{user_emails_regex}))).*" logger.debug(f"{by_pattern=}") deadline_pattern = r"^X-KernelTest-Deadline:\s*(.*)" # digg up since and until repo_path = args.repo_path if args.since: since = args.since.strftime("%Y-%m-%d") if args.until: until = args.until.strftime("%Y-%m-%d") logger.debug(f"{repo_path=}") git_date_from_until = [] if args.since: git_date_from_until = ["--since", since] if args.until: git_date_from_until.extend(["--until", until]) # Build the git log command with the --since option git_log_sha_command = [ "-C", repo_path, "log", "--format=format:%H:::%ae:::%s", "--grep", f"\[PATCH .* review$" ] git_log_sha_command.extend(git_date_from_until) logger.debug(f"{git_log_sha_command=}") # Get the git log try: #git_log_sha = run_git("-C", repo_path, "log", "{git_date_from_until}", "--format=format:%H") git_log_sha = run_git(git_log_sha_command) except sp.CalledProcessError as e: print(f"Error reading git log: {e}") return [] results = {} replies = "" release = "" extra_search_release = "" for line in git_log_sha.out.split("\n"): logger.debug(f"{line=}") sha, email, subj = line.split(":::") if subj.endswith(" review") and email in from_user_emails: replies = f"{line}" if replies == "" else f"{replies}\n{line}" if subj.startswith("[PATCH ") and subj.endswith(" review"): release = f"{line}" if release == "" else f"{release}\n{line}" print("asdf") print(replies) print("asdf") print(release) for line in replies.split("\n"): logger.debug(f"{line=}") sha, email, subj = line.split(":::") if line == "": continue version = re.sub(" review", "", re.sub(".*] ", "", subj)) #if version.startswith("unicode"): # print(version) # asdf # continue results[version] = { "Kernel Version": version, "From": email, "Sha": sha, } for line in release.split("\n"): logger.debug(f"{line=}") sha, email, subj = line.split(":::") version = re.sub(" review", "", re.sub(".*] ", "", subj)) logger.debug(f"{version=}") if version not in results: git_find_release_command = [ "-C", repo_path, "log", "--format=format:%H:::%ae:::%s", "--grep", f"^\[PATCH .* {version} review$" ] find_release = run_git(git_find_release_command) extra_search_release = f"{find_release.out}" if extra_search_release == "" else f"{extra_search_release}\n{find_release.out}" logger.debug(f"BAJS {find_release.out=}") _, _, subj = find_release.out.split(":::") version = re.sub(" review", "", re.sub(".*] ", "", subj)) results[version] = { "Kernel Version": version, } logger.debug(f"{version=}") results[version].update({ "Kernel Version": version, }) release = f"{release}\n{extra_search_release}" # Linaro's reply email logger.debug(f"{replies=}") for line in replies.split("\n"): sha,email,subj = line.split(":::") git_show_command = [ "-C", repo_path, "show", f"{sha}:m" ] try: git_show = run_git(git_show_command) except sp.CalledProcessError as e: print(f"Error reading git log: {e}") # Extract Reported-by and Message-ID by_ = None from_ = email #logger.debug(f"{git_show.out=}") for line in git_show.out.split("\n"): #logger.debug(f"{line=}") from_matches = re.match(from_pattern, line) if from_matches: from_ = from_matches.group(0) by_matches = re.match(by_pattern, line) if by_matches: by_ = by_matches.group(0) in_reply_to_matches = re.search(in_reply_to_pattern, line) if in_reply_to_matches: in_reply_to_ = in_reply_to_matches.group(0) message_id_matches = re.search(message_id_pattern, line) if message_id_matches: message_id_ = message_id_matches.group(0) subject_matches = re.match(subject_pattern, line) if subject_matches: subject_ = subject_matches.group(0) date_matches = re.match(date_pattern, line) if date_matches: date_ = date_matches.group(0) logger.debug(f"{message_id_=}") logger.debug(f"{subject_=}") logger.debug(f"{date_=}") if by_ and from_ != None: logger.debug(f"{from_=}") logger.debug(f"{by_=}") kernel_version = re.sub(" review", "", re.sub(".*] ", "", subject_)) message_id_ = re.sub(">", "", re.sub("Message-I(D|d): <", "", message_id_)) in_reply_to_ = re.sub(">", "", re.sub("In-Reply-To: <", "", in_reply_to_)) subject_ = re.sub("Subject: ", "", subject_) date_ = date_.strip("Date: ") #date_obj = datetime.strptime(date_, "%a, %d %b %Y %H:%M:%S %z") results[kernel_version] = { "In-Reply-To": in_reply_to_, "Message-ID": message_id_, "url": f"https://lore.kernel.org/stable/{message_id_}", "Subject": subject_, "Reply Date": date_, "Tag": by_, } # gregh's RC email logger.debug(f"{release=}") for line in release.split("\n"): sha, email, subj = line.split(":::") logger.debug(f"SATAN: {subj=}") git_show_command = [ "-C", repo_path, "show", f"{sha}:m" ] try: git_show = run_git(git_show_command) except sp.CalledProcessError as e: print(f"Error reading git log: {e}") # Extract Reported-by and Message-ID by_ = None from_ = email deadline_ = None message_id_ = None subject_ = subj date_ = None for line in git_show.out.split("\n"): #logger.debug(f"{line=}") deadline_matches = re.search(deadline_pattern, line) if deadline_matches: deadline_ = deadline_matches.group(0) message_id_matches = re.search(message_id_pattern, line) if message_id_matches: message_id_ = message_id_matches.group(0) # subject_matches = re.match(subject_pattern, line) # if subject_matches: # subject_ = subject_matches.group(0) date_matches = re.match(date_pattern, line) if date_matches: date_ = date_matches.group(0) #logger.debug(f"{message_id_=}") #logger.debug(f"{subject_=}") #logger.debug(f"{date_=}") try: if "unicode" in subject_: print(sha) print(line) print(subject_) print(message_id_) print(deadline_) asdf kernel_version = re.sub(" review", "", re.sub(".*] ", "", subj)) message_id_ = re.sub(">", "", re.sub("Message-I(D|d): <", "", message_id_)) #subject_ = re.sub("Subject: ", "", subject_) if deadline_ is not None: deadline_ = re.sub("X-KernelTest-Deadline: ", "", deadline_) date_ = date_.strip("Date: ") results[kernel_version].update({ "Kernel Version": kernel_version, "Release Subject": subj, "Release Message-ID": message_id_, "Release Sha": sha, "Release Date": date_, "Deadline Date": deadline_, }) except KeyError as e: logger.info(f"{kernel_version=}") logger.info(f"{subject_=}") return results def valid_date(s: str) -> datetime: try: return datetime.strptime(s, "%Y-%m-%d") except ValueError: raise argparse.ArgumentTypeError(f"not a valid date: {s!r}") def parse_args(): # Argument parser setup parser = argparse.ArgumentParser(description="Search git tree for 'Reported-by: Linux Kernel Functional Testing'") parser.add_argument( "--repo-path", type=str, required=True, help="Path to the git repository" ) parser.add_argument( "--since", help="The Start Date - format YYYY-MM-DD", type=valid_date ) parser.add_argument( "--until", help="The End Date - format YYYY-MM-DD", default=None, type=valid_date ) parser.add_argument( "--debug", action="store_true", default=False, help="Display debug messages", ) return parser.parse_args() def get_sla(date1_str, date2_str): logger.debug(f"get_sla: {date1_str=}, {date2_str=}") # Parse the first date (ISO format with timezone) date1_main = date1_str[:-6] # Extract the main datetime part date1_tz = date1_str[-6:] # Extract the timezone part date1 = datetime.strptime(date1_main, "%Y-%m-%dT%H:%M") or datetime.strptime(date2_main, "%a, %d %b %Y %H:%M:%S") tz_sign = 1 if date1_tz[0] == '+' else -1 tz_hours = int(date1_tz[1:3]) tz_minutes = int(date1_tz[4:6]) date1 = date1 - timedelta(hours=tz_hours * tz_sign, minutes=tz_minutes * tz_sign) date2_main = date2_str[:-6] # Extract the main datetime part date2_tz = date2_str[-5:] # Extract the timezone part date2 = datetime.strptime(date2_main, "%a, %d %b %Y %H:%M:%S") tz_sign = 1 if date2_tz[0] == '+' else -1 tz_hours = int(date2_tz[1:3]) tz_minutes = int(date2_tz[3:5]) date2 = date2 - timedelta(hours=tz_hours * tz_sign, minutes=tz_minutes * tz_sign) time_difference = date1 - date2 total_seconds = time_difference.total_seconds() hours_difference = int(time_difference.total_seconds() // 3600) minutes_difference = int((total_seconds % 3600) // 60) return hours_difference, minutes_difference def fix_dict(results): for key, value in results.items(): hours = 0 logger.debug(f"{key=}") try: release_date = results[key]['Deadline Date'] if results[key]['Deadline Date'] is not None else results[key]['Release Date'] hours, minutes = get_sla(results[key]['Deadline Date'], results[key]['Reply Date']) except KeyError as e: continue logger.debug(f"{hours=}") if hours <= 8: sla = "<8h" elif hours <= 24: sla = "<24h" elif hours <= 48: sla = "<48h" else: sla = ">48h" if hours != 0: results[key].update({ "sla_time": f"{hours}h{minutes}m", "sla": sla, }) return results def main(): args = parse_args() if args.debug: logger.setLevel(level=logging.DEBUG) # Search for reported-by entries in the Git tree results = search_git_tree(args) if results: results = fix_dict(results) print(json.dumps(results,indent=2)) #logger.info(f"{json.dumps(results,indent=2)=}") print("Search Results:") for key, value in results.items(): try: #logger.debug(f"{key=}") #logger.debug(f"{value=}") #print(f"url: {value['url']}, sla: {value['sla']}, hours: {value['sla_time']}") pass except KeyError as e: logger.info(f"{value=}") else: print("No matches found.") if __name__ == "__main__": main()