125 lines
2.9 KiB
Python
125 lines
2.9 KiB
Python
import logging
|
|
import requests
|
|
from typing import Annotated
|
|
from fastapi import FastAPI, Header, HTTPException, status
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import FileResponse
|
|
from pydantic import BaseModel
|
|
from . import myice
|
|
|
|
origins = ["*"]
|
|
|
|
app = FastAPI()
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=origins,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
block_endpoints = ["/health"]
|
|
|
|
|
|
class LogFilter(logging.Filter):
|
|
def filter(self, record):
|
|
if record.args and len(record.args) >= 3:
|
|
if record.args[2] in block_endpoints:
|
|
return False
|
|
return True
|
|
|
|
|
|
uvicorn_logger = logging.getLogger("uvicorn.access")
|
|
uvicorn_logger.addFilter(LogFilter())
|
|
|
|
|
|
class AuthHeaders(BaseModel):
|
|
Authorization: str
|
|
|
|
def token(self) -> str:
|
|
return self.Authorization.split(" ")[1]
|
|
|
|
def authorized(self) -> bool:
|
|
if self.token() == "abc":
|
|
return True
|
|
return False
|
|
|
|
|
|
class HealthCheck(BaseModel):
|
|
"""Response model to validate and return when performing a health check."""
|
|
|
|
status: str = "OK"
|
|
|
|
|
|
@app.get("/")
|
|
async def home():
|
|
return FileResponse("index.html")
|
|
|
|
|
|
@app.get(
|
|
"/health",
|
|
response_model=HealthCheck,
|
|
status_code=status.HTTP_200_OK,
|
|
)
|
|
def get_health() -> HealthCheck:
|
|
return HealthCheck(status="OK")
|
|
|
|
|
|
@app.get("/favicon.ico")
|
|
async def favico():
|
|
return FileResponse("favicon.ico")
|
|
|
|
|
|
@app.get("/schedule")
|
|
async def schedule(
|
|
headers: Annotated[AuthHeaders, Header()],
|
|
):
|
|
if not headers.authorized():
|
|
raise HTTPException(401, detail="get out")
|
|
username, password, userid, existing_token = myice.get_login()
|
|
if existing_token:
|
|
myice.userdata = {
|
|
"id": userid,
|
|
"id_club": 186,
|
|
"token": existing_token,
|
|
}
|
|
else:
|
|
myice.userdata = myice.mobile_login()
|
|
return myice.refresh_data()["club_games"]
|
|
|
|
|
|
@app.get("/game/{game_id}")
|
|
async def game(
|
|
headers: Annotated[AuthHeaders, Header()],
|
|
game_id: int,
|
|
):
|
|
username, password, userid, existing_token = myice.get_login()
|
|
if existing_token:
|
|
myice.userdata = {
|
|
"id": userid,
|
|
"id_club": 186,
|
|
"token": existing_token,
|
|
}
|
|
else:
|
|
myice.userdata = myice.mobile_login()
|
|
|
|
# data = refresh_data()
|
|
with requests.post(
|
|
"https://app.myice.hockey/api/mobilerest/getevent",
|
|
headers=myice.mobile_headers,
|
|
data="&".join(
|
|
[
|
|
f"token={myice.userdata['token']}",
|
|
f"id_event={game_id}",
|
|
"type=games",
|
|
f"id_player={myice.userdata['id']}",
|
|
f"id_club={myice.userdata['id_club']}",
|
|
"language=FR",
|
|
]
|
|
),
|
|
# verify=False,
|
|
) as r:
|
|
data = r.json()["eventData"]
|
|
return data
|