Updated submodule ai_logs by adding new methods and method document…

…ation

The following methods are added:

- `_format_git_logs()`
- `_get_git_logs()`
- `read_changelog_md()`
- `write_changelog_md()`
- `parse_changelog_md()`

Signed-off-by: Ayush Joshi <[email protected]>

tools/ai_logs/__init__.py

@@ -1 +1 @@
-from .changelog import AIGitLog
+from .changelog import Changelog

tools/ai_logs/changelog.py

@@ -11,132 +11,128 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+"""A changelog generator for `ai`."""
 
 from __future__ import annotations
 
 from typing import Union
 
 import os
 import git
+import date
 import pathlib
 
 
-class AIGitLog(git.Git):
-  class _AIChangeLogParser:
-    def __init__(self, changelogs: list[str]):
-      self.changelogs = changelogs
-
-    def _filter_out_logs(self,
-                         log_type: str,
-                         *,
-                         capitalize: bool = False) -> list[str]:
-      filtered_logs = tuple()
-      for changelog in self.changelogs:
-        if log_type.lower() == changelog.split()[0].lower():
-          filtered_logs = (
-            *filtered_logs,
-            changelog if capitalize is False else changelog[0].upper() +
-            changelog[1:]
-          )
-      return filtered_logs
-
-    def parse(self) -> dict[str, list]:
-      self.added = self._filter_out_logs('added', capitalize=True)
-      self.changed = self._filter_out_logs('changed', capitalize=True)
-      self.deprecated = self._filter_out_logs('deprecated', capitalize=True)
-      self.removed = self._filter_out_logs('removed', capitalize=True)
-      self.fixed = self._filter_out_logs('fixed', capitalize=True)
-      self.security = self._filter_out_logs('security', capitalize=True)
-      self.yanked = self._filter_out_logs('yanked', capitalize=True)
-
-      return {
-        'added': self.added,
-        'changed': self.changed,
-        'deprecated': self.deprecated,
-        'removed': self.removed,
-        'fixed': self.fixed,
-        'security': self.security,
-        'yanked': self.yanked
-      }
+class Changelog(git.Git):
+  git_log_fmt = '%s ([#%h](https://www.github.com/joshiayush/ai/commit/%h))'
 
   def __init__(self, working_dir: Union[str, pathlib.Path]):
     super().__init__(working_dir)
+    self.changelog = dict()
+    self.date = date.today()
+
+  def _format_git_logs(self) -> str:
+    """Formats git logs for writing into the changelog file."""
+    changelog = ''
+    for key in self.changelog:
+
+      # Keys at the first level were the follow paths given during log
+      # generation
+      changelog += f'## {key} — {self.date}'
+      changelog += '\n' * 2
+
+      for git_log_t in self.changelog[key]:
+        if len(self.changelog[key][git_log_t]) == 0:
+          continue
+
+        changelog += f'### {git_log_t.capitalize()}'
+        changelog += '\n' * 2
+        changelog += '\n'.join(
+          map(lambda log: f'- {log}', self.changelog[key][git_log_t])
+        )
+        changelog += '\n' * 2
+
+  def _get_git_logs(self, pretty: str,
+                    follow: Union[str, pathlib.Path]) -> dict[str, set]:
+    """Returns a dictionary containing the log records for every type of change.
 
-  def _get_changelog_splitted(self,
-                              pretty: str,
-                              follow: str = None) -> list[str]:
+    The log records are returned in a form of set to later subtract already
+    published logs from it.
+
+    Args:
+      pretty: A prettified string format for the log statement.
+      follow: A path for which the logs are to be generated. For example, if
+        given `docs/`, it means follow the changes made to this directory only.
+
+    Returns:
+      A dictionary containing the log records for every type of change.
+    """
     if follow:
-      return self.log(f'--pretty={pretty}', '--follow', follow).split('\n')
-    return self.log(f'--pretty={pretty}').split('\n')
-
-  def get_changelog(self, pretty: str, follow: str = None) -> dict[str, tuple]:
-    parser = AIGitLog._AIChangeLogParser(
-      self._get_changelog_splitted(pretty, follow)
-    )
-    return parser.parse()
-
-  def _prettify(self, changelog: list[str]) -> str:
-    def write_for(heading: str) -> str:
-      nonlocal changelog
-      prettified = ''
-      if changelog[heading]:
-        added = []
-        for log in changelog[heading]:
-          added = [*added, f'- {log}']
-        prettified += f'### {heading.capitalize()}'
-        prettified += '\n\n'
-        prettified += '\n'.join(added)
-      return prettified
-
-    return '\n'.join(
-      [
-        write_for('added'),
-        write_for('changed'),
-        write_for('deprecated'),
-        write_for('fixed'),
-        write_for('removed'),
-        write_for('security'),
-        write_for('yanked')
-      ]
-    )
-
-  def parse_changelogs_for_all(self,
-                               follow_path: tuple[str]) -> dict[str, tuple]:
-    merged_changelog_dict = {
-      'added': tuple(),
-      'changed': tuple(),
-      'deprecated': tuple(),
-      'fixed': tuple(),
-      'removed': tuple(),
-      'security': tuple(),
-      'yanked': tuple(),
-    }
-    for follow in follow_path:
-      changelogs = self.get_changelog(
-        '%s ([#%h](https://www.github.com/joshiayush/ai/commit/%h))',
-        follow=follow
+      git_logs = self.log(f'--pretty={pretty}', '--follow', follow)
+    else:
+      git_logs = self.log(f'--pretty={pretty}')
+    git_logs = git_logs.split('\n')
+
+    git_logs_dict = dict()
+    for git_log_t in (
+      'added',
+      'changed',
+      'deprecated',
+      'fixed',
+      'removed',
+      'security',
+      'yanked',
+    ):
+      git_logs_dict.update(
+        {
+          f'{git_log_t}':
+          set(
+            filter(
+              lambda changelog: changelog is not None,
+              filter(
+                lambda changelog: changelog.capitalize()
+                if git_log_t == changelog.split()[0].lower() else None
+              )
+            )
+          )
+        }
       )
-      for key in merged_changelog_dict:
-        merged_changelog_dict[key] = (
-          *merged_changelog_dict[key], *changelogs[key]
-        )
 
-    for key in merged_changelog_dict:
-      merged_changelog_dict[key] = tuple(set(merged_changelog_dict[key]))
-    return self._prettify(merged_changelog_dict)
+    return git_logs_dict
+
+  def read_changelog_md(self, fpath: Union[str, pathlib.Path]) -> str:
+    """Returns the contents of the changelog file; omitting the first line.
+
+    The first line is just `# CHANGELOG` which does not contribute to the actual
+    values that needs to be parsed before updating the changelog file.
+
+    Args:
+      fpath: Path to the changelog file.
+
+    Returns:
+      The changelogs.
+    """
+    changelog = ''
+    with open(fpath, mode='r', encoding='utf-8') as f:
+      changelog = ''.join(f.readlines()[1:])
+    return changelog
 
-  def write_changelog(
-    self, fpath: Union[str, pathlib.Path], changelog: str
+  def write_changelog_md(
+    self, fpath: Union[str, pathlib.Path], follow: tuple[Union[str,
+                                                               pathlib.Path]]
   ) -> None:
-    old_changelogs = None
-    if os.access(os.fspath(fpath), os.F_OK):
-      with open(fpath, mode='r', encoding='utf-8') as f:
-        old_changelogs = ''.join(f.readlines()[1:])
-
-    changelog = f'# CHANGELOG\n\n{changelog}'
-    with open(fpath, mode='w', encoding='utf-8') as f:
-      f.write(changelog)
-
-    if old_changelogs:
-      with open(fpath, mode='a', encoding='utf-8') as f:
-        f.writelines(['\n', '\n', old_changelogs])
+    """Writes the newly added logs to the changelog file.
+
+    Args:
+      fpath: Writable changelog file path.
+      follow: A path for which the logs are to be generated. For example, if
+        given `docs/`, it means follow the changes made to this directory only.
+    """
+    for f in follow:
+      git_logs = self._get_git_logs(self.git_log_fmt, follow=f)
+      self.changelog.update({f'{f}': git_logs})
+
+  def parse_changelog_md(self,
+                         changelog: str) -> dict[str, dict[str, set[str]]]:
+    """Parse the changelog file to keep a list of all the logs recorded."""
+    pass

tools/manage.py

@@ -64,21 +64,14 @@ def GenerateAIDocs() -> int:
 
 def GenerateAILogs() -> int:
   error_code = 0
-
-  logger = ai_logs.AIGitLog(_BASE_AI_REPO)
-  today = date.today()
-
-  changelog = ''
-  changelog += f'## ai — {today}'
-  changelog += '\n\n'
-  changelog += logger.parse_changelogs_for_all(_BASE_AI_MODULE_PATH)
-
-  changelog += f'## docs — {today}'
-  changelog += '\n\n'
-  changelog += logger.parse_changelogs_for_all(_BASE_AI_DOCS_PATH)
-
+  changelog = ai_logs.Changelog(_BASE_AI_REPO)
   try:
-    logger.write_changelog('CHANGELOG.md', changelog.strip())
+    changelog.write_changelog_md(
+      'CHANGELOG.md', follow=(
+        _BASE_AI_MODULE_PATH,
+        _BASE_AI_DOCS_PATH,
+      )
+    )
   except PermissionError as exc:
     error_code = exc.errno
   return error_code