forked from machaao/mistral-7b-chatbot
-
Notifications
You must be signed in to change notification settings - Fork 0
/
machaao_utils.py
105 lines (78 loc) · 2.76 KB
/
machaao_utils.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
import json
from base64 import b64decode
from datetime import datetime
import requests
from requests.structures import CaseInsensitiveDict
c = 'UTF-8'
def get_recent(
base_url, api_token,
server_session_create_time,
user_id, current_session=True):
limit = 5
e = "L3YxL2NvbnZlcnNhdGlvbnMvaGlzdG9yeS8="
check = b64decode(e).decode(c)
url = f"{base_url}{check}{user_id}/{limit}"
headers = CaseInsensitiveDict()
headers["api_token"] = api_token
headers["Content-Type"] = "application/json"
resp = requests.get(url, headers=headers, timeout=10)
if resp.status_code == 200:
messages = resp.json()
if current_session:
filtered_messages = list()
for message in messages:
create_time_stamp = message.get("_created_at")
create_time = datetime.strptime(create_time_stamp, "%Y-%m-%dT%H:%M:%S.%fZ")
if create_time > server_session_create_time:
filtered_messages.append(message)
while len(filtered_messages) > 0 and (filtered_messages[0].get("type") == "outgoing"):
_ = filtered_messages.pop(0)
return filtered_messages
else:
return messages
def send(url, headers, payload=None):
if payload:
print("sending post to platform: " + str(payload))
response = requests.request("POST", url, data=json.dumps(payload), headers=headers)
print("response from the platform: " + str(response.text))
else:
response = requests.request("GET", url, headers=headers)
return response
def get_details(api_token, base_url):
_cache_ts_param = str(datetime.now().timestamp())
e = "L3YxL2JvdHMvY29uZmlnP3Y9"
check = b64decode(e).decode(c)
url = f"{base_url}{check}{_cache_ts_param}"
headers = {
"api_token": api_token,
"Content-Type": "application/json",
}
response = send(url, headers)
if response and response.status_code == 200:
return response.json()
else:
return {}
def check_balance(
base_url, api_token,
user_id, DEFAULT_HTTP_TIMEOUT=10):
balance = 0
e = 'L3YxL2NvaW5zL2JhbGFuY2UvY2hlY2s='
check = b64decode(e).decode(c)
url = f"{base_url}{check}"
headers = CaseInsensitiveDict()
headers["api_token"] = api_token
headers["Content-Type"] = "application/json"
data = {
"userId": user_id,
"coins": 1
}
resp = requests.post(
url, data=json.dumps(data),
headers=headers, timeout=DEFAULT_HTTP_TIMEOUT
)
if resp.status_code == 200:
out = resp.json()
if out and out["balance"]:
balance = out["balance"]
print(f"balance: {balance}, user_id: {user_id}")
return balance