-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtui.py
170 lines (145 loc) · 6.59 KB
/
tui.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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
from hubspot_client import HubSpotClient
import os
from dotenv import dotenv_values
import subprocess
import threading
from concurrent.futures import ThreadPoolExecutor
from tqdm import tqdm
import time
class TUI:
def __init__(self, hs_client):
self.hs_client = hs_client
command = 'gum style --border double --margin "1" --padding "1 2" --border-foreground "#FF7A59" "👋 Welcome to $(gum style --foreground "#FF7A59" \'HubSpot TUI\')"'
process = self.run_command(command)
print(process.stdout)
self.contacts, self.companies = self.get_assets()
@staticmethod
def parse_output(gum_output):
return gum_output.stdout.strip().split("\n")
@staticmethod
def run_command(command):
process = subprocess.run(command, shell=True, stdout=subprocess.PIPE, text=True)
return process
def prompt(self, input_placehoder):
command = f'gum input --placeholder "{input_placehoder}"'
value = self.parse_output(self.run_command(command))
return value[0]
def create_contact(self):
email = self.prompt("email")
firstname = self.prompt("firstname")
lastname = self.prompt("lastname")
return self.hs_client.create_contact(email, firstname, lastname)
def create_company(self):
name = self.prompt("name")
domain = self.prompt("domain")
return self.hs_client.create_company(name, domain)
def create_deal(self):
command = 'gum confirm "Do you want to create a company?"'
process = self.run_command(command)
if process.returncode == 0:
company_id = self.create_company()
else:
company_id = self.get_company()
command = 'gum confirm "Do you want to create a contact?"'
process = self.run_command(command)
if process.returncode == 0:
contact_id = self.create_contact()
else:
contact_id = self.get_contact()
deal_name = self.prompt("dealname")
deal_stage = self.get_stage()
pipeline_id = self.hs_client.get_pipeline().get("pipeline_id")
deal_id = self.hs_client.create_deal(deal_name, deal_stage, pipeline_id, company_id, contact_id)
print(f"🖇️ https://app-eu1.hubspot.com/contacts/{self.hs_client.account_id}/record/0-3/{deal_id}")
def get_assets(self):
command = 'gum spin --spinner meter --title "Loading assets (companies, contacts)..." -- sleep 5'
command_thread = threading.Thread(target=self.run_command, args=(command,))
command_thread.start()
with ThreadPoolExecutor(max_workers=2) as executor:
future_contact = executor.submit(self.hs_client.get_contact)
future_company = executor.submit(self.hs_client.get_company)
# Retrieve the results once they're available
contacts = future_contact.result()
companies = future_company.result()
command_thread.join()
return contacts, companies
def get_company(self):
display_companies = []
for c in self.companies:
try:
item = {
"name": c.get("properties").get("name").replace("'", ""),
"domain": c.get("properties").get("domain"),
"id": c.get("id")
}
display_companies.append(
f'{item.get("name")} - {item.get("domain")} - {item.get("id")}'
)
except:
pass
items_string = "\\n".join(display_companies)
command = f"echo '{items_string}' | gum filter"
process = self.run_command(command)
company_id = process.stdout.split("-")[-1].strip()
print(f"🖇️ https://app-eu1.hubspot.com/contacts/{self.hs_client.account_id}/record/0-2/{company_id}")
return company_id
def get_contact(self):
display_contacts = []
for c in self.contacts:
try:
item = {
"firstname": c.get("properties").get("firstname").replace("'", ""),
"lastname": c.get("properties").get("lastname").replace("'", ""),
"email": c.get("properties").get("email"),
"id": c.get("id")
}
display_contacts.append(
f'{item.get("firstname")} - {item.get("lastname")} - {item.get("email")} - {item.get("id")}'
)
except:
pass
items_string = "\\n".join(display_contacts)
command = f"echo '{items_string}' | gum filter"
process = self.run_command(command)
contact_id = process.stdout.split("-")[-1].strip()
print(f"🖇️ https://app-eu1.hubspot.com/contacts/{self.hs_client.account_id}/record/0-1/{contact_id}")
return contact_id
def get_stage(self):
stages = [ f'{s.get("stage_id")} - {s.get("label")}' for s in self.hs_client.get_pipeline().get("stages")]
items_string = "\\n".join(stages)
command = f"echo '{items_string}' | gum filter"
process = self.run_command(command)
return process.stdout.split("-")[0].strip()
def launch_menu(self):
command = 'gum choose --cursor="> " "0. Create Deal 🤝" "1. Create Company 🏢" "2. Create Contact 📞" "3. Get Company" "4. Get Contact" "5. Refresh data 🆕"'
process = self.run_command(command)
id = process.stdout.split(".")[0].strip()
if id == "0":
self.create_deal()
if id == "1":
self.create_company()
if id == "2":
self.create_contact()
if id == "3":
self.get_company()
if id == "4":
self.get_contact()
if id == "5":
self.contacts, self.companies = self.get_assets()
command = 'gum confirm "Do you want to stay in the TUI?"'
process = self.run_command(command)
if process.returncode == 0:
self.launch_menu()
else:
exit(0)
if __name__ == "__main__":
os.environ['GUM_INPUT_PROMPT_FOREGROUND'] = "#FF7A59"
os.environ['GUM_CHOOSE_PROMPT_FOREGROUND'] = "#FF7A59"
os.environ['GUM_CHOOSE_CURSOR_FOREGROUND'] = "#FF7A59"
os.environ['GUM_FILTER_PROMPT_FOREGROUND'] = "#FF7A59"
os.environ['GUM_FILTER_MATCH_FOREGROUND'] = "#FF7A59"
os.environ['GUM_SPIN_SPINNER_FOREGROUND'] = "#FF7A59"
env_variables = dotenv_values('.env')
hs_client = HubSpotClient(env_variables.get('HS_API_TOKEN'), env_variables.get("HS_ACCOUNT_ID"))
tui = TUI(hs_client)
tui.launch_menu()