import time import logging import openai import re from models import get_db_connection, add_ques_llm, create_quiz_by_id import os logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", filename=os.path.join('/tmp', 'app.log'), handlers=[ logging.StreamHandler() ] ) client = openai.OpenAI( base_url="https://api.groq.com/openai/v1", api_key=os.getenv("GROQ_API_KEY") ) def generate_question_random(theme): ques_prompt = ( f"Generate only one unique, factually accurate quiz question on the theme of {theme}. \n " "Make sure each time the question generated is very unique and from a new topic. Make the question as innovative as possible so that it doesn't get repeated." "Generate question on a different theme each time involve literature, sports, science, mathematics, technology and other themes" "Question generated in each chat has to be a new question. Don't generate even a similar question to previous one." "Ensure that:\n" "1. The question is unique.\n" "2. Options: A, B, C, D.\n" "3. Correct answer should be marked.\n" "4. Difficulty: Easy, Medium, Hard.\n" "Provide response in this format:\n" "Theme: [Theme]\n" "Question: [Your question here]\n" "A) [Option A]\n" "B) [Option B]\n" "C) [Option C]\n" "D) [Option D]\n" "Correct answer: [Correct option letter]\n" "Difficulty level: [Easy/Medium/Hard]" ) response = client.chat.completions.create( model="llama-3.3-70b-versatile", messages=[ {"role": "system", "content": "You are a helpful assistant who generates quizzes."}, {"role": "user", "content": ques_prompt} ], temperature=0.5, max_tokens=1024 ) return response.choices[0].message.content def parse_question(response_text): """Extracts theme, question, options, correct answer, and difficulty from LLM response.""" question_data = {} question_match = re.search(r'Question:\s*(.*)', response_text) options_matches = re.findall(r'([A-D])\)\s*(.*)', response_text) correct_answer_match = re.search(r'Correct answer:\s*([A-D])', response_text) difficulty_match = re.search(r'Difficulty level:\s*(.*)', response_text) if question_match and options_matches and correct_answer_match: question = question_match.group(1).strip() question_data["question"] = question options = [f"{opt[0]}) {opt[1].strip()}" for opt in options_matches] question_data["options"] = options correct_option = correct_answer_match.group(1).strip().upper() question_data["correct_answer"] = correct_option difficulty = difficulty_match.group(1).strip() question_data["difficulty"] = difficulty logging.debug("Parsed question: %s", question) logging.debug("Parsed options: %s", options) logging.debug("Parsed correct option: %s", correct_option) logging.debug("Parsed difficulty: %s", difficulty) if correct_option not in ['A', 'B', 'C', 'D']: raise ValueError("The correct option is invalid.") return question_data if "question" in question_data else None else: raise ValueError("Question format is incorrect or missing correct option.") def generate_quiz_daily(): connection = get_db_connection() if not connection: logging.error("Failed to connect to DB in generate_quiz_daily") return try: cursor = connection.cursor() cursor.execute("SELECT theme FROM themes") themes = [row[0] for row in cursor.fetchall()] for theme in themes: theme = theme.lower() try: logging.info("Creating daily quiz from bank for theme: %s", theme) result = create_quiz_by_id(theme, 5) logging.info("Quiz created: %s", result) except Exception as e: logging.error("Failed to create quiz for theme %s: %s", theme, e) finally: cursor.close() connection.close() def continuous_generation(): generation_count = 0 last_quiz_creation = 0 while True: logging.info("Generating a new quiz question...") connection = get_db_connection() if not connection: logging.error("Failed to connect to DB in generate_quiz_daily") return try: cursor = connection.cursor() cursor.execute("SELECT theme FROM themes") themes = [row[0] for row in cursor.fetchall()] for theme in themes: theme = theme.lower() try: logging.info("Creating daily question from bank for theme: %s", theme) question_text = generate_question_random(theme) if question_text: try: question_data = parse_question(question_text) if question_data: # theme = question_data["theme"].lower() # added = add_theme_if_not_exists(theme) logging.info( "Storing Question - Theme: %s | Question: %s | Answer: %s | Difficulty: %s", theme, question_data["question"], question_data["correct_answer"], question_data["difficulty"] ) result = add_ques_llm( theme, question_data["question"], question_data["options"], question_data["correct_answer"], question_data["difficulty"] ) if result == "True": logging.info("New quiz question added successfully.") else: logging.warning("Failed to add question to DB.") continue else: logging.warning("Could not parse generated question.") except Exception as e: logging.error("Error processing quiz data: %s", e) logging.info("Question created") except Exception as e: logging.error("Failed to generate question for theme %s: %s", theme, e) finally: cursor.close() connection.close() generation_count += 1 if time.time() - last_quiz_creation >= 86400: logging.info("Running daily quiz creation for all themes...") generate_quiz_daily() last_quiz_creation = time.time() time.sleep(20) if __name__ == '__main__': import time import traceback while True: try: continuous_generation() except Exception as e: print("continuous_generation crashed:", e) traceback.print_exc() print("Restarting in 5 seconds...") time.sleep(5) # from app import app # if __name__ == '__main__': # import time # import traceback # with app.app_context(): # while True: # try: # continuous_generation() # except Exception as e: # print("continuous_generation crashed:", e) # traceback.print_exc() # print("Restarting in 5 seconds...") # time.sleep(5)