Allow to remove specific collection changelog entries from Ansible changelog

changelogs/fragments/639-changelog-cleanup.yml

@@ -0,0 +1,2 @@
+minor_changes:
+  - "Allow to remove collection changelog entries from the Ansible changelog (https://github.com/ansible-community/antsibull-build/pull/639)."

src/antsibull_build/build_data_lint.py

@@ -12,10 +12,43 @@
 
 import os
 
+import pydantic as p
 from antsibull_changelog.lint import lint_changelog_yaml as _lint_changelog_yaml
 from antsibull_core import app_context
 from antsibull_core.collection_meta import lint_collection_meta as _lint_collection_meta
 from antsibull_core.dependency_files import parse_pieces_file
+from antsibull_core.pydantic import forbid_extras, get_formatted_error_messages
+from semantic_version import Version as SemVer
+
+from .changelog import RemoveCollectionChangelogEntries
+
+
+def _lint_rcce(rcce: dict, errors: list[str]) -> None:
+    forbid_extras(RemoveCollectionChangelogEntries)
+    try:
+        rcce_obj = RemoveCollectionChangelogEntries.model_validate(rcce)
+        for collection_name, versions in rcce_obj.root.items():
+            for version in versions.root:
+                try:
+                    SemVer(version)
+                except ValueError:
+                    errors.append(
+                        f"remove_collection_changelog_entries -> {collection_name}"
+                        f" -> {version}: {version!r} is not a valid semantic version"
+                    )
+    except p.ValidationError as e:
+        for msg in get_formatted_error_messages(e):
+            errors.append(f"remove_collection_changelog_entries -> {msg}")
+
+
+def _lint_changelog_extra_data(data: dict, errors: list[str]) -> None:
+    rcce = data.pop("remove_collection_changelog_entries", None)  # noqa: F841
+    if isinstance(rcce, dict):
+        _lint_rcce(rcce, errors)
+    elif rcce is not None:
+        errors.append(
+            "remove_collection_changelog_entries: Input should be a valid dictionary"
+        )
 
 
 def lint_build_data() -> int:
@@ -38,7 +71,10 @@ def lint_build_data() -> int:
     # Lint changelog.yaml
     changelog_path = os.path.join(data_dir, "changelog.yaml")
     for path, _, __, message in _lint_changelog_yaml(
-        changelog_path, no_semantic_versioning=True, strict=True
+        changelog_path,
+        no_semantic_versioning=True,
+        strict=True,
+        preprocess_data=lambda data: _lint_changelog_extra_data(data, errors),
     ):
         errors.append(f"{path}: {message}")
 

src/antsibull_build/changelog.py

@@ -17,9 +17,11 @@
 import tempfile
 import typing as t
 from collections import defaultdict
+from collections.abc import Callable
 
 import aiohttp
 import asyncio_pool  # type: ignore[import]
+import pydantic as p
 from antsibull_changelog.changes import ChangesData, add_release
 from antsibull_changelog.config import (
     ChangelogConfig,
@@ -51,6 +53,37 @@
 mlog = log.fields(mod=__name__)
 
 
+class RemoveCollectionVersionSchema(p.BaseModel):
+    changes: dict[str, list[str]]
+
+
+RemoveCollectionVersionsSchema = p.RootModel[dict[str, RemoveCollectionVersionSchema]]
+
+RemoveCollectionChangelogEntries = p.RootModel[
+    dict[str, RemoveCollectionVersionsSchema]
+]
+
+
+def _extract_extra_data(
+    data: dict,
+    store: Callable[[dict[str, dict[SemVer, RemoveCollectionVersionSchema]]], None],
+) -> None:
+    try:
+        rcce_obj = RemoveCollectionChangelogEntries.model_validate(
+            data.get("remove_collection_changelog_entries") or {}
+        )
+        rcce = {
+            collection_name: {
+                SemVer(version): data for version, data in versions.root.items()
+            }
+            for collection_name, versions in rcce_obj.root.items()
+        }
+    except (p.ValidationError, ValueError):
+        # ignore error; linting should complain, not us
+        rcce = {}
+    store(rcce)
+
+
 class ChangelogData:
     """
     Data for a single changelog (for a collection, for ansible-core, for Ansible)
@@ -62,6 +95,10 @@ class ChangelogData:
     generator: ChangelogGenerator
     generator_flatmap: bool
 
+    remove_collection_changelog_entries: (
+        dict[str, dict[SemVer, RemoveCollectionVersionSchema]] | None
+    ) = None
+
     def __init__(
         self,
         paths: PathsConfig,
@@ -74,6 +111,7 @@ def __init__(
         self.changes = changes
         self.generator_flatmap = flatmap
         self.generator = ChangelogGenerator(self.config, self.changes, flatmap=flatmap)
+        self.remove_collection_changelog_entries = None
 
     @classmethod
     def collection(
@@ -114,13 +152,28 @@ def ansible(
         config.release_tag_re = r"""(v(?:[\d.ab\-]|rc)+)"""
         config.pre_release_tag_re = r"""(?P<pre_release>(?:[ab]|rc)+\d*)$"""
 
+        remove_collection_changelog_entries = {}
+
+        def store_extra_data(
+            rcce: dict[str, dict[SemVer, RemoveCollectionVersionSchema]]
+        ) -> None:
+            remove_collection_changelog_entries.update(rcce)
+
         changelog_path = ""
         if directory is not None:
             changelog_path = os.path.join(directory, "changelog.yaml")
-        changes = ChangesData(config, changelog_path)
+        changes = ChangesData(
+            config,
+            changelog_path,
+            extra_data_extractor=lambda data: _extract_extra_data(
+                data, store_extra_data
+            ),
+        )
         if output_directory is not None:
             changes.path = os.path.join(output_directory, "changelog.yaml")
-        return cls(paths, config, changes, flatmap=True)
+        result = cls(paths, config, changes, flatmap=True)
+        result.remove_collection_changelog_entries = remove_collection_changelog_entries
+        return result
 
     @classmethod
     def concatenate(cls, changelogs: list[ChangelogData]) -> ChangelogData:
@@ -159,7 +212,16 @@ def add_ansible_release(
             release_date["changes"]["release_summary"] = release_summary
 
     def save(self):
-        self.changes.save()
+        extra_data = {}
+        if self.remove_collection_changelog_entries is not None:
+            extra_data["remove_collection_changelog_entries"] = {
+                collection_name: {
+                    str(version): changes.model_dump()
+                    for version, changes in versions.items()
+                }
+                for collection_name, versions in self.remove_collection_changelog_entries.items()
+            }
+        self.changes.save(extra_data=extra_data or None)
 
 
 def read_file(tarball_path: str, matcher: t.Callable[[str], bool]) -> bytes | None:
@@ -763,6 +825,63 @@ def _populate_ansible_changelog(
                 )
 
 
+def _cleanup_collection_version(
+    collection_name: str,
+    collection_data: dict[str, RemoveCollectionVersionSchema],
+    changelog: ChangelogData,
+) -> None:
+    flog = mlog.fields(func="_cleanup_collection_version")
+    for version, data in collection_data.items():
+        release = changelog.changes.releases.get(str(version))
+        changes = (release or {}).get("changes")
+        if not changes:
+            flog.warning(
+                f"Trying to remove changelog entries from {collection_name} {version},"
+                " but found no release"
+            )
+            continue
+        for category, entries in data.changes.items():
+            if category not in changes:
+                flog.warning(
+                    f"Trying to remove {category!r} changelog entries from"
+                    f" {collection_name} {version}, but found no entries"
+                )
+                continue
+            for entry in entries:
+                try:
+                    changes[category].remove(entry)
+                except ValueError:
+                    flog.warning(
+                        f"Cannot find {category!r} changelog entry for"
+                        f" {collection_name} {version}: {entry!r}"
+                    )
+
+
+def _cleanup_collection_changelogs(
+    ansible_changelog: ChangelogData,
+    collection_collectors: list[CollectionChangelogCollector],
+) -> None:
+    flog = mlog.fields(func="_populate_ansible_changelog")
+    rcce = ansible_changelog.remove_collection_changelog_entries
+    if not rcce:
+        return
+
+    for collection_collector in collection_collectors:
+        collection_data = rcce.get(collection_collector.collection)
+        if not collection_data:
+            continue
+        changelog = collection_collector.changelog
+        if not changelog:
+            flog.warning(
+                f"Trying to remove changelog entries from {collection_collector.collection},"
+                " but found no changelog"
+            )
+            continue
+        _cleanup_collection_version(
+            collection_collector.collection, collection_data, changelog
+        )
+
+
 def get_changelog(
     ansible_version: PypiVer,
     deps_dir: str | None,
@@ -821,6 +940,8 @@ def get_changelog(
         collect_changelogs(collectors, core_collector, collection_cache, galaxy_context)
     )
 
+    _cleanup_collection_changelogs(ansible_changelog, collectors)
+
     changelog = []
 
     sorted_versions = collect_versions(versions, ansible_changelog.config)