75 lines
2.5 KiB
Python
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)
|