giants-maps-api/app/routers/maps.py
2021-12-22 15:56:35 +01:00

80 lines
2.7 KiB
Python

import base64
import struct
import io
import zipfile
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from ..dependencies import get_db
from ..schemas import *
from ..utils import crc32, map_model_to_out_schema
from ..models import *
from ..config import config
router = APIRouter()
MAX_UPLOAD_SIZE = config["max_file_size"]
@router.get("/maps", tags=["maps"], response_model=List[MapOut])
async def get_all_maps(db: Session = Depends(get_db)):
return [map_model_to_out_schema(m) for m in db.query(Map).all()]
@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),
db: Session = Depends(get_db)):
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:
crc_int = struct.unpack(">L", bytes.fromhex(human_crc))[0]
else:
crc_int = crc
existing_map = db.query(Map).filter(Map.crc == crc_int).first()
if not existing_map:
raise HTTPException(status_code=404, detail="Map not found")
return map_model_to_out_schema(existing_map)
@router.post("/maps", tags=["maps"], response_model=MapOut)
async def upload_map(map_in: MapIn, db: Session = Depends(get_db)):
filename = map_in.name
if not filename.lower().endswith(".gck"):
raise HTTPException(status_code=400, detail="Invalid file")
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 = db.query(Map).filter(Map.crc == crc).first()
if existing_map:
return map_model_to_out_schema(existing_map)
else:
uploaded_filename = "%s.gck" % crc
with open("%s%s" % (config["upload_path"], uploaded_filename), "wb") as fp:
fp.write(map_bytes)
uploaded_map = Map()
uploaded_map.crc = crc
uploaded_map.name = map_in.name
uploaded_map.filename = uploaded_filename
uploaded_map.size = len(map_bytes)
db.add(uploaded_map)
db.commit()
db.refresh(uploaded_map)
return map_model_to_out_schema(uploaded_map)