140 lines
4 KiB
Python
140 lines
4 KiB
Python
import time
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import requests
|
|
# from django.conf import settings
|
|
# from django.core.cache import cache
|
|
from django.db import models
|
|
from httpx import Request
|
|
|
|
from gotify_rest_api_client import AuthenticatedClient
|
|
from gotify_rest_api_client.models import *
|
|
from gotify_rest_api_client.api.message import create_message
|
|
from gotify_rest_api_client.api.user import create_user
|
|
from gotify_rest_api_client.api.client import create_client
|
|
from gotify_rest_api_client.api.application import create_app
|
|
|
|
ADMIN_CLIENT_TOKEN = 'CbWdsJrmRTDzbuD'
|
|
|
|
import logging
|
|
|
|
# logging.basicConfig(
|
|
# format="%(levelname)s [%(asctime)s] %(name)s - %(message)s",
|
|
# datefmt="%Y-%m-%d %H:%M:%S",
|
|
# level=logging.DEBUG
|
|
# )
|
|
logger = logging.getLogger('utils.gotify_clients')
|
|
|
|
|
|
def log_request(request: Request):
|
|
logger.debug(f'authorization: {request.headers["authorization"]}')
|
|
logger.debug(
|
|
f"Request event hook: {request.method} {request.url} {request.method} {request.headers} - Waiting for response")
|
|
|
|
|
|
def log_response(response):
|
|
request = response.request
|
|
logger.debug(f"Response event hook: {request.method} {request.url} - Status {response.status_code}")
|
|
# logger.debug(f"Response event hook: {response.read()}")
|
|
|
|
|
|
def get_client(access_token):
|
|
client = AuthenticatedClient(base_url="https://push.gooyal.com", token=access_token,
|
|
httpx_args={"event_hooks": {"request": [log_request], "response": [log_response]}},
|
|
)
|
|
return client
|
|
|
|
|
|
def get_basic_client(username, password):
|
|
auth = httpx.BasicAuth(username=username, password=password)
|
|
prefix, token = auth._auth_header.split()
|
|
|
|
# httpx_client = httpx.Client(auth=auth, base_url="https://push.gooyal.com")
|
|
client = AuthenticatedClient(base_url="https://push.gooyal.com", token=token, prefix=prefix,
|
|
httpx_args={
|
|
"event_hooks": {"request": [log_request], "response": [log_response]},
|
|
# "auth": auth,
|
|
# "base_url": "https://push.gooyal.com"
|
|
}
|
|
)
|
|
# client.set_httpx_client(httpx_client)
|
|
return client
|
|
|
|
|
|
def send_notif(data):
|
|
class Tmp():
|
|
def to_dict(self):
|
|
return data
|
|
|
|
final_data = Tmp()
|
|
client = get_client()
|
|
response: Any = create_message.sync(client=client, body=final_data)
|
|
|
|
return response
|
|
|
|
|
|
def submit_user(user_uuid):
|
|
data_dict = {
|
|
"admin": False,
|
|
"name": user_uuid,
|
|
"pass": user_uuid
|
|
}
|
|
|
|
class Tmp():
|
|
def to_dict(self):
|
|
return data_dict
|
|
|
|
final_data = Tmp()
|
|
client = get_client(ADMIN_CLIENT_TOKEN)
|
|
response: Any = create_user.sync(client=client, body=final_data)
|
|
|
|
return response
|
|
|
|
|
|
def submit_client(user_uuid):
|
|
data_dict = {
|
|
"name": user_uuid,
|
|
}
|
|
|
|
class Tmp():
|
|
def to_dict(self):
|
|
return data_dict
|
|
|
|
final_data = Tmp()
|
|
client = get_basic_client(user_uuid, user_uuid)
|
|
response: Any = create_client.sync(client=client, body=final_data)
|
|
|
|
return response
|
|
|
|
|
|
def submit_application(token, user_uuid):
|
|
data_dict = {
|
|
"defaultPriority": 5,
|
|
"description": None,
|
|
"name": user_uuid
|
|
}
|
|
|
|
class Tmp():
|
|
def to_dict(self):
|
|
return data_dict
|
|
|
|
final_data = Tmp()
|
|
client = get_client(token)
|
|
response: Any = create_app.sync(client=client, body=final_data)
|
|
|
|
return response
|
|
|
|
|
|
sample_data = {
|
|
"message": "**Backup** was successfully finished.",
|
|
"priority": 2,
|
|
"title": "Backup"
|
|
}
|
|
|
|
# ClientModel(id=20, name='820c1110-a3bd-47ce-a3f4-f8d2dd70fa9f', token='CbNecn03yWNIl93', last_used=None, additional_properties={})
|
|
|
|
# send_notif(sample_data)
|
|
# print(submit_user('820c1110-a3bd-47ce-a3f4-f8d2dd70fa9f'))
|
|
# print(submit_client('820c1110-a3bd-47ce-a3f4-f8d2dd70fa9f'))
|
|
print(submit_application('CbNecn03yWNIl93','820c1110-a3bd-47ce-a3f4-f8d2dd70fa9f'))
|