Refactored services

Author: G1NG3R-exe

app/services/__init__.py

@@ -0,0 +1,19 @@
+from .test_services import (
+    generate_test_service,
+    evaluate_answers_service,
+    calculate_percentage_score_service
+)
+
+from .course_services import (
+    generate_knowledge_assessment_service,
+    create_course_service,
+    generate_improved_course_name
+)
+
+from .lesson_services import (
+    generate_lesson_content_service
+)
+
+from .tutor_services import get_tutor_response_service
+
+from .edit_services import edit_course_service

app/services/course_services.py

@@ -0,0 +1,64 @@
+from app.ai_clients import ask_ai, ask_gemini
+from app.models import db, Course, Lesson
+from models.json_extractor import JsonExtractor
+from models.prompt_builders import CoursePromptBuilder
+from .utils import update_token_count
+from .test_services import calculate_percentage_score_service
+
+def generate_knowledge_assessment_service(detailed_results):
+    prompt = ("Provide a concise, one or two paragraph assessment of the user's knowledge "
+              "of the topic based on the test results below. Do not address the user directly; "
+              "use third-person phrasing. Do not use markdown.\n\n")
+    for result in detailed_results:
+        prompt += f"Q: {result['question']}\nA: {result['answer']}\nAssessment: {result['assessment']}\n\n"
+
+    assessment_text, tokens = ask_ai(prompt, model="gpt-4o-mini", json_mode=False)
+    update_token_count(tokens)
+    return assessment_text if assessment_text and "Error:" not in assessment_text else "Could not generate assessment."
+
+
+def create_course_service(user, topic, knowledge_assessment, assessed_answers):
+    prompt = CoursePromptBuilder.build_course_structure_prompt(
+        topic, knowledge_assessment, assessed_answers, user.language, user
+    )
+    raw_output, tokens = ask_ai(prompt, json_mode=True)
+    update_token_count(tokens)
+
+    if not raw_output or "Error:" in raw_output:
+        print(f"Failed to generate course structure. AI response: {raw_output}")
+        return None
+
+    course_data = JsonExtractor.extract_json(raw_output)
+    if not course_data:
+        print("Failed to parse course structure from AI response.")
+        return None
+
+    from .course_services import generate_improved_course_name
+    original_name = course_data.get("course_title", topic)
+    improved_name = generate_improved_course_name(original_name)
+    course_data['course_title'] = improved_name
+
+    course = Course(user_id=user.id, course_title=improved_name, course_data=course_data)
+    db.session.add(course)
+    db.session.commit()
+
+    for unit in course_data.get('units', []):
+        for lesson_data in unit.get('lessons', []):
+            new_lesson = Lesson(course_id=course.id,
+                                unit_title=unit.get('unit_title'),
+                                lesson_title=lesson_data.get('lesson_title'))
+            db.session.add(new_lesson)
+    db.session.commit()
+    return course
+
+
+def generate_improved_course_name(original_name):
+    prompt = (f"Improve this course name to be more concise and informative. Keep it under 60 characters. "
+              f"Original name: {original_name}. Return ONLY the improved name, no quotes or additional text.")
+    improved_name, tokens = ask_gemini(prompt)
+    if not improved_name or "Error:" in improved_name:
+        improved_name, tokens = ask_ai(prompt)
+    update_token_count(tokens)
+
+    improved_name = improved_name.strip('"\'').strip()
+    return improved_name if improved_name else original_name

app/services/edit_services.py

@@ -0,0 +1,76 @@
+from app.models import db, Lesson
+from app.ai_clients import ask_ai
+from models.json_extractor import JsonExtractor
+from models.prompt_builders import CourseEditorPromptBuilder
+from .utils import update_token_count
+
+def edit_course_service(course, user_request, language):
+    print(f"[DEBUG] edit_course_service called with request: {user_request}")
+
+    title_keywords = ["title", "name", "rename", "call this"]
+    is_title_update = any(k in user_request.lower() for k in title_keywords)
+
+    if is_title_update and "course_title" in course.course_data:
+        print("[DEBUG] Processing title update request")
+        prompt = CourseEditorPromptBuilder.build_title_improvement_prompt(user_request, language)
+        new_title, tokens = ask_ai(prompt, model="gpt-4o-mini", json_mode=False)
+        update_token_count(tokens)
+
+        if new_title and "Error:" not in new_title:
+            new_title = new_title.strip('\'" ')
+            course_data = course.course_data
+            course_data["course_title"] = new_title
+            course.course_data = course_data
+            db.session.commit()
+            return course_data, None
+        else:
+            err = f"Failed to generate an improved title. Response: {new_title}"
+            print(f"[ERROR] {err}")
+            return None, err
+
+    print("[DEBUG] Processing full course edit request")
+    prompt = CourseEditorPromptBuilder.build_edit_prompt(course.course_data, user_request, language)
+    new_course_str, tokens = ask_ai(prompt, model="gpt-4o", json_mode=True)
+    update_token_count(tokens)
+
+    if not new_course_str or "Error:" in new_course_str:
+        err = f"AI failed to generate a new course structure. Response: {new_course_str}"
+        print(f"[ERROR] {err}")
+        return None, err
+
+    try:
+        new_course_json = JsonExtractor.extract_json(new_course_str)
+        if 'course_title' not in new_course_json or 'units' not in new_course_json:
+            raise ValueError(f"Invalid JSON structure returned by AI: {new_course_json}")
+    except Exception as e:
+        err = f"Error parsing new course structure: {str(e)}"
+        print(f"[ERROR] {err}")
+        return None, err
+
+    try:
+        course.course_data = new_course_json
+        new_titles = {l['lesson_title'] for u in new_course_json['units'] for l in u.get('lessons', []) if 'lesson_title' in l}
+
+        existing_lessons = {l.lesson_title: l for l in Lesson.query.filter_by(course_id=course.id).all()}
+        for title, obj in existing_lessons.items():
+            if title not in new_titles:
+                db.session.delete(obj)
+
+        for u in new_course_json['units']:
+            for l in u.get('lessons', []):
+                t = l.get('lesson_title')
+                if t and t not in existing_lessons:
+                    db.session.add(Lesson(course_id=course.id, unit_title=u['unit_title'], lesson_title=t))
+
+        course.completed_lessons = Lesson.query.filter_by(course_id=course.id, is_completed=True).count()
+        db.session.commit()
+        print("[DEBUG] Course update completed successfully")
+        return new_course_json, None
+
+    except Exception as e:
+        db.session.rollback()
+        err = f"Database error during course update: {e}"
+        print(f"[ERROR] {err}")
+        import traceback
+        traceback.print_exc()
+        return None, err

app/services/lesson_services.py

@@ -0,0 +1,75 @@
+import re
+import markdown
+from flask import url_for
+from app.ai_clients import ask_ai_stream
+from app.models import db, Lesson
+from models.prompt_builders import LessonPromptBuilder
+
+def generate_lesson_content_service(lesson, user):
+    user_profile = {'age': user.age, 'bio': user.bio}
+    course_structure = lesson.course.course_data
+
+    prompt = LessonPromptBuilder.build_lesson_content_prompt(
+        lesson.lesson_title, lesson.unit_title, user.language, user.preferred_lesson_length,
+        user_profile=user_profile, course_structure=course_structure
+    )
+
+    def content_generator():
+        response_stream = ask_ai_stream(prompt, model="gpt-4o")
+        full_markdown_chunks = []
+
+        for chunk in response_stream:
+            yield chunk
+            full_markdown_chunks.append(chunk)
+
+        full_markdown_text = "".join(full_markdown_chunks)
+        if "Error:" in full_markdown_text:
+            lesson.html_content = "<p>Error generating lesson content. Please try again later.</p>"
+        else:
+            next_up_link_md = _generate_next_up_link(lesson, user)
+            final_md = full_markdown_text + next_up_link_md
+            processed_text = re.sub(r'\[IMAGE_PROMPT:\s*"(.*?)"\]', r'<i>[Image Prompt: "\1"]</i>', final_md)
+            lesson.html_content = markdown.markdown(processed_text, extensions=["fenced_code", "tables"])
+
+        if not lesson.is_completed:
+            lesson.is_completed = True
+            course = lesson.course
+            course.completed_lessons = Lesson.query.filter_by(course_id=course.id, is_completed=True).count()
+
+        db.session.commit()
+
+    return content_generator()
+
+
+def _generate_next_up_link(lesson, user):
+    course = lesson.course
+    lang = user.language
+    try:
+        units = course.course_data.get('units', [])
+        current_unit_index, current_unit = next(
+            ((i, u) for i, u in enumerate(units) if u['unit_title'] == lesson.unit_title), (None, None))
+        if not current_unit:
+            return ""
+
+        lessons = current_unit.get('lessons', [])
+        current_index = next(
+            (i for i, l in enumerate(lessons) if l['lesson_title'] == lesson.lesson_title), None)
+        if current_index is None:
+            return ""
+
+        if current_index + 1 < len(lessons):
+            next_lesson = lessons[current_index + 1]
+            next_lesson_obj = Lesson.query.filter_by(course_id=course.id,
+                                                     lesson_title=next_lesson.get('lesson_title')).first()
+            if next_lesson_obj:
+                url = url_for('loading_lesson', lesson_id=next_lesson_obj.id, _external=True)
+                return f"\n\n<hr>\n\n### 👉 {'Далее' if lang == 'russian' else 'Next up'}: [{next_lesson['lesson_title']}]({url})"
+
+        test_data = current_unit.get('test')
+        if test_data and test_data.get('test_title'):
+            url = url_for('loading_unit_test', course_id=course.id,
+                          unit_title=current_unit['unit_title'], test_title=test_data['test_title'], _external=True)
+            return f"\n\n<hr>\n\n### 👉 {'Далее' if lang == 'russian' else 'Next up'}: [{test_data['test_title']}]({url})"
+    except Exception as e:
+        print(f"Error generating 'Next up' link: {e}")
+    return ''

app/services/test_services.py

@@ -0,0 +1,91 @@
+import json
+from models.json_extractor import JsonExtractor
+from models.fulltest import Test
+from models.question import Question
+from app.ai_clients import ask_gemini
+from .utils import update_token_count
+from models.prompt_builders import TestPromptBuilder, AnswerPromptBuilder
+
+def generate_test_service(topic, format_type, additional_context, language, user_profile=None, lesson_content_context=""):
+    prompt = TestPromptBuilder.build_multiple_choice_prompt(
+        topic, additional_context, language,
+        user_profile=user_profile,
+        lesson_content_context=lesson_content_context
+    )
+    raw_output, tokens = ask_gemini(prompt, json_mode=True)
+    update_token_count(tokens)
+
+    if not raw_output or "Error:" in raw_output:
+        print(f"Failed to generate test with Gemini. AI response: {raw_output}")
+        return None
+
+    test_data = JsonExtractor.extract_json(raw_output)
+    questions = [
+        Question(question=q.get("question"), options=q.get("options", {}))
+        for q in test_data.get("questions", [])
+    ]
+
+    return Test(
+        test_name=test_data.get("test-name", "Unnamed Test"),
+        topic=test_data.get("topic", topic),
+        questions=questions
+    )
+
+
+def evaluate_answers_service(questions, user_answers, language):
+    qa_list = []
+    for q, ua in zip(questions, user_answers):
+        answer_key = ua.get('answer_value', ua.get('answer', ''))
+        answer_text = ua.get('answer', '')
+        if hasattr(q, 'options') and answer_key in q.options:
+            answer_text = q.options[answer_key]
+        qa_list.append({
+            "question": q.question,
+            "answer": answer_text,
+            "original_answer": answer_key
+        })
+
+    from app.ai_clients import ask_gemini
+    prompt = AnswerPromptBuilder.build_batch_check_prompt(
+        [{"question": qa["question"], "answer": qa["answer"]} for qa in qa_list],
+        language
+    )
+    response_text, tokens = ask_gemini(prompt, json_mode=True)
+    update_token_count(tokens)
+
+    if not response_text or "Error:" in response_text:
+        return []
+
+    try:
+        response_json = JsonExtractor.extract_json(response_text)
+        assessments = response_json.get("assessments", [])
+        detailed_results = []
+
+        for i, qa in enumerate(qa_list):
+            assessment = next(
+                (item['assessment'] for item in assessments if item['id'] == i),
+                "Evaluation Error"
+            )
+            detailed_results.append({
+                "question": qa["question"],
+                "answer": qa["answer"],
+                "original_answer": qa["original_answer"],
+                "assessment": assessment
+            })
+        return detailed_results
+
+    except (ValueError, KeyError) as e:
+        print(f"Error parsing Gemini batch assessment response: {e}")
+        return []
+
+
+def calculate_percentage_score_service(detailed_results):
+    from app.ai_clients import ask_gemini
+    prompt = "Based on these answers and evaluations, give a final score from 0-100:\n\n"
+    for result in detailed_results:
+        prompt += f"Q: {result['question']}\nA: {result['answer']}\nAssessment: {result['assessment']}\n\n"
+    prompt += "Return just the number."
+    result_text, tokens = ask_gemini(prompt, json_mode=False)
+    update_token_count(tokens)
+
+    return int(''.join(filter(str.isdigit, result_text))) if result_text and "Error:" not in result_text else 0

app/services/tutor_services.py

@@ -0,0 +1,13 @@
+from app.ai_clients import ask_gemini_stream
+from models.prompt_builders import ChatPromptBuilder
+
+def get_tutor_response_service(lesson, chat_history, user_question, user):
+    lesson_content = lesson.html_content or "Lesson content has not been generated yet."
+    prompt = ChatPromptBuilder.build_tutor_prompt(
+        lesson_content=lesson_content,
+        unit_title=lesson.unit_title,
+        chat_history=chat_history,
+        user_question=user_question,
+        language=user.language
+    )
+    return ask_gemini_stream(prompt)

app/services/utils.py

@@ -0,0 +1,8 @@
+from flask_login import current_user
+from app.models import db
+
+def update_token_count(tokens_to_add):
+    """Helper function to add tokens to the current user's total."""
+    if tokens_to_add > 0:
+        current_user.tokens_used += tokens_to_add
+        db.session.commit()