feat: extract CI lifecycle and Prow job management skills from openshift/release

skills/_shared/rhdh_lifecycle/__init__.py

@@ -0,0 +1 @@
+"""RHDH lifecycle data from external APIs (Red Hat, endoflife.date)."""

skills/_shared/rhdh_lifecycle/ocp.py

@@ -0,0 +1,102 @@
+#!/usr/bin/env python3
+"""OCP lifecycle phase classification.
+
+Classifies OCP versions from the Red Hat Product Life Cycles API into
+lifecycle phases (Full Support, Maintenance, EUS, End of life).
+
+Usage:
+    from rhdh_lifecycle.ocp import classify_ocp_versions
+    from rhdh_lifecycle.redhat import fetch_api
+    api_data = fetch_api("Red Hat OpenShift Container Platform")
+    versions = classify_ocp_versions(api_data, "2025-05-13")
+"""
+
+import re
+
+from rhdh_lifecycle.redhat import _is_date, _to_date
+
+
+def classify_ocp_versions(api_data, today):
+    """Classify OCP versions from the Red Hat Product Life Cycles API.
+
+    Args:
+        api_data: Raw API response dict (the full JSON response).
+        today: Date string in YYYY-MM-DD format.
+
+    Returns:
+        List of dicts with keys: version, ocp_supported, phase,
+        ga_date, end_of_support_date. Sorted by version.
+    """
+    versions = api_data.get("data", [{}])[0].get("versions", [])
+
+    # Keep only clean X.Y version names, skip variants like "4.6 EUS" or "3"
+    versions = [v for v in versions if re.match(r"^\d+\.\d+$", v.get("name", ""))]
+
+    # Filter to OCP 4.x and above (future-proof for 5.x+)
+    versions = [v for v in versions if int(v["name"].split(".")[0]) >= 4]
+
+    phase_order = [
+        "Extended update support Term 2",
+        "Extended update support",
+        "Maintenance support",
+        "Full support",
+    ]
+
+    results = []
+    for ver in versions:
+        phases = ver.get("phases", [])
+
+        # Find latest end-of-support date across all support phases
+        end_dates = []
+        for pname in phase_order:
+            for p in phases:
+                if p.get("name") == pname:
+                    d = _to_date(p.get("end_date"))
+                    if d:
+                        end_dates.append(d)
+        end_of_support = max(end_dates) if end_dates else None
+
+        # GA date
+        ga_raw = None
+        for p in phases:
+            if p.get("name") == "General availability":
+                ga_raw = p.get("end_date", "N/A")
+                break
+
+        # Determine current phase
+        current_phase = "End of life"
+        for pname in phase_order:
+            for p in phases:
+                if p.get("name") != pname:
+                    continue
+                start = _to_date(p.get("start_date"))
+                end_raw = p.get("end_date")
+                end = _to_date(end_raw)
+                if start and start <= today:
+                    if end and end >= today:
+                        current_phase = pname
+                        break
+                    elif not _is_date(end_raw) and end_raw not in (
+                        "N/A",
+                        "",
+                        None,
+                    ):
+                        # Non-date end value (e.g., "Ongoing") means still active
+                        current_phase = pname
+                        break
+            if current_phase != "End of life":
+                break
+
+        results.append(
+            {
+                "version": ver["name"],
+                "ocp_supported": current_phase != "End of life",
+                "phase": current_phase,
+                "ga_date": _to_date(ga_raw) if _is_date(ga_raw) else "N/A",
+                "end_of_support_date": end_of_support or "N/A",
+            }
+        )
+
+    # Sort by version
+    results.sort(key=lambda v: [int(x) for x in v["version"].split(".")])
+    return results

skills/_shared/rhdh_lifecycle/redhat.py

@@ -0,0 +1,205 @@
+#!/usr/bin/env python3
+"""Unified client for the Red Hat Product Life Cycles API.
+
+Fetches lifecycle data for any Red Hat product (RHDH, OCP, RHBK, Quay, etc.)
+and returns a consistent structure. Product-specific post-processing functions
+handle cases like RHBK major version grouping or RHDH OCP compatibility.
+
+Usage:
+    from rhdh_lifecycle.redhat import fetch_product_lifecycle
+
+    versions = fetch_product_lifecycle("rhbk")
+    versions = fetch_product_lifecycle("Red Hat Quay")
+    versions = fetch_product_lifecycle("ocp", filter_version="4.16")
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import sys
+import urllib.error
+import urllib.request
+
+LIFECYCLE_API_URL = "https://access.redhat.com/product-life-cycles/api/v1/products"
+
+PRODUCT_ALIASES = {
+    "rhdh": "Red Hat Developer Hub",
+    "ocp": "Red Hat OpenShift Container Platform",
+    "rhbk": "Red Hat build of Keycloak",
+    "quay": "Red Hat Quay",
+    "rosa": "Red Hat OpenShift Service on AWS",
+    "osd": "Red Hat OpenShift Dedicated",
+}
+
+
+def _is_date(val):
+    """Return True if val looks like a YYYY-MM-DD date string."""
+    if not val or not isinstance(val, str):
+        return False
+    return bool(re.match(r"^\d{4}-\d{2}-\d{2}", val))
+
+
+def _to_date(val):
+    """Extract YYYY-MM-DD from a date string, or None."""
+    if _is_date(val):
+        return val[:10]
+    return None
+
+
+def _phase_date(phases, phase_name):
+    """Extract the end_date for a named phase, formatted as YYYY-MM-DD or raw string."""
+    for p in phases:
+        if p.get("name") == phase_name:
+            d = p.get("end_date", "N/A")
+            if d and isinstance(d, str) and _is_date(d):
+                return d[:10]
+            return str(d) if d else "N/A"
+    return "N/A"
+
+
+def _ver_sort_key(version_str):
+    """Sort key for version strings like '4.16' or '26.2'."""
+    try:
+        return [int(x) for x in version_str.split(".")]
+    except ValueError:
+        return [0]
+
+
+def resolve_product_name(product):
+    """Resolve a product alias to the full API product name."""
+    return PRODUCT_ALIASES.get(product.lower(), product)
+
+
+def fetch_api(product_name):
+    """Fetch raw lifecycle data from the Red Hat Product Life Cycles API."""
+    url = f"{LIFECYCLE_API_URL}?name={product_name.replace(' ', '+')}"
+    req = urllib.request.Request(
+        url, headers={"Accept": "application/json", "User-Agent": "rhdh-skill"}
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=30) as resp:
+            return json.loads(resp.read().decode("utf-8"))
+    except (urllib.error.URLError, OSError) as exc:
+        print(f"ERROR: Failed to fetch lifecycle data for {product_name}: {exc}", file=sys.stderr)
+        sys.exit(1)
+
+
+def parse_versions(api_data, filter_version=None):
+    """Parse raw API response into a consistent list of version dicts.
+
+    Returns a list of dicts with keys: version, type, supported, ga_date,
+    end_date, phases (dict of phase_name -> end_date), extra (dict for
+    product-specific fields like ocp_versions).
+    """
+    data_list = api_data.get("data", [])
+    if not data_list:
+        return []
+    versions_raw = data_list[0].get("versions", [])
+
+    results = []
+    for ver in versions_raw:
+        name = ver.get("name", "")
+        if filter_version and name != filter_version:
+            continue
+        vtype = ver.get("type", "")
+        raw_phases = ver.get("phases", [])
+
+        # Build phases dict
+        phases = {}
+        for p in raw_phases:
+            pname = p.get("name", "")
+            if pname:
+                phases[pname] = _phase_date(raw_phases, pname)
+
+        # GA date
+        ga_date = phases.get("General availability", "N/A")
+
+        # Latest end-of-support date across all non-GA phases
+        end_dates = [_to_date(d) for d in phases.values() if _to_date(d) and d != ga_date]
+        end_date = max(end_dates) if end_dates else "N/A"
+
+        # Product-specific extra fields
+        extra = {}
+        ocp_compat = ver.get("openshift_compatibility", "")
+        if ocp_compat:
+            extra["ocp_versions"] = [v.strip() for v in ocp_compat.split(",") if v.strip()]
+
+        results.append(
+            {
+                "version": name,
+                "type": vtype,
+                "supported": vtype != "End of life",
+                "ga_date": ga_date,
+                "end_date": end_date,
+                "phases": phases,
+                "extra": extra,
+            }
+        )
+
+    results.sort(key=lambda v: _ver_sort_key(v["version"]))
+    return results
+
+
+def fetch_product_lifecycle(product, filter_version=None):
+    """Fetch and parse lifecycle data for a Red Hat product.
+
+    Args:
+        product: Product alias ("rhbk", "quay", "rhdh", "ocp") or full name.
+        filter_version: Optional version string to filter to.
+
+    Returns:
+        List of version dicts with consistent shape.
+    """
+    full_name = resolve_product_name(product)
+    api_data = fetch_api(full_name)
+    return parse_versions(api_data, filter_version)
+
+
+def rhbk_major_versions(versions):
+    """Group RHBK minor versions into major version summaries.
+
+    A major version is "active" if at least one of its minor releases
+    is not end-of-life.
+
+    Returns:
+        List of dicts: {major_version, active, ga_date, end_date, minor_releases}
+    """
+    groups = {}
+    for v in versions:
+        # Skip umbrella entries like "26.x"
+        if "x" in v["version"] or not re.match(r"^\d+\.\d+$", v["version"]):
+            continue
+        major = v["version"].split(".")[0]
+        if major not in groups:
+            groups[major] = {
+                "minor_releases": [],
+                "any_active": False,
+                "ga_dates": [],
+                "end_dates": [],
+            }
+        groups[major]["minor_releases"].append(v["version"])
+        if v["supported"]:
+            groups[major]["any_active"] = True
+        if v["ga_date"] != "N/A":
+            groups[major]["ga_dates"].append(v["ga_date"])
+        if v["end_date"] != "N/A":
+            groups[major]["end_dates"].append(v["end_date"])
+
+    results = []
+    for major, info in sorted(groups.items(), key=lambda x: int(x[0])):
+        results.append(
+            {
+                "major_version": major,
+                "active": info["any_active"],
+                "ga_date": min(info["ga_dates"]) if info["ga_dates"] else "N/A",
+                "end_date": max(info["end_dates"]) if info["end_dates"] else "N/A",
+                "minor_releases": sorted(info["minor_releases"], key=_ver_sort_key),
+            }
+        )
+    return results
+
+
+def list_known_products():
+    """Return sorted list of known product aliases and their full names."""
+    return sorted(PRODUCT_ALIASES.items())

skills/_shared/rhdh_lifecycle/rhdh.py

@@ -0,0 +1,45 @@
+#!/usr/bin/env python3
+"""RHDH release lifecycle data -- wrapper around the generic Red Hat API client.
+
+Usage:
+    from rhdh_lifecycle.rhdh import fetch_rhdh_lifecycle
+    versions = fetch_rhdh_lifecycle()
+"""
+
+from __future__ import annotations
+
+from rhdh_lifecycle.redhat import fetch_api, fetch_product_lifecycle, parse_versions
+
+
+def fetch_rhdh_lifecycle(filter_version=None):
+    """Fetch and parse RHDH lifecycle data."""
+    versions = fetch_product_lifecycle("rhdh", filter_version)
+    # Flatten extra.ocp_versions into top-level for convenience
+    for v in versions:
+        v["ocp_versions"] = v.get("extra", {}).get("ocp_versions", [])
+        v["full_support_end"] = v.get("phases", {}).get("Full support", "N/A")
+        v["maintenance_end"] = v.get("phases", {}).get("Maintenance support", "N/A")
+    return versions
+
+
+def fetch_lifecycle_api(product_name):
+    """Fetch raw API data. Delegates to redhat module."""
+    return fetch_api(product_name)
+
+
+def parse_rhdh_versions(api_data, filter_version=None):
+    """Parse RHDH versions from raw API data."""
+    versions = parse_versions(api_data, filter_version)
+    for v in versions:
+        v["ocp_versions"] = v.get("extra", {}).get("ocp_versions", [])
+        v["full_support_end"] = v.get("phases", {}).get("Full support", "N/A")
+        v["maintenance_end"] = v.get("phases", {}).get("Maintenance support", "N/A")
+    return versions
+
+
+def rhdh_supported_ocp_versions(rhdh_data):
+    """Return sorted list of OCP versions supported by any active RHDH release."""
+    return sorted(
+        {ocp for v in rhdh_data if v["supported"] for ocp in v.get("ocp_versions", [])},
+        key=lambda x: [int(n) for n in x.split(".")],
+    )

skills/_shared/rhdh_prow/__init__.py

@@ -0,0 +1 @@
+"""Access openshift/release repo data (CI configs, cluster pools)."""