@samitouri / QOSami_Agents_Zero-1
Code Commits Issues Pulls Sessions CI/CD
main
main
py 166 lines 6.21 KB
Raw
1 from __future__ import annotations
2
3 from helpers.api import ApiHandler, Request, Response
4
5 from plugins._commands.helpers import commands as commands_helper
6
7
8 class Commands(ApiHandler):
9 async def process(self, input: dict, request: Request) -> dict | Response:
10 action = str(input.get("action", "") or "").strip()
11
12 if action == "list_effective":
13 return self._list_effective(input)
14 if action == "list_scope":
15 return self._list_scope(input)
16 if action == "get":
17 return self._get(input)
18 if action == "save":
19 return self._save(input)
20 if action == "delete":
21 return self._delete(input)
22 if action == "duplicate":
23 return self._duplicate(input)
24 if action == "scope_info":
25 return self._scope_info(input)
26 if action == "resolve":
27 return await self._resolve(input)
28
29 return Response(status=400, response=f"Unknown action: {action}")
30
31 def _list_effective(self, input: dict) -> dict | Response:
32 context_scope = commands_helper.get_context_scope(str(input.get("context_id", "") or ""))
33 commands, scope = commands_helper.list_effective_commands(
34 project_name=context_scope["project_name"],
35 )
36 commands = [
37 command
38 for command in commands
39 if not command.get("frontmatter_extra", {}).get("webui_hidden")
40 ]
41 return {
42 "ok": True,
43 "commands": commands,
44 "scope": scope,
45 }
46
47 def _list_scope(self, input: dict) -> dict | Response:
48 commands, scope = commands_helper.list_scope_commands(
49 project_name=str(input.get("project_name", "") or ""),
50 )
51 return {
52 "ok": True,
53 "commands": commands,
54 "builtin_commands": commands_helper.list_builtin_commands(),
55 "scope": scope,
56 }
57
58 def _get(self, input: dict) -> dict | Response:
59 path = str(input.get("path", "") or "")
60 if not path:
61 return Response(status=400, response="Missing path")
62
63 try:
64 command = commands_helper.get_command(
65 path,
66 project_name=str(input.get("project_name", "") or ""),
67 )
68 except FileNotFoundError:
69 return Response(status=404, response="Command not found")
70 except ValueError as error:
71 return Response(status=400, response=str(error))
72
73 return {"ok": True, "command": command}
74
75 def _save(self, input: dict) -> dict | Response:
76 try:
77 command = commands_helper.save_command(
78 project_name=str(input.get("project_name", "") or ""),
79 existing_path=str(input.get("existing_path", "") or ""),
80 name=str(input.get("name", "") or ""),
81 description=str(input.get("description", "") or ""),
82 argument_hint=str(input.get("argument_hint", "") or ""),
83 command_type=str(input.get("command_type", "text") or "text"),
84 body=str(input.get("body", "") or ""),
85 include_history=bool(input.get("include_history", False)),
86 extra_frontmatter=input.get("extra_frontmatter", {}) or {},
87 )
88 except FileExistsError as error:
89 return Response(status=409, response=str(error))
90 except ValueError as error:
91 return Response(status=400, response=str(error))
92
93 return {"ok": True, "command": command}
94
95 def _delete(self, input: dict) -> dict | Response:
96 path = str(input.get("path", "") or "")
97 if not path:
98 return Response(status=400, response="Missing path")
99
100 try:
101 commands_helper.delete_command(
102 path,
103 project_name=str(input.get("project_name", "") or ""),
104 )
105 except FileNotFoundError:
106 return Response(status=404, response="Command not found")
107 except ValueError as error:
108 return Response(status=400, response=str(error))
109
110 return {"ok": True}
111
112 def _duplicate(self, input: dict) -> dict | Response:
113 path = str(input.get("path", "") or "")
114 if not path:
115 return Response(status=400, response="Missing path")
116
117 try:
118 command = commands_helper.duplicate_command(
119 path,
120 project_name=str(input.get("project_name", "") or ""),
121 )
122 except FileNotFoundError:
123 return Response(status=404, response="Command not found")
124 except ValueError as error:
125 return Response(status=400, response=str(error))
126
127 return {"ok": True, "command": command}
128
129 def _scope_info(self, input: dict) -> dict | Response:
130 explicit_project = str(input.get("project_name", "") or "")
131 context_scope = commands_helper.get_context_scope(str(input.get("context_id", "") or ""))
132
133 project_name = explicit_project if "project_name" in input else context_scope["project_name"]
134
135 scope = commands_helper.get_scope_payload(
136 project_name=project_name,
137 ensure_directory=bool(input.get("ensure_directory", False)),
138 )
139 return {
140 "ok": True,
141 "scope": commands_helper.strip_private_scope(scope),
142 "context_scope": context_scope,
143 }
144
145 async def _resolve(self, input: dict) -> dict | Response:
146 path = str(input.get("path", "") or "")
147 if not path:
148 return Response(status=400, response="Missing path")
149
150 slash_text = str(input.get("slash_text", "") or "")
151 if not slash_text:
152 return Response(status=400, response="Missing slash_text")
153
154 try:
155 resolution = await commands_helper.resolve_command_invocation(
156 path=path,
157 slash_text=slash_text,
158 project_name=str(input.get("project_name", "") or ""),
159 context_id=str(input.get("context_id", "") or ""),
160 )
161 except FileNotFoundError:
162 return Response(status=404, response="Command not found")
163 except ValueError as error:
164 return Response(status=400, response=str(error))
165
166 return {"ok": True, "resolution": resolution}