172 lines
5 KiB
Python
172 lines
5 KiB
Python
import json
|
|
import math
|
|
import time
|
|
|
|
from django.core.handlers.wsgi import WSGIRequest
|
|
from django.db.models import Q
|
|
from django.http import HttpResponse
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
|
|
from tmessage.models import tMMessage, tMUser
|
|
|
|
from .helper import get_user_object, get_username
|
|
|
|
|
|
def _json_response(data: dict | list, /, *, status: int=200, content_type: str="application/json") -> HttpResponse:
|
|
return HttpResponse(
|
|
json.dumps(data),
|
|
status=status,
|
|
content_type=content_type
|
|
)
|
|
|
|
def _add_qF(original: None | Q, new: Q) -> Q:
|
|
if original is None:
|
|
return new
|
|
return original | new
|
|
|
|
RESPONSE_400 = _json_response({ "success": False }, status=400)
|
|
RESPONSE_401 = _json_response({ "success": False }, status=401)
|
|
RESPONSE_403 = _json_response({ "success": False }, status=403)
|
|
|
|
@csrf_exempt
|
|
def api_messages(request: WSGIRequest) -> HttpResponse:
|
|
username = get_username(request)
|
|
|
|
if username is None:
|
|
return RESPONSE_401
|
|
|
|
user = get_user_object(username, i_promise_this_user_exists=True)
|
|
|
|
if request.method == "POST":
|
|
body = json.loads(request.body)
|
|
reply = body["content"].strip()
|
|
message_id = body["id"]
|
|
print(reply)
|
|
|
|
if not (isinstance(message_id, int) and isinstance(reply, str) and reply):
|
|
print("1")
|
|
return RESPONSE_400
|
|
|
|
try:
|
|
message = tMMessage.objects.get(message_id=message_id)
|
|
except tMMessage.DoesNotExist:
|
|
print("2")
|
|
return RESPONSE_400
|
|
|
|
if username != message.u_to.username:
|
|
return RESPONSE_403
|
|
|
|
if message.response is not None:
|
|
print("3")
|
|
return RESPONSE_400
|
|
|
|
message.response = reply
|
|
message.response_timestamp = math.floor(time.time())
|
|
message.save()
|
|
|
|
return _json_response({
|
|
"success": True,
|
|
"content": reply,
|
|
"timestamp": message.response_timestamp
|
|
})
|
|
|
|
elif request.method == "DELETE":
|
|
body = json.loads(request.body)
|
|
message_id = body["id"]
|
|
|
|
if not isinstance(message_id, int):
|
|
return RESPONSE_400
|
|
|
|
try:
|
|
message = tMMessage.objects.get(message_id=message_id)
|
|
except tMMessage.DoesNotExist:
|
|
return RESPONSE_400
|
|
|
|
if username != message.u_to.username:
|
|
return RESPONSE_403
|
|
|
|
message.delete()
|
|
|
|
return _json_response({
|
|
"success": True
|
|
})
|
|
|
|
queryFilter = None
|
|
|
|
if "offset" in request.GET and (request.GET.get("offset") or "").isdigit():
|
|
queryFilter = _add_qF(queryFilter, Q(message_id__lt=int(request.GET.get("offset") or "")))
|
|
|
|
if "unread" in request.GET:
|
|
queryFilter = _add_qF(queryFilter, Q(response=None))
|
|
|
|
if queryFilter:
|
|
msgObjects = user.received.filter(queryFilter) # type: ignore
|
|
else:
|
|
msgObjects = user.received.all() # type: ignore
|
|
|
|
output = []
|
|
messages = msgObjects.order_by("-message_id")[:50].values_list(
|
|
"message_id", "content", "response", "anonymous", "u_from", "u_to", "sent_timestamp", "response_timestamp"
|
|
)
|
|
|
|
for message in messages:
|
|
output.append({
|
|
"id": message[0],
|
|
"content": message[1],
|
|
"response": message[2],
|
|
"from": message[4] if not message[3] and message[4] else None,
|
|
"to": message[5],
|
|
"timestamp": message[6],
|
|
"response_timestamp": message[7]
|
|
})
|
|
|
|
return _json_response({
|
|
"success": True,
|
|
"canRespond": True,
|
|
"messages": output,
|
|
"more": msgObjects.count() > 50
|
|
})
|
|
|
|
def api_user(request: WSGIRequest, username: str) -> HttpResponse:
|
|
queryFilter = {}
|
|
|
|
try:
|
|
user = get_user_object(username)
|
|
except tMUser.DoesNotExist:
|
|
return RESPONSE_400
|
|
|
|
queryFilter = None
|
|
|
|
if "offset" in request.GET and (request.GET.get("offset") or "").isdigit():
|
|
queryFilter = _add_qF(queryFilter, Q(message_id__lt=int(request.GET.get("offset") or "")))
|
|
|
|
if "all" not in request.GET:
|
|
queryFilter = _add_qF(queryFilter, ~Q(response=None))
|
|
|
|
if queryFilter:
|
|
msgObjects = user.received.filter(queryFilter) # type: ignore
|
|
else:
|
|
msgObjects = user.received.all() # type: ignore
|
|
|
|
output = []
|
|
messages = msgObjects.order_by("-sent_timestamp" if "all" in request.GET else "-response_timestamp")[:50].values_list(
|
|
"message_id", "content", "response", "anonymous", "u_from", "u_to", "sent_timestamp", "response_timestamp"
|
|
)
|
|
|
|
for message in messages:
|
|
output.append({
|
|
"id": message[0],
|
|
"content": message[1],
|
|
"response": message[2],
|
|
"from": message[4] if not message[3] and message[4] else None,
|
|
"to": message[5],
|
|
"timestamp": message[6],
|
|
"response_timestamp": message[7]
|
|
})
|
|
|
|
return _json_response({
|
|
"success": True,
|
|
"canRespond": username == get_username(request),
|
|
"messages": output,
|
|
"more": msgObjects.count() > 50
|
|
})
|