forked from agent0ai/agent-zero
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathprojects.py
More file actions
148 lines (130 loc) · 5.51 KB
/
Copy pathprojects.py
File metadata and controls
148 lines (130 loc) · 5.51 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
from python.helpers.api import ApiHandler, Input, Output, Request, Response
from python.helpers import projects
from python.helpers.notification import NotificationManager, NotificationType, NotificationPriority
class Projects(ApiHandler):
async def process(self, input: Input, request: Request) -> Output:
action = input.get("action", "")
ctxid = input.get("context_id", None)
if ctxid:
_context = self.use_context(ctxid)
try:
if action == "list":
data = self.get_active_projects_list()
elif action == "list_options":
data = self.get_active_projects_options()
elif action == "load":
data = self.load_project(input.get("name", None))
elif action == "create":
data = self.create_project(input.get("project", None))
elif action == "clone":
data = self.clone_project(input.get("project", None))
elif action == "update":
data = self.update_project(input.get("project", None))
elif action == "delete":
data = self.delete_project(input.get("name", None))
elif action == "activate":
data = self.activate_project(ctxid, input.get("name", None))
elif action == "deactivate":
data = self.deactivate_project(ctxid)
elif action == "file_structure":
data = self.get_file_structure(input.get("name", None), input.get("settings"))
else:
raise Exception("Invalid action")
return {
"ok": True,
"data": data,
}
except Exception as e:
return {
"ok": False,
"error": str(e),
}
def get_active_projects_list(self):
return projects.get_active_projects_list()
def get_active_projects_options(self):
items = projects.get_active_projects_list() or []
return [
{"key": p.get("name", ""), "label": p.get("title", "") or p.get("name", "")}
for p in items
if p.get("name")
]
def create_project(self, project: dict|None):
if project is None:
raise Exception("Project data is required")
data = projects.BasicProjectData(**project)
name = projects.create_project(project["name"], data)
return projects.load_edit_project_data(name)
def clone_project(self, project: dict|None):
if project is None:
raise Exception("Project data is required")
git_url = project.get("git_url", "")
git_token = project.get("git_token", "")
if not git_url:
raise Exception("Git URL is required")
# Progress notification
notification = NotificationManager.send_notification(
NotificationType.PROGRESS,
NotificationPriority.NORMAL,
f"Cloning repository...",
"Git Clone",
display_time=999,
group="git_clone"
)
try:
data = projects.BasicProjectData(**project)
name = projects.clone_git_project(project["name"], git_url, git_token, data)
# Success notification
NotificationManager.send_notification(
NotificationType.SUCCESS,
NotificationPriority.NORMAL,
f"Repository cloned successfully",
"Git Clone",
display_time=3,
group="git_clone"
)
return projects.load_edit_project_data(name)
except Exception as e:
# Error notification
NotificationManager.send_notification(
NotificationType.ERROR,
NotificationPriority.HIGH,
f"Clone failed: {str(e)}",
"Git Clone",
display_time=5,
group="git_clone"
)
raise
def load_project(self, name: str|None):
if name is None:
raise Exception("Project name is required")
return projects.load_edit_project_data(name)
def update_project(self, project: dict|None):
if project is None:
raise Exception("Project data is required")
data = projects.EditProjectData(**project)
name = projects.update_project(project["name"], data)
return projects.load_edit_project_data(name)
def delete_project(self, name: str|None):
if name is None:
raise Exception("Project name is required")
return projects.delete_project(name)
def activate_project(self, context_id: str|None, name: str|None):
if not context_id:
raise Exception("Context ID is required")
if not name:
raise Exception("Project name is required")
return projects.activate_project(context_id, name)
def deactivate_project(self, context_id: str|None):
if not context_id:
raise Exception("Context ID is required")
return projects.deactivate_project(context_id)
def get_file_structure(self, name: str|None, settings: dict|None):
if not name:
raise Exception("Project name is required")
# project data
basic_data = projects.load_basic_project_data(name)
# override file structure settings
if settings:
basic_data["file_structure"] = settings # type: ignore
# get structure
return projects.get_file_structure(name, basic_data)