Merge branch 'master' into dependabot/pip/werkzeug-3.0.6

Author: x10102

.gitignore

@@ -166,6 +166,7 @@ translations_new.json
 *.db-wal
 *.db-shm
 *.csv
+*.bak
 config.json
 czech.json
 temp

App.py

@@ -29,6 +29,7 @@
 from blueprints.api import ApiController
 from blueprints.rsspage import RssPageController
 from blueprints.oauth import OauthController
+from blueprints.embed import EmbedController
 
 from extensions import login_manager, dbs, sched, oauth, rss, webhook
 
@@ -165,9 +166,12 @@ def extensions_init() -> None:
     app.register_blueprint(ApiController)
     app.register_blueprint(OauthController)
     app.register_blueprint(RssPageController)
+    app.register_blueprint(EmbedController)
 
     # Create the admin user
     user_init()
+
+    # Load extensions and enable integrations based on config
     extensions_init()
 
     # Force oauthlib to allow insecure transport when debugging

blueprints/articles.py

@@ -1,5 +1,6 @@
 # Builtins
 from datetime import datetime
+from http import HTTPStatus
 from logging import info
 
 # External
@@ -77,7 +78,7 @@ def edit_article(aid: int):
 
     article = dbs.get_article(aid)
     if not article:
-        abort(404)
+        abort(HTTPStatus.NOT_FOUND)
 
     if request.method == "GET":
         fdata = {'title': article.name, 'words': article.words, 'bonus': article.bonus, 'link': article.link, 'translator': article.author.nickname}

blueprints/auth.py

@@ -17,7 +17,10 @@
 def login():
 
     if request.method == "GET":
-        session['login_next'] = request.referrer
+        # Saves the last URL, but prevents the user from staying on the logging page
+        # if the first login attempt fails
+        if not request.referrer.endswith('/login'):
+            session['login_next'] = request.referrer
         return render_template('auth/login.j2', form=LoginForm())
 
     form = LoginForm()
@@ -34,6 +37,7 @@ def login():
         return redirect(url_for('UserAuth.pw_change'))
     login_user(user)
     referrer = session['login_next']
+    info(referrer)
     del session['login_next']
     return redirect(referrer or url_for('index'))
 

blueprints/debug.py

@@ -1,8 +1,10 @@
-from logging import warning
-from flask import Blueprint, redirect, url_for, current_app, request
+from http import HTTPStatus
+from logging import critical, warning, error
+from flask import Blueprint, redirect, url_for, current_app, request, render_template, send_from_directory, flash, abort
 from flask_login import login_required, current_user
+from datetime import datetime
 
-from extensions import dbs, sched
+from extensions import sched, webhook
 
 DebugTools = Blueprint('DebugTools', __name__)
 
@@ -15,16 +17,49 @@ def log_debug_access():
 @login_required
 def nickupdate():
     sched.run_job('Fetch nicknames')
-    return redirect(url_for('index'))
+    flash("Aktualizace spuštěna na pozadí!")
+    return redirect(request.referrer or url_for('index'))
 
 @DebugTools.route('/debug/avupdate')
 @login_required
 def avdownload():
     sched.run_job('Download avatars')
-    return redirect(url_for('index'))
+    flash("Aktualizace spuštěna na pozadí!")
+    return redirect(request.referrer or url_for('index'))
 
 @DebugTools.route('/debug/rssupdate')
 @login_required
 def updaterss():
     sched.run_job('Fetch RSS updates')
-    return redirect(url_for('index'))
+    flash("Aktualizace spuštěna na pozadí!")
+    return redirect(request.referrer or url_for('index'))
+
+@DebugTools.route('/debug')
+def debug_index():
+    return render_template('debug/tools.j2')
+
+@DebugTools.route('/debug/test_webhook')
+def webhook_testing():
+    try:
+        webhook.send_text('TEST MESSAGE')
+        flash("Testovací webhook odeslán!")
+    except Exception as e:
+        flash("Testovací webhook se nepodařilo odeslat (zkontrolujte logy)")
+        error(f"Sending test webhook failed with error ({str(e)})")
+    return redirect(request.referrer or url_for('index'))
+    
+@DebugTools.route('/debug/db/export')
+def export_database():
+    download_name=datetime.strftime(datetime.now(), 'scp_%d_%m_%Y.db')
+    flash("Databáze exportována!")
+    return send_from_directory('data', 'scp.db', as_attachment=True, download_name=download_name)
+
+@DebugTools.route('/debug/raise_error')
+def raise_error():
+    error("Error handling test")
+    abort(HTTPStatus.INTERNAL_SERVER_ERROR)
+
+@DebugTools.route('/debug/raise_critical')
+def raise_critical_error():
+    critical("Critical error handling test")
+    abort(HTTPStatus.INTERNAL_SERVER_ERROR)

blueprints/embed.py

@@ -0,0 +1,47 @@
+from http import HTTPStatus
+from flask import render_template, Blueprint, abort, request
+from extensions import dbs
+from enum import StrEnum
+from os import getcwd, path, listdir
+
+EmbedController = Blueprint('EmbedController', __name__)
+EMBED_TEMPLATE_PATH = path.join(getcwd(), 'templates', 'embeds')
+
+class EmbedType(StrEnum):
+    TRANSLATOR = "translator"
+    WRITER = "writer"
+    CORRECTOR = "corrector"
+
+# TODO: Change this to a class-based blueprint
+writer_themes_installed = list()
+translator_themes_installed = list()
+
+# "record_once" runs the function only when the blueprint is first registered, don't think they could have chosen a worse name
+# The record function takes the BlueprintSetupState object but we discard it here as we don't need it
+@EmbedController.record_once
+def on_blueprint_load(_):
+    global writer_themes_installed
+    global translator_themes_installed
+    # Build a list of themes from files in the templates directory
+    writer_themes_installed = [theme.removesuffix('.j2') for theme in listdir(path.join(EMBED_TEMPLATE_PATH, 'writer'))]
+    translator_themes_installed = [theme.removesuffix('.j2') for theme in listdir(path.join(EMBED_TEMPLATE_PATH, 'translator'))]
+
+def get_template(template_type: EmbedType, theme: str = "default"):
+    if path.exists(path.join(EMBED_TEMPLATE_PATH, template_type, f"{theme}.j2")):
+        return f"embeds/{template_type}/{theme}.j2"
+    else:
+        # Don't abort on invalid theme, just use the default
+        return f"embeds/{template_type}/default.j2"
+
+@EmbedController.route('/user/<int:uid>/embed', methods=["GET"])
+def user_badge(uid: int):
+    embed_type = request.args.get("type", type=str, default=EmbedType.TRANSLATOR)
+    if embed_type not in list(EmbedType): abort(HTTPStatus.BAD_REQUEST) # Abort on invalid type
+    embed_theme = request.args.get("theme", type=str, default="default")
+    user = dbs.get_user(uid) or abort(HTTPStatus.NOT_FOUND) # Abort on invalid user
+    stats = dbs.get_user_stats(uid)
+    if embed_type == EmbedType.TRANSLATOR:
+        last = dbs.get_last_article(uid, False)
+    else:
+        last = dbs.get_last_article(uid, True)
+    return render_template(get_template(embed_type, embed_theme), user=user, stats=stats, last=last)

blueprints/users.py

@@ -1,3 +1,4 @@
+from http import HTTPStatus
 from flask import Blueprint, url_for, redirect, session, request, render_template, abort, flash
 from forms import NewUserForm, EditUserForm
 from flask_login import current_user, login_required
@@ -41,7 +42,7 @@ def add_user():
 @UserController.route('/user/<int:uid>/edit', methods=["GET", "POST"])
 @login_required
 def edit_user(uid: int):
-    user = dbs.get_user(uid) or abort(404)
+    user = dbs.get_user(uid) or abort(HTTPStatus.NOT_FOUND)
 
     if request.method == "GET":
         fdata = {'nickname': user.nickname, 'wikidot': user.wikidot, 'discord': user.discord, 'login': int(user.password is not None)}
@@ -60,7 +61,7 @@ def edit_user(uid: int):
 def user(uid: int):
     sort = request.args.get('sort', 'latest', str)
     page = request.args.get('p', 0, int)
-    user = dbs.get_user(uid) or abort(404)
+    user = dbs.get_user(uid) or abort(HTTPStatus.NOT_FOUND)
     corrections = dbs.get_corrections_by_user(uid)
     translations = dbs.get_translations_by_user(uid, sort, page)
     originals = dbs.get_originals_by_user(uid)
@@ -69,9 +70,10 @@ def user(uid: int):
 @UserController.route('/user/<int:uid>/delete', methods=["POST", "GET"])
 @login_required
 def delete_user(uid: int):
-    name = dbs.get_user(uid).nickname
+    user = dbs.get_user(uid) or abort(HTTPStatus.NOT_FOUND)
+    name = user.nickname
     dbs.delete_user(uid)
     info(f"User {name} deleted by {current_user.nickname} (ID: {current_user.uid})")
     flash(f'Uživatel {name} smazán')
 
-    return redirect(url_for('index'))
+    return redirect(url_for('index'))

connectors/rss.py

@@ -79,43 +79,6 @@ def get_rss_update_type(update: dict) -> RSSUpdateType:
             return RSSUpdateType.RSS_CORRECTION
         else:
             return RSSUpdateType.RSS_UNKNOWN
-        
-    @staticmethod
-    def en_page_exists(url: str) -> bool:
-        """
-        Converts a branch URL into an EN URL of the same page and checks if it exists on EN
-        """
-        try:
-            parsed_url = parse.urlparse(url)
-        except ValueError:
-            error(f'Cannot parse URL "{url}"')
-            return False
-        parsed_url = parsed_url._replace(scheme='https')._replace(netloc='scp-wiki.wikidot.com')
-        wl_parsed_url = parsed_url._replace(scheme='https')._replace(netloc='wanderers-library.wikidot.com')
-        en_url = parse.urlunparse(parsed_url)
-        wl_url = parse.urlunparse(wl_parsed_url)
-        try:
-            head_result = requests.head(en_url, headers={'User-Agent': USER_AGENT})
-        except requests.RequestException as e:
-            error(f'Request to {en_url} failed ({str(e)})')
-            return False
-        #TODO: Refactor this
-        match head_result.status_code:
-            case 200:
-                return True
-            case 404:
-                head_result = requests.head(wl_url, headers={'User-Agent': USER_AGENT})
-                match head_result.status_code:
-                    case 200:
-                        return True
-                    case 404:
-                        return False
-                    case _:
-                        warning(f'Got unusual status code ({head_result.status_code}) for URL {en_url}')
-                        return False
-            case _:
-                warning(f'Got unusual status code ({head_result.status_code}) for URL {en_url}')
-                return False
 
     def get_rss_update_author(self, update: dict) -> Optional[User]:
         update_description = update['description']
@@ -146,17 +109,11 @@ def _process_new_page(self, update) -> bool:
             info('Ignoring {title} in RSS feed (couldn\'t match wikidot username {author} to a user)')
             return False
         debug(f'Check {title} with ts {timestamp}, last db update was {self.__dbs.lastupdated}')
-        #if title.lower().endswith(IGNORE_BRANCH_TAG):
-        #    info(f'Ignoring {title} in RSS feed (not a translation)')
-        #    return False
 
         if timestamp+TIMEZONE_UTC_OFFSET > self.__dbs.lastupdated:
             if self.__dbs.get_article_by_link(update['link']):
                 info(f'Ignoring {title} in RSS feed (added manually)')
                 return False
-            #if not RSSMonitor.en_page_exists(update['link']):
-            #    info(f'Ignoring {title} in RSS feed (EN Wiki page doesn\'t exist)')
-            #    return False
             self.__updates.append(RSSUpdate(timestamp+TIMEZONE_UTC_OFFSET, update['link'], title, author, uuid4(), RSSUpdateType.RSS_NEWPAGE))
             return True
         return False
@@ -172,7 +129,6 @@ def _process_correction(self, update) -> bool:
             translation = self.__dbs.get_article_by_link(update['link'])
             if not translation:
                 self.__updates.append(RSSUpdate(timestamp+TIMEZONE_UTC_OFFSET, update['link'], real_title, author, uuid4(), RSSUpdateType.RSS_CORRECTION))
-                #self.__webhook.send_text(f'Korekci od {author.nickname} pro {real_title} nelze přiřadit k článku. Zapište manuálně.')
                 warning(f"Correction for {real_title} by {author.nickname} cannot be assigned to an article")
             else:
                 self.__dbs.assign_corrector(translation, author)
@@ -194,8 +150,6 @@ def _process_update(self, update) -> bool:
             case RSSUpdateType.RSS_CORRECTION:
                 self._process_correction(update)
 
-        
-
         return False
 
     def check(self):

db.py

@@ -340,6 +340,14 @@ def get_global_stats(self) -> StatisticsRow:
         query = "SELECT * FROM Statistics"
         data = self.__tryexec(query).fetchone()
         return StatisticsRow(*data)
+
+    def get_last_article(self, uid: int, original: bool = False) -> t.Optional[Article]:
+        query = "SELECT * FROM Article WHERE idauthor=? AND is_original=? ORDER BY added DESC, id DESC LIMIT 1"
+        data = (uid, original)
+        row = self.__tryexec(query, data).fetchone()
+        if not row:
+            return None
+        return self.__make_article(row)
 
     def add_article(self, a: Article) -> int:
         query = "INSERT INTO Article (name, words, bonus, added, link, idauthor, is_original) VALUES (?, ?, ?, ?, ?, ?, ?)"

requirements.txt

@@ -21,7 +21,7 @@ hypercorn==0.14.4
 hyperframe==6.0.1
 idna==3.4
 itsdangerous==2.1.2
-Jinja2==3.1.4
+Jinja2==3.1.5
 lxml==4.9.3
 MarkupSafe==2.1.3
 MechanicalSoup==1.3.0
@@ -40,7 +40,7 @@ tomli==2.0.1
 tzdata==2023.3
 tzlocal==5.0.1
 urllib3==2.2.2
-waitress==2.1.2
 Werkzeug==3.0.6
+waitress==3.0.1
 wsproto==1.2.0
 WTForms==3.0.1