forked from GoogleCloudPlatform/dfcx-scrapi
-
Notifications
You must be signed in to change notification settings - Fork 0
/
project.py
128 lines (103 loc) · 3.95 KB
/
project.py
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
"""Project level cross-region functions for CX."""
# Copyright 2023 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
import time
from typing import Dict
from dfcx_scrapi.core import scrapi_base
from dfcx_scrapi.core.agents import Agents
# logging config
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)-8s %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
class Project(scrapi_base.ScrapiBase):
"""Top Level class representing the Project level resources
when working on a Dialogflow CX project. This Class will allow you to
extract information about your GCP project as a whole in relation to
your CX agents.
"""
def __init__(
self,
creds_path: str = None,
creds_dict: Dict = None,
creds=None,
scope=False,
project_id: str = None,
):
super().__init__(
creds_path=creds_path,
creds_dict=creds_dict,
creds=creds,
scope=scope,
)
if project_id:
self.project_id = project_id
self.agents = Agents(creds=self.creds)
def list_agents(self, project_id=None):
"""Get list of all agents from all regions for a project.
Args:
project_id: (Optional) The GCP project id to list all agents
from. If no project_id is provided the library will look at
self.project_id or error.
Returns:
A List of Dicts that has all agent metadata info.
"""
region_list = [
"global",
"us-central1",
"us-east1",
"us-west1",
"asia-northeast1",
"australia-southeast1",
"northamerica-northeast1",
"europe-west1",
"europe-west2",
]
if not project_id:
project_id = self.project_id
all_agents = []
for region in region_list:
location_path = f"projects/{project_id}/locations/{region}"
all_agents += self.agents.list_agents(location_path)
return all_agents
def backup_all_agents(self, gcs_bucket: str, project_id: str = None):
"""Export all Agent files to GCS bucket.
Args:
gcs_bucket: The GCS bucket to backup all files to
Returns:
List of all LROs to reference backend job status
"""
if not project_id:
project_id = self.project_id
logging.info("==== Fetching all Agent IDs =====")
all_agents = self.list_agents(project_id=project_id)
logging.info("Received %s Agent IDs", len(all_agents))
logging.info("==== Starting Agent Backups ====")
lro_list = []
for agent in all_agents:
logging.info("Backing up Agent: %s", agent.display_name)
temp_display_name = agent.display_name
temp_display_name = temp_display_name.strip()
temp_display_name = temp_display_name.lower()
temp_display_name = temp_display_name.replace(" ", "_")
temp_display_name = temp_display_name.replace("/", "_")
temp_display_name = temp_display_name.replace(":", "_")
temp_display_name = temp_display_name.replace("-", "_")
temp_gcs_uri = f"gs://{gcs_bucket}/{temp_display_name}"
lro_list.append(self.agents.export_agent(agent.name, temp_gcs_uri))
time.sleep(1)
logging.info("==== Agent Backup Job Complete ====")
return lro_list