Merge pull request #55 from hrshdhgd/new-term-template

Add new term based on template

Author: hrshdhgd

src/ontobot_change_agent/api.py

@@ -97,6 +97,7 @@ def _extract_info_from_issue_object(issue: Issue) -> dict:
     important_info = {k: issue_as_dict[RAW_DATA][k] for k in ISSUE_KEYS}
     important_info["body"] = _make_sense_of_body(important_info["body"])
     important_info["user"] = issue_as_dict["_rawData"]["user"]["login"]
+    important_info["labels"] = [label["name"] for label in issue_as_dict["_rawData"]["labels"]]
     return important_info
 
 
@@ -163,3 +164,28 @@ def process_issue_via_oak(input: str, commands: list, output: str = None):
         impl_obj.apply_patch(change)
 
     impl_obj.dump(output, output_format)
+
+
+def process_new_term_template(body, prefix):
+    """Process an issue generated via new term request template."""
+    split_body = body.replace("\r", "").strip("#").split("\n\n###")
+    CURIE = prefix + ":XXXXXX"
+    body_as_dict = {}
+
+    for line in split_body:
+        if line.split("\n\n")[1].strip() not in ["_No response_", "None"]:
+            body_as_dict[line.split("\n\n")[0].strip()] = line.split("\n\n")[1].strip()
+
+    kgcl_command_list = [f"create node {CURIE} '{body_as_dict['Label']}'"]
+
+    if "Synonyms" in body_as_dict:
+        body_as_dict["Synonyms"] = body_as_dict["Synonyms"].split(",")
+        for synonym in body_as_dict["Synonyms"]:
+            if "Synonym type" in body_as_dict:
+                kgcl_command_list.append(
+                    f"create {body_as_dict['Synonym type']} synonym '{synonym.strip()}' for {CURIE}"
+                )
+            else:
+                kgcl_command_list.append(f"create synonym {synonym.strip()} for {CURIE}")
+
+    return (kgcl_command_list, body_as_dict)

src/ontobot_change_agent/cli.py

@@ -5,12 +5,18 @@
 import logging
 import os
 import re
-from typing import TextIO
+from typing import TextIO, Union
 
 import click
 
 from ontobot_change_agent import __version__
-from ontobot_change_agent.api import get_all_labels_from_repo, get_issues, process_issue_via_oak
+from ontobot_change_agent.api import (
+    get_all_labels_from_repo,
+    get_issues,
+    process_issue_via_oak,
+    process_new_term_template,
+)
+from ontobot_change_agent.constants import NEW_TERM_LABEL
 
 __all__ = [
     "main",
@@ -71,6 +77,8 @@ def main(verbose: int, quiet: bool):
     help="Filter based on a search for label of issue.",
 )
 
+prefix_option = click.option("-p", "--prefix", help="Assign prefix based on ontology resource.")
+
 state_option = click.option(
     "-s",
     "--state",
@@ -140,29 +148,35 @@ def get_labels(repo: str, token: str):
 @input_argument
 @repo_option
 @token_option
+@prefix_option
 @label_option
 @issue_number_option
 @state_option
 @output_option
 def process_issue(
-    input: str, repo: str, token: str, label: str, number: int, state: str, output: str
+    input: str, repo: str, prefix: str, token: str, label: str, number: int, state: str, output: str
 ):
     """Run processes based on issue label.
 
     :param repo: GitHub repository name [org/repo_name]
     :param label: Label of issues.
     :param state: State of issue ["open", "close" etc.]
     """
-    formatted_body = "The following commands were executed: </br> "
-
     for issue in get_issues(
         repository_name=repo, token=token, label=label, number=number, state=state
     ):
         # Make sure ontobot_change_agent needs to be triggered or no.
         if issue:
-            if re.match(r"(.*)ontobot(.*)apply(.*):(.*)", issue[BODY]):
+            KGCL_COMMANDS = []
+            if NEW_TERM_LABEL in issue["labels"]:
+                formatted_body = "The following input was provided: </br> "
+                KGCL_COMMANDS, body_as_dict = process_new_term_template(issue["body"], prefix)
+                formatted_body += _convert_to_markdown(body_as_dict)
+                formatted_body += "</br> The following commands were executed: </br> "
+
+            elif re.match(r"(.*)ontobot(.*)apply(.*):(.*)", issue[BODY]):
+                formatted_body = "The following commands were executed: </br> "
                 bullet_starters = ["* ", "- "]
-                KGCL_COMMANDS = []
                 for bullet in bullet_starters:
                     KGCL_COMMANDS.extend(
                         [
@@ -171,47 +185,54 @@ def process_issue(
                             if item.lstrip().startswith(bullet)
                         ]
                     )
-                if output:
-                    new_output = output
-                else:
-                    new_output = input
 
                 KGCL_COMMANDS = [x.strip() for x in KGCL_COMMANDS]
-                if issue["number"] == number and len(KGCL_COMMANDS) > 0:  # noqa W503  # noqa W503
-                    process_issue_via_oak(
-                        input=input,
-                        commands=KGCL_COMMANDS,
-                        output=new_output,
-                    )
 
-                    formatted_body += _list_to_markdown(KGCL_COMMANDS)
-                    formatted_body += "</br>Fixes #" + str(issue["number"])
-                    # TODO: remove `set-output` when env var setting is confirmed.
-                    if os.getenv("GITHUB_ENV"):
-                        with open(os.getenv("GITHUB_ENV"), "a") as env:  # type: ignore
-                            print(f"PR_BODY={formatted_body}", file=env)
-                            print(f"PR_TITLE={issue[TITLE]}", file=env)
-                            print(f"ISSUE_CREATOR={issue[USER]}", file=env)
-
-                    click.echo(
-                        f"""
-                        PR_BODY={formatted_body}
-                        PR_TITLE={issue[TITLE]}
-                        ISSUE_CREATOR={issue[USER]}
-                        """
-                    )
             else:
                 click.echo(f"""{issue[TITLE]} does not need ontobot's attention.""")
         else:
             click.echo(f"""Issue number:{number} is either closed or does not exist.""")
             break
 
-
-def _list_to_markdown(list: list) -> str:
+        if output:
+            new_output = output
+        else:
+            new_output = input
+
+        if issue["number"] == number and len(KGCL_COMMANDS) > 0:  # noqa W503  # noqa W503
+            process_issue_via_oak(
+                input=input,
+                commands=KGCL_COMMANDS,
+                output=new_output,
+            )
+
+            formatted_body += _convert_to_markdown(KGCL_COMMANDS)
+            formatted_body += "</br>Fixes #" + str(issue["number"])
+
+            if os.getenv("GITHUB_ENV"):
+                with open(os.getenv("GITHUB_ENV"), "a") as env:  # type: ignore
+                    print(f"PR_BODY={formatted_body}", file=env)
+                    print(f"PR_TITLE={issue[TITLE]}", file=env)
+                    print(f"ISSUE_CREATOR={issue[USER]}", file=env)
+
+            click.echo(
+                f"""
+                PR_BODY={formatted_body}
+                PR_TITLE={issue[TITLE]}
+                ISSUE_CREATOR={issue[USER]}
+                """
+            )
+
+
+def _convert_to_markdown(item: Union[list, dict]) -> str:
     bullet = "* "
     md = ""
-    for line in list:
-        md += bullet + line + "</br>"
+    if isinstance(item, list):
+        for line in item:
+            md += bullet + line + "</br>"
+    elif isinstance(item, dict):
+        for k, v in item.items():
+            md += bullet + k + ":" + str(v) + "</br>"
     return md
 
 

src/ontobot_change_agent/constants.py

@@ -0,0 +1,3 @@
+"""Constants."""
+NEW_TERM_LABEL = "New term request"
+SYNONYM_LABEL = "synonym"

tests/test_api.py

@@ -28,20 +28,19 @@ def test_get_issues_with_label(self):
         self.assertTrue(type(issues[0]), Issue)
         self.assertTrue(issues[0]["number"], 2)
         self.assertTrue(issues[0]["title"], self.issue_title)
-        self.assertTrue(issues[0]["labels"][0]["name"], self.label)
+        self.assertTrue(issues[0]["labels"][0], self.label)
 
     def test_get_issues_with_title(self):
         """Test if 'get_issues' returns the correct title."""
         issues = []
         for issue in get_issues(repository_name=self.repo_name, title_search=self.issue_title):
             if issue:
                 issues.append(issue)
-
         self.assertEqual(len(issues), 1)
         self.assertTrue(type(issues[0]), Issue)
         self.assertTrue(issues[0]["number"], 2)
         self.assertTrue(issues[0]["title"], self.issue_title)
-        self.assertTrue(issues[0]["labels"][0]["name"], self.label)
+        self.assertTrue(issues[0]["labels"][0], self.label)
 
     def test_get_issues_with_number(self):
         """Test return the correct issue by number."""