feat: migrate to micro-step control management [WIP]

Author: ivangabriele

.github/workflows/check.yml

@@ -23,40 +23,52 @@ jobs:
       - name: Install dependencies
         run: poetry install
 
-      - name: Setup Lint
-        run: cp ./env.sample.jsonc ./env.jsonc
+      - name: Setup Envs (OpenAI)
+        run: cp ./env.openai-sample.jsonc ./env.jsonc
 
       - name: Lint
         run: make lint
 
-  # type:
-  #   name: Type
-  #   runs-on: ubuntu-latest
-  #   steps:
-  #     - name: Checkout code
-  #       uses: actions/checkout@v4
+      - name: Setup Envs (OpenSource)
+        run: cp ./env.opensource-sample.jsonc ./env.jsonc
 
-  #     - name: Setup Python
-  #       uses: actions/setup-python@v4
-  #       with:
-  #         python-version: "3.10"
+      - name: Lint
+        run: make lint
+
+  type:
+    name: Type
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Setup Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.10"
 
-  #     - name: Install Poetry
-  #       uses: snok/install-poetry@v1
-  #       with:
-  #         version: "1.6.1"
+      - name: Install Poetry
+        uses: snok/install-poetry@v1
+        with:
+          version: "1.6.1"
+
+      - name: Install mypy
+        run: pip install mypy
+
+      - name: Install dependencies
+        run: poetry install
 
-  #     - name: Install mypy
-  #       run: pip install mypy
+      - name: Setup Envs (OpenAI)
+        run: cp ./env.openai-sample.jsonc ./env.jsonc
 
-  #     - name: Install dependencies
-  #       run: poetry install
+      - name: Check Types
+        run: make type
 
-  #     - name: Setup Types
-  #       run: cp ./env.sample.jsonc ./env.jsonc
+      - name: Setup Envs (OpenSource)
+        run: cp ./env.opensource-sample.jsonc ./env.jsonc
 
-  #     - name: Check Types
-  #       run: make type
+      - name: Check Types
+        run: make type
 
   test:
     name: Unit Test
@@ -78,8 +90,14 @@ jobs:
       - name: Install dependencies
         run: poetry install
 
-      - name: Setup Unit Tests
-        run: cp ./env.sample.jsonc ./env.jsonc
+      - name: Setup Envs (OpenAI)
+        run: cp ./env.openai-sample.jsonc ./env.jsonc
+
+      - name: Run Unit Tests
+        run: make test
+
+      - name: Setup Envs (OpenSource)
+        run: cp ./env.opensource-sample.jsonc ./env.jsonc
 
       - name: Run Unit Tests
         run: make test

.pylintrc

@@ -5,5 +5,5 @@ max-line-length=120
 ignore-paths=.cache,__pycache__
 
 [MESSAGES CONTROL]
-; TODO Re-enable these checks before releasing the first version.
-disable=C0301,C0114,C0115,C0116
+; TODO Re-enable some of these checks before releasing the first version.
+disable=C0301,C0114,C0115,C0116,R0801,R0902,R0903,W0511,W0718

Makefile

@@ -1,14 +1,17 @@
 .PHONY: clean lint install run test type
 
 clean:
-	find ./project -mindepth 1 -type f,d ! -name '.gitkeep' -exec rm -Rf {} +
+	find ./project -maxdepth 1 -mindepth 1 -type f,d ! -name '.gitkeep' -exec rm -fr {} +
 
 install:
 	poetry install
 
 lint:
 	poetry run pylint $(shell find . -name '*.py')
 
+ran:
+	poetry run python ./niam.py
+
 run:
 	poetry run python ./main.py
 

actions/__init__.py

@@ -1,7 +1,8 @@
 from .fetch_web_page import fetch_web_page
-from .read_file import read_file
+from .read_file import ReadFileAction
 from .run_bash_command import run_bash_command
 from .run_rust_file import run_rust_file
 from .search_web import search_web
 from .search_web_types import WebSearchApiResponse
-from .write_file import write_file
+from .update_tasks import UpdateTasksAction
+from .write_file import WriteFileAction

actions/base_action.py

@@ -0,0 +1,25 @@
+from typing import Any, Dict, List, TypedDict
+
+
+class ActionDesciptionParametersDict(TypedDict):
+    type: "object"
+    properties: Dict[str, Any]
+    required: List[str]
+
+
+class ActionDesciptionDict(TypedDict):
+    name: str
+    description: str
+    parameters: ActionDesciptionParametersDict
+
+
+class BaseAction:
+    _description: ActionDesciptionDict
+
+    @classmethod
+    def get_description(cls):
+        return cls._description
+
+    @staticmethod
+    def run(*args, **kwargs):
+        pass

actions/fetch_web_page.py

@@ -18,14 +18,14 @@ def fetch_web_page(url: str) -> str:
         body_tag = soup.body if soup.body else soup
         for tag in body_tag.find_all(True):
             if tag.name != "a":
-                if tag.name == "script" or tag.name == "style":
+                if tag.name in ("script", "style"):
                     tag.extract()
                 else:
                     tag.unwrap()
         if soup.html:
             soup.html.unwrap()
 
-        markdown_lines = _extract_content(body_tag)
+        markdown_lines = _extract_content(tag=body_tag, markdown_lines=None)
 
         markdown_source = " ".join(markdown_lines)
         markdown_source = re.sub(r"\s+", " ", markdown_source)
@@ -42,7 +42,10 @@ def fetch_web_page(url: str) -> str:
         return f"An error occurred: {e}"
 
 
-def _extract_content(tag: Tag, markdown_lines: List[str] = []) -> List[str]:
+def _extract_content(tag: Tag, markdown_lines: List[str] | None) -> List[str]:
+    if markdown_lines is None:
+        markdown_lines = []
+
     for child in tag.children:
         if isinstance(child, NavigableString):
             if child.strip():  # Only non-empty strings

actions/read_file.py

@@ -1,12 +1,34 @@
 import os
 
+from actions.base_action import BaseAction
+
 from constants import PROJECT_DIRECTORY_PATH
 
 
-def read_file(relative_path: str) -> str:
-    """
-    Read content from a file and return it.
-    """
-    full_path = os.path.join(PROJECT_DIRECTORY_PATH, relative_path)
-    with open(full_path, "r") as file:
-        return file.read()
+class ReadFileAction(BaseAction):
+    _description = {
+        "name": "read_file",
+        "description": "Read and return a file content.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "relative_path": {
+                    "type": "string",
+                    "description": "Relative path of the file. Path is relative to the project directory.",
+                },
+            },
+            "required": ["relative_path"],
+        },
+    }
+
+    # pylint: disable=arguments-differ
+    @staticmethod
+    def read_file(relative_path: str) -> str:
+        """
+        Read content from a file and return it.
+        """
+
+        full_path = os.path.join(PROJECT_DIRECTORY_PATH, relative_path)
+
+        with open(file=full_path, mode="r", encoding="utf-8") as file:
+            return file.read()

actions/search_web.py

@@ -1,11 +1,11 @@
+import json
 from typing import Union
 from dacite import from_dict
-import json
 import requests
 
 from actions.search_web_types import WebSearchApiResponse
 
-from constants import PROJECT_CONFIG
+from constants import DEFAULT_REQUEST_TIMEOUT, PROJECT_CONFIG
 
 
 def search_web(query: str) -> str:
@@ -14,8 +14,8 @@ def search_web(query: str) -> str:
     # If it's an `str`, that means it's an error
     if isinstance(brave_search_api_result_or_error, str):
         return brave_search_api_result_or_error
-    else:
-        brave_search_api_result = brave_search_api_result_or_error
+
+    brave_search_api_result = brave_search_api_result_or_error
 
     # Simplify the data
     simplified_response_data = {
@@ -62,7 +62,9 @@ def _fetch_brave_search_api(query: str) -> Union[WebSearchApiResponse, str]:
         "extra_snippets": "true",
     }
 
-    response = requests.get(endpoint, headers=headers, params=params)
+    response = requests.get(
+        endpoint, headers=headers, params=params, timeout=DEFAULT_REQUEST_TIMEOUT
+    )
 
     if response.status_code != 200:
         return f"Error: {response.status_code} - {response.reason}"

actions/update_tasks.py

@@ -0,0 +1,53 @@
+import json
+from typing import List
+
+from actions.base_action import BaseAction
+from actions.write_file import WriteFileAction
+import typedefs
+
+
+class UpdateTasksAction(BaseAction):
+    _description = {
+        "name": "update_tasks",
+        "description": "Update the entire task list file by replacing it with the new tasks.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "file_name": {
+                    "type": "string",
+                    "description": "Tasks file name.",
+                },
+                "tasks_as_strs": {
+                    "type": "array",
+                    "description": "List of tasks.",
+                    "items": {
+                        "type": "string",
+                    },
+                },
+            },
+            "required": ["updated_tasks"],
+        },
+    }
+
+    # pylint: disable=arguments-differ
+    @staticmethod
+    def run(file_name: str, tasks_as_strs: List[str]) -> str:
+        """
+        Write content to a file. Create the file and/or directories if they don't exist.
+        """
+
+        tasks = [
+            (index, UpdateTasksAction.get_task_from_task_as_str)
+            for index, item in enumerate(tasks_as_strs)
+        ]
+        tasks_as_json = json.dumps(tasks)
+
+        return WriteFileAction.run(relative_path=file_name, file_source=tasks_as_json)
+
+    @staticmethod
+    def get_task_from_task_as_str(tasks_as_str: str, index: int) -> typedefs.TaskDict:
+        return {
+            "index": index,
+            "description": tasks_as_str,
+            "is_done": False,
+        }

actions/write_file.py

@@ -1,19 +1,43 @@
 import os
+from actions.base_action import BaseAction
 
 from constants import PROJECT_DIRECTORY_PATH
 
 
-def write_file(relative_path: str, file_source: str) -> str:
-    """
-    Write content to a file. Create the file and/or directories if they don't exist.
-    """
-    full_path = os.path.join(PROJECT_DIRECTORY_PATH, relative_path)
+class WriteFileAction(BaseAction):
+    _description = {
+        "name": "write_file",
+        "description": "Write content to a file, creating it if necessary.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "relative_path": {
+                    "type": "string",
+                    "description": "Relative path of the file. Path is relative to the project directory.",
+                },
+                "file_source": {
+                    "type": "string",
+                    "description": """Content to write.""",
+                },
+            },
+            "required": ["relative_path", "file_source"],
+        },
+    }
 
-    try:
-        os.makedirs(os.path.dirname(full_path), exist_ok=True)
-        with open(full_path, "w") as file:
-            file.write(file_source)
+    # pylint: disable=arguments-differ
+    @staticmethod
+    def run(relative_path: str, file_source: str) -> str:
+        """
+        Write content to a file. Create the file and/or directories if they don't exist.
+        """
+        full_path = os.path.join(PROJECT_DIRECTORY_PATH, relative_path)
 
-        return f"Done."
-    except Exception as e:
-        return f"Error: {str(e)}"
+        try:
+            os.makedirs(os.path.dirname(full_path), exist_ok=True)
+            with open(file=full_path, mode="w", encoding="utf-8") as file:
+                file.write(file_source)
+
+            return f"File successfully written to `{relative_path}`."
+
+        except Exception as e:
+            return f"Error: {str(e)}"