Use libdnf5 for getting the sourcestopic/dnf5

This is based on:
- getting the packages cited in Source RPM field,
- extracting the name
- searching the package in Testing reposotories having said name
- adding the source package names removing .src at end.
File dnfmadbbase comes from MADb
config.py.in has to be copied in config.py and adapted to the location
where dnf cache will be written and to the mirror to use.

3 files changed, 217 insertions, 21 deletions

diff --git a/config.py.in b/config.py.in
new file mode 100644
index 0000000..6764135
--- /dev/null
+++ b/config.py.in
@@ -0,0 +1,24 @@
+# the last release
+TOP_RELEASE = 9
+# be sure that these directories are writable for the user running the application
+DATA_PATH = "/var/lib/madb"
+LOG_PATH = DATA_PATH
+MIRROR_URL = "https://fr2.rpmfind.net/linux/mageia/distrib/"
+# Name of the development version
+DEV_NAME = "cauldron"
+# Level of logging
+LOG_LEVEL = "DEBUG"
+BUGZILLA_URL = "https://bugs.mageia.org"
+# List of architectures managed
+ARCHES = {
+    "x86_64": "x86 64bits",
+    "i586": "x86 32bits",
+    "aarch64": "Arm 64bits",
+    "armv7hl": "Arm 32bits v7hl",
+}
+# Used as filter in search bar
+DISTRIBUTION = {
+    DEV_NAME: "Mageia cauldron",
+    str(TOP_RELEASE): "Mageia " + str(TOP_RELEASE),
+    str(TOP_RELEASE - 1): "Mageia " + str(TOP_RELEASE -1),
+}
diff --git a/dnf5madbbase.py b/dnf5madbbase.py
new file mode 100644
index 0000000..e3f0859
--- /dev/null
+++ b/dnf5madbbase.py
@@ -0,0 +1,167 @@
+import libdnf5
+import os
+from datetime import datetime, timedelta
+from libdnf5.common import QueryCmp_GLOB as GLOB
+import config
+
+
+class Dnf5MadbBase():
+
+    def __init__(self, release, arch, root, refresh=False):
+
+        self.release = release
+        self.arch = arch
+        self.root = root
+
+        # Create a new Base object
+        self._base = libdnf5.base.Base()
+        self._base_config = self._base.get_config()
+        self._base_config.installroot = root
+        # https://github.com/rpm-software-management/dnf5/issues/412#issuecomment-1493782078
+        self._base_config.optional_metadata_types = ['filelists', 'other']
+        self._base.config_file_path = os.path.join(root, "dnf/dnf.conf")
+        self._base.load_config()
+        vars = self._base.get_vars().get()
+        vars.set('releasever', release)
+        vars.set('arch', arch)
+        self._base_config.logdir = os.path.join(config.LOG_PATH)
+        self._base_config.cachedir = os.path.join(root, "dnf", "cache")
+        self._base_config.reposdir = os.path.join(root, "dnf", "etc","distro.repos.d")
+        log_router = self._base.get_logger()
+        logger = libdnf5.logger.create_file_logger(self._base)
+        log_router.add_logger(logger)
+        self._base_config.module_platform_id = f"Mageia:{release}"
+        self._base_config.metadata_expire = 20 if refresh else -1
+        self._base.setup()
+        self._repo_sack = self._base.get_repo_sack()
+        repos = {}
+        for section in ("core", "nonfree", "tainted"):
+            for cl in ("backports", "backports_testing", "release", "updates", "updates_testing"):
+                repo_name = f"{release}-{arch}-{section}-{cl}"
+                repo_srpms_name = f"{release}-SRPMS-{section}-{cl}"
+                repos[repo_name] = self._repo_sack.create_repo(repo_name)
+                repos[repo_name].get_config().baseurl = os.path.join(config.MIRROR_URL, release, arch, "media", section, cl)
+                repos[repo_name].get_config().name = f"{config.DISTRIBUTION[release]} {arch} {section.capitalize()} {cl.capitalize()}"
+                repos[repo_srpms_name] = self._repo_sack.create_repo(repo_srpms_name)
+                repos[repo_srpms_name].get_config().baseurl = os.path.join(config.MIRROR_URL, release, "SRPMS", section, cl)
+                repos[repo_srpms_name].get_config().name = f"{config.DISTRIBUTION[release]} SRPMS {section.capitalize()} {cl.capitalize()}"
+        self._repo_sack.update_and_load_enabled_repos(False)
+
+
+    def search_name(self, values, graphical=False, repo=None):
+        """Search in a list of package attributes for a list of keys.
+
+        :param values: the values to match
+        :params graphical: boolean, filter on *.desktop files in /usr/share/applications
+        :param repo: name of the repository to search in. Accept wildcards.
+        :return: a list of package objects
+        """
+        query = libdnf5.rpm.PackageQuery(self._base)
+        #query.filter_arch([self.arch, "noarch"])
+        query.filter_name(values, GLOB)
+        if graphical:
+            query.filter_file(["/usr/share/applications/*.desktop"], GLOB)
+        if repo:
+            query.filter_repo_id([repo], GLOB)
+        return query
+
+    def search_nevra(self, values, graphical=False, repo=None):
+        """Search in a list of package attributes for a list of keys.
+
+        :param values: the values to match
+        :params graphical: boolean, filter on *.desktop files in /usr/share/applications
+        :param repo: name of the repository to search in. Accept wildcards.
+        :return: a list of package objects
+        """
+        query = libdnf5.rpm.PackageQuery(self._base)
+        #query.filter_arch([self.arch, "noarch"])
+        query.filter_nevra(values, GLOB)
+        if graphical:
+            query.filter_file(["/usr/share/applications/*.desktop"], GLOB)
+        if repo:
+            query.filter_repo_id([repo], GLOB)
+        return query
+
+    def search_in_group(self, value, graphical=False, repo=None):
+        """Search a list of package in a group.
+
+        :param values: the values to match
+        :return: a list of package objects
+        """
+        query = libdnf5.rpm.PackageQuery(self._base)
+        query.filter_arch([self.arch, "noarch"])
+        if graphical and graphical == "1":
+            query.filter_file(["/usr/share/applications/*.desktop"], GLOB)
+        if repo:
+            query.filter_repo_id([repo])
+        return [rpm for rpm in query if rpm.get_group().startswith(value)]
+
+    def search_updates(self, backports=False, last=False, testing=True, graphical=False):
+        query = libdnf5.rpm.PackageQuery(self._base)
+        query.filter_arch([self.arch, "noarch"])
+        if backports:
+            repo = "*backports"
+            days = config.RECENT_BACKPORTS_DURATION
+        else:
+            repo = "*updates"
+            days = config.RECENT_UPDATES_DURATION
+        if testing:
+            repo += "_testing"
+        query.filter_repo_id([repo], GLOB)
+        if graphical:
+            query.filter_file(["/usr/share/applications/*.desktop"], GLOB)
+        if last:
+            query.filter_recent(int((datetime.now() - timedelta(days=days)).timestamp()))
+        return query
+
+    def search_by_sources(self, values, repo=None):
+        query = libdnf5.rpm.PackageQuery(self._base)
+        query.filter_arch([self.arch, "noarch"])
+        if repo:
+            query.filter_repo_id([repo], GLOB)
+        query.filter_sourcerpm(values, GLOB)
+        return query
+
+    def provides_requires(self, rpm_list):
+        query = libdnf5.rpm.PackageQuery(self._base)
+        query.filter_arch([self.arch, "noarch"])
+        query.filter_repo_id([self.release + "*"], GLOB)
+        query.filter_provides(rpm_list)
+        return query
+
+    def search_provides(self, rpm_list):
+        query = libdnf5.rpm.PackageQuery(self._base)
+        query.filter_arch([self.arch, "noarch"])
+        query.filter_repo_id([self.release + "*"], GLOB)
+        query.filter_provides(rpm_list)
+        return query
+
+    def search(self, search_type, search_list, graphical=False, repo=""):
+        query = libdnf5.rpm.PackageQuery(self._base)
+        query.filter_arch([self.arch, "noarch"])
+        if repo == "":
+            query.filter_repo_id([self.release + "*"], GLOB)
+        else:
+            query.filter_repo_id([repo], GLOB)
+        #search_list = self.search_name(rpm_list)
+        if search_type == "requires":
+            query.filter_requires(search_list)
+        elif search_type == "recommends":
+            query.filter_recommends(search_list)
+        elif search_type == "suggests":
+            query.filter_suggests(search_list)
+        elif search_type == "supplements":
+            query.filter_supplements(search_list)
+        elif search_type == "provides":
+            query.filter_provides(search_list)
+        else:
+            query.filter_name(search_list, GLOB)
+        if graphical:
+            query.filter_file(["/usr/share/applications/*.desktop"], GLOB)
+        return query
+
+
+    def repo_enabled(self):
+        query = libdnf5.repo.RepoQuery(self._base)
+        query.filter_enabled(True)
+        return query
diff --git a/mga-advisor.py b/mga-advisor.py
index 8fbcf73..6743bbd 100644
--- a/mga-advisor.py
+++ b/mga-advisor.py
@@ -9,6 +9,9 @@ import requests
 from textwrap import wrap
 import time
 
+import config as config
+from dnf5madbbase import Dnf5MadbBase
+
 from PySide6.QtWidgets import (
     QApplication,
     QWidget,
@@ -111,27 +114,12 @@ class Widget(QWidget):
         r = requests.get(url, headers=headers)
         if r.status_code == 200 and r.json()["faults"] == []:
             desc =""
-            for pkg in  re.split(';|,| ', r.json()['bugs'][0]['cf_rpmpkg']):
-                pkg = pkg.strip()
-                if pkg == "":
-                    continue
-                analyze = re.search(r"([\w\-\+_]+)-\d", pkg)
-                if analyze is not None:
-                    pkg = analyze.group(1)
-                sources = self.src_populate(pkg)
-                for source in sources:
-                    suffix = ".mga" + source["mga_release"]
-                    if source["repo"] in ("tainted", "nonfree"):
-                        suffix +=  "." + source["repo"]
-                    self.ui.list_src.addItem(
-                        " ".join((source["mga_release"],
-                                  source["repo"],
-                                  source["package"] +
-                                  "-" + source["version"] +
-                                  "-" + source["release"] +
-                                  suffix,
-                        )))
-                desc += f" {pkg}"
+            sources = self._srpms(r.json()['bugs'][0]['cf_rpmpkg'])
+            for rel in sources.keys():
+                    print(f"Adding {rel}")
+                    for item in sources[rel]:
+                        self.ui.list_src.addItem(item)
+                        desc += f" {item}"
             for cve in  re.split(';|,| ', r.json()['bugs'][0]['cf_cve']):
                 cve = cve.strip()
                 if cve != "":
@@ -321,6 +309,23 @@ class Widget(QWidget):
         self.ui.bug_le.setText(re.sub(r'\D', '', self.ui.bug_le.text()))
         return True
 
+    def _srpms(self, field):
+        """
+        Return a set with the names of source packages in 2 latest releases
+        """
+        results = {}
+        for release in (config.TOP_RELEASE, config.TOP_RELEASE - 1):
+            
+            distro =  Dnf5MadbBase(str(release), "x86_64", config.DATA_PATH, refresh = True)
+            # extract list from bug report field, removing extra src.rpm
+            srpms = [srpm.strip().removesuffix(".rpm").removesuffix(".src") + "*" for srpm in re.split(';|,| ', field) if srpm.strip() != ""]
+            # get only the source package names
+            srpms_names = [x.get_name() for x in distro.search_nevra(srpms, repo=f"{release}-SRPMS-*")]
+            # We want the same source rpm, but from Testing repo
+            results[str(release)] = list(set([x.get_nevra().removesuffix(".src") for x in distro.search_name(srpms_names, repo=f"{release}-SRPMS-*testing*")]))
+            # print(f"For {release} {field} : {results[str(release)]}")
+        return results
+
     def src_populate(self, package):
         # retrieve information with repo, release from package name
         cmd = ["mgarepo", "rpmlog"]