update es.py
This commit is contained in:
redtea committed
1 parent
1e844948c7
commit
23c2a9fa18
1 file changed
+443
@@ -0,0 +1,443 @@
|
||||
# pip install requests
|
||||
from typing import Dict
|
||||
|
||||
import requests
|
||||
import json
|
||||
from enum import Enum
|
||||
|
||||
|
||||
# class syntax
|
||||
class ConfidentialType(Enum):
|
||||
ALL = "ALL"
|
||||
FRIENDS = "FRIENDS"
|
||||
NOBODY = "NOBODY"
|
||||
|
||||
|
||||
class ESProjectInvite:
|
||||
def __init__(self, id: int, project: int, user: int):
|
||||
self.id = id
|
||||
self.project = project
|
||||
self.user = user
|
||||
|
||||
def __str__(self):
|
||||
return '{ ' + f'"id": {self.id},' + f'"project": \"{self.project}\",' + f'"user": {self.user}' + ' }'
|
||||
|
||||
|
||||
class ESProject:
|
||||
def __init__(self, id: int, name: str, owner_id: int, tags: list[str], members: tuple[int]):
|
||||
self.id = id
|
||||
self.name = name
|
||||
self.owner_id = owner_id
|
||||
self.tags = tags
|
||||
self.members = members
|
||||
|
||||
def __str__(self):
|
||||
t = ', '.join(['"' + s + '"' for s in self.tags])
|
||||
return '{ ' + f'"id": {self.id},' + f'"name": \"{self.name}\",' + f'"owner_id": {self.owner_id},' + f'"tags": [{t}]' + f', "members": {self.members}' + ' }'
|
||||
|
||||
|
||||
class ESAccount:
|
||||
def __init__(self, date_of_birth, description, follows: tuple, followers: tuple):
|
||||
self.date_of_birth = date_of_birth
|
||||
self.description = description
|
||||
self._follows = follows
|
||||
self._followers = followers
|
||||
|
||||
def follows(self) -> tuple:
|
||||
return self._follows
|
||||
|
||||
def followers(self) -> tuple:
|
||||
return self._followers
|
||||
|
||||
def __str__(self):
|
||||
return '{ ' + f'"date_of_birth": \"{self.date_of_birth}\",' + f'"description": \"{self.description}\",' + f'"follows": {self._follows},' + f'"followers": {self._followers}' + ' }'
|
||||
|
||||
|
||||
class ESUser:
|
||||
account: ESAccount
|
||||
confidencial: map
|
||||
|
||||
def __init__(self, id, username, first_name, last_name, picture_url, email, access, telegram_id, google_id,
|
||||
token="", account=None, confidencial=None):
|
||||
self.id = id
|
||||
self.username = username
|
||||
self.first_name = first_name
|
||||
self.last_name = last_name
|
||||
self.picture_url = picture_url
|
||||
self.email = email
|
||||
self.access = access
|
||||
self.telegram_id = telegram_id
|
||||
self.google_id = google_id
|
||||
if token != "":
|
||||
self.token = token
|
||||
if account != None:
|
||||
self.account: ESAccount = account
|
||||
if confidencial != None:
|
||||
self.confidencial: map = confidencial
|
||||
|
||||
def __str__(self):
|
||||
return '{ ' + f'"id": {self.id}, "username": "{self.username}", "first_name": "{self.first_name}", "last_name": "{self.last_name}", "picture_url": "{self.picture_url}", "email": "{self.email}", "access": "{self.access}", "telegram_id": "{self.telegram_id}", "google_id": "{self.google_id}"' + ' }'
|
||||
|
||||
|
||||
class ESCleint:
|
||||
token: str
|
||||
|
||||
def __init__(self, token):
|
||||
self.token = token
|
||||
|
||||
def get_self(self) -> ESUser:
|
||||
url = 'https://auth.easystartup.su/api/user/self'
|
||||
|
||||
response = requests.get(url, headers=headers(self.token))
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
data = response.json()
|
||||
return ESUser(data["id"], data["username"], data["first_name"], data["last_name"], data["picture_url"],
|
||||
data["email"], data["access"], data["telegram_id"], data["google_id"],
|
||||
token=self.token, account=self._get_self_account(), confidencial=self._get_self_confidential())
|
||||
|
||||
def follow(self, id: int):
|
||||
url = 'https://account.easystartup.su/api/following/follow/' + str(id)
|
||||
|
||||
response = requests.put(url, headers=headers(self.token))
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
def unfollow(self, id: int):
|
||||
url = 'https://account.easystartup.su/api/following/unfollow/' + str(id)
|
||||
|
||||
response = requests.put(url, headers=headers(self.token))
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
def set_follows(self, user: ESUser, follows: tuple):
|
||||
user.account._follows = follows
|
||||
self._edit_account(user)
|
||||
|
||||
def set_followers(self, user: ESUser, followers: tuple):
|
||||
user.account._followers = followers
|
||||
self._edit_account(user)
|
||||
|
||||
def _get_self_account(self) -> ESAccount:
|
||||
url = 'https://account.easystartup.su/api/account'
|
||||
|
||||
response = requests.get(url, headers=headers(self.token))
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
data = response.json()
|
||||
follows = tuple()
|
||||
followers = tuple()
|
||||
if data["follows"] != None:
|
||||
follows = tuple(data["follows"])
|
||||
if data["followers"] != None:
|
||||
followers = tuple(data["followers"])
|
||||
return ESAccount(data["date_of_birth"], data["description"], follows, followers)
|
||||
|
||||
def _edit_self_account(self, user: ESUser):
|
||||
url = 'https://account.easystartup.su/api/account'
|
||||
|
||||
data = {
|
||||
"date_of_birth": user.account.date_of_birth,
|
||||
"description": user.account.description,
|
||||
}
|
||||
response = requests.post(url, headers=headers(self.token), json=data)
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
def _edit_self_confidential(self, user: ESUser):
|
||||
url = 'https://account.easystartup.su/api/confidential'
|
||||
|
||||
data = user.confidencial
|
||||
for k in data:
|
||||
data[k] = data[k].value
|
||||
response = requests.post(url, headers=headers(self.token), json=data)
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
def _get_self_confidential(self) -> map:
|
||||
url = 'https://account.easystartup.su/api/confidential'
|
||||
|
||||
response = requests.get(url, headers=headers(self.token))
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
data = response.json()
|
||||
for k in data:
|
||||
data[k] = ConfidentialType(data[k])
|
||||
return data
|
||||
|
||||
def edit_self(self, user: ESUser):
|
||||
url = 'https://auth.easystartup.su/api/user/self'
|
||||
|
||||
data = {
|
||||
"email": user.email,
|
||||
"first_name": user.first_name,
|
||||
"last_name": user.last_name,
|
||||
"username": user.username
|
||||
}
|
||||
response = requests.post(url, headers=headers(self.token), json=data)
|
||||
|
||||
self._edit_self_account(user)
|
||||
self._edit_self_confidential(user)
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
def _get_account(self, id: int) -> ESAccount:
|
||||
url = 'https://account.easystartup.su/api/account/' + str(id)
|
||||
|
||||
response = requests.get(url, headers=headers(self.token))
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
data = response.json()
|
||||
follows = tuple()
|
||||
followers = tuple()
|
||||
if data["follows"] != None:
|
||||
follows = tuple(data["follows"])
|
||||
if data["followers"] != None:
|
||||
followers = tuple(data["followers"])
|
||||
return ESAccount(data["date_of_birth"], data["description"], follows, followers)
|
||||
|
||||
def _edit_account(self, user: ESUser):
|
||||
url = 'https://account.easystartup.su/api/confidential/' + str(user.id)
|
||||
|
||||
data = {
|
||||
"date_of_birth": user.account.date_of_birth,
|
||||
"description": user.account.description,
|
||||
}
|
||||
response = requests.post(url, headers=headers(self.token), json=data)
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
def _edit_confidential(self, user: ESUser):
|
||||
url = 'https://account.easystartup.su/api/confidential/' + str(user.id)
|
||||
|
||||
data = user.confidencial
|
||||
for k in data:
|
||||
data[k] = data.value
|
||||
response = requests.post(url, headers=headers(self.token), json=data)
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
def _get_confidential(self, id: int) -> map:
|
||||
url = 'https://account.easystartup.su/api/confidential/' + str(id)
|
||||
|
||||
response = requests.get(url, headers=headers(self.token))
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
data = response.json()
|
||||
for k in data:
|
||||
data[k] = ConfidentialType(data[k])
|
||||
return data
|
||||
|
||||
# open('myfile.png', 'rb')
|
||||
def picture(self, file):
|
||||
multipart_form_data = {
|
||||
'file': ('pic.png', file)
|
||||
}
|
||||
response = requests.post('https://account.easystartup.su/api/account/picture', files=multipart_form_data,
|
||||
headers=headers(self.token))
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
def create(self, user: ESUser) -> ESUser:
|
||||
url = 'https://auth.easystartup.su/api/user/'
|
||||
jsn = json.loads(str(user))
|
||||
response = requests.put(url, headers=headers(self.token), json=jsn)
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
data = response.json()
|
||||
return ESUser(data["id"], data["username"], data["first_name"], data["last_name"], data["picture_url"],
|
||||
data["email"], data["access"], data["telegram_id"], data["google_id"], token=self.token,
|
||||
account=self._get_account(data["id"]),
|
||||
confidencial=self._get_confidential(data["id"]))
|
||||
|
||||
def edit(self, user: ESUser):
|
||||
url = 'https://auth.easystartup.su/api/user/' + str(user.id)
|
||||
jsn = json.loads(str(user))
|
||||
response = requests.post(url, headers=headers(self.token), json=jsn)
|
||||
self._edit_account(user)
|
||||
self._edit_confidential(user)
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
def delete(self, user: ESUser):
|
||||
url = 'https://auth.easystartup.su/api/user/' + str(user.id)
|
||||
response = requests.delete(url, headers=headers(self.token))
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
url = 'https://auth.easystartup.su/api/account/' + str(user.id)
|
||||
response = requests.delete(url, headers=headers(self.token))
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
def users(self, start: int = 0, limit: int = -1) -> list[int]:
|
||||
url = f'https://auth.easystartup.su/api/user/all?start={start}&limit={limit}'
|
||||
|
||||
response = requests.get(url, headers=headers(self.token))
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
return response.json()
|
||||
|
||||
def public(self, id) -> list:
|
||||
url = f'https://account.easystartup.su/api/public/' + str(id)
|
||||
|
||||
response = requests.get(url)
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
return response.json()
|
||||
|
||||
def users_count(self) -> list:
|
||||
url = f'https://auth.easystartup.su/api/user/count'
|
||||
|
||||
response = requests.get(url, headers=headers(self.token))
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
return response.json()["count"]
|
||||
|
||||
def get(self, id: int) -> ESUser:
|
||||
url = 'https://auth.easystartup.su/api/user/' + str(id)
|
||||
|
||||
response = requests.get(url, headers=headers(self.token))
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
data = response.json()
|
||||
return ESUser(data["id"], data["username"], data["first_name"], data["last_name"], data["picture_url"],
|
||||
data["email"], data["access"], data["telegram_id"], data["google_id"], token=self.token,
|
||||
account=self._get_account(data["id"]), confidencial=self._get_confidential(data["id"]))
|
||||
|
||||
|
||||
def create_project(self, name: str, tags: list[str], owner_id: int) -> ESProject:
|
||||
url = 'https://project.easystartup.su/api/project/'
|
||||
tags_json = ', '.join(['"' + s + '"' for s in tags])
|
||||
jsn = json.loads('{' + f"""
|
||||
"name": "{name}",
|
||||
"tags": [{tags_json}],
|
||||
"owner_id": {owner_id}
|
||||
""" + '}')
|
||||
response = requests.put(url, headers=headers(self.token), json=jsn)
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
data = response.json()
|
||||
return ESProject(data["id"], data["name"], data["owner_id"], data["tags"], data["members"])
|
||||
|
||||
def edit_project(self, project: ESProject):
|
||||
url = 'https://project.easystartup.su/api/project/' + str(project.id)
|
||||
jsn = json.loads(str(project))
|
||||
response = requests.post(url, headers=headers(self.token), json=jsn)
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
|
||||
def projects_count(self) -> list:
|
||||
url = f'https://project.easystartup.su/api/project/count'
|
||||
|
||||
response = requests.get(url, headers=headers(self.token))
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
return response.json()["count"]
|
||||
|
||||
def projects(self, start: int = 0, limit: int = -1) -> list[int]:
|
||||
url = f'https://project.easystartup.su/api/project/all?start={start}&limit={limit}'
|
||||
|
||||
response = requests.get(url, headers=headers(self.token))
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
return response.json()
|
||||
|
||||
def my_projects(self, start: int = 0, limit: int = -1) -> list[int]:
|
||||
url = f'https://project.easystartup.su/api/project/my?start={start}&limit={limit}'
|
||||
|
||||
response = requests.get(url, headers=headers(self.token))
|
||||
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
return response.json()
|
||||
|
||||
def get_project(self, id: int) -> ESProject:
|
||||
url = 'https://project.easystartup.su/api/project/' + str(id)
|
||||
response = requests.get(url, headers=headers(self.token))
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
data = response.json()
|
||||
return ESProject(data["id"], data["name"], data["owner_id"], data["tags"], data["members"])
|
||||
|
||||
def delete_project(self, id: int):
|
||||
url = 'https://project.easystartup.su/api/project/' + str(id)
|
||||
response = requests.delete(url, headers=headers(self.token))
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
def invite_to_project(self, project: int, user: int):
|
||||
url = 'https://project.easystartup.su/api/member/invite'
|
||||
jsn = json.loads('{' + f"""
|
||||
"project": {project},
|
||||
"user": {user}
|
||||
""" + '}')
|
||||
response = requests.post(url, headers=headers(self.token), json=jsn)
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
|
||||
def accept_invite_to_project(self, invite: int):
|
||||
url = 'https://project.easystartup.su/api/member/invite/accept/' + str(invite)
|
||||
response = requests.post(url, headers=headers(self.token))
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
def my_invites_to_project(self):
|
||||
url = 'https://project.easystartup.su/api/member/invite/my/'
|
||||
response = requests.get(url, headers=headers(self.token))
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
return response.json()
|
||||
def kick_from_project(self, project: int, user: int):
|
||||
url = 'https://project.easystartup.su/api/member/kick'
|
||||
jsn = json.loads('{' + f"""
|
||||
"project": "{project}",
|
||||
"user": "{user}"
|
||||
""" + '}')
|
||||
response = requests.post(url, headers=headers(self.token))
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
def get_invite(self, id: int) -> ESProjectInvite:
|
||||
url = 'https://project.easystartup.su/api/member/invite/' + str(id)
|
||||
response = requests.get(url, headers=headers(self.token))
|
||||
if response.status_code != 200:
|
||||
raise Exception("Bad response code: " + str(response.status_code))
|
||||
data = response.json()
|
||||
return ESProjectInvite(data["id"], data["project"], data["user"])
|
||||
|
||||
|
||||
def headers(token: str) -> dict[str, str]:
|
||||
return {
|
||||
'Authorization': 'Bearer ' + token,
|
||||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36 Edg/129.0.0.0',
|
||||
'Accept': 'application/json'
|
||||
}
|
||||
Reference in new issue
Block a user