lewangoalski/FlaskWebProject/FlaskWebProject/lewy_api_v1.py

# API is expected to return a tuple of:
# - HTTP status code,
# - human-readable status message,
# - json with appropriate data
from datetime import datetime
from flask_sqlalchemy import SQLAlchemy
from fs_scraper import scraper
from functools import wraps
from lewy_globals import getDb, colors as c
import flask, json, time
import lewy_db as ldb
import lewy_globals

def require_authentication(func):
	"""
	Ten dekorator służy do wymuszenia parametru "token"
	podczas obsługi zapytania. Powinien on zostać doklejony
	do żądania, np. /api/v1/halt?token=XXX...
	Wartość tokenu jest pobierana z pola api_key w config.toml.
	Jeżeli skrypt jej tam nie znajdzie, jest generowana losowo
	na starcie i drukowana w terminalu.
	"""
	@wraps(func)
	def wrapper(*args, **kwargs):
		token = kwargs["r"].args.get('token')
		if token == lewy_globals.config['api']['api_key']:
			try:
				status, received, data = func(*args, **kwargs)
				return status, received, data
			except:
				raise AssertionError(f"Function \"{func.__name__}\" does not return status, code, and data as it should!")
		else:
			increment_bad_requests()
			return 401, "error", {'error_msg': "Unauthorized"}
	return wrapper

def increment_bad_requests():
	"""
	Zwiększa globalny, tymczasowy licznik nieprawidłowych zapytań.
	"""
	lewy_globals.apiFailedRequests += 1

def not_implemented(data):
	"""
	Zwraca kod 501 wraz z endpointem, który wywołał błąd.
	
	:param      data:  Ścieżka zapytania
	:type       data:  list
	"""
	# TODO: change list to string -> data, not data[0]
	return 501, f"not recognised/implemented: {data[0]}", []

def __czy_x_istnieje(typ, **id):
	rekord = getDb().simple_select_all(typ, **id)
	if rekord is not None and rekord != []:
		return True
	else:
		return False

def czy_sportowiec_istnieje(id_zawodnika: str):
	return __czy_x_istnieje("sportowcy", id_zawodnika=id_zawodnika)

# GET /api/v1
def stub_hello():
	"""
	Prosta funkcja witająca użytkowników w /api/v1
	"""
	return 200, 'hello from v1! stats are at /api/v1/stats', []

def epoch_to_date(epoch):
	"""
	Zamienia Unix'owy epoch na lokalny czas,
	w formacie przypominającym format ISO.
	
	:param      epoch:  Epoch - sekundy po 1. stycznia 1970
	:type       epoch:  int
	"""
	return time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(epoch))

# GET /api/v1/
# GET /api/v1/stats
def stats():
	"""
	Zwraca ogólne statystyki serwera.
	"""
	data_to_send = {
		"start_time": lewy_globals.starttime,
		"uptime": lewy_globals.getUptime(),
		"real_uptime": lewy_globals.realUptime,
		"total_api_requests": lewy_globals.apiRequests,
		"failed_api_requests": lewy_globals.apiFailedRequests,
		"outside_api_requests": lewy_globals.outsideApiHits,
		"local_api_requests": lewy_globals.apiRequests - lewy_globals.outsideApiHits
	}
	return 200, "ok", data_to_send

# GET /api/v1/matches
def get_matches(r = None, id_zawodnika: str | None = None, rok: int | None = None):
	"""
	Zwraca mecze.
	Przykład wywołania:
	get_matches(r, id_zawodnika=1), tożsame z GET /api/v1/matches?id_zawodnika=1
	get_matches(r, rok=2024),       tożsame z GET /api/v1/matches?rok=2024
	get_matches(r),                 tożsame z GET /api/v1/matches
	"""
	response_json = []
	mecze = None

	if id_zawodnika is None:
		# Gdy nie podano id wprost, sprawdź, czy podano je przez parametr.
		id_zawodnika = r.args.get('id_zawodnika', -1)

	if rok is None:
		# Gdy nie podano roku wprost, sprawdź, czy podano je przez parametr.
		# Jeśli nie, przyjmij None (2025).
		rok = r.args.get('rok', None)

	# Sprawdź, czy podano jakiekolwiek ID sportowca. Jeżeli nie, wypisz wszystkie mecze.
	if id_zawodnika == -1:
		mecze = getDb().get_sportsman_matches(year=rok)

	# Sprawdź, czy sportowiec o podanym (lub niepodanym) id istnieje.
	# Jeśli nie istnieje, wypisz wszystkie mecze.
	elif not czy_sportowiec_istnieje(id_zawodnika=id_zawodnika):
		return 404, "error", {"error_msg": "This sportsman has not been found in the database. Try: id_zawodnika=1"}

	# Gdy sportowiec istnieje, wypisz jego mecze.
	else:
		mecze = getDb().get_sportsman_matches(id_zawodnika=id_zawodnika, year=rok)

	for mecz in mecze:
		response_json.append(mecz.jsonify())
	# print(f"zwracam mecze: {response_json}")
	return 200, "ok", response_json

# GET /api/v1/debugger_halt?token=XXX...
@require_authentication
def debugger_halt(r):
	"""
	Zatrzymuje wykonywanie skryptu, aby pozwolić
	administratorowi na wykonywanie dowolnego polecenia z konsoli.
	"""
	if lewy_globals.config['general']['is_proxied']:
		print(f"{c.WARNING}[{epoch_to_date(time.time())}]{c.ENDC} {r.headers['X-Forwarded-For']} triggered a debugger halt!")
	else:
		print(f"{c.WARNING}[{epoch_to_date(time.time())}]{c.ENDC} {r.remote_addr} triggered a debugger halt!")
	breakpoint()
	return 200, "ok", []

def last_goal_for(sportowiec: str = "MVC8zHZD"):
	"""
	Pobierz klub, dla którego sportowiec (domyślnie Robert) oddał ostatni gol.
	
	:returns:   Klub, lub None
	:rtype:     kluby | None
	"""

	sportowiec = getDb().simple_select_all("sportowcy", zewnetrzne_id_zawodnika=sportowiec)
	if sportowiec != []:
		return sportowiec[0].ostatni_gol_dla

	return None

def lookup(data, request):
	"""
	Obsługuje zapytania zwrócone do /api/v1/...
	
	:param      data:     Lista ze ścieżką zapytania
	:type       data:     list
	:param      request:  Zapytanie
	:type       request:  flask.request
	
	:returns:   Wartość zwróconą przez którąś z przywołanych funkcji.
	"""
	if data == []:
		return stub_hello()
	match data[0].lower():
		case 'stats' | '':
			return stats()
		case 'user':
			return stub_hello()
		case 'info':
			return stub_hello()
		case 'halt':
			return debugger_halt(r = request)
		case 'matches':
			return get_matches(r = request)
		case _:
			increment_bad_requests()
			return not_implemented(data)