Initial text report generation and other misc changes

pyproject.toml

@@ -8,7 +8,9 @@ version = "0"
 description = "NG tooling for handling the Fedora Orphaned Packages Process"
 requires-python = ">=3.12"
 dependencies = [
-  "click",
-  "fedrq",
+    "click",
+    "fedrq",
+    "pydantic",  # Also pulled in by fedrq
+    "texttable",
 ]
 scripts.endorphans = "endorphans.main:main"

src/endorphans/__init__.py

@@ -1,3 +1,9 @@
 # SPDX-License-Identifier: GPL-2.0-or-later
 # Copyright (C) 2026 Maxwell G <maxwell@gtmx.me>
 # Copyright (c) Fedora Project contributors
+
+import logging
+
+fmt = "{levelname}:{name}:{lineno}: {message}"
+logging.basicConfig(format=fmt, style="{", level=logging.INFO)
+logging.getLogger(__name__)

src/endorphans/depsolve.py

@@ -50,6 +50,7 @@ components with the orphan(s) on which those components depend(s).
 """
 
 import json
+import logging
 import os
 import subprocess
 import typing as t
@@ -65,6 +66,8 @@ from fedrq._utils import get_source_name
 
 from .err import EndorphansDepSolveError as DepsolveError
 
+logger = logging.getLogger(__name__)
+
 StrPath = str | os.PathLike[str]
 DepTuple = tuple[str, str, str]
 
@@ -137,8 +140,8 @@ class _Dnf5Er:
         cmd = [
             "dnf5",
             "--disablerepo=*",
-            f"--setopt=cachedir={self._bm.conf.cachedir}",
-            "--cacheonly",
+            f"--setopt=cachedir={self._bm.conf.cachedir}",  # pyright: ignore[reportAttributeAccessIssue]
+            # "--cacheonly",
         ]
         if r := self._bm.vars.get_value("releasever"):
             cmd.append(f"--releasever={r}")
@@ -281,6 +284,7 @@ class DepSolveResult:
     # which orphaned packages the maintainer of package foo would need to pick
     # up to unbreak foo.
     affected_pkg_to_orphans: dict[str, set[str]]
+    orphans: list[str]
 
 
 def depsolve(branch: str, repo: str, orphans: t.Collection[str]) -> DepSolveResult:
@@ -337,5 +341,6 @@ def depsolve(branch: str, repo: str, orphans: t.Collection[str]) -> DepSolveResu
                     (RPMPackage.from_package(orphan_pkg), reldep)
                 )
                 affected_pkg_to_orphans[nevra_srcname].add(orphan_srcname)
-
-    return DepSolveResult(orphan_to_affected_pkgs, affected_pkg_to_orphans)
+    return DepSolveResult(
+        orphan_to_affected_pkgs, affected_pkg_to_orphans, sorted(orphans)
+    )

src/endorphans/distgit.py

@@ -34,6 +34,7 @@ class DistgitClient(t.Protocol):
         """
         Get a package's maintainers, including group maintainers.
         Group maintainers are prefixed with an @ sign.
+        Maintainers lists are sorted().
         """
         ...
 
@@ -91,6 +92,8 @@ class _PagureDistgitClient:
         watchers_data = self.get_extra("bz")["rpms"]
         result = {}
         for package in packages:
+            maints: list[str]
+            watchers: list[str]
             try:
                 maints = owner_alias_data[package]
                 watchers = watchers_data[package]
@@ -101,6 +104,7 @@ class _PagureDistgitClient:
             for watcher in watchers:
                 if watcher[0] == "@":
                     maints.append(watcher)
+            maints.sort()
             result[package] = maints
         return result
 

src/endorphans/koji.py

@@ -2,6 +2,7 @@
 # Copyright (C) 2026 Maxwell G <maxwell@gtmx.me>
 # Copyright (c) Fedora Project contributors
 
+import datetime as dt
 import typing as t
 
 import koji
@@ -11,14 +12,31 @@ from ._utils import ORPHAN
 KOJIHUB = "https://koji.fedoraproject.org/kojihub"
 
 
-def get_unblocked_orphan_packages(tag_id: str | int) -> list[str]:
-    """
-    Returns a list of packages in a tag that are not `blocked` and are owned by orphan.
-    """
-    session = koji.ClientSession(KOJIHUB)
-    pkgs = session.listPackages(tagID=tag_id, inherited=True)
-    result = []
-    for pkg in pkgs:
-        if pkg["owner_name"] == ORPHAN and not pkg["blocked"]:
-            result.append(pkg["package_name"])
-    return result
+class KojiClient:
+    def __init__(self, kojihub: str | None = None) -> None:
+        kojihub = kojihub or KOJIHUB
+        self.session = koji.ClientSession(KOJIHUB)
+
+    def get_unblocked_orphan_packages(self, tag_id: str | int) -> list[str]:
+        """
+        Returns a list of packages in a tag that are not `blocked` and are owned by orphan.
+        """
+        pkgs = self.session.listPackages(tagID=tag_id, inherited=True)
+        result = []
+        for pkg in pkgs:
+            if pkg["owner_name"] == ORPHAN and not pkg["blocked"]:
+                result.append(pkg["package_name"])
+        return result
+
+    def get_orphaned_datetime(
+        self, tag_id: str | int, packages: t.Collection[str]
+    ) -> dict[str, dt.datetime]:
+        result: dict[str, dt.datetime] = {}
+        for package in packages:
+            hist = self.session.queryHistory(
+                tables=["tag_package_owners"], tag=tag_id, package=package
+            )
+            result[package] = dt.datetime.fromtimestamp(
+                hist["tag_package_owners"][0]["create_ts"], dt.UTC
+            )
+        return result

src/endorphans/main.py

@@ -17,13 +17,13 @@
 import dataclasses
 import json
 import os
-import subprocess
 import typing as t
 
 import click
+import pydantic
 
-from . import depsolve
-from .koji import get_unblocked_orphan_packages
+from . import depsolve, distgit, textreport
+from .koji import KojiClient
 
 StrPath = str | os.PathLike[str]
 
@@ -34,9 +34,34 @@ def rl(fp: t.IO[str]):
             yield line.rstrip("\n")
 
 
-@click.command(
+@click.group(
     context_settings={"show_default": True, "help_option_names": ["-h", "--help"]}
 )
+def main():
+    pass
+
+
+# TODO: Make an `all` command that does the depsolving and text (and JSON) report generation in one go.
+
+
+# TODO: Make koji_tag a constant or better yet, find the rawhide version dynamically
+@click.option("--koji-tag", default="f45")
+@click.option("-j", "--depsolve-json", required=True, type=click.File("rb"))
+@main.command(name="textreport")
+def textreport_(koji_tag: str, depsolve_json: t.IO[bytes]) -> None:
+    depsolve_result = pydantic.TypeAdapter(depsolve.DepSolveResult).validate_json(
+        depsolve_json.read()
+    )
+    depsolve_json.close()
+    koji_client = KojiClient()
+    dg = distgit.get_client()
+    dts = koji_client.get_orphaned_datetime(koji_tag, depsolve_result.orphans)
+    maints = dg.get_package_maintainers(depsolve_result.orphans)
+    report = textreport.generate(depsolve_result, maints, dts)
+    print(report)
+
+
+@main.command(name="depsolve")
 @click.option("-b", "--branch", default="rawhide")
 @click.option("-r", "--repo", default="@buildroot")
 @click.option(
@@ -44,11 +69,15 @@ def rl(fp: t.IO[str]):
     type=click.File(),
     help="List of packages; defaults to getting list of orphans from goorphans",
 )
-@click.option("--koji-tag", default="rawhide")
-def main(branch: str, repo: str, package_list: t.IO[str] | None, koji_tag: str) -> None:
+# TODO: Make koji_tag a constant or better yet, find the rawhide version dynamically
+@click.option("--koji-tag", default="f45")
+def depsolve_(
+    branch: str, repo: str, package_list: t.IO[str] | None, koji_tag: str
+) -> None:
+    koji_client = KojiClient()
     if package_list is None:
         # Get list of orphans
-        orphans = get_unblocked_orphan_packages(koji_tag)
+        orphans = koji_client.get_unblocked_orphan_packages(koji_tag)
     else:
         orphans = list(rl(package_list))
     depsolve_result = depsolve.depsolve(branch, repo, orphans)
@@ -60,7 +89,7 @@ def main(branch: str, repo: str, package_list: t.IO[str] | None, koji_tag: str)
 
 def json_default(d: object) -> t.Any:
     if isinstance(d, depsolve.RPMPackage):
-        return str(d)
+        return dataclasses.asdict(d)
     if isinstance(d, set):
         return sorted(d)
     raise TypeError

src/endorphans/textreport.py

@@ -0,0 +1,36 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+# Copyright (C) 2026 Maxwell G <maxwell@gtmx.me>
+# Copyright (c) Fedora Project contributors
+
+"""
+Generate plain text report based on dependency resolution findings
+"""
+
+import datetime as dt
+import typing as t
+
+import jinja2
+import texttable  # type: ignore[import]
+
+from .depsolve import DepSolveResult
+
+
+def generate(
+    depsolve_result: DepSolveResult,
+    maintainers_map: t.Mapping[str, t.Sequence[str]],
+    orphaned_datetime: t.Mapping[str, dt.datetime],
+) -> str:
+    # TODO:
+    result: list[str] = []
+    table = texttable.Texttable(max_width=80)
+    table.header(["Package", "Maintainers", "Orphaned duration"])
+    table.set_deco(table.HEADER)
+    now = dt.datetime.now(dt.UTC)
+    for package in depsolve_result.orphans:
+        maintainers = ", ".join(maintainers_map[package])
+        delta = now - orphaned_datetime[package]
+        weeks = delta.days // 7
+        orphaned_since = f"{weeks} week{'s' if weeks > 1 else ''}"
+        table.add_row([package, maintainers, orphaned_since])
+    result.append(t.cast(str, table.draw()))
+    return "\n".join(result)