forked from Hay1tsme/artemis
341 lines
13 KiB
Python
341 lines
13 KiB
Python
import logging, coloredlogs
|
|
from typing import Optional, Dict, List
|
|
from sqlalchemy.orm import scoped_session, sessionmaker
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy import create_engine
|
|
from logging.handlers import TimedRotatingFileHandler
|
|
import importlib, os
|
|
import secrets, string
|
|
import bcrypt
|
|
from hashlib import sha256
|
|
|
|
from core.config import CoreConfig
|
|
from core.data.schema import *
|
|
from core.utils import Utils
|
|
|
|
|
|
class Data:
|
|
def __init__(self, cfg: CoreConfig) -> None:
|
|
self.config = cfg
|
|
|
|
if self.config.database.sha2_password:
|
|
passwd = sha256(self.config.database.password.encode()).digest()
|
|
self.__url = f"{self.config.database.protocol}://{self.config.database.username}:{passwd.hex()}@{self.config.database.host}/{self.config.database.name}?charset=utf8mb4"
|
|
else:
|
|
self.__url = f"{self.config.database.protocol}://{self.config.database.username}:{self.config.database.password}@{self.config.database.host}/{self.config.database.name}?charset=utf8mb4"
|
|
|
|
self.__engine = create_engine(self.__url, pool_recycle=3600)
|
|
session = sessionmaker(bind=self.__engine, autoflush=True, autocommit=True)
|
|
self.session = scoped_session(session)
|
|
|
|
self.user = UserData(self.config, self.session)
|
|
self.arcade = ArcadeData(self.config, self.session)
|
|
self.card = CardData(self.config, self.session)
|
|
self.base = BaseData(self.config, self.session)
|
|
self.current_schema_version = 4
|
|
|
|
log_fmt_str = "[%(asctime)s] %(levelname)s | Database | %(message)s"
|
|
log_fmt = logging.Formatter(log_fmt_str)
|
|
self.logger = logging.getLogger("database")
|
|
|
|
# Prevent the logger from adding handlers multiple times
|
|
if not getattr(self.logger, "handler_set", None):
|
|
fileHandler = TimedRotatingFileHandler(
|
|
"{0}/{1}.log".format(self.config.server.log_dir, "db"),
|
|
encoding="utf-8",
|
|
when="d",
|
|
backupCount=10,
|
|
)
|
|
fileHandler.setFormatter(log_fmt)
|
|
|
|
consoleHandler = logging.StreamHandler()
|
|
consoleHandler.setFormatter(log_fmt)
|
|
|
|
self.logger.addHandler(fileHandler)
|
|
self.logger.addHandler(consoleHandler)
|
|
|
|
self.logger.setLevel(self.config.database.loglevel)
|
|
coloredlogs.install(
|
|
cfg.database.loglevel, logger=self.logger, fmt=log_fmt_str
|
|
)
|
|
self.logger.handler_set = True # type: ignore
|
|
|
|
def create_database(self):
|
|
self.logger.info("Creating databases...")
|
|
try:
|
|
metadata.create_all(self.__engine.connect())
|
|
except SQLAlchemyError as e:
|
|
self.logger.error(f"Failed to create databases! {e}")
|
|
return
|
|
|
|
games = Utils.get_all_titles()
|
|
for game_dir, game_mod in games.items():
|
|
try:
|
|
if hasattr(game_mod, "database") and hasattr(
|
|
game_mod, "current_schema_version"
|
|
):
|
|
game_mod.database(self.config)
|
|
metadata.create_all(self.__engine.connect())
|
|
|
|
self.base.set_schema_ver(
|
|
game_mod.current_schema_version, game_mod.game_codes[0]
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.warning(
|
|
f"Could not load database schema from {game_dir} - {e}"
|
|
)
|
|
|
|
self.logger.info(f"Setting base_schema_ver to {self.current_schema_version}")
|
|
self.base.set_schema_ver(self.current_schema_version)
|
|
|
|
self.logger.info(
|
|
f"Setting user auto_incrememnt to {self.config.database.user_table_autoincrement_start}"
|
|
)
|
|
self.user.reset_autoincrement(
|
|
self.config.database.user_table_autoincrement_start
|
|
)
|
|
|
|
def recreate_database(self):
|
|
self.logger.info("Dropping all databases...")
|
|
self.base.execute("SET FOREIGN_KEY_CHECKS=0")
|
|
try:
|
|
metadata.drop_all(self.__engine.connect())
|
|
except SQLAlchemyError as e:
|
|
self.logger.error(f"Failed to drop databases! {e}")
|
|
return
|
|
|
|
for root, dirs, files in os.walk("./titles"):
|
|
for dir in dirs:
|
|
if not dir.startswith("__"):
|
|
try:
|
|
mod = importlib.import_module(f"titles.{dir}")
|
|
|
|
try:
|
|
if hasattr(mod, "database"):
|
|
mod.database(self.config)
|
|
metadata.drop_all(self.__engine.connect())
|
|
|
|
except Exception as e:
|
|
self.logger.warning(
|
|
f"Could not load database schema from {dir} - {e}"
|
|
)
|
|
|
|
except ImportError as e:
|
|
self.logger.warning(
|
|
f"Failed to load database schema dir {dir} - {e}"
|
|
)
|
|
break
|
|
|
|
self.base.execute("SET FOREIGN_KEY_CHECKS=1")
|
|
|
|
self.create_database()
|
|
|
|
def migrate_database(self, game: str, version: Optional[int], action: str) -> None:
|
|
old_ver = self.base.get_schema_ver(game)
|
|
sql = ""
|
|
if version is None:
|
|
if not game == "CORE":
|
|
titles = Utils.get_all_titles()
|
|
|
|
for folder, mod in titles.items():
|
|
if not mod.game_codes[0] == game:
|
|
continue
|
|
|
|
if hasattr(mod, "current_schema_version"):
|
|
version = mod.current_schema_version
|
|
|
|
else:
|
|
self.logger.warn(
|
|
f"current_schema_version not found for {folder}"
|
|
)
|
|
|
|
else:
|
|
version = self.current_schema_version
|
|
|
|
if version is None:
|
|
self.logger.warn(
|
|
f"Could not determine latest version for {game}, please specify --version"
|
|
)
|
|
|
|
if old_ver is None:
|
|
self.logger.error(
|
|
f"Schema for game {game} does not exist, did you run the creation script?"
|
|
)
|
|
return
|
|
|
|
if old_ver == version:
|
|
self.logger.info(
|
|
f"Schema for game {game} is already version {old_ver}, nothing to do"
|
|
)
|
|
return
|
|
|
|
if action == "upgrade":
|
|
for x in range(old_ver, version):
|
|
if not os.path.exists(
|
|
f"core/data/schema/versions/{game.upper()}_{x + 1}_{action}.sql"
|
|
):
|
|
self.logger.error(
|
|
f"Could not find {action} script {game.upper()}_{x + 1}_{action}.sql in core/data/schema/versions folder"
|
|
)
|
|
return
|
|
|
|
with open(
|
|
f"core/data/schema/versions/{game.upper()}_{x + 1}_{action}.sql",
|
|
"r",
|
|
encoding="utf-8",
|
|
) as f:
|
|
sql = f.read()
|
|
|
|
result = self.base.execute(sql)
|
|
if result is None:
|
|
self.logger.error("Error execuing sql script!")
|
|
return None
|
|
|
|
else:
|
|
for x in range(old_ver, version, -1):
|
|
if not os.path.exists(
|
|
f"core/data/schema/versions/{game.upper()}_{x - 1}_{action}.sql"
|
|
):
|
|
self.logger.error(
|
|
f"Could not find {action} script {game.upper()}_{x - 1}_{action}.sql in core/data/schema/versions folder"
|
|
)
|
|
return
|
|
|
|
with open(
|
|
f"core/data/schema/versions/{game.upper()}_{x - 1}_{action}.sql",
|
|
"r",
|
|
encoding="utf-8",
|
|
) as f:
|
|
sql = f.read()
|
|
|
|
result = self.base.execute(sql)
|
|
if result is None:
|
|
self.logger.error("Error execuing sql script!")
|
|
return None
|
|
|
|
result = self.base.set_schema_ver(version, game)
|
|
if result is None:
|
|
self.logger.error("Error setting version in schema_version table!")
|
|
return None
|
|
|
|
self.logger.info(f"Successfully migrated {game} to schema version {version}")
|
|
|
|
def create_owner(self, email: Optional[str] = None) -> None:
|
|
pw = "".join(
|
|
secrets.choice(string.ascii_letters + string.digits) for i in range(20)
|
|
)
|
|
hash = bcrypt.hashpw(pw.encode(), bcrypt.gensalt())
|
|
|
|
user_id = self.user.create_user(email=email, permission=255, password=hash)
|
|
if user_id is None:
|
|
self.logger.error(f"Failed to create owner with email {email}")
|
|
return
|
|
|
|
card_id = self.card.create_card(user_id, "00000000000000000000")
|
|
if card_id is None:
|
|
self.logger.error(f"Failed to create card for owner with id {user_id}")
|
|
return
|
|
|
|
self.logger.warn(
|
|
f"Successfully created owner with email {email}, access code 00000000000000000000, and password {pw} Make sure to change this password and assign a real card ASAP!"
|
|
)
|
|
|
|
def migrate_card(self, old_ac: str, new_ac: str, should_force: bool) -> None:
|
|
if old_ac == new_ac:
|
|
self.logger.error("Both access codes are the same!")
|
|
return
|
|
|
|
new_card = self.card.get_card_by_access_code(new_ac)
|
|
if new_card is None:
|
|
self.card.update_access_code(old_ac, new_ac)
|
|
return
|
|
|
|
if not should_force:
|
|
self.logger.warn(
|
|
f"Card already exists for access code {new_ac} (id {new_card['id']}). If you wish to continue, rerun with the '--force' flag."
|
|
f" All exiting data on the target card {new_ac} will be perminently erased and replaced with data from card {old_ac}."
|
|
)
|
|
return
|
|
|
|
self.logger.info(
|
|
f"All exiting data on the target card {new_ac} will be perminently erased and replaced with data from card {old_ac}."
|
|
)
|
|
self.card.delete_card(new_card["id"])
|
|
self.card.update_access_code(old_ac, new_ac)
|
|
|
|
hanging_user = self.user.get_user(new_card["user"])
|
|
if hanging_user["password"] is None:
|
|
self.logger.info(f"Delete hanging user {hanging_user['id']}")
|
|
self.user.delete_user(hanging_user["id"])
|
|
|
|
def delete_hanging_users(self) -> None:
|
|
"""
|
|
Finds and deletes users that have not registered for the webui that have no cards assocated with them.
|
|
"""
|
|
unreg_users = self.user.get_unregistered_users()
|
|
if unreg_users is None:
|
|
self.logger.error("Error occoured finding unregistered users")
|
|
|
|
for user in unreg_users:
|
|
cards = self.card.get_user_cards(user["id"])
|
|
if cards is None:
|
|
self.logger.error(f"Error getting cards for user {user['id']}")
|
|
continue
|
|
|
|
if not cards:
|
|
self.logger.info(f"Delete hanging user {user['id']}")
|
|
self.user.delete_user(user["id"])
|
|
|
|
def autoupgrade(self) -> None:
|
|
all_game_versions = self.base.get_all_schema_vers()
|
|
if all_game_versions is None:
|
|
self.logger.warn("Failed to get schema versions")
|
|
return
|
|
|
|
all_games = Utils.get_all_titles()
|
|
all_games_list: Dict[str, int] = {}
|
|
for _, mod in all_games.items():
|
|
if hasattr(mod, "current_schema_version"):
|
|
all_games_list[mod.game_codes[0]] = mod.current_schema_version
|
|
|
|
for x in all_game_versions:
|
|
failed = False
|
|
game = x["game"].upper()
|
|
update_ver = int(x["version"])
|
|
latest_ver = all_games_list.get(game, 1)
|
|
if game == "CORE":
|
|
latest_ver = self.current_schema_version
|
|
|
|
if update_ver == latest_ver:
|
|
self.logger.info(f"{game} is already latest version")
|
|
continue
|
|
|
|
for y in range(update_ver + 1, latest_ver + 1):
|
|
if os.path.exists(f"core/data/schema/versions/{game}_{y}_upgrade.sql"):
|
|
with open(
|
|
f"core/data/schema/versions/{game}_{y}_upgrade.sql",
|
|
"r",
|
|
encoding="utf-8",
|
|
) as f:
|
|
sql = f.read()
|
|
|
|
result = self.base.execute(sql)
|
|
if result is None:
|
|
self.logger.error(
|
|
f"Error execuing sql script for game {game} v{y}!"
|
|
)
|
|
failed = True
|
|
break
|
|
else:
|
|
self.logger.warning(f"Could not find script {game}_{y}_upgrade.sql")
|
|
failed = True
|
|
|
|
if not failed:
|
|
self.base.set_schema_ver(latest_ver, game)
|
|
|
|
def show_versions(self) -> None:
|
|
all_game_versions = self.base.get_all_schema_vers()
|
|
for ver in all_game_versions:
|
|
self.logger.info(f"{ver['game']} -> v{ver['version']}")
|