Refactor workflow file loading and error handling

devchat/__init__.py

@@ -1,4 +1,4 @@
 import os
 
 script_dir = os.path.dirname(os.path.realpath(__file__))
-os.environ["TIKTOKEN_CACHE_DIR"] = os.path.join(script_dir, "tiktoken_cache")
+os.environ["TIKTOKEN_CACHE_DIR"] = os.path.join(script_dir, "tiktoken_cache")

devchat/_cli/router.py

@@ -145,7 +145,7 @@ def llm_route(content: Optional[str], parent: Optional[str], reference: Optional
         workflow = Workflow.load(name) if name else None
         if workflow:
             print(assistant.prompt.formatted_header())
-            
+
             return_code = 0
             if workflow.should_show_help(user_input):
                 doc = workflow.get_help_doc(user_input)

devchat/workflow/command/config.py

@@ -1,7 +1,6 @@
 import json
 
 from pathlib import Path
-from typing import NamedTuple, List, Set, Tuple
 import click
 import oyaml as yaml
 
@@ -11,12 +10,12 @@
 @click.command(help="Workflow configuration.", name="config")
 @click.option("--json", "in_json", is_flag=True, help="Output in json format.")
 def config_cmd(in_json: bool):
-    
+
     config_path = Path(WORKFLOWS_BASE) / WORKFLOWS_CONFIG_FILENAME
     config_content = {}
     if config_path.exists():
-        with open(config_path, "r", encoding="utf-8") as f:
-            config_content = yaml.safe_load(f.read())
+        with open(config_path, "r", encoding="utf-8") as file:
+            config_content = yaml.safe_load(file.read())
 
     if not in_json:
         click.echo(config_content)

devchat/workflow/command/list.py

@@ -1,15 +1,19 @@
 import json
 from pathlib import Path
-from typing import NamedTuple, List, Set, Tuple, Optional, Dict
+from typing import List, Set, Tuple, Dict
 from dataclasses import dataclass, asdict, field
 
 import click
 import oyaml as yaml
+import yaml as pyyaml
 
 from devchat.workflow.namespace import get_prioritized_namespace_path
 from devchat.workflow.path import COMMAND_FILENAMES
+from devchat.utils import get_logger
 
 
+logger = get_logger(__name__)
+
 @dataclass
 class WorkflowMeta:
     name: str
@@ -41,28 +45,33 @@ def iter_namespace(
     interest_files = set(COMMAND_FILENAMES)
     result = []
     unique_names = set(existing_names)
-    for f in root.rglob("*"):
-        if f.is_file() and f.name in interest_files:
-            rel_path = f.relative_to(root)
-            parts = rel_path.parts
-            workflow_name = ".".join(parts[:-1])
-            is_first = workflow_name not in unique_names
-            unique_names.add(workflow_name)
-
-            # load the config content from f
-            with open(f, "r", encoding="utf-8") as fi:
-                yaml_content = fi.read()
-                command_conf = yaml.safe_load(yaml_content)
-                # pop the "steps" field
-                command_conf.pop("steps", None)
-
-            workflow = WorkflowMeta(
-                name=workflow_name,
-                namespace=root.name,
-                active=is_first,
-                command_conf=command_conf,
-            )
-            result.append(workflow)
+    for file in root.rglob("*"):
+        try:
+            if file.is_file() and file.name in interest_files:
+                rel_path = file.relative_to(root)
+                parts = rel_path.parts
+                workflow_name = ".".join(parts[:-1])
+                is_first = workflow_name not in unique_names
+
+                # load the config content from file
+                with open(file, "r", encoding="utf-8") as file_handle:
+                    yaml_content = file_handle.read()
+                    command_conf = yaml.safe_load(yaml_content)
+                    # pop the "steps" field
+                    command_conf.pop("steps", None)
+
+                workflow = WorkflowMeta(
+                    name=workflow_name,
+                    namespace=root.name,
+                    active=is_first,
+                    command_conf=command_conf,
+                )
+                unique_names.add(workflow_name)
+                result.append(workflow)
+        except pyyaml.scanner.ScannerError as err:
+            logger.error("Failed to load %s: %s", rel_path, err)
+        except Exception as err:
+            logger.error("Unknown error when loading %s: %s", rel_path, err)
 
     return result, unique_names
 
@@ -75,19 +84,19 @@ def list_cmd(in_json: bool):
     workflows: List[WorkflowMeta] = []
     visited_names = set()
     for ns_path in namespace_paths:
-        ws, visited_names = iter_namespace(ns_path, visited_names)
-        workflows.extend(ws)
+        ws_names, visited_names = iter_namespace(ns_path, visited_names)
+        workflows.extend(ws_names)
 
     if not in_json:
         # print basic info
-        active_count = len([w for w in workflows if w.active])
+        active_count = len([workflow for workflow in workflows if workflow.active])
         total_count = len(workflows)
         click.echo(f"workflows (active/total): {active_count}/{total_count}")
-        for w in workflows:
-            click.echo(w)
+        for workflow in workflows:
+            click.echo(workflow)
 
     else:
         # convert workflows to json
-        data = [asdict(w) for w in workflows]
+        data = [asdict(workflow) for workflow in workflows]
         json_format = json.dumps(data)
         click.echo(json_format)

devchat/workflow/command/update.py

@@ -57,7 +57,7 @@ def _backup(workflow_base: Path, n: int = 5) -> Optional[Path]:
     """
 
     if not workflow_base.exists():
-        return
+        return None
 
     backup_dir = workflow_base.parent / ".backup"
     backup_dir.mkdir(exist_ok=True)
@@ -180,7 +180,7 @@ def _clone_repo_to_dir(candidates: List[Tuple[str, str]], dst_dir: Path) -> bool
     clone_ok = False
     for url, branch in candidates:
         try:
-            repo = Repo.clone_from(url, to_path=dst_dir, branch=branch)
+            Repo.clone_from(url, to_path=dst_dir, branch=branch)
             click.echo(f"Cloned from {url}|{branch} to {dst_dir}")
             clone_ok = True
             break
@@ -274,8 +274,6 @@ def update_by_git(workflow_base: Path):
             )
         except GitCommandError as e:
             click.echo(f"Failed to update to the latest main: {e}. Skip update.")
-            return
-        return
 
 
 def copy_workflows_usr():

devchat/workflow/namespace.py

@@ -34,14 +34,14 @@ def _load_custom_config() -> CustomConfig:
     if not os.path.exists(CUSTOM_CONFIG_FILE):
         return config
 
-    with open(CUSTOM_CONFIG_FILE, "r", encoding="utf-8") as f:
-        content = f.read()
+    with open(CUSTOM_CONFIG_FILE, "r", encoding="utf-8") as file:
+        content = file.read()
         yaml_content = yaml.safe_load(content)
         try:
             if yaml_content:
                 config = CustomConfig.parse_obj(yaml_content)
-        except ValidationError as e:
-            logger.warning(f"Invalid custom config file: {e}")
+        except ValidationError as err:
+            logger.warning("Invalid custom config file: %s", err)
 
     return config
 
@@ -66,8 +66,8 @@ def get_prioritized_namespace_path() -> List[str]:
 
 def main():
     paths = get_prioritized_namespace_path()
-    for p in paths:
-        print(p)
+    for pathv in paths:
+        print(pathv)
 
 
 if __name__ == "__main__":

devchat/workflow/step.py

@@ -53,6 +53,7 @@ def _setup_env(
         """
         Setup the environment variables for the subprocess.
         """
+        _1 = wf_config
         command_raw = self.command_raw
 
         env = os.environ.copy()
@@ -98,7 +99,7 @@ def _validate_and_interpolate(
                     "The command uses $workflow_python, "
                     "but the workflow_python is not set yet."
                 )
-        
+
         args = []
         for p in parts:
             arg = p
@@ -116,7 +117,7 @@ def _validate_and_interpolate(
 
             if p.startswith(BuiltInVars.command_path):
                 # NOTE: 在文档中说明 command.yml 中表示路径采用 POSIX 标准
-                # 即，使用 / 分隔路径，而非 \ (Windows) 
+                # 即，使用 / 分隔路径，而非 \ (Windows)
                 path_parts = p.split("/")
                 # replace "$command_path" with the root path in path_parts
                 arg = os.path.join(wf_config.root_path, *path_parts[1:])
@@ -125,9 +126,9 @@ def _validate_and_interpolate(
                 arg = arg.replace(BuiltInVars.user_input, rt_param.user_input)
 
             args.append(arg)
-        
+
         return args
-            
+
 
     def run(
         self, wf_config: WorkflowConfig, rt_param: RuntimeParameter
@@ -175,4 +176,3 @@ def _pipe_reader(pipe, data, out_file):
             proc.wait()
             return_code = proc.returncode
             return return_code, stdout_data["data"], stderr_data["data"]
-

devchat/workflow/user_setting.py

@@ -12,13 +12,12 @@ def _load_user_settings() -> UserSettings:
     if not settings_path.exists():
         return UserSettings()
 
-    with open(settings_path, "r", encoding="utf-8") as f:
-        content = yaml.safe_load(f.read())
-    
+    with open(settings_path, "r", encoding="utf-8") as file:
+        content = yaml.safe_load(file.read())
+
     if content:
         return UserSettings.parse_obj(content)
 
     return UserSettings()
 
 USER_SETTINGS = _load_user_settings()
-