146 lines
4.8 KiB
Python
146 lines
4.8 KiB
Python
# -*- encoding: utf-8 -*-
|
|
|
|
"""
|
|
The REST endpoints.
|
|
"""
|
|
|
|
from aiohttp import web
|
|
import datetime
|
|
import logging
|
|
import typing as T
|
|
|
|
from aiohttp_session import get_session, Session
|
|
|
|
from bot_z.async_operator import AsyncOperator
|
|
from api.async_bot import login, logout, checkin, checkout, status
|
|
from api import BASE_URI, DEBUG
|
|
|
|
|
|
alog = logging.getLogger("api")
|
|
routes = web.RouteTableDef()
|
|
OPERATORS = {} # type: T.Dict[T.Text, AsyncOperator]
|
|
|
|
|
|
async def get_set_operator(
|
|
request: web.Request, user: T.Text, password: T.Text
|
|
) -> T.Tuple[AsyncOperator, Session]:
|
|
session = await get_session(request)
|
|
op = None
|
|
if "async_operator" in session:
|
|
op = OPERATORS.get(session["async_operator"])
|
|
base_uri = request.app["base_uri"]
|
|
debug = request.app["debug"]
|
|
headless = request.app["headless"]
|
|
if op is None:
|
|
op = AsyncOperator(base_uri, name=user, headless=headless, debug=debug)
|
|
session["async_operator"] = user
|
|
OPERATORS[user] = op
|
|
return op, session
|
|
|
|
|
|
@routes.get("/")
|
|
@routes.get("/login")
|
|
@routes.get("/badge")
|
|
async def routing_handler(request: web.Request) -> web.Response:
|
|
alog.debug("(%s) %s", request.method, request.path)
|
|
session = await get_session(request)
|
|
op = session.get("async_operator")
|
|
_logged_in = True if op else False
|
|
alog.info("%s - is%s in session", request.path, " NOT" if not _logged_in else "")
|
|
return web.json_response({"logged_in": _logged_in})
|
|
|
|
|
|
@routes.post("/login")
|
|
async def login_handler(request: web.Request) -> web.Response:
|
|
data = await request.json()
|
|
user = data.get("username")
|
|
password = data.get("password")
|
|
if not user or not password:
|
|
alog.debug("login - missing username or password: %s", data)
|
|
return web.json_response({"error": "Missing username or password"}, status=403)
|
|
op, session = await get_set_operator(request, user, password)
|
|
alog.debug("login - user: %s, password: %s", user, password)
|
|
res = await login(op, user, password)
|
|
alog.debug("login result: %s", res)
|
|
if not res:
|
|
session.invalidate()
|
|
alog.info("Login failed; session invalidated.")
|
|
return web.json_response({"logged_in": res}, status=200)
|
|
|
|
|
|
@routes.post("/logout")
|
|
async def logout_handler(request: web.Request) -> web.Response:
|
|
alog.debug("logout")
|
|
session = await get_session(request)
|
|
op_key = session.get("async_operator")
|
|
op = OPERATORS.get(op_key)
|
|
if not op:
|
|
return web.json_response(
|
|
{"error": "No session", "logged_in": False}, status=403
|
|
)
|
|
res = await logout(op)
|
|
alog.debug("logout result: %s", res)
|
|
# FIX: assess if better to invalidate session and dump the browser instance.
|
|
del OPERATORS[op_key]
|
|
return web.json_response({"logged_in": res}, status=200)
|
|
|
|
|
|
@routes.post("/checkin")
|
|
async def checkin_handler(request: web.Request) -> web.Response:
|
|
alog.debug("checkin")
|
|
session = await get_session(request)
|
|
op = OPERATORS.get(session.get("async_operator"))
|
|
if not op:
|
|
return web.json_response(
|
|
{"error": "No session", "logged_in": False}, status=403
|
|
)
|
|
res = await checkin(op)
|
|
alog.debug("checkin result: %s", res)
|
|
return web.json_response({"checked_in": res, "logged_in": True}, status=200)
|
|
|
|
|
|
@routes.post("/checkout")
|
|
async def checkout_handler(request: web.Request) -> web.Response:
|
|
alog.debug("checkout")
|
|
session = await get_session(request)
|
|
op = OPERATORS.get(session.get("async_operator"))
|
|
if not op:
|
|
return web.json_response(
|
|
{"error": "No session", "logged_in": False}, status=403
|
|
)
|
|
res = await checkout(op)
|
|
alog.debug("checkout result: %s", res)
|
|
return web.json_response({"checked_in": res, "logged_in": True}, status=200)
|
|
|
|
|
|
@routes.get("/movements")
|
|
async def movements_handle(request: web.Request) -> web.Response:
|
|
alog.debug("movements")
|
|
session = await get_session(request)
|
|
op = OPERATORS.get(session.get("async_operator"))
|
|
if not op:
|
|
return web.json_response(
|
|
{"error": "No session", "logged_in": False}, status=403
|
|
)
|
|
res = await status(op)
|
|
alog.debug("movements result: %s", res)
|
|
if not res:
|
|
return web.json_response(
|
|
{"error": "No movements found", "logged_in": True}, status=404
|
|
)
|
|
movements = []
|
|
for r in res:
|
|
if r and len(r) == 2:
|
|
movements.append({"time": r[1], "type": r[0]})
|
|
resp_data = {"movements": movements}
|
|
resp_data["logged_in"] = True # type: ignore
|
|
return web.json_response(resp_data, status=200)
|
|
|
|
|
|
@routes.get("/status")
|
|
async def status_handle(request: web.Request) -> web.Response:
|
|
alog.debug("status")
|
|
session = await get_session(request)
|
|
_logged_in = session.get("async_operator") in OPERATORS
|
|
return web.json_response({"logged_in": _logged_in}, status=200)
|