From 0de3cb3487bb79c38b5bff4a542541576e84fa65 Mon Sep 17 00:00:00 2001 From: WorldTeacher <41587052+WorldTeacher@users.noreply.github.com> Date: Tue, 21 May 2024 13:20:45 +0200 Subject: [PATCH] small bugfix based on changed structure --- src/backend/database.py | 2395 ++++++++++++++++++++------------------- 1 file changed, 1198 insertions(+), 1197 deletions(-) diff --git a/src/backend/database.py b/src/backend/database.py index d48d89e..92f9367 100644 --- a/src/backend/database.py +++ b/src/backend/database.py @@ -1,1197 +1,1198 @@ -import datetime -import os -import re -import sqlite3 as sql -import tempfile -from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple, Union - -# from icecream import ic -from omegaconf import OmegaConf - -from src.backend.db import ( - CREATE_TABLE_APPARAT, - CREATE_TABLE_APPKONTOS, - CREATE_TABLE_FILES, - CREATE_TABLE_MEDIA, - CREATE_TABLE_MESSAGES, - CREATE_TABLE_PROF, - CREATE_TABLE_SUBJECTS, - CREATE_TABLE_USER, -) -from src.errors import AppPresentError, NoResultError -from src.logic.constants import SEMAP_MEDIA_ACCOUNTS -from src.logic.dataclass import ApparatData, BookData -from src.logic.log import MyLogger -from src.utils import create_blob, dump_pickle, load_pickle - -config = OmegaConf.load("config.yaml") -logger = MyLogger(__name__) - - -class Database: - """ - Initialize the database and create the tables if they do not exist. - """ - - def __init__(self, db_path: str = None): - """ - Default constructor for the database class - - Args: - db_path (str, optional): Optional Path for testing / specific purposes. Defaults to None. - """ - if db_path is None: - self.db_path = config.database.path + config.database.name - self.db_path = self.db_path.replace("~", str(Path.home())) - else: - self.db_path = db_path - if self.get_db_contents() is None: - logger.log_critical("Database does not exist, creating tables") - self.create_tables() - - def get_db_contents(self) -> Union[List[Tuple], None]: - """ - Get the contents of the - - Returns: - Union[List[Tuple], None]: _description_ - """ - try: - with sql.connect(self.db_path) as conn: - cursor = conn.cursor() - cursor.execute("SELECT * FROM sqlite_master WHERE type='table'") - return cursor.fetchall() - except sql.OperationalError: - return None - - def connect(self) -> sql.Connection: - """ - Connect to the database - - Returns: - sql.Connection: The active connection to the database - """ - print(self.db_path) - return sql.connect(self.db_path) - - def close_connection(self, conn: sql.Connection): - """ - closes the connection to the database - - Args: - ---- - - conn (sql.Connection): the connection to be closed - """ - conn.close() - - def create_tables(self): - """ - Create the tables in the database - """ - conn = self.connect() - cursor = conn.cursor() - cursor.execute(CREATE_TABLE_APPARAT) - cursor.execute(CREATE_TABLE_MESSAGES) - cursor.execute(CREATE_TABLE_MEDIA) - cursor.execute(CREATE_TABLE_APPKONTOS) - cursor.execute(CREATE_TABLE_FILES) - cursor.execute(CREATE_TABLE_PROF) - cursor.execute(CREATE_TABLE_USER) - cursor.execute(CREATE_TABLE_SUBJECTS) - conn.commit() - self.close_connection(conn) - - def insertInto(self, query: str, params: Tuple) -> None: - """ - Insert sent data into the database - - Args: - query (str): The query to be executed - params (Tuple): the parameters to be inserted into the database - """ - conn = self.connect() - cursor = conn.cursor() - logger.log_info(f"Inserting {params} into database with query {query}") - cursor.execute(query, params) - conn.commit() - self.close_connection(conn) - - def query_db( - self, query: str, args: Tuple = (), one: bool = False - ) -> Union[Tuple, List[Tuple]]: - """ - Query the Database for the sent query. - - Args: - query (str): The query to be executed - args (Tuple, optional): The arguments for the query. Defaults to (). - one (bool, optional): Return the first result only. Defaults to False. - - Returns: - Union[Typle|List[Tuple]]: Returns the result of the query - """ - conn = self.connect() - cursor = conn.cursor() - logger.log_info(f"Querying database with query {query}, args: {args}") - cursor.execute(query, args) - rv = cursor.fetchall() - conn.commit() - self.close_connection(conn) - return (rv[0] if rv else None) if one else rv - - # Books - def addBookToDatabase( - self, bookdata: BookData, app_id: Union[str, int], prof_id: Union[str, int] - ): - """ - Add books to the database. Both app_id and prof_id are required to add the book to the database, as the app_id and prof_id are used to select the books later on. - - Args: - bookdata (BookData): The metadata of the book to be added - app_id (str): The apparat id where the book should be added to - prof_id (str): The id of the professor where the book should be added to. - """ - conn = self.connect() - cursor = conn.cursor() - t_query = ( - f"SELECT bookdata FROM media WHERE app_id={app_id} AND prof_id={prof_id}" - ) - # print(t_query) - result = cursor.execute(t_query).fetchall() - result = [load_pickle(i[0]) for i in result] - if bookdata in result: - print("Bookdata already in database") - # check if the book was deleted in the apparat - query = ( - "SELECT deleted FROM media WHERE app_id=? AND prof_id=? AND bookdata=?" - ) - params = (app_id, prof_id, dump_pickle(bookdata)) - result = cursor.execute(query, params).fetchone() - if result[0] == 1: - print("Book was deleted, updating bookdata") - query = "UPDATE media SET deleted=0 WHERE app_id=? AND prof_id=? AND bookdata=?" - params = (app_id, prof_id, dump_pickle(bookdata)) - cursor.execute(query, params) - conn.commit() - return - - query = ( - "INSERT INTO media (bookdata, app_id, prof_id,deleted) VALUES (?, ?, ?,?)" - ) - converted = dump_pickle(bookdata) - params = (converted, app_id, prof_id, 0) - cursor.execute(query, params) - conn.commit() - self.close_connection(conn) - - def getBookIdBasedOnSignature( - self, app_id: Union[str, int], prof_id: Union[str, int], signature: str - ) -> int: - """ - Get a book id based on the signature of the book. - - Args: - app_id (str): The apparat id the book should be associated with - prof_id (str): The professor id the book should be associated with - signature (str): The signature of the book - - Returns: - int: The id of the book - """ - result = self.query_db( - "SELECT bookdata, id FROM media WHERE app_id=? AND prof_id=?", - (app_id, prof_id), - ) - books = [(load_pickle(i[0]), i[1]) for i in result] - book = [i for i in books if i[0].signature == signature][0][1] - return book - - def getBookBasedOnSignature( - self, app_id: Union[str, int], prof_id: Union[str, int], signature: str - ) -> BookData: - """ - Get the book based on the signature of the book. - - Args: - app_id (str): The apparat id the book should be associated with - prof_id (str): The professor id the book should be associated with - signature (str): The signature of the book - - Returns: - BookData: The total metadata of the book wrapped in a BookData object - """ - result = self.query_db( - "SELECT bookdata FROM media WHERE app_id=? AND prof_id=?", (app_id, prof_id) - ) - books = [load_pickle(i[0]) for i in result] - book = [i for i in books if i.signature == signature][0] - return book - - def getLastBookId(self) -> int: - """ - Get the last book id in the database - - Returns: - int: ID of the last book in the database - """ - return self.query_db("SELECT id FROM media ORDER BY id DESC", one=True)[0] - - def searchBook(self, data: dict[str, str]) -> list[tuple[BookData, int]]: - """ - Search a book in the database based on the sent data. - - Args: - data (dict[str, str]): A dictionary containing the data to be searched for. The dictionary can contain the following: - - signature: The signature of the book - - title: The title of the book - - Returns: - list[tuple[BookData, int]]: A list of tuples containing the wrapped Metadata and the id of the book - """ - rdata = self.query_db("SELECT * FROM media WHERE deleted=0") - # ic(rdata, len(rdata)) - mode = 0 - if len(data) == 1: - if "signature" in data.keys(): - mode = 1 - elif "title" in data.keys(): - mode = 2 - elif len(data) == 2: - mode = 3 - else: - return None - ret = [] - for book in rdata: - bookdata = load_pickle(book[1]) - app_id = book[2] - prof_id = book[3] - if mode == 1: - if data["signature"] in bookdata.signature: - ret.append((bookdata, app_id, prof_id)) - elif mode == 2: - if data["title"] in bookdata.title: - ret.append((bookdata, app_id, prof_id)) - elif mode == 3: - if ( - data["signature"] in bookdata.signature - and data["title"] in bookdata.title - ): - ret.append((bookdata, app_id, prof_id)) - # ic(ret) - return ret - - def setAvailability(self, book_id: str, available: str): - """ - Set the availability of a book in the database - - Args: - book_id (str): The id of the book - available (str): The availability of the book - """ - self.query_db("UPDATE media SET available=? WHERE id=?", (available, book_id)) - - def getBookId( - self, bookdata: BookData, app_id: Union[str, int], prof_id: Union[str, int] - ) -> int: - """ - Get the id of a book based on the metadata of the book - - Args: - bookdata (BookData): The wrapped metadata of the book - app_id (str): The apparat id the book should be associated with - prof_id (str): The professor id the book should be associated with - - Returns: - int: ID of the book - """ - result = self.query_db( - "SELECT id FROM media WHERE bookdata=? AND app_id=? AND prof_id=?", - (dump_pickle(bookdata), app_id, prof_id), - one=True, - ) - return result[0] - - def getBook(self, book_id: int) -> BookData: - """ - Get the book based on the id in the database - - Args: - book_id (int): The id of the book - - Returns: - BookData: The metadata of the book wrapped in a BookData object - """ - return load_pickle( - self.query_db( - "SELECT bookdata FROM media WHERE id=?", (book_id,), one=True - )[0] - ) - - def getBooks( - self, app_id: Union[str, int], prof_id: Union[str, int], deleted=0 - ) -> list[dict[int, BookData, int]]: - """ - Get the Books based on the apparat id and the professor id - - Args: - app_id (str): The ID of the apparat - prof_id (str): The ID of the professor - deleted (int, optional): The state of the book. Set to 1 to include deleted ones. Defaults to 0. - - Returns: - list[dict[int, BookData, int]]: A list of dictionaries containing the id, the metadata of the book and the availability of the book - """ - qdata = self.query_db( - f"SELECT id,bookdata,available FROM media WHERE (app_id={app_id} AND prof_id={prof_id}) AND (deleted={deleted if deleted == 0 else '1 OR deleted=0'})" - ) - ret_result = [] - for result_a in qdata: - data = {"id": int, "bookdata": BookData, "available": int} - data["id"] = result_a[0] - data["bookdata"] = load_pickle(result_a[1]) - data["available"] = result_a[2] - ret_result.append(data) - return ret_result - - def updateBookdata(self, book_id, bookdata: BookData): - """ - Update the bookdata in the database - - Args: - book_id (str): The id of the book - bookdata (BookData): The new metadata of the book - """ - self.query_db( - "UPDATE media SET bookdata=? WHERE id=?", (dump_pickle(bookdata), book_id) - ) - - def deleteBook(self, book_id): - """ - Delete a book from the database - - Args: - book_id (str): ID of the book - """ - self.query_db("UPDATE media SET deleted=1 WHERE id=?", (book_id,)) - - # File Interactions - def getBlob(self, filename, app_id: Union[str, int]): - """ - Get a blob from the database - - Args: - filename (str): The name of the file - app_id (str): ID of the apparat - - Returns: - bytes: The file stored in - """ - return self.query_db( - "SELECT fileblob FROM files WHERE filename=? AND app_id=?", - (filename, app_id), - one=True, - )[0] - - def insertFile( - self, file: list[dict], app_id: Union[str, int], prof_id: Union[str, int] - ): - """Instert a list of files into the database - - Args: - file (list[dict]): a list containing all the files to be inserted - Structured: [{"name": "filename", "path": "path", "type": "filetype"}] - app_id (int): the id of the apparat - prof_id (str): the id of the professor - """ - for f in file: - filename = f["name"] - path = f["path"] - filetyp = f["type"] - if path == "Database": - continue - blob = create_blob(path) - query = "INSERT OR IGNORE INTO files (filename, fileblob, app_id, filetyp,prof_id) VALUES (?, ?, ?, ?,?)" - self.query_db(query, (filename, blob, app_id, filetyp, prof_id)) - - def recreateFile( - self, filename: str, app_id: Union[str, int], filetype: str - ) -> str: - """Recreate a file from the database - - Args: - filename (str): the name of the file - app_id (Union[str,int]): the id of the apparat - filetype (str): the extension of the file to be created - - Returns: - str: The filename of the recreated file - """ - blob = self.getBlob(filename, app_id) - tempdir = config.database.tempdir - tempdir = tempdir.replace("~", str(Path.home())) - tempdir_path = Path(tempdir) - if not os.path.exists(tempdir_path): - os.mkdir(tempdir_path) - file = tempfile.NamedTemporaryFile( - delete=False, dir=tempdir_path, mode="wb", suffix=f".{filetype}" - ) - file.write(blob) - print("file created") - return file.name - - def getFiles(self, app_id: Union[str, int], prof_id: int) -> list[tuple]: - """Get all the files associated with the apparat and the professor - - Args: - app_id (Union[str,int]): The id of the apparat - prof_id (Union[str,int]): the id of the professor - - Returns: - list[tuple]: a list of tuples containing the filename and the filetype for the corresponding apparat and professor - """ - return self.query_db( - "SELECT filename, filetyp FROM files WHERE app_id=? AND prof_id=?", - (app_id, prof_id), - ) - - def getSemersters(self) -> list[str]: - """Return all the unique semesters in the database - - Returns: - list: a list of strings containing the semesters - """ - data = self.query_db("SELECT DISTINCT erstellsemester FROM semesterapparat") - return [i[0] for i in data] - - def getSubjects(self): - """Get all the subjects in the database - - Returns: - list[tuple]: a list of tuples containing the subjects - """ - return self.query_db("SELECT * FROM subjects") - - # Messages - def addMessage(self, message: dict, user: str, app_id: Union[str, int]): - """add a Message to the database - - Args: - message (dict): the message to be added - user (str): the user who added the message - app_id (Union[str,int]): the id of the apparat - """ - - def __getUserId(user): - return self.query_db( - "SELECT id FROM user WHERE username=?", (user,), one=True - )[0] - - user_id = __getUserId(user) - self.query_db( - "INSERT INTO messages (message, user_id, remind_at,appnr) VALUES (?,?,?,?)", - (message["message"], user_id, message["remind_at"], app_id), - ) - - def getMessages(self, date: str) -> list[dict[str, str, str, str]]: - """Get all the messages for a specific date - - Args: - date (str): a date.datetime object formatted as a string in the format "YYYY-MM-DD" - - Returns: - list[dict[str, str, str, str]]: a list of dictionaries containing the message, the user who added the message, the apparat id and the id of the message - """ - - def __get_user_name(user_id): - return self.query_db( - "SELECT username FROM user WHERE id=?", (user_id,), one=True - )[0] - - messages = self.query_db("SELECT * FROM messages WHERE remind_at=?", (date,)) - ret = [ - {"message": i[2], "user": __get_user_name(i[4]), "appnr": i[5], "id": i[0]} - for i in messages - ] - return ret - - def deleteMessage(self, message_id): - """Delete a message from the database - - Args: - message_id (str): the id of the message - """ - self.query_db("DELETE FROM messages WHERE id=?", (message_id,)) - - # Prof data - def getProfNameById(self, prof_id: Union[str, int], add_title: bool = False) -> str: - """Get a professor name based on the id - - Args: - prof_id (Union[str,int]): The id of the professor - add_title (bool, optional): wether to add the title or no. Defaults to False. - - Returns: - str: The name of the professor - """ - prof = self.query_db( - "SELECT fullname FROM prof WHERE id=?", (prof_id,), one=True - ) - if add_title: - return f"{self.getTitleById(prof_id)}{prof[0]}" - else: - return prof[0] - - def getTitleById(self, prof_id: Union[str, int]) -> str: - """get the title of a professor based on the id - - Args: - prof_id (Union[str,int]): the id of the professor - - Returns: - str: the title of the professor, with an added whitespace at the end, if no title is present, an empty string is returned - """ - title = self.query_db( - "SELECT titel FROM prof WHERE id=?", (prof_id,), one=True - )[0] - return f"{title} " if title is not None else "" - - def getProfByName(self, prof_name: str) -> tuple: - """get all the data of a professor based on the name - - Args: - prof_name (str): the name of the professor - - Returns: - tuple: the data of the professor - """ - return self.query_db( - "SELECT * FROM prof WHERE fullname=?", (prof_name,), one=True - ) - - def getProfId(self, prof_name: str) -> Optional[int]: - """Get the id of a professor based on the name - - Args: - prof_name (str): the name of the professor - - Returns: - Optional[int]: the id of the professor, if the professor is not found, None is returned - """ - - data = self.getProfByName(prof_name.replace(",", "")) - if data is None: - return None - else: - return data[0] - - def getSpecificProfData(self, prof_id: Union[str, int], fields: List[str]) -> tuple: - """A customisable function to get specific data of a professor based on the id - - Args: - prof_id (Union[str,int]): the id of the professor - fields (List[str]): a list of fields to be returned - - Returns: - tuple: a tuple containing the requested data - """ - query = "SELECT " - for field in fields: - query += f"{field}," - query = query[:-1] - query += " FROM prof WHERE id=?" - return self.query_db(query, (prof_id,), one=True)[0] - - def getProfData(self, profname: str): - """Get mail, telephone number and title of a professor based on the name - - Args: - profname (str): name of the professor - - Returns: - tuple: the mail, telephone number and title of the professor - """ - data = self.query_db( - "SELECT mail, telnr, titel FROM prof WHERE fullname=?", - (profname.replace(",", ""),), - one=True, - ) - return data - - def createProf(self, prof_details: dict): - """Create a professor in the database - - Args: - prof_details (dict): a dictionary containing the details of the professor - """ - prof_title = prof_details["prof_title"] - prof_fname = prof_details["profname"].split(",")[1] - prof_fname = prof_fname.strip() - prof_lname = prof_details["profname"].split(",")[0] - prof_lname = prof_lname.strip() - prof_fullname = prof_details["profname"].replace(",", "") - prof_mail = prof_details["prof_mail"] - prof_tel = prof_details["prof_tel"] - params = ( - prof_title, - prof_fname, - prof_lname, - prof_mail, - prof_tel, - prof_fullname, - ) - query = "INSERT OR IGNORE INTO prof (titel, fname, lname, mail, telnr, fullname) VALUES (?, ?, ?, ?, ?, ?)" - self.insertInto(query=query, params=params) - - def getProfs(self) -> list[tuple]: - """Return all the professors in the database - - Returns: - list[tuple]: a list containing all the professors in individual tuples - """ - return self.query_db("SELECT * FROM prof") - - # Apparat - def getAllAparats(self, deleted=0) -> list[tuple]: - """Get all the apparats in the database - - Args: - deleted (int, optional): Switch the result to use . Defaults to 0. - - Returns: - list[tuple]: a list of tuples containing the apparats - """ - return self.query_db( - "SELECT * FROM semesterapparat WHERE deletion_status=?", (deleted,) - ) - - def getApparatData(self, appnr, appname) -> ApparatData: - """Get the Apparat data based on the apparat number and the name - - Args: - appnr (str): the apparat number - appname (str): the name of the apparat - - Raises: - NoResultError: an error is raised if no result is found - - Returns: - ApparatData: the appended data of the apparat wrapped in an ApparatData object - """ - result = self.query_db( - "SELECT * FROM semesterapparat WHERE appnr=? AND name=?", - (appnr, appname), - one=True, - ) - if result is None: - raise NoResultError("No result found") - apparat = ApparatData() - apparat.appname = result[1] - apparat.appnr = result[4] - apparat.dauerapp = True if result[7] == 1 else False - prof_data = self.getProfData(self.getProfNameById(result[2])) - apparat.profname = self.getProfNameById(result[2]) - apparat.prof_mail = prof_data[0] - apparat.prof_tel = prof_data[1] - apparat.prof_title = prof_data[2] - apparat.app_fach = result[3] - apparat.erstellsemester = result[5] - apparat.semester = result[8] - apparat.deleted = result[9] - apparat.apparat_adis_id = result[11] - apparat.prof_adis_id = result[12] - return apparat - - def getUnavailableApparatNumbers(self) -> List[int]: - """Get a list of all the apparat numbers in the database that are currently in use - - Returns: - List[int]: the list of used apparat numbers - """ - numbers = self.query_db( - "SELECT appnr FROM semesterapparat WHERE deletion_status=0" - ) - numbers = [i[0] for i in numbers] - logger.log_info(f"Currently used apparat numbers: {numbers}") - return numbers - - def setNewSemesterDate(self, app_id: Union[str, int], newDate, dauerapp=False): - """Set the new semester date for an apparat - - Args: - app_id (Union[str,int]): the id of the apparat - newDate (str): the new date - dauerapp (bool, optional): if the apparat was changed to dauerapparat. Defaults to False. - """ - date = datetime.datetime.strptime(newDate, "%d.%m.%Y").strftime("%Y-%m-%d") - if dauerapp: - self.query_db( - "UPDATE semesterapparat SET verlängerung_bis=?, dauerapp=? WHERE appnr=?", - (date, dauerapp, app_id), - ) - else: - self.query_db( - "UPDATE semesterapparat SET endsemester=? WHERE appnr=?", (date, app_id) - ) - - def getApparatId(self, apparat_name) -> Optional[int]: - """get the id of an apparat based on the name - - Args: - apparat_name (str): the name of the apparat e.g. "Semesterapparat 1" - - Returns: - Optional[int]: the id of the apparat, if the apparat is not found, None is returned - """ - data = self.query_db( - "SELECT appnr FROM semesterapparat WHERE name=?", (apparat_name,), one=True - ) - if data is None: - return None - else: - return data[0] - - def createApparat(self, apparat: ApparatData) -> int: - """create the apparat in the database - - Args: - apparat (ApparatData): the wrapped metadata of the apparat - - Raises: - AppPresentError: an error describing that the apparats chosen id is already present in the database - - Returns: - Optional[int]: the id of the apparat - """ - - prof_id = self.getProfId(apparat.profname) - app_id = self.getApparatId(apparat.appname) - if app_id: - raise AppPresentError(app_id) - - self.createProf(apparat.get_prof_details()) - prof_id = self.getProfId(apparat.profname) - # ic(prof_id) - query = f"INSERT OR IGNORE INTO semesterapparat (appnr, name, erstellsemester, dauer, prof_id, fach,deletion_status,konto) VALUES ('{apparat.appnr}', '{apparat.appname}', '{apparat.semester}', '{apparat.dauerapp}', {prof_id}, '{apparat.app_fach}', '{0}', '{SEMAP_MEDIA_ACCOUNTS[apparat.appnr]}')" - logger.log_info(query) - self.query_db(query) - return self.getApparatId(apparat.appname) - - def getApparatsByProf(self, prof_id: Union[str, int]) -> list[tuple]: - """Get all apparats based on the professor id - - Args: - prof_id (Union[str,int]): the id of the professor - - Returns: - list[tuple]: a list of tuples containing the apparats - """ - return self.query_db( - "SELECT * FROM semesterapparat WHERE prof_id=?", (prof_id,) - ) - - def getApparatsBySemester(self, semester: str) -> dict[list]: - """get all apparats based on the semester - - Args: - semester (str): the selected semester - - Returns: - dict[list]: a list off all created and deleted apparats for the selected semester - """ - data = self.query_db( - "SELECT name, prof_id FROM semesterapparat WHERE erstellsemester=?", - (semester,), - ) - conn = self.connect() - cursor = conn.cursor() - c_tmp = [] - for i in data: - c_tmp.append((i[0], self.getProfNameById(i[1]))) - query = ( - f"SELECT name,prof_id FROM semesterapparat WHERE deleted_date='{semester}'" - ) - result = cursor.execute(query).fetchall() - d_tmp = [] - for i in result: - d_tmp.append((i[0], self.getProfNameById(i[1]))) - # group the apparats by prof - c_ret = {} - for i in c_tmp: - if i[1] not in c_ret.keys(): - c_ret[i[1]] = [i[0]] - else: - c_ret[i[1]].append(i[0]) - d_ret = {} - for i in d_tmp: - if i[1] not in d_ret.keys(): - d_ret[i[1]] = [i[0]] - else: - d_ret[i[1]].append(i[0]) - self.close_connection(conn) - return {"created": c_ret, "deleted": d_ret} - - def getApparatCountBySemester(self) -> tuple[list[str], list[int]]: - """get a list of all apparats created and deleted by semester - - Returns: - tuple[list[str],list[int]]: a tuple containing two lists, the first list contains the semesters, the second list contains the amount of apparats created and deleted for the corresponding semester - """ - conn = self.connect() - cursor = conn.cursor() - semesters = self.getSemersters() - created = [] - deleted = [] - for semester in semesters: - query = f"SELECT COUNT(*) FROM semesterapparat WHERE erstellsemester='{semester}'" - result = cursor.execute(query).fetchone() - created.append(result[0]) - query = f"SELECT COUNT(*) FROM semesterapparat WHERE deletion_status=1 AND deleted_date='{semester}'" - result = cursor.execute(query).fetchone() - deleted.append(result[0]) - # store data in a tuple - ret = [] - e_tuple = () - for sem in semesters: - e_tuple = ( - sem, - created[semesters.index(sem)], - deleted[semesters.index(sem)], - ) - ret.append(e_tuple) - self.close_connection(conn) - return ret - - def deleteApparat(self, app_id: Union[str, int], semester: str): - """Delete an apparat from the database - - Args: - app_id (Union[str, int]): the id of the apparat - semester (str): the semester the apparat should be deleted from - """ - self.query_db( - "UPDATE semesterapparat SET deletion_status=1, deleted_date=? WHERE appnr=?", - (semester, app_id), - ) - - def isEternal(self, id): - """check if the apparat is eternal (dauerapparat) - - Args: - id (int): the id of the apparat to be checked - - Returns: - int: the state of the apparat - """ - return self.query_db( - "SELECT dauer FROM semesterapparat WHERE appnr=?", (id,), one=True - ) - - def getApparatName(self, app_id: Union[str, int], prof_id: Union[str, int]): - """get the name of the apparat based on the id - - Args: - app_id (Union[str,int]): the id of the apparat - prof_id (Union[str,int]): the id of the professor - - Returns: - str: the name of the apparat - """ - return self.query_db( - "SELECT name FROM semesterapparat WHERE appnr=? AND prof_id=?", - (app_id, prof_id), - one=True, - )[0] - - def updateApparat(self, apparat_data: ApparatData): - """Update an apparat in the database - - Args: - apparat_data (ApparatData): the new metadata of the apparat - """ - query = "UPDATE semesterapparat SET name = ?, fach = ?, dauer = ?, prof_id = ?, prof_id_adis = ?, apparat_id_adis = ? WHERE appnr = ?" - params = ( - apparat_data.appname, - apparat_data.app_fach, - apparat_data.dauerapp, - self.getProfId(apparat_data.profname), - apparat_data.prof_adis_id, - apparat_data.apparat_adis_id, - apparat_data.appnr, - ) - logger.log_info(f"Updating apparat with query {query} and params {params}") - self.query_db(query, params) - - def checkApparatExists(self, apparat_name: str): - """check if the apparat is already present in the database based on the name - - Args: - apparat_name (str): the name of the apparat - - Returns: - bool: True if the apparat is present, False if not - """ - return ( - True - if self.query_db( - "SELECT appnr FROM semesterapparat WHERE name=?", - (apparat_name,), - one=True, - ) - else False - ) - - def checkApparatExistsById(self, app_id: Union[str, int]) -> bool: - """a check to see if the apparat is already present in the database, based on the id - - Args: - app_id (Union[str, int]): the id of the apparat - - Returns: - bool: True if the apparat is present, False if not - """ - return ( - True - if self.query_db( - "SELECT appnr FROM semesterapparat WHERE appnr=?", (app_id,), one=True - ) - else False - ) - - # Statistics - def statistic_request(self, **kwargs: Any): - """Take n amount of kwargs and return the result of the query""" - - def __query(query): - """execute the query and return the result - - Args: - query (str): the constructed query - - Returns: - list: the result of the query - """ - conn = self.connect() - cursor = conn.cursor() - result = cursor.execute(query).fetchall() - for result_a in result: - orig_value = result_a - prof_name = self.getProfNameById(result_a[2]) - # replace the prof_id with the prof_name - result_a = list(result_a) - result_a[2] = prof_name - result_a = tuple(result_a) - result[result.index(orig_value)] = result_a - self.close_connection(conn) - return result - - if "deletable" in kwargs.keys(): - query = f"SELECT * FROM semesterapparat WHERE deletion_status=0 AND dauer=0 AND (erstellsemester!='{kwargs['deletesemester']}' OR verlängerung_bis!='{kwargs['deletesemester']}')" - return __query(query) - if "dauer" in kwargs.keys(): - kwargs["dauer"] = kwargs["dauer"].replace("Ja", "1").replace("Nein", "0") - query = "SELECT * FROM semesterapparat WHERE " - for key, value in kwargs.items() if kwargs.items() is not None else {}: - print(key, value) - query += f"{key}='{value}' AND " - print(query) - # remove deletesemester part from normal query, as this will be added to the database upon deleting the apparat - if "deletesemester" in kwargs.keys(): - query = query.replace( - f"deletesemester='{kwargs['deletesemester']}' AND ", "" - ) - if "endsemester" in kwargs.keys(): - if "erstellsemester" in kwargs.keys(): - query = query.replace(f"endsemester='{kwargs['endsemester']}' AND ", "") - query = query.replace( - f"erstellsemester='{kwargs['erstellsemester']} AND ", "xyz" - ) - else: - query = query.replace( - f"endsemester='{kwargs['endsemester']}' AND ", "xyz" - ) - print("replaced") - query = query.replace( - "xyz", - f"(erstellsemester='{kwargs['endsemester']}' OR verlängerung_bis='{kwargs['endsemester']}') AND ", - ) - # remove all x="" parts from the query where x is a key in kwargs - query = query[:-5] - print(query) - return __query(query) - - # Admin data - def getUser(self): - """Get a single user from the database""" - return self.query_db("SELECT * FROM user", one=True) - - def getUsers(self) -> list[tuple]: - """Return a list of tuples of all the users in the database""" - return self.query_db("SELECT * FROM user") - - def login(self, user, hashed_password): - """try to login the user. - The salt for the user will be requested from the database and then added to the hashed password. The password will then be compared to the password in the database - - Args: - user (str): username that tries to login - hashed_password (str): the password the user tries to login with - - Returns: - bool: True if the login was successful, False if not - """ - salt = self.query_db( - "SELECT salt FROM user WHERE username=?", (user,), one=True - )[0] - if salt is None: - return False - hashed_password = salt + hashed_password - password = self.query_db( - "SELECT password FROM user WHERE username=?", (user,), one=True - )[0] - if password == hashed_password: - return True - else: - return False - - def changePassword(self, user, new_password): - """change the password of a user. - The password will be added with the salt and then committed to the database - - Args: - user (str): username - new_password (str): the hashed password - """ - salt = self.query_db( - "SELECT salt FROM user WHERE username=?", (user,), one=True - )[0] - new_password = salt + new_password - self.query_db( - "UPDATE user SET password=? WHERE username=?", (new_password, user) - ) - - def getRole(self, user): - """get the role of the user - - Args: - user (str): username - - Returns: - str: the name of the role - """ - return self.query_db( - "SELECT role FROM user WHERE username=?", (user,), one=True - )[0] - - def getRoles(self) -> list[tuple]: - """get all the roles in the database - - Returns: - list[str]: a list of all the roles - """ - return self.query_db("SELECT role FROM user") - - def checkUsername(self, user) -> bool: - """a check to see if the username is already present in the database - - Args: - user (str): the username - - Returns: - bool: True if the username is present, False if not - """ - data = self.query_db( - "SELECT username FROM user WHERE username=?", (user,), one=True - ) - return True if data is not None else False - - def createUser(self, user, password, role, salt): - """create an user from the AdminCommands class. - - Args: - user (str): the username of the user - password (str): a hashed password - role (str): the role of the user - salt (str): a salt for the password - """ - self.query_db( - "INSERT OR IGNORE INTO user (username, password, role, salt) VALUES (?,?,?,?)", - (user, password, role, salt), - ) - - def deleteUser(self, user): - """delete an unser - - Args: - user (str): username of the user - """ - self.query_db("DELETE FROM user WHERE username=?", (user,)) - - def updateUser(self, username, data: dict[str, str]): - """changge the data of a user - - Args: - username (str): the username of the user - data (dict[str, str]): the data to be changed - """ - conn = self.connect() - cursor = conn.cursor() - query = "UPDATE user SET " - for key, value in data.items(): - if key == "username": - continue - query += f"{key}='{value}'," - query = query[:-1] - query += " WHERE username=?" - params = (username,) - cursor.execute(query, params) - conn.commit() - self.close_connection(conn) - - def getFacultyMember(self, name: str) -> tuple: - """get a faculty member based on the name - - Args: - name (str): the name to be searched for - - Returns: - tuple: a tuple containing the data of the faculty member - """ - return self.query_db( - "SELECT titel, fname,lname,mail,telnr,fullname FROM prof WHERE fullname=?", - (name,), - one=True, - ) - - def updateFacultyMember(self, data: dict, oldlname: str, oldfname: str): - """update the data of a faculty member - - Args: - data (dict): a dictionary containing the data to be updated - oldlname (str): the old last name of the faculty member - oldfname (str): the old first name of the faculty member - """ - placeholders = ", ".join([f"{i}=:{i} " for i in data.keys()]) - query = f"UPDATE prof SET {placeholders} WHERE lname = :oldlname AND fname = :oldfname" - data["oldlname"] = oldlname - data["oldfname"] = oldfname - self.query_db(query, data) - - def getFacultyMembers(self): - """get a list of all faculty members - - Returns: - list[tuple]: a list of tuples containing the faculty members - """ - return self.query_db("SELECT titel, fname,lname,mail,telnr,fullname FROM prof") - - def restoreApparat(self, app_id: Union[str, int]): - """restore an apparat from the database - - Args: - app_id (Union[str, int]): the id of the apparat - """ - return self.query_db( - "UPDATE semesterapparat SET deletion_status=0, deleted_date=NULL WHERE appnr=?", - (app_id,), - ) +import datetime +import os +import re +import sqlite3 as sql +import tempfile +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple, Union + +# from icecream import ic +from omegaconf import OmegaConf + +from src.backend.db import ( + CREATE_TABLE_APPARAT, + CREATE_TABLE_APPKONTOS, + CREATE_TABLE_FILES, + CREATE_TABLE_MEDIA, + CREATE_TABLE_MESSAGES, + CREATE_TABLE_PROF, + CREATE_TABLE_SUBJECTS, + CREATE_TABLE_USER, +) +from src.errors import AppPresentError, NoResultError +from src.logic.constants import SEMAP_MEDIA_ACCOUNTS +from src.logic.dataclass import ApparatData, BookData +from src.logic.log import MyLogger +from src.utils import create_blob, dump_pickle, load_pickle + +config = OmegaConf.load("config.yaml") +logger = MyLogger(__name__) + + +class Database: + """ + Initialize the database and create the tables if they do not exist. + """ + + def __init__(self, db_path: str = None): + """ + Default constructor for the database class + + Args: + db_path (str, optional): Optional Path for testing / specific purposes. Defaults to None. + """ + if db_path is None: + self.db_path = config.database.path + config.database.name + self.db_path = self.db_path.replace("~", str(Path.home())) + else: + self.db_path = db_path + if self.get_db_contents() is None: + logger.log_critical("Database does not exist, creating tables") + self.create_tables() + + def get_db_contents(self) -> Union[List[Tuple], None]: + """ + Get the contents of the + + Returns: + Union[List[Tuple], None]: _description_ + """ + try: + with sql.connect(self.db_path) as conn: + cursor = conn.cursor() + cursor.execute("SELECT * FROM sqlite_master WHERE type='table'") + return cursor.fetchall() + except sql.OperationalError: + return None + + def connect(self) -> sql.Connection: + """ + Connect to the database + + Returns: + sql.Connection: The active connection to the database + """ + print(self.db_path) + return sql.connect(self.db_path) + + def close_connection(self, conn: sql.Connection): + """ + closes the connection to the database + + Args: + ---- + - conn (sql.Connection): the connection to be closed + """ + conn.close() + + def create_tables(self): + """ + Create the tables in the database + """ + conn = self.connect() + cursor = conn.cursor() + cursor.execute(CREATE_TABLE_APPARAT) + cursor.execute(CREATE_TABLE_MESSAGES) + cursor.execute(CREATE_TABLE_MEDIA) + cursor.execute(CREATE_TABLE_APPKONTOS) + cursor.execute(CREATE_TABLE_FILES) + cursor.execute(CREATE_TABLE_PROF) + cursor.execute(CREATE_TABLE_USER) + cursor.execute(CREATE_TABLE_SUBJECTS) + conn.commit() + self.close_connection(conn) + + def insertInto(self, query: str, params: Tuple) -> None: + """ + Insert sent data into the database + + Args: + query (str): The query to be executed + params (Tuple): the parameters to be inserted into the database + """ + conn = self.connect() + cursor = conn.cursor() + logger.log_info(f"Inserting {params} into database with query {query}") + cursor.execute(query, params) + conn.commit() + self.close_connection(conn) + + def query_db( + self, query: str, args: Tuple = (), one: bool = False + ) -> Union[Tuple, List[Tuple]]: + """ + Query the Database for the sent query. + + Args: + query (str): The query to be executed + args (Tuple, optional): The arguments for the query. Defaults to (). + one (bool, optional): Return the first result only. Defaults to False. + + Returns: + Union[Typle|List[Tuple]]: Returns the result of the query + """ + conn = self.connect() + cursor = conn.cursor() + logger.log_info(f"Querying database with query {query}, args: {args}") + cursor.execute(query, args) + rv = cursor.fetchall() + conn.commit() + self.close_connection(conn) + return (rv[0] if rv else None) if one else rv + + # Books + def addBookToDatabase( + self, bookdata: BookData, app_id: Union[str, int], prof_id: Union[str, int] + ): + """ + Add books to the database. Both app_id and prof_id are required to add the book to the database, as the app_id and prof_id are used to select the books later on. + + Args: + bookdata (BookData): The metadata of the book to be added + app_id (str): The apparat id where the book should be added to + prof_id (str): The id of the professor where the book should be added to. + """ + conn = self.connect() + cursor = conn.cursor() + t_query = ( + f"SELECT bookdata FROM media WHERE app_id={app_id} AND prof_id={prof_id}" + ) + # print(t_query) + result = cursor.execute(t_query).fetchall() + result = [load_pickle(i[0]) for i in result] + if bookdata in result: + print("Bookdata already in database") + # check if the book was deleted in the apparat + query = ( + "SELECT deleted FROM media WHERE app_id=? AND prof_id=? AND bookdata=?" + ) + params = (app_id, prof_id, dump_pickle(bookdata)) + result = cursor.execute(query, params).fetchone() + if result[0] == 1: + print("Book was deleted, updating bookdata") + query = "UPDATE media SET deleted=0 WHERE app_id=? AND prof_id=? AND bookdata=?" + params = (app_id, prof_id, dump_pickle(bookdata)) + cursor.execute(query, params) + conn.commit() + return + + query = ( + "INSERT INTO media (bookdata, app_id, prof_id,deleted) VALUES (?, ?, ?,?)" + ) + converted = dump_pickle(bookdata) + params = (converted, app_id, prof_id, 0) + cursor.execute(query, params) + conn.commit() + self.close_connection(conn) + + def getBookIdBasedOnSignature( + self, app_id: Union[str, int], prof_id: Union[str, int], signature: str + ) -> int: + """ + Get a book id based on the signature of the book. + + Args: + app_id (str): The apparat id the book should be associated with + prof_id (str): The professor id the book should be associated with + signature (str): The signature of the book + + Returns: + int: The id of the book + """ + result = self.query_db( + "SELECT bookdata, id FROM media WHERE app_id=? AND prof_id=?", + (app_id, prof_id), + ) + books = [(load_pickle(i[0]), i[1]) for i in result] + book = [i for i in books if i[0].signature == signature][0][1] + return book + + def getBookBasedOnSignature( + self, app_id: Union[str, int], prof_id: Union[str, int], signature: str + ) -> BookData: + """ + Get the book based on the signature of the book. + + Args: + app_id (str): The apparat id the book should be associated with + prof_id (str): The professor id the book should be associated with + signature (str): The signature of the book + + Returns: + BookData: The total metadata of the book wrapped in a BookData object + """ + result = self.query_db( + "SELECT bookdata FROM media WHERE app_id=? AND prof_id=?", (app_id, prof_id) + ) + books = [load_pickle(i[0]) for i in result] + book = [i for i in books if i.signature == signature][0] + return book + + def getLastBookId(self) -> int: + """ + Get the last book id in the database + + Returns: + int: ID of the last book in the database + """ + return self.query_db("SELECT id FROM media ORDER BY id DESC", one=True)[0] + + def searchBook(self, data: dict[str, str]) -> list[tuple[BookData, int]]: + """ + Search a book in the database based on the sent data. + + Args: + data (dict[str, str]): A dictionary containing the data to be searched for. The dictionary can contain the following: + - signature: The signature of the book + - title: The title of the book + + Returns: + list[tuple[BookData, int]]: A list of tuples containing the wrapped Metadata and the id of the book + """ + rdata = self.query_db("SELECT * FROM media WHERE deleted=0") + # ic(rdata, len(rdata)) + mode = 0 + if len(data) == 1: + if "signature" in data.keys(): + mode = 1 + elif "title" in data.keys(): + mode = 2 + elif len(data) == 2: + mode = 3 + else: + return None + ret = [] + for book in rdata: + bookdata = load_pickle(book[1]) + app_id = book[2] + prof_id = book[3] + if mode == 1: + if data["signature"] in bookdata.signature: + ret.append((bookdata, app_id, prof_id)) + elif mode == 2: + if data["title"] in bookdata.title: + ret.append((bookdata, app_id, prof_id)) + elif mode == 3: + if ( + data["signature"] in bookdata.signature + and data["title"] in bookdata.title + ): + ret.append((bookdata, app_id, prof_id)) + # ic(ret) + return ret + + def setAvailability(self, book_id: str, available: str): + """ + Set the availability of a book in the database + + Args: + book_id (str): The id of the book + available (str): The availability of the book + """ + self.query_db("UPDATE media SET available=? WHERE id=?", (available, book_id)) + + def getBookId( + self, bookdata: BookData, app_id: Union[str, int], prof_id: Union[str, int] + ) -> int: + """ + Get the id of a book based on the metadata of the book + + Args: + bookdata (BookData): The wrapped metadata of the book + app_id (str): The apparat id the book should be associated with + prof_id (str): The professor id the book should be associated with + + Returns: + int: ID of the book + """ + result = self.query_db( + "SELECT id FROM media WHERE bookdata=? AND app_id=? AND prof_id=?", + (dump_pickle(bookdata), app_id, prof_id), + one=True, + ) + return result[0] + + def getBook(self, book_id: int) -> BookData: + """ + Get the book based on the id in the database + + Args: + book_id (int): The id of the book + + Returns: + BookData: The metadata of the book wrapped in a BookData object + """ + return load_pickle( + self.query_db( + "SELECT bookdata FROM media WHERE id=?", (book_id,), one=True + )[0] + ) + + def getBooks( + self, app_id: Union[str, int], prof_id: Union[str, int], deleted=0 + ) -> list[dict[int, BookData, int]]: + """ + Get the Books based on the apparat id and the professor id + + Args: + app_id (str): The ID of the apparat + prof_id (str): The ID of the professor + deleted (int, optional): The state of the book. Set to 1 to include deleted ones. Defaults to 0. + + Returns: + list[dict[int, BookData, int]]: A list of dictionaries containing the id, the metadata of the book and the availability of the book + """ + qdata = self.query_db( + f"SELECT id,bookdata,available FROM media WHERE (app_id={app_id} AND prof_id={prof_id}) AND (deleted={deleted if deleted == 0 else '1 OR deleted=0'})" + ) + ret_result = [] + for result_a in qdata: + data = {"id": int, "bookdata": BookData, "available": int} + data["id"] = result_a[0] + data["bookdata"] = load_pickle(result_a[1]) + data["available"] = result_a[2] + ret_result.append(data) + return ret_result + + def updateBookdata(self, book_id, bookdata: BookData): + """ + Update the bookdata in the database + + Args: + book_id (str): The id of the book + bookdata (BookData): The new metadata of the book + """ + self.query_db( + "UPDATE media SET bookdata=? WHERE id=?", (dump_pickle(bookdata), book_id) + ) + + def deleteBook(self, book_id): + """ + Delete a book from the database + + Args: + book_id (str): ID of the book + """ + self.query_db("UPDATE media SET deleted=1 WHERE id=?", (book_id,)) + + # File Interactions + def getBlob(self, filename, app_id: Union[str, int]): + """ + Get a blob from the database + + Args: + filename (str): The name of the file + app_id (str): ID of the apparat + + Returns: + bytes: The file stored in + """ + return self.query_db( + "SELECT fileblob FROM files WHERE filename=? AND app_id=?", + (filename, app_id), + one=True, + )[0] + + def insertFile( + self, file: list[dict], app_id: Union[str, int], prof_id: Union[str, int] + ): + """Instert a list of files into the database + + Args: + file (list[dict]): a list containing all the files to be inserted + Structured: [{"name": "filename", "path": "path", "type": "filetype"}] + app_id (int): the id of the apparat + prof_id (str): the id of the professor + """ + for f in file: + filename = f["name"] + path = f["path"] + filetyp = f["type"] + if path == "Database": + continue + blob = create_blob(path) + query = "INSERT OR IGNORE INTO files (filename, fileblob, app_id, filetyp,prof_id) VALUES (?, ?, ?, ?,?)" + self.query_db(query, (filename, blob, app_id, filetyp, prof_id)) + + def recreateFile( + self, filename: str, app_id: Union[str, int], filetype: str + ) -> str: + """Recreate a file from the database + + Args: + filename (str): the name of the file + app_id (Union[str,int]): the id of the apparat + filetype (str): the extension of the file to be created + + Returns: + str: The filename of the recreated file + """ + blob = self.getBlob(filename, app_id) + tempdir = config.database.tempdir + tempdir = tempdir.replace("~", str(Path.home())) + tempdir_path = Path(tempdir) + if not os.path.exists(tempdir_path): + os.mkdir(tempdir_path) + file = tempfile.NamedTemporaryFile( + delete=False, dir=tempdir_path, mode="wb", suffix=f".{filetype}" + ) + file.write(blob) + print("file created") + return file.name + + def getFiles(self, app_id: Union[str, int], prof_id: int) -> list[tuple]: + """Get all the files associated with the apparat and the professor + + Args: + app_id (Union[str,int]): The id of the apparat + prof_id (Union[str,int]): the id of the professor + + Returns: + list[tuple]: a list of tuples containing the filename and the filetype for the corresponding apparat and professor + """ + return self.query_db( + "SELECT filename, filetyp FROM files WHERE app_id=? AND prof_id=?", + (app_id, prof_id), + ) + + def getSemersters(self) -> list[str]: + """Return all the unique semesters in the database + + Returns: + list: a list of strings containing the semesters + """ + data = self.query_db("SELECT DISTINCT erstellsemester FROM semesterapparat") + return [i[0] for i in data] + + def getSubjects(self): + """Get all the subjects in the database + + Returns: + list[tuple]: a list of tuples containing the subjects + """ + return self.query_db("SELECT * FROM subjects") + + # Messages + def addMessage(self, message: dict, user: str, app_id: Union[str, int]): + """add a Message to the database + + Args: + message (dict): the message to be added + user (str): the user who added the message + app_id (Union[str,int]): the id of the apparat + """ + + def __getUserId(user): + return self.query_db( + "SELECT id FROM user WHERE username=?", (user,), one=True + )[0] + + user_id = __getUserId(user) + self.query_db( + "INSERT INTO messages (message, user_id, remind_at,appnr) VALUES (?,?,?,?)", + (message["message"], user_id, message["remind_at"], app_id), + ) + + def getMessages(self, date: str) -> list[dict[str, str, str, str]]: + """Get all the messages for a specific date + + Args: + date (str): a date.datetime object formatted as a string in the format "YYYY-MM-DD" + + Returns: + list[dict[str, str, str, str]]: a list of dictionaries containing the message, the user who added the message, the apparat id and the id of the message + """ + + def __get_user_name(user_id): + return self.query_db( + "SELECT username FROM user WHERE id=?", (user_id,), one=True + )[0] + + messages = self.query_db("SELECT * FROM messages WHERE remind_at=?", (date,)) + ret = [ + {"message": i[2], "user": __get_user_name(i[4]), "appnr": i[5], "id": i[0]} + for i in messages + ] + return ret + + def deleteMessage(self, message_id): + """Delete a message from the database + + Args: + message_id (str): the id of the message + """ + self.query_db("DELETE FROM messages WHERE id=?", (message_id,)) + + # Prof data + def getProfNameById(self, prof_id: Union[str, int], add_title: bool = False) -> str: + """Get a professor name based on the id + + Args: + prof_id (Union[str,int]): The id of the professor + add_title (bool, optional): wether to add the title or no. Defaults to False. + + Returns: + str: The name of the professor + """ + prof = self.query_db( + "SELECT fullname FROM prof WHERE id=?", (prof_id,), one=True + ) + if add_title: + return f"{self.getTitleById(prof_id)}{prof[0]}" + else: + return prof[0] + + def getTitleById(self, prof_id: Union[str, int]) -> str: + """get the title of a professor based on the id + + Args: + prof_id (Union[str,int]): the id of the professor + + Returns: + str: the title of the professor, with an added whitespace at the end, if no title is present, an empty string is returned + """ + title = self.query_db( + "SELECT titel FROM prof WHERE id=?", (prof_id,), one=True + )[0] + return f"{title} " if title is not None else "" + + def getProfByName(self, prof_name: str) -> tuple: + """get all the data of a professor based on the name + + Args: + prof_name (str): the name of the professor + + Returns: + tuple: the data of the professor + """ + return self.query_db( + "SELECT * FROM prof WHERE fullname=?", (prof_name,), one=True + ) + + def getProfId(self, prof_name: str) -> Optional[int]: + """Get the id of a professor based on the name + + Args: + prof_name (str): the name of the professor + + Returns: + Optional[int]: the id of the professor, if the professor is not found, None is returned + """ + + data = self.getProfByName(prof_name.replace(",", "")) + if data is None: + return None + else: + return data[0] + + def getSpecificProfData(self, prof_id: Union[str, int], fields: List[str]) -> tuple: + """A customisable function to get specific data of a professor based on the id + + Args: + prof_id (Union[str,int]): the id of the professor + fields (List[str]): a list of fields to be returned + + Returns: + tuple: a tuple containing the requested data + """ + query = "SELECT " + for field in fields: + query += f"{field}," + query = query[:-1] + query += " FROM prof WHERE id=?" + return self.query_db(query, (prof_id,), one=True)[0] + + def getProfData(self, profname: str): + """Get mail, telephone number and title of a professor based on the name + + Args: + profname (str): name of the professor + + Returns: + tuple: the mail, telephone number and title of the professor + """ + data = self.query_db( + "SELECT mail, telnr, titel FROM prof WHERE fullname=?", + (profname.replace(",", ""),), + one=True, + ) + return data + + def createProf(self, prof_details: dict): + """Create a professor in the database + + Args: + prof_details (dict): a dictionary containing the details of the professor + """ + prof_title = prof_details["prof_title"] + prof_fname = prof_details["profname"].split(",")[1] + prof_fname = prof_fname.strip() + prof_lname = prof_details["profname"].split(",")[0] + prof_lname = prof_lname.strip() + prof_fullname = prof_details["profname"].replace(",", "") + prof_mail = prof_details["prof_mail"] + prof_tel = prof_details["prof_tel"] + params = ( + prof_title, + prof_fname, + prof_lname, + prof_mail, + prof_tel, + prof_fullname, + ) + query = "INSERT OR IGNORE INTO prof (titel, fname, lname, mail, telnr, fullname) VALUES (?, ?, ?, ?, ?, ?)" + self.insertInto(query=query, params=params) + + def getProfs(self) -> list[tuple]: + """Return all the professors in the database + + Returns: + list[tuple]: a list containing all the professors in individual tuples + """ + return self.query_db("SELECT * FROM prof") + + # Apparat + def getAllAparats(self, deleted=0) -> list[tuple]: + """Get all the apparats in the database + + Args: + deleted (int, optional): Switch the result to use . Defaults to 0. + + Returns: + list[tuple]: a list of tuples containing the apparats + """ + return self.query_db( + "SELECT * FROM semesterapparat WHERE deletion_status=?", (deleted,) + ) + + def getApparatData(self, appnr, appname) -> ApparatData: + """Get the Apparat data based on the apparat number and the name + + Args: + appnr (str): the apparat number + appname (str): the name of the apparat + + Raises: + NoResultError: an error is raised if no result is found + + Returns: + ApparatData: the appended data of the apparat wrapped in an ApparatData object + """ + result = self.query_db( + "SELECT * FROM semesterapparat WHERE appnr=? AND name=?", + (appnr, appname), + one=True, + ) + if result is None: + raise NoResultError("No result found") + apparat = ApparatData() + apparat.appname = result[1] + apparat.appnr = result[4] + apparat.dauerapp = True if result[7] == 1 else False + prof_data = self.getProfData(self.getProfNameById(result[2])) + apparat.profname = self.getProfNameById(result[2]) + apparat.prof_mail = prof_data[0] + apparat.prof_tel = prof_data[1] + apparat.prof_title = prof_data[2] + apparat.app_fach = result[3] + apparat.erstellsemester = result[5] + apparat.semester = result[8] + apparat.deleted = result[9] + apparat.apparat_adis_id = result[11] + apparat.prof_adis_id = result[12] + return apparat + + def getUnavailableApparatNumbers(self) -> List[int]: + """Get a list of all the apparat numbers in the database that are currently in use + + Returns: + List[int]: the list of used apparat numbers + """ + numbers = self.query_db( + "SELECT appnr FROM semesterapparat WHERE deletion_status=0" + ) + numbers = [i[0] for i in numbers] + logger.log_info(f"Currently used apparat numbers: {numbers}") + return numbers + + def setNewSemesterDate(self, app_id: Union[str, int], newDate, dauerapp=False): + """Set the new semester date for an apparat + + Args: + app_id (Union[str,int]): the id of the apparat + newDate (str): the new date + dauerapp (bool, optional): if the apparat was changed to dauerapparat. Defaults to False. + """ + + if dauerapp: + self.query_db( + "UPDATE semesterapparat SET verlängerung_bis=?, dauerapp=? WHERE appnr=?", + (newDate, dauerapp, app_id), + ) + else: + self.query_db( + "UPDATE semesterapparat SET verlängerung_bis=? WHERE appnr=?", + (newDate, app_id), + ) + + def getApparatId(self, apparat_name) -> Optional[int]: + """get the id of an apparat based on the name + + Args: + apparat_name (str): the name of the apparat e.g. "Semesterapparat 1" + + Returns: + Optional[int]: the id of the apparat, if the apparat is not found, None is returned + """ + data = self.query_db( + "SELECT appnr FROM semesterapparat WHERE name=?", (apparat_name,), one=True + ) + if data is None: + return None + else: + return data[0] + + def createApparat(self, apparat: ApparatData) -> int: + """create the apparat in the database + + Args: + apparat (ApparatData): the wrapped metadata of the apparat + + Raises: + AppPresentError: an error describing that the apparats chosen id is already present in the database + + Returns: + Optional[int]: the id of the apparat + """ + + prof_id = self.getProfId(apparat.profname) + app_id = self.getApparatId(apparat.appname) + if app_id: + raise AppPresentError(app_id) + + self.createProf(apparat.get_prof_details()) + prof_id = self.getProfId(apparat.profname) + # ic(prof_id) + query = f"INSERT OR IGNORE INTO semesterapparat (appnr, name, erstellsemester, dauer, prof_id, fach,deletion_status,konto) VALUES ('{apparat.appnr}', '{apparat.appname}', '{apparat.semester}', '{apparat.dauerapp}', {prof_id}, '{apparat.app_fach}', '{0}', '{SEMAP_MEDIA_ACCOUNTS[apparat.appnr]}')" + logger.log_info(query) + self.query_db(query) + return self.getApparatId(apparat.appname) + + def getApparatsByProf(self, prof_id: Union[str, int]) -> list[tuple]: + """Get all apparats based on the professor id + + Args: + prof_id (Union[str,int]): the id of the professor + + Returns: + list[tuple]: a list of tuples containing the apparats + """ + return self.query_db( + "SELECT * FROM semesterapparat WHERE prof_id=?", (prof_id,) + ) + + def getApparatsBySemester(self, semester: str) -> dict[list]: + """get all apparats based on the semester + + Args: + semester (str): the selected semester + + Returns: + dict[list]: a list off all created and deleted apparats for the selected semester + """ + data = self.query_db( + "SELECT name, prof_id FROM semesterapparat WHERE erstellsemester=?", + (semester,), + ) + conn = self.connect() + cursor = conn.cursor() + c_tmp = [] + for i in data: + c_tmp.append((i[0], self.getProfNameById(i[1]))) + query = ( + f"SELECT name,prof_id FROM semesterapparat WHERE deleted_date='{semester}'" + ) + result = cursor.execute(query).fetchall() + d_tmp = [] + for i in result: + d_tmp.append((i[0], self.getProfNameById(i[1]))) + # group the apparats by prof + c_ret = {} + for i in c_tmp: + if i[1] not in c_ret.keys(): + c_ret[i[1]] = [i[0]] + else: + c_ret[i[1]].append(i[0]) + d_ret = {} + for i in d_tmp: + if i[1] not in d_ret.keys(): + d_ret[i[1]] = [i[0]] + else: + d_ret[i[1]].append(i[0]) + self.close_connection(conn) + return {"created": c_ret, "deleted": d_ret} + + def getApparatCountBySemester(self) -> tuple[list[str], list[int]]: + """get a list of all apparats created and deleted by semester + + Returns: + tuple[list[str],list[int]]: a tuple containing two lists, the first list contains the semesters, the second list contains the amount of apparats created and deleted for the corresponding semester + """ + conn = self.connect() + cursor = conn.cursor() + semesters = self.getSemersters() + created = [] + deleted = [] + for semester in semesters: + query = f"SELECT COUNT(*) FROM semesterapparat WHERE erstellsemester='{semester}'" + result = cursor.execute(query).fetchone() + created.append(result[0]) + query = f"SELECT COUNT(*) FROM semesterapparat WHERE deletion_status=1 AND deleted_date='{semester}'" + result = cursor.execute(query).fetchone() + deleted.append(result[0]) + # store data in a tuple + ret = [] + e_tuple = () + for sem in semesters: + e_tuple = ( + sem, + created[semesters.index(sem)], + deleted[semesters.index(sem)], + ) + ret.append(e_tuple) + self.close_connection(conn) + return ret + + def deleteApparat(self, app_id: Union[str, int], semester: str): + """Delete an apparat from the database + + Args: + app_id (Union[str, int]): the id of the apparat + semester (str): the semester the apparat should be deleted from + """ + self.query_db( + "UPDATE semesterapparat SET deletion_status=1, deleted_date=? WHERE appnr=?", + (semester, app_id), + ) + + def isEternal(self, id): + """check if the apparat is eternal (dauerapparat) + + Args: + id (int): the id of the apparat to be checked + + Returns: + int: the state of the apparat + """ + return self.query_db( + "SELECT dauer FROM semesterapparat WHERE appnr=?", (id,), one=True + ) + + def getApparatName(self, app_id: Union[str, int], prof_id: Union[str, int]): + """get the name of the apparat based on the id + + Args: + app_id (Union[str,int]): the id of the apparat + prof_id (Union[str,int]): the id of the professor + + Returns: + str: the name of the apparat + """ + return self.query_db( + "SELECT name FROM semesterapparat WHERE appnr=? AND prof_id=?", + (app_id, prof_id), + one=True, + )[0] + + def updateApparat(self, apparat_data: ApparatData): + """Update an apparat in the database + + Args: + apparat_data (ApparatData): the new metadata of the apparat + """ + query = "UPDATE semesterapparat SET name = ?, fach = ?, dauer = ?, prof_id = ?, prof_id_adis = ?, apparat_id_adis = ? WHERE appnr = ?" + params = ( + apparat_data.appname, + apparat_data.app_fach, + apparat_data.dauerapp, + self.getProfId(apparat_data.profname), + apparat_data.prof_adis_id, + apparat_data.apparat_adis_id, + apparat_data.appnr, + ) + logger.log_info(f"Updating apparat with query {query} and params {params}") + self.query_db(query, params) + + def checkApparatExists(self, apparat_name: str): + """check if the apparat is already present in the database based on the name + + Args: + apparat_name (str): the name of the apparat + + Returns: + bool: True if the apparat is present, False if not + """ + return ( + True + if self.query_db( + "SELECT appnr FROM semesterapparat WHERE name=?", + (apparat_name,), + one=True, + ) + else False + ) + + def checkApparatExistsById(self, app_id: Union[str, int]) -> bool: + """a check to see if the apparat is already present in the database, based on the id + + Args: + app_id (Union[str, int]): the id of the apparat + + Returns: + bool: True if the apparat is present, False if not + """ + return ( + True + if self.query_db( + "SELECT appnr FROM semesterapparat WHERE appnr=?", (app_id,), one=True + ) + else False + ) + + # Statistics + def statistic_request(self, **kwargs: Any): + """Take n amount of kwargs and return the result of the query""" + + def __query(query): + """execute the query and return the result + + Args: + query (str): the constructed query + + Returns: + list: the result of the query + """ + conn = self.connect() + cursor = conn.cursor() + result = cursor.execute(query).fetchall() + for result_a in result: + orig_value = result_a + prof_name = self.getProfNameById(result_a[2]) + # replace the prof_id with the prof_name + result_a = list(result_a) + result_a[2] = prof_name + result_a = tuple(result_a) + result[result.index(orig_value)] = result_a + self.close_connection(conn) + return result + + if "deletable" in kwargs.keys(): + query = f"SELECT * FROM semesterapparat WHERE deletion_status=0 AND dauer=0 AND (erstellsemester!='{kwargs['deletesemester']}' OR verlängerung_bis!='{kwargs['deletesemester']}')" + return __query(query) + if "dauer" in kwargs.keys(): + kwargs["dauer"] = kwargs["dauer"].replace("Ja", "1").replace("Nein", "0") + query = "SELECT * FROM semesterapparat WHERE " + for key, value in kwargs.items() if kwargs.items() is not None else {}: + print(key, value) + query += f"{key}='{value}' AND " + print(query) + # remove deletesemester part from normal query, as this will be added to the database upon deleting the apparat + if "deletesemester" in kwargs.keys(): + query = query.replace( + f"deletesemester='{kwargs['deletesemester']}' AND ", "" + ) + if "endsemester" in kwargs.keys(): + if "erstellsemester" in kwargs.keys(): + query = query.replace(f"endsemester='{kwargs['endsemester']}' AND ", "") + query = query.replace( + f"erstellsemester='{kwargs['erstellsemester']} AND ", "xyz" + ) + else: + query = query.replace( + f"endsemester='{kwargs['endsemester']}' AND ", "xyz" + ) + print("replaced") + query = query.replace( + "xyz", + f"(erstellsemester='{kwargs['endsemester']}' OR verlängerung_bis='{kwargs['endsemester']}') AND ", + ) + # remove all x="" parts from the query where x is a key in kwargs + query = query[:-5] + print(query) + return __query(query) + + # Admin data + def getUser(self): + """Get a single user from the database""" + return self.query_db("SELECT * FROM user", one=True) + + def getUsers(self) -> list[tuple]: + """Return a list of tuples of all the users in the database""" + return self.query_db("SELECT * FROM user") + + def login(self, user, hashed_password): + """try to login the user. + The salt for the user will be requested from the database and then added to the hashed password. The password will then be compared to the password in the database + + Args: + user (str): username that tries to login + hashed_password (str): the password the user tries to login with + + Returns: + bool: True if the login was successful, False if not + """ + salt = self.query_db( + "SELECT salt FROM user WHERE username=?", (user,), one=True + )[0] + if salt is None: + return False + hashed_password = salt + hashed_password + password = self.query_db( + "SELECT password FROM user WHERE username=?", (user,), one=True + )[0] + if password == hashed_password: + return True + else: + return False + + def changePassword(self, user, new_password): + """change the password of a user. + The password will be added with the salt and then committed to the database + + Args: + user (str): username + new_password (str): the hashed password + """ + salt = self.query_db( + "SELECT salt FROM user WHERE username=?", (user,), one=True + )[0] + new_password = salt + new_password + self.query_db( + "UPDATE user SET password=? WHERE username=?", (new_password, user) + ) + + def getRole(self, user): + """get the role of the user + + Args: + user (str): username + + Returns: + str: the name of the role + """ + return self.query_db( + "SELECT role FROM user WHERE username=?", (user,), one=True + )[0] + + def getRoles(self) -> list[tuple]: + """get all the roles in the database + + Returns: + list[str]: a list of all the roles + """ + return self.query_db("SELECT role FROM user") + + def checkUsername(self, user) -> bool: + """a check to see if the username is already present in the database + + Args: + user (str): the username + + Returns: + bool: True if the username is present, False if not + """ + data = self.query_db( + "SELECT username FROM user WHERE username=?", (user,), one=True + ) + return True if data is not None else False + + def createUser(self, user, password, role, salt): + """create an user from the AdminCommands class. + + Args: + user (str): the username of the user + password (str): a hashed password + role (str): the role of the user + salt (str): a salt for the password + """ + self.query_db( + "INSERT OR IGNORE INTO user (username, password, role, salt) VALUES (?,?,?,?)", + (user, password, role, salt), + ) + + def deleteUser(self, user): + """delete an unser + + Args: + user (str): username of the user + """ + self.query_db("DELETE FROM user WHERE username=?", (user,)) + + def updateUser(self, username, data: dict[str, str]): + """changge the data of a user + + Args: + username (str): the username of the user + data (dict[str, str]): the data to be changed + """ + conn = self.connect() + cursor = conn.cursor() + query = "UPDATE user SET " + for key, value in data.items(): + if key == "username": + continue + query += f"{key}='{value}'," + query = query[:-1] + query += " WHERE username=?" + params = (username,) + cursor.execute(query, params) + conn.commit() + self.close_connection(conn) + + def getFacultyMember(self, name: str) -> tuple: + """get a faculty member based on the name + + Args: + name (str): the name to be searched for + + Returns: + tuple: a tuple containing the data of the faculty member + """ + return self.query_db( + "SELECT titel, fname,lname,mail,telnr,fullname FROM prof WHERE fullname=?", + (name,), + one=True, + ) + + def updateFacultyMember(self, data: dict, oldlname: str, oldfname: str): + """update the data of a faculty member + + Args: + data (dict): a dictionary containing the data to be updated + oldlname (str): the old last name of the faculty member + oldfname (str): the old first name of the faculty member + """ + placeholders = ", ".join([f"{i}=:{i} " for i in data.keys()]) + query = f"UPDATE prof SET {placeholders} WHERE lname = :oldlname AND fname = :oldfname" + data["oldlname"] = oldlname + data["oldfname"] = oldfname + self.query_db(query, data) + + def getFacultyMembers(self): + """get a list of all faculty members + + Returns: + list[tuple]: a list of tuples containing the faculty members + """ + return self.query_db("SELECT titel, fname,lname,mail,telnr,fullname FROM prof") + + def restoreApparat(self, app_id: Union[str, int]): + """restore an apparat from the database + + Args: + app_id (Union[str, int]): the id of the apparat + """ + return self.query_db( + "UPDATE semesterapparat SET deletion_status=0, deleted_date=NULL WHERE appnr=?", + (app_id,), + )