giants-maps-api/app/routers/maps.py
Hipstercat 7b74f1240f
All checks were successful
continuous-integration/drone/push Build is passing
dropped db, works with plain file structure now
Signed-off-by: Hipstercat <tasty@hipstercat.fr>
2021-12-22 23:25:46 +01:00

75 lines
2.5 KiB
Python

import base64
import io
import os.path
import pathlib
import zipfile
from typing import List, Optional
from fastapi import APIRouter, HTTPException, Query
from ..schemas import *
from ..utils import crc32, read_all_maps, map_file_to_dict
from ..config import config
import re
router = APIRouter()
MAX_UPLOAD_SIZE = config["max_file_size"]
MAPS = read_all_maps()
@router.get("/maps", tags=["maps"], response_model=List[MapOut])
async def get_all_maps():
return [MapOut(**d) for d in MAPS]
@router.get("/map", tags=["maps"], response_model=MapOut)
async def get_map_by_crc(human_crc: Optional[str] = Query(None, min_length=8, max_length=8),
crc: Optional[int] = Query(None)):
if not human_crc and not crc:
raise HTTPException(status_code=400, detail="Please use crc or human_crc but not both")
if human_crc and crc:
raise HTTPException(status_code=400, detail="Please use crc or human_crc but not both")
if human_crc:
maps = [gmap for gmap in MAPS if gmap["crc_human"] == human_crc.upper()]
else:
maps = [gmap for gmap in MAPS if gmap["crc"] == crc]
if not maps:
raise HTTPException(status_code=404, detail="Map not found")
return MapOut(**maps[0])
@router.post("/maps", tags=["maps"], response_model=MapOut)
async def upload_map(map_in: MapIn):
allowed_name = re.compile("[a-zA-Z-.\d()[] ]+.gck")
filename = map_in.name
if not allowed_name.fullmatch(filename):
raise HTTPException(status_code=400, detail="Invalid filename")
map_bytes = base64.b64decode(map_in.b64_data.encode("utf8"))
if len(map_bytes) > MAX_UPLOAD_SIZE:
raise HTTPException(status_code=400, detail="File too big")
map_io = io.BytesIO(map_bytes)
try:
zipfile.ZipFile(map_io)
except zipfile.BadZipfile:
raise HTTPException(status_code=400, detail="File is not a valid map")
crc = crc32(map_bytes)
existing_map = [gmap for gmap in MAPS if gmap["crc"] == crc]
if existing_map:
return MapOut(**existing_map[0])
else:
uploaded_filename = f"{filename}.gck"
i = 0
while os.path.exists(f"{config['upload_path']}{uploaded_filename}"):
uploaded_filename = f"{filename}-{i}.gck"
i += 1
with open(f"{config['upload_path']}{uploaded_filename}", "wb") as fp:
fp.write(map_bytes)
gmap = map_file_to_dict(pathlib.Path(f"{config['upload_path']}{uploaded_filename}"))
MAPS.append(gmap)
return MapOut(**gmap)