firewallissues/update_products_from_issues.py

#!/usr/bin/env python3

from __future__ import annotations

import argparse
import json
import re
from dataclasses import dataclass
from datetime import date
from pathlib import Path
from typing import Any


ISSUE_TYPES = {"addressed", "known"}
FILENAME_RE = re.compile(r"^(?P<version>[^_]+)_(?P<date>\d{4}-\d{2}-\d{2})\.[^.]+$")
HOTFIX_RE = re.compile(r"^(?P<base>\d+(?:\.\d+)*)(?:-h(?P<hotfix>\d+))?$")


@dataclass(frozen=True)
class FileIdentity:
    product: str
    issue_type: str
    version: str


@dataclass(frozen=True)
class FileRecord:
    identity: FileIdentity
    filename: str
    release_date: date


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Update products.json from issue files, keeping latest date per version."
    )
    parser.add_argument(
        "--issues-dir",
        default="web/data/issues",
        help="Directory containing product issue files.",
    )
    parser.add_argument(
        "--products-json",
        default="web/data/products.json",
        help="Path to products.json.",
    )
    return parser.parse_args()


def collect_latest_issue_files(issues_dir: Path) -> dict[FileIdentity, FileRecord]:
    latest: dict[FileIdentity, FileRecord] = {}

    for file_path in sorted(p for p in issues_dir.rglob("*") if p.is_file()):
        relative = file_path.relative_to(issues_dir)
        if len(relative.parts) != 3:
            continue

        product, issue_type, filename = relative.parts
        if issue_type not in ISSUE_TYPES:
            continue

        parsed = parse_issue_filename(filename)
        if parsed is None:
            continue

        version, release_date = parsed
        identity = FileIdentity(product=product, issue_type=issue_type, version=version)
        record = FileRecord(identity=identity, filename=filename, release_date=release_date)

        current = latest.get(identity)
        if current is None or (record.release_date, record.filename) > (
            current.release_date,
            current.filename,
        ):
            latest[identity] = record

    return latest


def parse_issue_filename(filename: str) -> tuple[str, date] | None:
    match = FILENAME_RE.match(filename)
    if not match:
        return None

    version = match.group("version")
    try:
        release_date = date.fromisoformat(match.group("date"))
    except ValueError:
        return None

    return version, release_date


def clear_leaf_issue_arrays(node: Any) -> None:
    if isinstance(node, dict):
        if "addressed" in node or "known" in node:
            node["addressed"] = []
            node["known"] = []
            return

        for value in node.values():
            clear_leaf_issue_arrays(value)


def is_issue_leaf(node: Any) -> bool:
    return isinstance(node, dict) and ("addressed" in node or "known" in node)


def pick_target_leaf_path(products: dict[str, Any], product: str, version: str) -> list[str] | None:
    parts = version.split(".")
    if not parts:
        return None

    major = parts[0]
    minor = f"{parts[0]}.{parts[1]}" if len(parts) > 1 else parts[0]

    candidates = [
        [product, major, minor],
        [product, major],
        [product],
    ]

    for path in candidates:
        node = get_path(products, path)
        if is_issue_leaf(node):
            return path

    return None


def get_path(data: Any, path: list[str]) -> Any:
    current = data
    for key in path:
        if not isinstance(current, dict) or key not in current:
            return None
        current = current[key]
    return current


def parse_release_prefix(prefix: str) -> tuple[tuple[int, ...], int]:
    match = HOTFIX_RE.match(prefix)
    if not match:
        return tuple(), -1

    base_parts = tuple(int(part) for part in match.group("base").split("."))
    hotfix_group = match.group("hotfix")
    hotfix = int(hotfix_group) if hotfix_group is not None else -1
    return base_parts, hotfix


def filename_sort_key(filename: str) -> tuple[tuple[int, ...], int, int, str]:
    prefix = filename.split("_", 1)[0]
    base_parts, hotfix = parse_release_prefix(prefix)
    is_hotfix = 1 if hotfix >= 0 else 0
    return base_parts, is_hotfix, hotfix, filename


def add_file_to_leaf(products: dict[str, Any], path: list[str], issue_type: str, filename: str) -> None:
    leaf = get_path(products, path)
    if not isinstance(leaf, dict):
        return

    files = leaf.get(issue_type)
    if not isinstance(files, list):
        files = []

    files.append(filename)
    leaf[issue_type] = sorted(set(files), key=filename_sort_key)


def update_products(issues_dir: Path, products_path: Path) -> None:
    if not issues_dir.is_dir():
        raise FileNotFoundError(f"Issues directory not found: {issues_dir}")
    if not products_path.is_file():
        raise FileNotFoundError(f"products.json not found: {products_path}")

    with products_path.open("r", encoding="utf-8") as f:
        products = json.load(f)

    if not isinstance(products, dict):
        raise ValueError("products.json must contain a top-level object")

    clear_leaf_issue_arrays(products)

    latest_files = collect_latest_issue_files(issues_dir)
    for record in sorted(latest_files.values(), key=lambda r: (r.identity.product, r.identity.issue_type, r.identity.version)):
        path = pick_target_leaf_path(products, record.identity.product, record.identity.version)
        if path is None:
            continue
        add_file_to_leaf(products, path, record.identity.issue_type, record.filename)

    with products_path.open("w", encoding="utf-8") as f:
        json.dump(products, f, indent=2)
        f.write("\n")


def main() -> int:
    args = parse_args()
    repo_root = Path(__file__).resolve().parent
    issues_dir = (repo_root / args.issues_dir).resolve()
    products_path = (repo_root / args.products_json).resolve()

    update_products(issues_dir=issues_dir, products_path=products_path)
    print(f"Updated {products_path} from {issues_dir}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())