add elsa functions
This commit is contained in:
@@ -10,6 +10,10 @@ from typing import Any, Dict, List, Optional, Tuple, Union
|
||||
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,
|
||||
@@ -99,6 +103,10 @@ class Database:
|
||||
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)
|
||||
|
||||
@@ -129,7 +137,7 @@ class Database:
|
||||
one (bool, optional): Return the first result only. Defaults to False.
|
||||
|
||||
Returns:
|
||||
Union[Typle|List[Tuple]]: Returns the result of the query
|
||||
Union[Tuple | List[Tuple]]: Returns the result of the query
|
||||
"""
|
||||
conn = self.connect()
|
||||
cursor = conn.cursor()
|
||||
@@ -648,6 +656,7 @@ class 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")
|
||||
|
||||
@@ -1196,3 +1205,134 @@ class Database:
|
||||
"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,)
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user