Select Git revision
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
course.py 11.75 KiB
from api.database import *
from api.miscellaneous import *
from api.authentication import get_authentication_methods
import api
COURSE_SECONDARY_COLUMN_VISIBLE = "course_visible"
COURSE_SECONDARY_COLUMN_DOWNLOADABLE = "course_downloadable"
COURSE_SECONDARY_COLUMN_EMBED_INVISIBLE = "course_embedinvisible"
COURSE_SECONDARY_DB_SELECTION = f"""
"courses"."handle" AS "course_handle",
"courses"."title" AS "course_title",
"courses"."short" AS "course_short",
"courses"."organizer" AS "course_organizer",
"courses"."subject" AS "course_subject",
"courses"."description" AS "course_description",
"courses"."coursechapters" AS "course_coursechapters",
"courses"."semester" AS "course_semester",
"courses"."login_info" AS "course_login_info",
"courses"."listed" AS "course_listed",
"courses"."visible" AS "{COURSE_SECONDARY_COLUMN_VISIBLE}",
"courses"."embedinvisible" AS "{COURSE_SECONDARY_COLUMN_EMBED_INVISIBLE}",
"courses"."downloadable" AS "{COURSE_SECONDARY_COLUMN_DOWNLOADABLE}"
"""
def _semester_db_to_json(semester_db: str):
if semester_db is None or len(semester_db) == 0 or semester_db.isspace():
return "none" # pragma: no cover
from api.objects.type import SEMESTER_STRING_PATTERN
if SEMESTER_STRING_PATTERN.fullmatch(semester_db) is None: # pragma: no cover
print(f"Warning: Invalid semester string in database: {truncate_string(semester_db)}")
return "none"
return semester_db
def course_list_db_to_json_no_lectures(courses_db: [], auth_db: [], is_mod: bool):
"""
:param courses_db: Must be sorted by id ascending
:param auth_db: authentication methods. Must be sorted by id ascending
:param is_mod:
:return:
"""
auth_i = 0
courses_json = []
for course_db in courses_db:
auth_i, course_auth_db = db_collect_id_sorted_data(auth_db, auth_i, "course_id", course_db["id"])
courses_json.append(course_db_to_json_no_lectures(course_db, course_auth_db, is_mod))
return courses_json
def course_db_to_json_no_auth_lectures(course_db: {}, is_mod: bool):
course_json = {
"id": course_db["id"],
"id_string": course_db["handle"],
"full_name": course_db["title"],
"short_name": course_db["short"],
"semester": _semester_db_to_json(course_db["semester"]),
"organizer": course_db["organizer"],
"topic": course_db["subject"],
"description": course_db["description"],
"show_chapters_on_course": bool(course_db["coursechapters"])
}
if len(course_db["login_info"]) > 0:
course_json["authentication_information"] = course_db["login_info"]
if is_mod:
course_json["is_listed"] = bool(course_db["listed"])
course_json["is_visible"] = bool(course_db["visible"])
return course_json
def course_db_to_json_no_lectures(course_db: {}, auth_db: [], is_mod: bool):
course_json = course_db_to_json_no_auth_lectures(course_db, is_mod)
course_json["default_authentication_methods"] = get_authentication_methods(auth_db, False)
return course_json
def course_secondary_db_to_json_no_lectures(secondary_course_db: {}, auth_db: [], is_mod: bool):
course_json = {
"id": secondary_course_db["course_id"],
"id_string": secondary_course_db["course_handle"],
"full_name": secondary_course_db["course_title"],
"short_name": secondary_course_db["course_short"],
"organizer": secondary_course_db["course_organizer"],
"topic": secondary_course_db["course_subject"],
"description": secondary_course_db["course_description"],
"show_chapters_on_course": bool(secondary_course_db["course_coursechapters"]),
"default_authentication_methods": get_authentication_methods(auth_db, False)
}
if "course_semester" in secondary_course_db:
course_json["semester"] = _semester_db_to_json(secondary_course_db["course_semester"])
if "course_login_info" in secondary_course_db and len(secondary_course_db["course_login_info"]) > 0:
course_json["authentication_information"] = secondary_course_db["course_login_info"]
if is_mod:
course_json["is_listed"] = bool(secondary_course_db["course_listed"])
course_json["is_visible"] = bool(secondary_course_db["course_visible"])
return course_json
_SQL_GET_COURSE_AUTH = PreparedStatement("""
SELECT * FROM "perm"
WHERE "perm"."course_id" = ?
AND (NOT "deleted")
""")
def course_query_auth(course_id: int, transaction: AbstractTransaction or None = None) -> DbResultSet:
"""
Returns a result even if course is not visible
"""
if transaction is None:
with db_pool.start_read_transaction() as trans:
return trans.execute_statement(_SQL_GET_COURSE_AUTH, course_id)
else:
return transaction.execute_statement(_SQL_GET_COURSE_AUTH, course_id)
def course_queue_query_auth(transaction: AbstractTransaction, course_id: int) -> FilledStatement:
"""
Returns a result even if course is not visible
"""
return transaction.queue_statement(_SQL_GET_COURSE_AUTH, course_id)
def lecture_db_to_json_no_chapters_media(lecture_db: {}, auth_db: [], course_allow_embed: bool, is_mod: bool):
lecture_id: int = lecture_db["id"]
lecture_json = {
"id": lecture_id,
"course_id": lecture_db["course_id"],
"title": lecture_db["title"],
"speaker": lecture_db["speaker"],
"location": lecture_db["place"],
"time": lecture_db["time"].replace(tzinfo=None).isoformat(sep="T", timespec="seconds"),
"duration": lecture_db["duration"],
"description": lecture_db["comment"],
"thumbnail_url": api.config["FILE_PATH_PREFIX"] + "thumbnail/l_" + str(lecture_id) + ".jpg",
"no_recording": bool(lecture_db["norecording"]),
"livestream_planned": bool(lecture_db["live"] or lecture_db["stream_job"]),
"authentication_methods": get_authentication_methods(auth_db, True),
"allow_embed": course_allow_embed
}
if lecture_db["stream_job"]:
lecture_json["livestream_url"] = "pub/hls/" + str(lecture_id) + ".m3u8"
if is_mod:
lecture_json["is_visible"] = bool(lecture_db["visible"])
lecture_json["internal_comment"] = lecture_db["internal"]
return lecture_json
_SQL_GET_LECTURE_AUTH_WITH_COURSE_ID = PreparedStatement("""
SELECT "perm".*
FROM "perm"
WHERE ("perm"."lecture_id" = ? OR "perm"."course_id" = ?)
AND (NOT "deleted")
""")
_SQL_GET_LECTURE_AUTH_NO_COURSE_ID = PreparedStatement("""
SELECT "perm".*
FROM "lectures"
JOIN "perm" ON ("perm"."lecture_id" = "lectures"."id" OR "perm"."course_id" = "lectures"."course_id")
WHERE
"lectures"."id" = ?
AND (NOT "perm"."deleted")
""")
def lecture_query_auth(lecture_id: int,
course_id: int = None,
transaction: AbstractTransaction = None) -> DbResultSet:
if course_id:
if transaction is None:
with db_pool.start_read_transaction() as trans:
return trans.execute_statement_and_close(
_SQL_GET_LECTURE_AUTH_WITH_COURSE_ID, lecture_id, course_id)
else:
return transaction.execute_statement(_SQL_GET_LECTURE_AUTH_WITH_COURSE_ID, lecture_id, course_id)
else:
if transaction is None:
with db_pool.start_read_transaction() as trans:
return trans.execute_statement_and_close(
_SQL_GET_LECTURE_AUTH_NO_COURSE_ID, lecture_id)
else:
return transaction.execute_statement(_SQL_GET_LECTURE_AUTH_NO_COURSE_ID, lecture_id)
def lecture_queue_query_auth(transaction: AbstractTransaction,
lecture_id: int,
course_id: int = None) -> FilledStatement:
if course_id:
return transaction.queue_statement(_SQL_GET_LECTURE_AUTH_WITH_COURSE_ID, lecture_id, course_id)
else:
return transaction.queue_statement(_SQL_GET_LECTURE_AUTH_NO_COURSE_ID, lecture_id)
def chapter_db_to_json(chapter_db: {}, is_mod: bool):
chapter_json = {
"start_time": chapter_db["time"],
"name": chapter_db["text"]
}
if is_mod:
chapter_json["id"] = chapter_db["id"]
chapter_json["is_visible"] = bool(chapter_db["visible"])
return chapter_json
def media_source_with_format_db_to_json(media_source_db: {}, course_is_downloadable: bool, is_mod: bool):
media_source_json = {
"quality": {
"name": media_source_db["format_description"],
"resolution": media_source_db["format_resolution"],
"aspect_ration": media_source_db["format_aspect"],
"priority": media_source_db["format_player_prio"]
},
"size": media_source_db["file_size"],
"comment": media_source_db["comment"],
"player_url": api.config["FILE_PATH_PREFIX"] + media_source_db["path"],
}
if media_source_db["downloadable"] == 1 and course_is_downloadable:
media_source_json["download_url"] = media_source_json["player_url"]
if is_mod:
media_source_json["id"] = media_source_db["id"]
media_source_json["is_visible"] = bool(media_source_db["visible"])
return media_source_json
def course_queue_search(transaction: ReadTransaction, search_term: str, is_mod: bool):
return _queue_search(
transaction,
"courses",
["title", "short", "handle", "organizer", "subject", "description"],
None,
None,
None if is_mod else 'WHERE "courses"."visible" AND "courses"."listed"',
'"courses"."semester" DESC',
20,
search_term
)
def lecture_queue_search(transaction: ReadTransaction, search_term: str, is_mod: bool):
return _queue_search(
transaction,
"lectures",
["title", "comment", "speaker"],
'JOIN "courses" ON ("lectures"."course_id" = "courses"."id")',
COURSE_SECONDARY_DB_SELECTION,
None if is_mod else 'WHERE "courses"."visible" AND "courses"."listed" AND "lectures"."visible"',
'"lectures"."time" DESC',
30,
search_term
)
def _queue_search(transaction: ReadTransaction,
table: str,
search_columns: list[str],
join_clause: str or None,
extra_select_columns: str or None,
where_clause: str or None,
extra_ordering: str or None,
limit: int,
search_term: str):
base_sub_query = f"""
SELECT "{table}"."id" AS "_id", CAST(%s AS INT) AS "_priority" FROM "{table}" WHERE {" OR ".join(
map(lambda column: f'LOWER("{table}"."{column}") LIKE ?',
search_columns))}
"""
words: list[str] = list(filter(lambda w: not w.isspace(), search_term.split(" ")))
if len(words) == 0:
raise ValueError("No words provided") # pragma: no cover
sub_queries: list[str] = []
all_values: list[DbValueType] = []
prio = len(words)
for word in words:
word = word.lower()
word = word.replace("%", "\\%").replace("_", "\\_")
word = "%" + word + "%"
sub_queries.append(base_sub_query % prio)
for _ in range(0, len(search_columns)):
all_values.append(word)
prio -= 1
return transaction.queue_statement(f"""
SELECT "{table}".* {"" if extra_select_columns is None else "," + extra_select_columns}
FROM "{table}"
JOIN (
SELECT "_id", CAST(SUM("_priority") AS INT) AS "_score"
FROM ({"UNION ALL".join(sub_queries)}) AS "_sub_result"
GROUP BY "_id"
) AS "_data" ON ("{table}"."id" = "_data"."_id")
{"" if join_clause is None else join_clause}
{"" if where_clause is None else where_clause}
ORDER BY "_data"."_score" DESC{"" if extra_ordering is None else ", " + extra_ordering} LIMIT {limit}
""", *all_values)