Move function-level stdlib imports to the top of the file

zeeguu/api/app.py

@@ -15,6 +15,9 @@
 
 # apimux is quite noisy; supress it's output
 import logging
+import subprocess
+import sys
+import traceback
 from apimux.log import logger
 
 # Suppress verbose Elasticsearch connection error logs
@@ -148,7 +151,6 @@ def create_app(testing=False):
     # Add SQL query listener to catch source_text queries
     from sqlalchemy import event
     from sqlalchemy.engine import Engine
-    import sys
 
     @event.listens_for(Engine, "before_cursor_execute")
     def detect_source_text_query(conn, cursor, statement, parameters, context, executemany):
@@ -159,7 +161,6 @@ def detect_source_text_query(conn, cursor, statement, parameters, context, execu
                     print(f"\n!!! FULL TABLE SCAN DETECTED ON source_text !!!", file=sys.stderr)
                     print(f"Statement: {statement[:500]}", file=sys.stderr)
                     print(f"Parameters: {parameters}", file=sys.stderr)
-                    import traceback
                     print("Call stack:", file=sys.stderr)
                     traceback.print_stack(file=sys.stderr)
                     sys.stderr.flush()
@@ -191,7 +192,6 @@ def shutdown_session_request(exception=None):
             # the session has already been cleaned up or doesn't need cleanup
             pass
         if exception:
-            import sys
             print(f"[DB-SESSION-TEARDOWN] Exception during request: {exception}", file=sys.stderr)
             sys.stderr.flush()
 
@@ -232,7 +232,6 @@ def serve_speech(filename):
     # print(app.config)
     # Log the current git commit hash
     try:
-        import subprocess
         git_hash = subprocess.check_output(
             ["git", "rev-parse", "HEAD"],
             cwd=os.path.dirname(__file__),

zeeguu/api/endpoints/accounts.py

@@ -1,4 +1,5 @@
 import hmac
+import re
 import sqlalchemy
 import flask
 
@@ -248,7 +249,6 @@ def request_email_verification():
     This avoids the "limbo state" where the user is no longer anonymous
     but hasn't verified their email yet.
     """
-    import re
     from zeeguu.core.emailer.email_confirmation import send_email_confirmation
 
     email = request.form.get("email", None)
@@ -388,8 +388,6 @@ def _email_change_error(user, new_email, password):
     by address, not by user, so the code only proves the inbox, and the
     password is what proves the person.
     """
-    import re
-
     if user.is_anonymous():
         return "Anonymous accounts add an email by upgrading the account"
     if not new_email or not password:
@@ -504,8 +502,6 @@ def send_code(email):
 @api.route("/reset_password/<email>", methods=["POST"])
 @cross_domain
 def reset_password(email):
-    from zeeguu.logging import log
-
     log(f"PASSWORD RESET ATTEMPT: email='{email}'")
 
     code = request.form.get("code", None)

zeeguu/api/endpoints/activity_tracking.py

@@ -1,3 +1,6 @@
+from datetime import datetime
+import json
+
 import flask
 from flask import request
 
@@ -59,8 +62,6 @@ def days_since_last_use():
     or an empty string in case there is no user activity event.
     """
 
-    from datetime import datetime
-
     last_active_time = UserActivityData.get_last_activity_timestamp(flask.g.user_id)
 
     if last_active_time:
@@ -82,7 +83,6 @@ def _update_reading_completion_on_scroll(user, form_data):
     try:
         from zeeguu.core.model import Article, UserArticle
         from zeeguu.core.behavioral_modeling import find_last_reading_percentage
-        import json
 
         article_id = int(form_data.get("article_id"))
         article = Article.find_by_id(article_id)
@@ -131,8 +131,6 @@ def _update_reading_completion_on_scroll(user, form_data):
 
         # Check if article is completed (>90%) and not already marked
         if completion_percentage > 0.9 and not user_article.completed_at:
-            from datetime import datetime
-
             user_article.completed_at = datetime.now()
 
             from zeeguu.core import events

zeeguu/api/endpoints/article.py

@@ -28,6 +28,8 @@
 import requests
 from io import BytesIO
 from urllib.parse import quote
+from datetime import datetime
+import traceback
 from markupsafe import escape
 from PIL import Image
 from zeeguu.config import ZEEGUU_DATA_FOLDER
@@ -311,8 +313,6 @@ def find_or_create_article():
 
         print_and_log_to_sentry(e)
 
-        import traceback
-
         traceback.print_exc()
 
         flask.abort(500)
@@ -498,7 +498,6 @@ def translate_and_adapt_article():
         JSON with translated article data and fragments
     """
     from zeeguu.core.llm_services.simplification_service import SimplificationService
-    from zeeguu.core.model import User
     from zeeguu.core.model.article import Article
     from zeeguu.core.content_retriever import download_and_parse
 
@@ -637,7 +636,6 @@ def translate_and_adapt_article():
         from zeeguu.api.endpoints.helpers.translated_article import (
             save_translated_article,
         )
-        from datetime import datetime
 
         # Create URL and language objects with unique translated URL
         translated_url = Url.find_or_create(db_session, translated_url_key)
@@ -740,7 +738,6 @@ def simplify_article(article_id):
         flask.abort(404, "Article not found")
 
     # Get the current user's CEFR level
-    from zeeguu.core.model import User
 
     user = User.find_by_id(flask.g.user_id)
     try:
@@ -812,8 +809,6 @@ def simplify_article(article_id):
             )
 
     except Exception as e:
-        from zeeguu.logging import log
-
         log(f"Simplification failed for article {article_id}: {str(e)}")
         return json_result({"status": "failed", "error": str(e)[:200]})
 

zeeguu/api/endpoints/article_upload.py

@@ -1,4 +1,6 @@
 """Endpoints for ArticleUpload — lightweight per-user ingestion entity."""
+from datetime import datetime
+
 import flask
 from flask import request
 
@@ -245,7 +247,6 @@ def article_upload_translate_and_adapt(upload_id):
     from zeeguu.api.endpoints.helpers.translated_article import (
         save_translated_article,
     )
-    from datetime import datetime
 
     user = User.find_by_id(flask.g.user_id)
     upload = _find_upload_or_404(upload_id, user)

zeeguu/api/endpoints/audio_lessons.py

@@ -1,3 +1,5 @@
+from datetime import datetime, timedelta
+
 import flask
 from markupsafe import escape
 
@@ -134,7 +136,6 @@ def generate_daily_lesson():
     is_general_topic = False
     if suggestion and lesson_type in ("topic", "situation"):
         # Reuse canonical form if this user has typed the same thing before
-        from zeeguu.core.model import DailyAudioLesson
         cached_canonical = DailyAudioLesson.find_canonical_for_raw_suggestion(user, suggestion)
         if cached_canonical:
             canonical_suggestion = cached_canonical
@@ -528,7 +529,6 @@ def get_audio_lesson_generation_progress():
         return json_result({"progress": None})
 
     # Detect stuck generations (e.g., server restarted mid-generation)
-    from datetime import datetime, timedelta
     if progress.updated_at and datetime.utcnow() - progress.updated_at > timedelta(minutes=2):
         progress.mark_error("Generation appears to have stopped. Please try again.")
         db.session.commit()

zeeguu/api/endpoints/bookmarks_and_words.py

@@ -291,10 +291,6 @@ def delete_bookmark(bookmark_id):
 @cross_domain
 @requires_session
 def set_user_word_exercise_preference(bookmark_id):
-    from zeeguu.core.word_scheduling.basicSR.four_levels_per_word import (
-        FourLevelsPerWord,
-    )
-
     bookmark = get_bookmark_if_owned(bookmark_id)
     user_word = bookmark.user_word
 
@@ -673,7 +669,6 @@ def set_preferred_bookmark(user_word_id):
     }
     """
     from zeeguu.core.model import UserWord, Bookmark
-    from flask import request
 
     user = User.find_by_id(flask.g.user_id)
     user_word = UserWord.query.get(user_word_id)

zeeguu/api/endpoints/crawl_stats.py

@@ -4,6 +4,7 @@
 """
 
 from datetime import datetime, timedelta
+from collections import defaultdict
 
 from flask import request
 from markupsafe import escape
@@ -158,7 +159,6 @@ def crawl_stats_today():
 def crawl_stats_week():
     """Get stats for the last 7 days."""
     # Temporarily set days param
-    from flask import request
     request.args = {"days": "7"}
     return crawl_stats()
 
@@ -245,7 +245,6 @@ def crawl_stats_dashboard():
                 .all()
             )
             # Group by feed_id
-            from collections import defaultdict
             topics_by_feed = defaultdict(list)
             for feed_id, topic_title, count in feed_topic_stats:
                 topics_by_feed[feed_id].append((topic_title, count))
@@ -392,7 +391,6 @@ def crawl_stats_dashboard():
             <h2>🏷️ Simplified by Topic per Language (Today)</h2>"""
 
     # Group topic stats by language
-    from collections import defaultdict
     topics_by_lang = defaultdict(list)
     for lang_code, lang_name, topic_id, topic_title, count in topic_stats_by_lang:
         topics_by_lang[lang_code].append((lang_name, topic_id, topic_title, count))

zeeguu/api/endpoints/generated_examples.py

@@ -1,5 +1,6 @@
 import json
 import traceback
+import re
 
 import flask
 import sentry_sdk
@@ -697,7 +698,6 @@ def preview_learning_card():
 
         # Parse CEFR level from level_note (e.g., "appropriate for B1 - ..." → "B1")
         level_note = learning_card.get("level_note", "")
-        import re
         cefr_match = re.search(r'\b(A1|A2|B1|B2|C1|C2)\b', level_note)
         if cefr_match:
             meaning.word_cefr_level = cefr_match.group(1)

zeeguu/api/endpoints/own_texts.py

@@ -1,3 +1,5 @@
+import os
+
 import flask
 import sqlalchemy
 from flask import request
@@ -161,7 +163,6 @@ def estimate_article_cefr():
 
     if ml_assessment is None:
         # Check if model exists for this language
-        import os
         data_folder = os.environ.get("ZEEGUU_DATA_FOLDER")
         model_path = None
         if data_folder:

zeeguu/api/endpoints/speech.py

@@ -1,5 +1,6 @@
 import os.path
 import re
+import hashlib
 
 from flask import request
 
@@ -148,7 +149,6 @@ def _file_name_for_phrase(phrase, language_id):
 
 def _file_name_for_full_article(full_article_text, language_id, article_id):
     # create md5 hash of the phrase and return it
-    import hashlib
 
     m = hashlib.md5()
     m.update(full_article_text.encode("utf-8"))

zeeguu/api/endpoints/teacher_dashboard/cohorts.py

@@ -64,8 +64,6 @@ def remove_cohort(cohort_id):
     Also removes all students from the cohort.
 
     """
-    from zeeguu.core.model import TeacherCohortMap, UserCohortMap
-
     check_permission_for_cohort(cohort_id)
 
     try:
@@ -122,8 +120,6 @@ def _link_teacher_cohort(user_id, cohort_id):
         """
         Takes user_id and cohort_id and links them together in teacher_cohort_map table.
         """
-        from zeeguu.core.model import TeacherCohortMap
-
         user = User.find_by_id(user_id)
         cohort = Cohort.find(cohort_id)
         db.session.add(TeacherCohortMap(user, cohort))

zeeguu/api/endpoints/translation.py

@@ -1,6 +1,8 @@
 import os
 from string import punctuation
 from urllib.parse import unquote_plus
+import json
+import time
 
 import flask
 from flask import request
@@ -133,7 +135,6 @@ def translate_word(from_lang_code, to_lang_code):
     else:
         # TODO: must remove theurl, and title - they are not used in the calling method.
         from zeeguu.logging import log
-        import time
 
         correction = None if is_separated_mwe else _shared_correction(
             word_str, full_sentence_context or context, from_lang_code, to_lang_code
@@ -324,7 +325,6 @@ def get_translations_stream(from_lang_code, to_lang_code):
 
     :return: SSE stream with translation events
     """
-    import json
     from flask import Response
     from zeeguu.core.translation_services.translator import get_translations_streaming
 
@@ -376,7 +376,6 @@ def translation_alternatives(from_lang_code, to_lang_code):
     from zeeguu.core.translation_services.translator import (
         _vote_translation,
     )
-    from python_translators.translation_query import TranslationQuery
 
     word_str = request.json.get("word", "").strip(punctuation_extended)
     context = request.json.get("context", "").strip()
@@ -613,8 +612,6 @@ def update_bookmark_translation_only(bookmark_id):
     :param bookmark_id: ID of bookmark to update
     :return: Updated bookmark as JSON
     """
-    from zeeguu.api.utils.json_result import json_result
-
     # Find bookmark with proper error handling
     # Fixes ZEEGUU-WEB-B2: Bookmark.find() throws NoResultFound, not returns None
     try:

zeeguu/api/endpoints/user_article.py

@@ -9,6 +9,7 @@
 
 from datetime import datetime
 import json
+import traceback
 
 import newspaper
 
@@ -112,7 +113,6 @@ def generate():
                 yield f"event: complete\ndata: {json.dumps(article_info)}\n\n"
 
             except Exception as e:
-                import traceback
                 traceback.print_exc()
                 yield f"event: error\ndata: {json.dumps({'error': str(e)})}\n\n"
 

zeeguu/api/endpoints/user_articles.py

@@ -1,3 +1,5 @@
+import traceback
+
 import flask
 
 from zeeguu.core.content_recommender import (
@@ -320,8 +322,6 @@ def user_articles_foryou():
         articles = content_recommendations(user.id, user.learned_language_id)
         print("Sending CB recommendations")
     except Exception as e:
-        import traceback
-
         traceback.print_exc()
         print("Failed with: ", e)
         capture_exception(e)

zeeguu/api/endpoints/user_stats.py

@@ -232,8 +232,6 @@ def get_browsing_stats_for_user(user_id, start, end):
 
 def get_translations_for_user(user_id, start, end):
     """Get translations/bookmarks made by user in the given period."""
-    from zeeguu.core.model.bookmark import Bookmark
-
     bookmarks = (
         Bookmark.query.join(UserWord, Bookmark.user_word_id == UserWord.id)
         .filter(UserWord.user_id == user_id)