PHB/SemesterapparatsManager
2
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases 1 Wiki Activity

fix broken files after faulty update

Browse Source
This commit is contained in:
WorldTeacher
2025-01-14 08:51:58 +01:00
parent 598da9bfac
commit 997d618ff1
102 changed files with 2499 additions and 548 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/logic/__init__.py
View File

@@ -1,9 +1,9 @@
from .log import MyLogger, logger
from .dataclass import ApparatData, BookData, Prof, Apparat, ELSA
from .thread_bookgrabber import BookGrabber
from .threads_autoadder import AutoAdder
from .threads_availchecker import AvailChecker
from .c_sort import custom_sort
from .c_sort import custom_sort, sort_semesters_list
from .constants import APP_NRS, PROF_TITLES, SEMAP_MEDIA_ACCOUNTS
from .csvparser import csv_to_list
from .wordparser import elsa_word_to_csv, word_docx_to_csv

121
src/logic/c_sort.py
View File

@@ -1,59 +1,86 @@
from typing import List, Tuple
from natsort import natsorted
def custom_sort(unsorted: List[Tuple[str, int, int]]) -> List[Tuple[str, int, int]]:
"""Sort a list of semesters in the format "SoSe n" and "WiSe n/n+1" in the correct order.
Where n == year in 2 digit format
Args:
----
unsorted (list[tuple]): List of semesters in the format "SoSe n" and "WiSe n/n+1"
Returns:
-------
ret (list[tuple]): Sorted list in correct order of WiSe n/n+1 and SoSe n
def parse_semester(semester: str):
"""
summer = natsorted([i for i in unsorted if "SoSe" in i[0]])
winter = natsorted([i for i in unsorted if "WiSe" in i[0]])
summer = natsorted(summer, key=lambda x: x[0])
winter = natsorted(winter, key=lambda x: x[0])
Parses the semester string into a sortable format.
Returns a tuple of (year, type), where type is 0 for SoSe and 1 for WiSe.
"""
if semester.startswith("SoSe"):
return int(semester.split()[1]), 0
elif semester.startswith("WiSe"):
year_part = semester.split()[1]
start_year, _ = map(int, year_part.split("/"))
return start_year, 1
else:
raise ValueError(f"Invalid semester format: {semester}")
# Merge the lists
ret = []
i = 0
j = 0
while i < len(summer) and j < len(winter):
if summer[i][0][5:] <= winter[j][0][5:]:
ret.append(summer[i])
i += 1
def custom_sort(entries):
"""
Sorts the list of tuples based on the custom schema.
:param entries: List of tuples in the format (str, int, int).
:return: Sorted list of tuples.
"""
return sorted(
entries,
key=lambda entry: (
parse_semester(entry[0]), # Sort by semester parsed as (year, type)
entry[1], # Then by the second element of the tuple
entry[2], # Finally by the third element of the tuple
),
)
def parse_semester(semester: str):
"""
Parses the semester string into a sortable format.
Returns a tuple of (year, type), where type is 0 for SoSe and 1 for WiSe.
"""
if semester.startswith("SoSe"):
return int(semester.split()[1]), 0
elif semester.startswith("WiSe"):
year_part = semester.split()[1]
if "/" in year_part:
start_year, _ = map(int, year_part.split("/"))
else:
ret.append(winter[j])
j += 1
start_year = int(year_part)
return start_year, 1
else:
raise ValueError(f"Invalid semester format: {semester}")
# Append the remaining items
while i < len(summer):
ret.append(summer[i])
i += 1
while j < len(winter):
ret.append(winter[j])
j += 1
return ret
def sort_semesters_list(semesters: list) -> list:
"""
Sorts a list of semester strings based on year and type.
# Test the function
pass
:param semesters: List of semester strings (e.g., "SoSe 24", "WiSe 22/23").
:return: Sorted list of semester strings.
"""
return sorted(semesters, key=parse_semester)
if __name__ == "__main__":
unsorted = [
("WiSe 23/24", 7, 5),
("SoSe 23", 5, 0),
("SoSe 22", 1, 0),
("WiSe 22/23", 1, 0),
("SoSe 15", 1, 0),
"SoSe 24",
"WiSe 22/23",
"WiSe 23/24",
"WiSe 20/21",
"SoSe 23",
"SoSe 20",
"WiSe 7/8",
"WiSe 14/15",
"WiSe 13/14",
"SoSe 8",
"WiSe 19/20",
"WiSe 12/13",
"WiSe 21/22",
"WiSe 18/19",
"WiSe 11/12",
"WiSe 9/10",
"WiSe 6/7",
"SoSe 7",
"WiSe 16/17",
"WiSe 24/25",
"SoSe 25",
]
# print(custom_sort(unsorted))
print(sort_semesters_list(unsorted))

1
src/logic/constants.py
View File

@@ -1,3 +1,4 @@
from enum import Enum
APP_NRS = [i for i in range(1, 181)]

1
src/logic/csvparser.py
View File

@@ -1,3 +1,4 @@
import csv
import chardet

68
src/logic/dataclass.py
View File

@@ -1,18 +1,20 @@
import re
from dataclasses import dataclass, field
from enum import Enum
@dataclass
class Prof:
id: int = None
_title: str = None
firstname: str= None
lastname: str= None
fullname: str= None
mail: str= None
telnr: str= None
#add function that sets the data based on a dict
firstname: str = None
lastname: str = None
fullname: str = None
mail: str = None
telnr: str = None
# add function that sets the data based on a dict
def from_dict(self, data: dict):
for key, value in data.items():
if hasattr(self, key):
@@ -28,7 +30,8 @@ class Prof:
@title.setter
def title(self, value):
self._title = value
#add function that sets the data from a tuple
# add function that sets the data from a tuple
def from_tuple(self, data: tuple):
setattr(self, "id", data[0])
setattr(self, "_title", data[1])
@@ -40,40 +43,18 @@ class Prof:
return self
def name(self, comma=False):
if self.firstname is None and self.lastname is None:
if "," in self.fullname:
self.firstname = self.fullname.split(",")[1].strip()
self.lastname = self.fullname.split(",")[0].strip()
else:
return self.fullname
if comma:
return f"{self.lastname}, {self.firstname}"
return f"{self.lastname} {self.firstname}"
@dataclass
class ApparatData:
prof_title: str | None = None
profname: str | None = None
dauerapp: bool = False
appnr: int | None = None
appname: str | None = None
app_fach: str | None = None
semester: str | None = None
erstellsemester: str | None = None
prof_mail: str | None = None
prof_tel: int | None = None
deleted: int = 0
prof_adis_id: int | None = None
apparat_adis_id: int | None = None
@property
def prof_details(self) -> Prof:
return Prof().from_dict({
"title": self.prof_title,
"firstname": self.profname.split(',')[1].strip(),
"lastname": self.profname.split(',')[0].strip(),
"mail": self.prof_mail,
"telnr": self.prof_tel,
"fullname": f"{self.profname.split(',')[0].strip()} {self.profname.split(',')[1].strip()}",
})
def translateToFullname(self):
return f"{self.profname.split(',')[0].strip()} {self.profname.split(',')[1].strip()}"
@dataclass
class BookData:
ppn: str | None = None
@@ -132,7 +113,7 @@ class Subjects(Enum):
FRENCH = (6, "Französisch")
GEOGRAPHY = (7, "Geographie")
HISTORY = (8, "Geschichte")
HEALT_EDUCATION = (9, "Gesundheitspädagogik")
HEALTH_EDUCATION = (9, "Gesundheitspädagogik")
HTW = (10, "Haushalt / Textil")
ART = (11, "Kunst")
MATH_IT = (12, "Mathematik / Informatik")
@@ -197,6 +178,13 @@ class Apparat:
setattr(self, "konto", data[13])
return self
@property
def get_semester(self):
if self.extend_until is not None:
return self.extend_until
else:
return self.created_semester
@dataclass
class ELSA:
@@ -211,3 +199,7 @@ class ELSA:
setattr(self, "semester", data[2])
setattr(self, "prof_id", data[3])
return self
@dataclass
class ApparatData:
prof: Prof = field(default_factory=Prof)
apparat: Apparat = field(default_factory=Apparat)

1
src/logic/fileparser.py
View File

@@ -1,3 +1,4 @@
import csv
import pandas as pd

1
src/logic/mail.py
View File

@@ -0,0 +1 @@

1
src/logic/settings.py
View File

@@ -1,3 +1,4 @@
from dataclasses import dataclass, field
import yaml

41
src/logic/thread_bookgrabber.py
View File

@@ -3,7 +3,7 @@ import sqlite3
from PyQt6.QtCore import QThread
from PyQt6.QtCore import pyqtSignal as Signal
from src.backend.database import Database
from src.logic.log import MyLogger
from src.logic.webrequest import BibTextTransformer, WebRequest
@@ -16,8 +16,7 @@ class BookGrabber(QThread):
def __init__(self, appnr):
super(BookGrabber, self).__init__(parent=None)
self.is_Running = True
self.logger = MyLogger("Worker")
self.logger.log_info("Starting worker thread")
logger.info("Starting worker thread")
self.data = None
self.app_id = None
self.prof_id = None
@@ -35,9 +34,9 @@ class BookGrabber(QThread):
self.data = data
self.use_any = any_book
self.use_exact = exact
self.logger.log_info(f"Working on {len(self.data)} entries")
logger.info(f"Working on {len(self.data)} entries")
self.tstate = (self.app_id, self.prof_id, self.mode, self.data)
self.logger.log_debug("State: " + str(self.tstate))
logger.debug("State: " + str(self.tstate))
# print(self.tstate)
def run(self):
@@ -50,7 +49,7 @@ class BookGrabber(QThread):
for entry in iterdata:
# print(entry)
signature = str(entry)
self.logger.log_info("Processing entry: " + signature)
logger.info("Processing entry: " + signature)
webdata = WebRequest().set_apparat(self.appnr).get_ppn(entry)
if self.use_any:
@@ -79,7 +78,7 @@ class BookGrabber(QThread):
self.db.addBookToDatabase(bd, self.app_id, self.prof_id)
# get latest book id
self.book_id = self.db.getLastBookId()
self.logger.log_info("Added book to database")
logger.info("Added book to database")
state = 0
for result in transformer.RDS_DATA:
# print(result.RDS_LOCATION)
@@ -87,17 +86,17 @@ class BookGrabber(QThread):
state = 1
break
self.logger.log_info(f"State of {signature}: {state}")
logger.info(f"State of {signature}: {state}")
# print("updating availability of " + str(self.book_id) + " to " + str(state))
try:
self.db.setAvailability(self.book_id, state)
except sqlite3.OperationalError as e:
self.logger.log_error(f"Failed to update availability: {e}")
logger.error(f"Failed to update availability: {e}")
# time.sleep(5)
item += 1
self.updateSignal.emit(item, len(self.data))
self.logger.log_info("Worker thread finished")
logger.info("Worker thread finished")
# self.done.emit()
self.quit()
@@ -111,17 +110,17 @@ class BookGrabber(QThread):
# def __init__(self, app_id, prof_id, mode, data, parent=None):
# super(BookGrabber, self).__init__(parent=None)
# self.is_Running = True
# self.logger = MyLogger("Worker")
# self.logger.log_info("Starting worker thread")
# logger = MyLogger("Worker")
# logger.info("Starting worker thread")
# self.data = data
# self.logger.log_info(f"Working on {len(self.data)} entries")
# logger.info(f"Working on {len(self.data)} entries")
# self.app_id = app_id
# self.prof_id = prof_id
# self.mode = mode
# self.book_id = None
# self.state = (self.app_id, self.prof_id, self.mode, self.data)
# # print(self.state)
# self.logger.log_info("state: " + str(self.state))
# logger.info("state: " + str(self.state))
# # time.sleep(2)
# def resetValues(self):
@@ -140,7 +139,7 @@ class BookGrabber(QThread):
# for entry in iterdata:
# # print(entry)
# signature = str(entry)
# self.logger.log_info("Processing entry: " + signature)
# logger.info("Processing entry: " + signature)
# webdata = WebRequest().get_ppn(entry).get_data()
# if webdata == "error":
@@ -153,19 +152,19 @@ class BookGrabber(QThread):
# self.db.addBookToDatabase(bd, self.app_id, self.prof_id)
# # get latest book id
# self.book_id = self.db.getLastBookId()
# self.logger.log_info("Added book to database")
# logger.info("Added book to database")
# state = 0
# # print(len(rds.items))
# for rds_item in rds.items:
# sign = rds_item.superlocation
# loc = rds_item.location
# # ic(sign, loc)
# # ic(rds_item)
# # logger.debug(sign, loc)
# # logger.debug(rds_item)
# if self.app_id in sign or self.app_id in loc:
# state = 1
# break
# self.logger.log_info(f"State of {signature}: {state}")
# logger.info(f"State of {signature}: {state}")
# # print(
# "updating availability of "
# + str(self.book_id)
@@ -175,12 +174,12 @@ class BookGrabber(QThread):
# try:
# self.db.setAvailability(self.book_id, state)
# except sqlite3.OperationalError as e:
# self.logger.log_error(f"Failed to update availability: {e}")
# logger.error(f"Failed to update availability: {e}")
# # time.sleep(5)
# item += 1
# self.updateSignal.emit(item, len(self.data))
# self.logger.log_info("Worker thread finished")
# logger.info("Worker thread finished")
# # self.done.emit()
# self.stop()
# if not self.is_Running:

9
src/logic/threads_autoadder.py
View File

@@ -5,7 +5,7 @@ from PyQt6.QtCore import QThread
from PyQt6.QtCore import pyqtSignal as Signal
from src.backend.database import Database
from src.logic.log import MyLogger
# from src.transformers import RDS_AVAIL_DATA
@@ -18,7 +18,6 @@ class AutoAdder(QThread):
def __init__(self, data=None, app_id=None, prof_id=None, parent=None):
super().__init__(parent)
self.logger = MyLogger("AutoAdder")
self.data = data
self.app_id = app_id
self.prof_id = prof_id
@@ -29,7 +28,7 @@ class AutoAdder(QThread):
def run(self):
self.db = Database()
# show the dialog, start the thread to gather data and dynamically update progressbar and listwidget
self.logger.log_info("Starting worker thread")
logger.info("Starting worker thread")
item = 0
for entry in self.data:
try:
@@ -42,11 +41,11 @@ class AutoAdder(QThread):
except Exception as e:
# print(e)
self.logger.log_exception(
logger.exception(
f"The query failed with message {e} for signature {entry}"
)
continue
if item == len(self.data):
self.logger.log_info("Worker thread finished")
logger.info("Worker thread finished")
# teminate thread
self.finished.emit()

15
src/logic/threads_availchecker.py
View File

@@ -5,7 +5,7 @@ from PyQt6.QtCore import QThread
from PyQt6.QtCore import pyqtSignal as Signal
from src.backend.database import Database
from src.logic.log import MyLogger
from src.logic.webrequest import BibTextTransformer, WebRequest
# from src.transformers import RDS_AVAIL_DATA
@@ -21,9 +21,8 @@ class AvailChecker(QThread):
if links is None:
links = []
super().__init__(parent)
self.logger = MyLogger("AvailChecker")
self.logger.log_info("Starting worker thread")
self.logger.log_info(
logger.info("Starting worker thread")
logger.info(
"Checking availability for "
+ str(links)
+ " with appnumber "
@@ -33,7 +32,7 @@ class AvailChecker(QThread):
self.links = links
self.appnumber = appnumber
self.books = books
self.logger.log_info(
logger.info(
f"Started worker with appnumber: {self.appnumber} and links: {self.links} and {len(self.books)} books..."
)
time.sleep(2)
@@ -43,7 +42,7 @@ class AvailChecker(QThread):
state = 0
count = 0
for link in self.links:
self.logger.log_info("Processing entry: " + str(link))
logger.info("Processing entry: " + str(link))
data = WebRequest().set_apparat(self.appnumber).get_ppn(link).get_data()
transformer = BibTextTransformer("RDS")
rds = transformer.get_data(data).return_data("rds_availability")
@@ -60,14 +59,14 @@ class AvailChecker(QThread):
if book["bookdata"].signature == link:
book_id = book["id"]
break
self.logger.log_info(f"State of {link}: " + str(state))
logger.info(f"State of {link}: " + str(state))
# print("Updating availability of " + str(book_id) + " to " + str(state))
self.db.setAvailability(book_id, state)
count += 1
self.updateProgress.emit(count, len(self.links))
self.updateSignal.emit(item.callnumber, state)
self.logger.log_info("Worker thread finished")
logger.info("Worker thread finished")
# teminate thread
self.quit()

19
src/logic/webrequest.py
View File

@@ -5,10 +5,9 @@ from bs4 import BeautifulSoup
from ratelimit import limits, sleep_and_retry
from src.logic.dataclass import BookData
from src.logic.log import MyLogger
from src.transformers import ARRAYData, BibTeXData, COinSData, RDSData, RISData
logger = MyLogger(__name__)
API_URL = "https://rds.ibs-bw.de/phfreiburg/opac/RDSIndexrecord/{}/"
PPN_URL = "https://rds.ibs-bw.de/phfreiburg/opac/RDSIndex/Search?type0%5B%5D=allfields&lookfor0%5B%5D=&join=AND&bool0%5B%5D=AND&type0%5B%5D=au&lookfor0%5B%5D=&join=AND&bool0%5B%5D=AND&type0%5B%5D=ti&lookfor0%5B%5D=&join=AND&bool0%5B%5D=AND&type0%5B%5D=ct&lookfor0%5B%5D=&join=AND&bool0%5B%5D=AND&type0%5B%5D=isn&lookfor0%5B%5D=&join=AND&bool0%5B%5D=AND&type0%5B%5D=ta&lookfor0%5B%5D=&join=AND&bool0%5B%5D=AND&type0%5B%5D=co&lookfor0%5B%5D=&join=AND&bool0%5B%5D=AND&type0%5B%5D=py&lookfor0%5B%5D=&join=AND&bool0%5B%5D=AND&type0%5B%5D=pp&lookfor0%5B%5D=&join=AND&bool0%5B%5D=AND&type0%5B%5D=pu&lookfor0%5B%5D=&join=AND&bool0%5B%5D=AND&type0%5B%5D=si&lookfor0%5B%5D={}&join=AND&bool0%5B%5D=AND&type0%5B%5D=zr&lookfor0%5B%5D=&join=AND&bool0%5B%5D=AND&type0%5B%5D=cc&lookfor0%5B%5D=&join=AND&bool0%5B%5D=AND"
@@ -37,19 +36,19 @@ class WebRequest:
self.ppn = None
self.data = None
self.timeout = 5
logger.log_info("Initialized WebRequest")
logger.info("Initialized WebRequest")
@property
def use_any_book(self):
"""use any book that matches the search term"""
self.use_any = True
logger.log_info("Using any book")
logger.info("Using any book")
return self
def set_apparat(self, apparat):
self.apparat = apparat
if int(self.apparat) < 10:
self.apparat = f"0{self.apparat}"
logger.log_info(f"Set apparat to {self.apparat}")
logger.info(f"Set apparat to {self.apparat}")
return self
def get_ppn(self, signature):
@@ -81,7 +80,7 @@ class WebRequest:
response = requests.get(link, timeout=self.timeout)
return response.text
except requests.exceptions.RequestException as e:
logger.log_error(f"Request failed: {e}")
logger.error(f"Request failed: {e}")
return None
def get_data(self):
links = self.get_book_links(self.ppn)
@@ -105,7 +104,7 @@ class WebRequest:
return_data.append(data)
return return_data
else:
logger.log_error("No <pre> tag found")
logger.error("No <pre> tag found")
raise ValueError("No <pre> tag found")
if f"Semesterapparat-{self.apparat}" in item_location:
pre_tag = soup.find_all("pre")
@@ -116,7 +115,7 @@ class WebRequest:
return_data.append(data)
return return_data
else:
logger.log_error("No <pre> tag found")
logger.error("No <pre> tag found")
return return_data
def get_data_elsa(self):
@@ -137,7 +136,7 @@ class WebRequest:
return_data.append(data)
return return_data
else:
logger.log_error("No <pre> tag found")
logger.error("No <pre> tag found")
return return_data
@@ -155,7 +154,7 @@ class BibTextTransformer:
self.field = None
self.signature = None
if mode not in self.valid_modes:
logger.log_error(f"Mode {mode} not valid")
logger.error(f"Mode {mode} not valid")
raise ValueError(f"Mode {mode} not valid")
self.data = None
# self.bookdata = BookData(**self.data)

1
src/logic/wordparser.py
View File

@@ -1,3 +1,4 @@
import pandas as pd
from docx import Document
import re

1
src/logic/zotero.py
View File

@@ -1,3 +1,4 @@
from pyzotero import zotero
from dataclasses import dataclass
from src.logic.webrequest import WebRequest, BibTextTransformer