move storage classes into server folder and fix error in localStorage when winner_snake is none
This commit is contained in:
+1
-1
@@ -2,7 +2,7 @@ from server.Files import read_file
|
||||
from server.GameBoard import GameBoard
|
||||
from server.SnakeBuilder import SnakeBuilder
|
||||
|
||||
from statestorage.StorageLoader import StorageLoader
|
||||
from server.storage.StorageLoader import StorageLoader
|
||||
|
||||
from flask import Flask
|
||||
from flask import request
|
||||
|
||||
@@ -0,0 +1,93 @@
|
||||
from server.GameBoard import GameBoard
|
||||
|
||||
from datetime import datetime
|
||||
import edgedb, json
|
||||
|
||||
class EdgeDB:
|
||||
def __init__(self, **kwargs):
|
||||
self.client = edgedb.create_client(
|
||||
tls_security="insecure"
|
||||
)
|
||||
|
||||
def insert_game_type(self, name:str, is_ladder:bool):
|
||||
return self.client.query_required_single("""
|
||||
insert GameType {
|
||||
name := <str>$name,
|
||||
is_ladder := <bool>$is_ladder
|
||||
}""",
|
||||
name=name,
|
||||
is_ladder=is_ladder
|
||||
)
|
||||
|
||||
def insert(self, game_board:GameBoard):
|
||||
game_type = game_board._get_type_of_gameboard()
|
||||
|
||||
self.client.query("""
|
||||
insert GameBoard {
|
||||
id := <uuid>$id,
|
||||
url := <str>$url,
|
||||
created_at := <datetime>$created_at,
|
||||
turns := <int32>$turns,
|
||||
map := <str>$map,
|
||||
winner := <str>$winner,
|
||||
moves := (
|
||||
with input_data := <array <tuple <turn: int32, `move`: str, game_board: json>> >$moves
|
||||
for data in array_unpack(input_data)
|
||||
insert Moves {
|
||||
turn := data.turn,
|
||||
snake_move := data.`move`,
|
||||
game_board := data.game_board
|
||||
}
|
||||
),
|
||||
type := (
|
||||
insert GameType {
|
||||
name := <str>$game_type,
|
||||
is_ladder := <bool>$is_ladder
|
||||
}
|
||||
),
|
||||
ruleset := (
|
||||
insert Ruleset {
|
||||
name := <str>$ruleset,
|
||||
version := <str>$version,
|
||||
settings := to_json(<str>$settings)
|
||||
}
|
||||
),
|
||||
snake := (
|
||||
insert Snake {
|
||||
type := <str>$snake_type,
|
||||
calculations := (
|
||||
with input_data := <array <tuple <turn: int32, data: array<json>>> >$calculations
|
||||
for tupel_data in array_unpack(input_data)
|
||||
insert Calculations {
|
||||
turn := tupel_data.turn,
|
||||
data := tupel_data.data
|
||||
}
|
||||
)
|
||||
}
|
||||
)
|
||||
}""",
|
||||
|
||||
id=game_board.id,
|
||||
url=game_board.url if game_board.url else "",
|
||||
created_at=datetime.fromtimestamp(game_board.now_date.timestamp(), game_board.now_date.astimezone().tzinfo),
|
||||
turns=game_board.turn,
|
||||
map=game_board.map,
|
||||
winner=', '.join(game_board.winner_snake_names) if game_board.winner_snake_names else "",
|
||||
moves=[ tuple([ele["turn"], ele["move"], json.dumps(ele["game_board"])]) for ele in game_board.turns ],
|
||||
|
||||
game_type=game_type["name"],
|
||||
is_ladder=game_type["is_ladder"],
|
||||
|
||||
ruleset=game_board.ruleset["name"],
|
||||
version=game_board.ruleset["version"],
|
||||
settings=json.dumps(game_board.ruleset["settings"]),
|
||||
|
||||
snake_type=game_board.snake_class.__class__.__name__,
|
||||
calculations=[ tuple([ele["turn"], [json.dumps(calc) for calc in ele["data"]]]) for ele in game_board.snake_class.get_history() ],
|
||||
)
|
||||
|
||||
def save(self, game_board:GameBoard):
|
||||
self.insert(game_board)
|
||||
|
||||
def __del__(self):
|
||||
self.client.close()
|
||||
@@ -0,0 +1,56 @@
|
||||
from server.GameBoard import GameBoard
|
||||
from server.Files import save_file
|
||||
|
||||
import json, os
|
||||
|
||||
class LocalStorage:
|
||||
def __init__(self, file_path:str):
|
||||
self.save_folder_dict = {
|
||||
"standart": "01_Standart",
|
||||
"duel": "02_Duels",
|
||||
"constrictor": "04_Constrictor",
|
||||
}
|
||||
|
||||
self.file_path = file_path
|
||||
|
||||
def _get_correct_folder_for_save_file(self, game_board:GameBoard, file_name:str, game_type:str, leader_board:bool, winner:bool):
|
||||
storage_folder = self.file_path
|
||||
if leader_board:
|
||||
storage_folder = os.path.join(storage_folder, "00_Leaderboards")
|
||||
|
||||
storage_folder = os.path.join(storage_folder, self.save_folder_dict[game_type])
|
||||
storage_folder = os.path.join(storage_folder, game_board.now_date.strftime('%Y'), game_board.now_date.strftime('%m_%B'), game_board.now_date.strftime('%d'))
|
||||
|
||||
if winner:
|
||||
storage_folder = os.path.join(storage_folder, "Winner")
|
||||
else:
|
||||
storage_folder = os.path.join(storage_folder, "Lost")
|
||||
|
||||
return os.path.join(storage_folder, file_name)
|
||||
|
||||
def save(self, game_board:GameBoard):
|
||||
game_type = game_board._get_type_of_gameboard()
|
||||
save_file_path = self._get_correct_folder_for_save_file(
|
||||
game_board,
|
||||
f"{game_board.snake_class.__class__.__name__}_{game_board.now_date.strftime('%H-%M-%S')}_{game_board.id}.json",
|
||||
game_type["name"],
|
||||
game_type["is_ladder"],
|
||||
True if game_board.winner_snake_names and "me" in game_board.winner_snake_names else False
|
||||
)
|
||||
|
||||
save_file(save_file_path, {
|
||||
"winner": game_board.winner_snake_names,
|
||||
"game": {
|
||||
"url": game_board.url,
|
||||
"id": game_board.id,
|
||||
"final_turns": game_board.turn,
|
||||
"map": game_board.map,
|
||||
"type": game_type,
|
||||
"ruleset": game_board.ruleset,
|
||||
},
|
||||
"moves": game_board.turns,
|
||||
"snake": {
|
||||
"type": game_board.snake_class.__class__.__name__,
|
||||
"calculations": game_board.snake_class.get_history(),
|
||||
},
|
||||
}, callback=json.dump, indent=2, ensure_ascii=False)
|
||||
@@ -0,0 +1,7 @@
|
||||
|
||||
class StorageLoader:
|
||||
@classmethod
|
||||
def build(self, selected_storage:str):
|
||||
storage_module = __import__(f'server.storage.{selected_storage}', fromlist=[selected_storage])
|
||||
storage_class = getattr(storage_module, selected_storage)
|
||||
return storage_class
|
||||
Reference in New Issue
Block a user