Skip to content
Snippets Groups Projects
Select Git revision
  • a75cf5a8062bde1b7afac806b21ec312679e1b04
  • main default
  • full_migration
  • v1.0.9 protected
  • v1.0.8 protected
  • v1.0.7 protected
  • v1.0.6 protected
  • v1.0.5 protected
  • v1.0.4 protected
  • v1.0.3 protected
  • v1.0.2 protected
  • v1.0.1 protected
  • v1.0 protected
13 results

course.py

Blame
  • 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)