generated from justuser-31/mrl_v1_license
131 lines
4.5 KiB
Python
131 lines
4.5 KiB
Python
import requests
|
|
|
|
from config import load_config
|
|
|
|
|
|
def create_account(
|
|
api_url: str, api_token: str, username: str, password: str, drive_quota: int
|
|
) -> dict:
|
|
"""
|
|
Call the create account API endpoint.
|
|
|
|
Args:
|
|
api_url: Base URL of the API server (e.g., 'http://localhost:8000')
|
|
api_token: Security token for API authentication
|
|
username: Username for the new account
|
|
password: Password for the new account
|
|
drive_quota: Storage quota in MB
|
|
|
|
Returns:
|
|
Dictionary with 'status_code' and 'message' keys
|
|
"""
|
|
url = f"{api_url.rstrip('/')}/create_account"
|
|
data = {
|
|
"token": api_token,
|
|
"username": username,
|
|
"password": password,
|
|
"drive_quota": str(drive_quota),
|
|
}
|
|
|
|
try:
|
|
response = requests.post(url, data=data)
|
|
return {"status_code": response.status_code, "message": response.text}
|
|
except requests.RequestException as e:
|
|
return {"status_code": -1, "message": f"Request failed: {str(e)}"}
|
|
|
|
|
|
def set_quota(api_url: str, api_token: str, username: str, drive_quota: int) -> dict:
|
|
"""
|
|
Call the set quota API endpoint.
|
|
|
|
Args:
|
|
api_url: Base URL of the API server (e.g., 'http://localhost:8000')
|
|
api_token: Security token for API authentication
|
|
username: Username whose quota should be updated
|
|
drive_quota: New storage quota in MB
|
|
|
|
Returns:
|
|
Dictionary with 'status_code' and 'message' keys
|
|
"""
|
|
url = f"{api_url.rstrip('/')}/set_quota"
|
|
data = {"token": api_token, "username": username, "drive_quota": str(drive_quota)}
|
|
|
|
try:
|
|
response = requests.post(url, data=data)
|
|
return {"status_code": response.status_code, "message": response.text}
|
|
except requests.RequestException as e:
|
|
return {"status_code": -1, "message": f"Request failed: {str(e)}"}
|
|
|
|
|
|
def delete_account(api_url: str, api_token: str, username: str) -> dict:
|
|
"""
|
|
Call the delete account API endpoint.
|
|
|
|
Args:
|
|
api_url: Base URL of the API server (e.g., 'http://localhost:8000')
|
|
api_token: Security token for API authentication
|
|
username: Username of the account to delete
|
|
|
|
Returns:
|
|
Dictionary with 'status_code' and 'message' keys
|
|
"""
|
|
url = f"{api_url.rstrip('/')}/delete_account"
|
|
data = {"token": api_token, "username": username}
|
|
|
|
try:
|
|
response = requests.post(url, data=data)
|
|
return {"status_code": response.status_code, "message": response.text}
|
|
except requests.RequestException as e:
|
|
return {"status_code": -1, "message": f"Request failed: {str(e)}"}
|
|
|
|
|
|
def get_all_users(api_url: str, api_token: str) -> dict:
|
|
"""
|
|
Call the get all users API endpoint.
|
|
|
|
Args:
|
|
api_url: Base URL of the API server (e.g., 'http://localhost:8000')
|
|
api_token: Security token for API authentication
|
|
|
|
Returns:
|
|
Dictionary with 'status_code' and 'data' keys.
|
|
If successful, 'data' will contain a list of users.
|
|
If failed, 'data' will contain an error message.
|
|
"""
|
|
url = f"{api_url.rstrip('/')}/users"
|
|
|
|
try:
|
|
response = requests.get(url, params={"token": api_token})
|
|
if response.status_code == 200:
|
|
return {"status_code": response.status_code, "data": response.json()}
|
|
else:
|
|
return {"status_code": response.status_code, "data": response.text}
|
|
except requests.RequestException as e:
|
|
return {"status_code": -1, "data": f"Request failed: {str(e)}"}
|
|
except ValueError as e:
|
|
# Handle case where response isn't valid JSON
|
|
return {"status_code": -1, "data": f"Failed to parse response: {str(e)}"}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
CONFIG = load_config()
|
|
# API_URL = "https://sfs.del.pw"
|
|
API_URL = "http://127.0.0.1:8000"
|
|
|
|
# print(set_quota(API_URL, CONFIG["security"]["api_token"], "sans", 10000))
|
|
# print(delete_account(API_URL, CONFIG["security"]["api_token"], "test"))
|
|
# print(create_account(API_URL, CONFIG["security"]["api_token"] , "test", "test", 999))
|
|
# print(get_all_users(API_URL, CONFIG["security"]["api_token"]))
|
|
|
|
users = get_all_users(API_URL, CONFIG["security"]["api_token"])["data"]
|
|
print(users)
|
|
print("-------------------------")
|
|
for user in users:
|
|
quota = user["quota_mb"]
|
|
quota += 100 # +100 MB to each user
|
|
print(
|
|
set_quota(API_URL, CONFIG["security"]["api_token"], user["username"], quota)
|
|
)
|
|
print("-------------------------")
|
|
print(get_all_users(API_URL, CONFIG["security"]["api_token"]))
|