import os import inspect 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_ELSA_FILES_TABLE, CREATE_ELSA_MEDIA_TABLE, CREATE_ELSA_PROF_TABLE, CREATE_ELSA_TABLE, 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") ascii_lowercase = "abcdefghijklmnopqrstuvwxyz0123456789)" 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. """ caller_frame = inspect.stack()[1] script_name = ( caller_frame.filename.replace("\\", "/").split("/")[-1].split(".")[0] ) print(script_name) name = f"Database.{script_name}" self.logger = MyLogger(name) 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() == []: self.logger.log_critical("Database does not exist, creating tables") self.create_tables() self.insertSubjects() 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 """ 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) cursor.execute(CREATE_ELSA_TABLE) cursor.execute(CREATE_ELSA_PROF_TABLE) cursor.execute(CREATE_ELSA_FILES_TABLE) cursor.execute(CREATE_ELSA_MEDIA_TABLE) 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() self.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[Tuple | List[Tuple]]: Returns the result of the query """ conn = self.connect() cursor = conn.cursor() self.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 insertSubjects(self): print("Inserting subjects") subjects = [ "Biologie", "Chemie", "Deutsch", "Englisch", "Erziehungswissenschaft", "Französisch", "Geographie", "Geschichte", "Gesundheitspädagogik", "Haushalt / Textil", "Kunst", "Mathematik / Informatik", "Medien in der Bildung", "Musik", "Philosophie", "Physik", "Poitikwissenschaft", "Prorektorat Lehre und Studium", "Psychologie", "Soziologie", "Sport", "Technik", "Theologie", "Wirtschaftslehre", ] conn = self.connect() cursor = conn.cursor() for subject in subjects: cursor.execute("INSERT INTO subjects (name) VALUES (?)", (subject,)) conn.commit() self.close_connection(conn) 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 getAllMessages(self) -> list[dict[str, str, str, str]]: """Get all the messages in the database 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") ret = [ { "message": i[2], "user": __get_user_name(i[4]), "appnr": i[5], "id": i[0], "remind_at": i[3], } for i in messages ] return ret 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.logger.log_info(f"Deleting message with id {message_id}") 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 tuple: (id, titel, fname, lname, fullname, mail, telnr) """ 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] self.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]}')" self.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, ) self.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) self.logger.log_info(f"Query result: {result}") 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] query = query.strip() # check if query ends with lowercase letter or a '. if not, remove last symbol and try again while query[-1] not in ascii_lowercase and query[-1] != "'": query = query[:-1] query = query.strip() print(query) res = __query(query) print(res) return res # 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,), ) # ELSA def createElsaProf(self, name): """create a new professor in the database for the ELSA system Args: name (str): the name of the professor """ self.query_db("INSERT INTO elsa_prof (name) VALUES (?)", (name,)) def createElsaApparat(self, date, prof_id, semester): """create a new apparat in the database for the ELSA system Args: date (str): the name of the apparat prof_id (int): the id of the professor semester (str): the semester the apparat is created in """ self.query_db( "INSERT INTO elsa (date, prof_id, semester) VALUES (?,?,?)", (date, prof_id, semester), ) def addElsaMedia(self, data: dict, elsa_id: int): """add a media to the ELSA system Args: data (dict): a dictionary containing the data of the media, Structured: {"chapter":str, "title":str, "signature":str, "pages":str} """ self.query_db( "INSERT INTO elsa_media (chapter, title, signature, pages, elsa_id) VALUES (?,?,?,?,?)", (data["chapter"], data["title"], data["signature"], data["pages"], elsa_id), ) def getElsaMedia(self, elsa_id: int): """get all the media of an ELSA apparat Args: elsa_id (int): the id of the ELSA apparat Returns: list[tuple]: a list of tuples containing the media """ return self.query_db("SELECT * FROM elsa_media WHERE elsa_id=?", (elsa_id,)) def insertElsaFile(self, file: list[dict], elsa_id: int): """Instert a list of files into the ELSA system Args: file (list[dict]): a list containing all the files to be inserted Structured: [{"name": "filename", "path": "path", "type": "filetype"}] elsa_id (int): the id of the ELSA apparat """ 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 elsa_files (filename, fileblob, elsa_id, filetyp) VALUES (?, ?, ?, ?)" self.query_db(query, (filename, blob, elsa_id, filetyp)) def recreateElsaFile(self, filename: str, filetype: str) -> str: """Recreate a file from the ELSA system Args: filename (str): the name of the file elsa_id (int): the id of the ELSA apparat filetype (str): the extension of the file to be created Returns: str: The filename of the recreated file """ blob = self.query_db( "SELECT fileblob FROM elsa_files WHERE filename=?", (filename,), one=True )[0] print(blob) 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 getElsaApparats(self): """Get all the ELSA apparats in the database Returns: list[tuple]: a list of tuples containing the ELSA apparats """ return self.query_db("SELECT * FROM elsa") def getElsaId(self, prof, semester, date): """get the id of an ELSA apparat based on the professor, semester and date Args: prof (str): the name of the professor semester (str): the semester date (str): the date of the apparat Returns: int: the id of the ELSA apparat """ data = self.query_db( "SELECT id FROM elsa WHERE prof_id=? AND semester=? AND date=?", (prof, semester, date), one=True, ) if data is None: return None return data[0] def getElsaFiles(self, elsa_id: int): """get all the files of an ELSA apparat Args: elsa_id (int): the id of the ELSA apparat Returns: list[tuple]: a list of tuples containing the files """ return self.query_db( "SELECT filename, filetyp FROM elsa_files WHERE elsa_id=?", (elsa_id,) )