From 23c2a9fa183b941287678eefa7b2ea3bf04c78ad Mon Sep 17 00:00:00 2001 From: Denis Istratov Date: Sun, 20 Oct 2024 12:00:00 +0300 Subject: [PATCH] update es.py --- es.py | 443 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 443 insertions(+) create mode 100644 es.py diff --git a/es.py b/es.py new file mode 100644 index 0000000..133272c --- /dev/null +++ b/es.py @@ -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' + }