from fastapi import HTTPException import os from ..utils.hashing import HashGenerator from user_journey_service.crew import UserJourney from pathlib import Path import whisper import json import re class EvaluationService: def __init__(self,input_data): self.crew_instance = UserJourney() self.input_data = input_data self.input_hash = HashGenerator.generate_input_hash(input_data) self.parsed = Path(f"parsed_course_content/{self.input_hash}.json") self.session_audio = Path(f"audio/stage/{self.input_hash}/") self.session_question = Path(f"audio/questions/{self.input_hash}/") self.session_content = Path(f"updated_content/{self.input_hash}/") self.session_question_md = Path(f"question/{self.input_hash}/") def verify_stage_md_wav(self): md_files = [f for f in os.listdir(self.session_content) if f.endswith(".md")] wav_files = [f for f in os.listdir(self.session_audio) if f.endswith(".wav")] wav_stages_set = set(os.path.splitext(f)[0] for f in wav_files) missing = [] for md_file in md_files: stage_name = os.path.splitext(md_file)[0] # e.g., 'stage1' if stage_name not in wav_stages_set: missing.append(stage_name) if missing: return False,missing return True, missing def get_audio_files(self): if not os.path.exists(self.session_audio) and not os.path.exists(self.session_question): return {"status": "failure", "message": "Audio files are not generated."} audio_wav_files = sorted(self.session_audio.glob('*.wav')) print(audio_wav_files) question_wav_files = sorted(self.session_question.glob('*.wav')) print(question_wav_files) # return {"content":audio_wav_files,"question":question_wav_files} return audio_wav_files import os def generate_stage_file_mapping(self): match,missing = self.verify_stage_md_wav() if match: # Open and load JSON # with open(self.parsed, "r") as file: # data = json.load(file) md_files = sorted([f for f in os.listdir(self.session_content) if f.endswith(".md")]) stage_data = [] for md_file in md_files: stage_name = os.path.splitext(md_file)[0] # 'stage1' wav_file = f"{stage_name}.wav" md_path = os.path.join(self.session_content, md_file) wav_path = os.path.join(self.session_audio, wav_file) stage_data.append({ "stage": stage_name, "md_path": md_path, "wav_path": wav_path }) return stage_data else: return missing def update_parsed_course_content(self): # Open and load JSON with open(self.parsed, "r") as file: data = json.load(file) # Add md_path and wav_path to each stage for i, stage in enumerate(data["stages"], start=1): wav_file = f"stage{i}.wav" md_file = f"stage{i}.md" stage["stage_no"] = f"stage{i}" stage["md_path"] = os.path.join(self.session_content, md_file) stage["wav_path"] = os.path.join(self.session_audio, wav_file) return data["stages"] def display_content(self,stage_data): md_path = stage_data["md_path"] print(f"The md path is : {md_path}") audio_content = stage_data["wav_path"] with open(md_path, 'r', encoding='utf-8') as f: content = f.read() return {"content":content,"audio":audio_content} def display_question(self,file): # Extract the stage (the folder name before the filename) p = Path("audio/questions/f3aafd4c95c9b2d02af0b6e68e65745d/stage1/1.wav") hash_id = p.parent.parent.name # "f3aafd4c95c9b2d02af0b6e68e65745d" stage = p.parent.name # "stage1" prefix = p.stem # "1" print("Hash ID:", hash_id) print("Stage:", stage) print("Prefix:", prefix) stage = stage + ".md" qn_stage_file = self.session_question_md / stage with open(qn_stage_file, 'r', encoding='utf-8') as f: content = f.read() try: # Regex to capture Q{prefix} pattern = rf"### Q{prefix}:\s*(.*)" match = re.search(pattern, content) if match: question = match.group(1).strip() print(f"Question for prefix {prefix}:", question) except: question = self.wav_to_text_whisper(file) return {"question":question, "audio":file} def list_qns(self,stage_data): inputs = self.input_data.dict() print(f"The inputs are : {inputs}") # audio_question = Path(f'audio/questions/{self.input_hash}/stage{idx}/') all_qns_answer ={} stage_content = stage_data["md_path"] audio_content = stage_data["wav_path"] stage_value = stage_data["stage_no"] audio_question = str(self.session_question) + "/" + stage_value + "/" print(f"the audio questions are in : {audio_question}") wav_files = sorted(Path(audio_question).glob('*.wav')) print(f"the question wav files are : {wav_files}") return wav_files def evaluate_individual_qn(self,wav_file): inputs = self.input_data.dict() print("Just before wav print") print(f"The inputs are : {inputs}") wav_path = Path(wav_file) print(wav_file) question = self.wav_to_text_whisper(wav_file) input("🎤 Press Enter when you're ready to answer...") print("📢 Listening to your answer...") answer = str(input("Enter your answer: ")) inputs["question"] = question inputs["answer"] = answer # inputs = {"question":question,"answer":answer} # all_qns_answer[question] = answer crew = self.crew_instance.evaluator_crew() feedback = crew.kickoff(inputs=inputs) return feedback.raw # inputs["users_response"] = all_qns_answer # crew = self.crew_instance.assessment_crew() # crew.kickoff(inputs=inputs) def evaluate_individual_qn_ans_pair(self,question,answer): #### added on Nov 13th 2025 inputs = self.input_data.dict() # print("Just before wav print") # print(f"The inputs are : {inputs}") # wav_path = Path(wav_file) # print(wav_file) # question = self.wav_to_text_whisper(wav_file) # input("🎤 Press Enter when you're ready to answer...") # print("📢 Listening to your answer...") # answer = str(input("Enter your answer: ")) inputs["question"] = question inputs["answer"] = answer # inputs = {"question":question,"answer":answer} # all_qns_answer[question] = answer crew = self.crew_instance.evaluator_crew() feedback = crew.kickoff(inputs=inputs) return feedback.raw # inputs["users_response"] = all_qns_answer # crew = self.crew_instance.assessment_crew() # crew.kickoff(inputs=inputs) def run_evaluation(self,stage_data): inputs = self.input_data.dict() print(f"The inputs are : {inputs}") # audio_question = Path(f'audio/questions/{self.input_hash}/stage{idx}/') all_qns_answer ={} stage_content = stage_data["md_path"] audio_content = stage_data["wav_path"] stage_value = stage_data["stage"] audio_question = str(self.session_question) + "/" + stage_value + "/" print(f"the audio questions are in : {audio_question}") wav_files = sorted(Path(audio_question).glob('*.wav')) print(f"the question wav files are : {wav_files}") for wav_file in wav_files: print(wav_file) question = self.wav_to_text_whisper(wav_file) input("🎤 Press Enter when you're ready to answer...") print("📢 Listening to your answer...") answer = str(input("Enter your answer: ")) inputs["question"] = question inputs["answer"] = answer # inputs = {"question":question,"answer":answer} all_qns_answer[question] = answer crew = self.crew_instance.evaluator_crew() crew.kickoff(inputs=inputs) inputs["users_response"] = all_qns_answer crew = self.crew_instance.assessment_crew() crew.kickoff(inputs=inputs) def wav_to_text_whisper(self,wav_file): print("inside wave to text conversion") model = whisper.load_model("base") result = model.transcribe(str(wav_file)) # print("Transcription:", result["text"]) print("\U0001F3A4 Transcription:", result["text"]) return result["text"]