1514 lines
53 KiB
Python
1514 lines
53 KiB
Python
import os
|
|
import inspect
|
|
import sqlite3 as sql
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Tuple, Union
|
|
import shutil
|
|
# 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
|
|
from src.backend.semester import generateSemesterByDate
|
|
from icecream import ic
|
|
|
|
config = OmegaConf.load("config.yaml")
|
|
ascii_lowercase = "abcdefghijklmnopqrstuvwxyz0123456789)"
|
|
caller_frame = inspect.stack()[1]
|
|
# get the line that called the function
|
|
|
|
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]
|
|
)
|
|
name = f"Database.{script_name}"
|
|
self.name = 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
|
|
self.checkDatabaseStatus()
|
|
|
|
def checkDatabaseStatus(self):
|
|
path = config.database.path
|
|
path = path.replace("~", str(Path.home()))
|
|
# print(path)
|
|
path = os.path.abspath(path)
|
|
if not os.path.exists(path):
|
|
# create path
|
|
# print(path)
|
|
os.makedirs(path)
|
|
if self.get_db_contents() == []:
|
|
self.logger.log_critical("Database does not exist, creating tables")
|
|
self.create_tables()
|
|
self.insertSubjects()
|
|
|
|
def get_caller_line(self) -> str:
|
|
# get the line from the script that called the function
|
|
caller_frame = inspect.stack()
|
|
# get the line that called the function based on self.name and caller_frame
|
|
script_name = self.name
|
|
for frame in caller_frame:
|
|
if script_name in frame.filename:
|
|
caller_frame = frame
|
|
break
|
|
line = f"{caller_frame.function}:{caller_frame.lineno} ->"
|
|
return line
|
|
|
|
def getElsaMediaID(self, work_author, signature, pages):
|
|
query = (
|
|
"SELECT id FROM elsa_media WHERE work_author=? AND signature=? AND pages=?"
|
|
)
|
|
params = (work_author, signature, pages)
|
|
result = self.query_db(query, params, one=True)
|
|
if result is None:
|
|
return NoResultError(
|
|
f"work_author: {work_author}, signature: {signature}, pages: {pages}"
|
|
).__str__()
|
|
return result[0]
|
|
|
|
def getElsaMediaType(self, id):
|
|
query = "SELECT type FROM elsa_media WHERE id=?"
|
|
return self.query_db(query, (id,), one=True)[0]
|
|
|
|
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()
|
|
log_message = f"{self.get_caller_line()} Querying database with query {query}, args: {args}"
|
|
# if "INSERT" in query:
|
|
# log_message = f"Querying database with query {query}"
|
|
|
|
self.logger.log_info(log_message)
|
|
try:
|
|
cursor.execute(query, args)
|
|
rv = cursor.fetchall()
|
|
conn.commit()
|
|
self.close_connection(conn)
|
|
except sql.OperationalError as e:
|
|
self.logger.log_error(f"Error in query: {e}")
|
|
return None
|
|
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}"
|
|
)
|
|
self.logger.log_info(t_query)
|
|
# # 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)
|
|
logMessage = f"Added book with signature {bookdata.signature} to database, data: {converted}"
|
|
self.logger.log_info(logMessage)
|
|
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 = []
|
|
if qdata is None:
|
|
return []
|
|
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
|
|
"""
|
|
query = "UPDATE media SET bookdata= ? WHERE id=?"
|
|
book = dump_pickle(bookdata)
|
|
self.query_db(query, (book, 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 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 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=?, dauer=? 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
|
|
"""
|
|
ic(apparat)
|
|
prof_id = self.getProfByName(apparat.get_prof_details()["fullname"])
|
|
ic(prof_id, apparat.profname)
|
|
if prof_id:
|
|
prof_id = prof_id[0]
|
|
app_id = self.getApparatId(apparat.appname)
|
|
if app_id:
|
|
return AppPresentError(app_id)
|
|
if not prof_id:
|
|
prof_id = self.createProf(apparat.get_prof_details())
|
|
# 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 None
|
|
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, app_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=?",
|
|
(app_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
|
|
"""
|
|
ic(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']}' AND verlängerung_bis IS NULL) OR
|
|
(erstellsemester!='{kwargs['deletesemester']}' AND verlängerung_bis!='{kwargs['deletesemester']}' AND verlängerung_bis!='{generateSemesterByDate(True)}')
|
|
)"""
|
|
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 OR IGNORE INTO elsa (date, prof_id, semester) VALUES (?,?,?)",
|
|
(date, prof_id, semester),
|
|
)
|
|
|
|
def updateElsaApparat(self, date, prof_id, semester, elsa_id):
|
|
"""update an ELSA apparat in the database
|
|
|
|
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
|
|
elsa_id (int): the id of the ELSA apparat
|
|
"""
|
|
self.query_db(
|
|
"UPDATE elsa SET date=?, prof_id=?, semester=? WHERE id=?",
|
|
(date, prof_id, semester, elsa_id),
|
|
)
|
|
|
|
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,
|
|
elsa_id (int): the id of the ELSA apparat
|
|
"""
|
|
headers = []
|
|
entries = []
|
|
for key, value in data.items():
|
|
headers.append(key)
|
|
entries.append(value)
|
|
headers.append("elsa_id")
|
|
entries.append(elsa_id)
|
|
query = f"INSERT INTO elsa_media ({', '.join(headers)}) VALUES ({', '.join(['?' for i in range(len(headers))])})"
|
|
self.query_db(query, entries)
|
|
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
|
|
"""
|
|
media = self.query_db("SELECT * FROM elsa_media WHERE elsa_id=?", (elsa_id,))
|
|
# convert the media to a list of dictionaries
|
|
ret = []
|
|
table_fields = self.query_db("PRAGMA table_info(elsa_media)")
|
|
for m in media:
|
|
tmp = {}
|
|
for i in range(len(m)):
|
|
tmp[table_fields[i][1]] = m[i]
|
|
ret.append(tmp)
|
|
return ret
|
|
|
|
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,)
|
|
)
|
|
###
|
|
|
|
|
|
|
|
def createProf(self, profdata):
|
|
conn = self.connect()
|
|
cursor = conn.cursor()
|
|
fname = profdata["profname"].split(", ")[1].strip()
|
|
lname = profdata["profname"].split(", ")[0].strip()
|
|
fullname = f"{lname} {fname}"
|
|
mail = profdata["prof_mail"]
|
|
telnr = profdata["prof_tel"]
|
|
title = profdata["title"]
|
|
|
|
query = f"INSERT INTO prof (fname, lname, fullname, mail, telnr,titel) VALUES ('{fname}','{lname}','{fullname}','{mail}','{telnr}','{title}')"
|
|
self.logger.log_info(query)
|
|
cursor.execute(query)
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
return self.getProfId(profdata)
|
|
|
|
def getProfId(self, profdata):
|
|
conn = self.connect()
|
|
cursor = conn.cursor()
|
|
fname = profdata["profname"].split(", ")[1].strip()
|
|
lname = profdata["profname"].split(", ")[0].strip()
|
|
fullname = f"{lname} {fname}"
|
|
query = f"SELECT id FROM prof WHERE fullname = '{fullname}'"
|
|
self.logger.log_info(query)
|
|
|
|
cursor.execute(query)
|
|
result = cursor.fetchone()
|
|
if result:
|
|
return result[0]
|
|
else: return None
|
|
|
|
def getProfByName(self, fullname):
|
|
'''Get all Data of the prof based on fullname
|
|
|
|
Args:
|
|
fullname (str): The full name of the prof
|
|
'''
|
|
conn = self.connect()
|
|
cursor = conn.cursor()
|
|
query = f"SELECT * FROM prof WHERE fullname = '{fullname}'"
|
|
self.logger.log_info(query)
|
|
|
|
result = cursor.execute(query).fetchone()
|
|
if result:
|
|
return result
|
|
else: return None
|
|
def getProfNameByApparat(self, apprarat_id):
|
|
query = f"SELECT prof_id from semesterapparat WHERE appnr = '{apprarat_id}'"
|
|
data = self.query_db(query)
|
|
if data:
|
|
return data[0][0]
|
|
else: return None |