| from sqlmodel import SQLModel, Field, create_engine, Session, select |
| from datetime import datetime |
| from typing import Optional |
| import os |
| import uuid |
| import subprocess |
| import re |
| import json |
| import matplotlib.pyplot as plt |
| from faster_whisper import WhisperModel |
| from openai import OpenAI |
|
|
| |
| db_path = "/tmp/chatter_sessions.db" |
| engine = create_engine(f"sqlite:///{db_path}") |
| SQLModel.metadata.create_all(engine) |
|
|
| openai_api_key = os.getenv("OPENAI_API_KEY") |
| client = OpenAI(api_key=openai_api_key) |
|
|
| |
| LANG_CODES = { |
| "English": "en", "Spanish": "es", "Hindi": "hi", "French": "fr", "German": "de", |
| "Arabic": "ar", "Chinese": "zh", "Portuguese": "pt", "Japanese": "ja", "Korean": "ko" |
| } |
|
|
| |
| class SessionEntry(SQLModel, table=True): |
| id: Optional[int] = Field(default=None, primary_key=True) |
| user: str |
| timestamp: str |
| transcript: str |
| feedback: str |
| language: str |
|
|
| |
| def save_to_db(user, transcript, feedback, language): |
| session = Session(engine) |
| entry = SessionEntry( |
| user=user, |
| timestamp=datetime.now().strftime("%Y-%m-%d %H:%M"), |
| transcript=transcript, |
| feedback=feedback, |
| language=language |
| ) |
| session.add(entry) |
| session.commit() |
| session.close() |
|
|
| def fetch_user_sessions(user): |
| session = Session(engine) |
| statement = select(SessionEntry).where(SessionEntry.user == user) |
| results = session.exec(statement).all() |
| session.close() |
| return results |
|
|
| |
| model = WhisperModel("base", compute_type="int8") |
|
|
| def convert_to_wav(input_file): |
| output_wav = f"/tmp/{uuid.uuid4()}.wav" |
| command = ["ffmpeg", "-y", "-i", input_file, "-ar", "16000", "-ac", "1", output_wav] |
| subprocess.run(command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) |
| return output_wav |
|
|
| def transcribe_audio(audio_path): |
| segments, _ = model.transcribe(audio_path) |
| return " ".join([segment.text for segment in segments]) |
|
|
| |
| def parse_scores_from_feedback(feedback_text): |
| try: |
| json_match = re.search(r"\{.*?\}", feedback_text, re.DOTALL) |
| if json_match: |
| score_block = json.loads(json_match.group(0)) |
| return {k.strip(): int(v) for k, v in score_block.items() if str(v).isdigit()} |
| except Exception as e: |
| print("Score parsing failed:", e) |
| return {} |
|
|
| def generate_feedback(transcript, language, goal="general improvement", focus_areas=None, previous_transcript=None): |
| focus_str = ", ".join(focus_areas) if focus_areas else "Clarity, Structure, Fluency, Content Relevance, and Tone" |
| history_section = f"\n\nFor reference, their previous transcript was:\n{previous_transcript}" if previous_transcript else "" |
|
|
| prompt = f""" |
| You are a supportive communication coach helping a learner whose goal is: **{goal}**. |
| |
| Evaluate the user's current speech based on the following areas: |
| {focus_str} |
| |
| Give a score out of 10 and a short explanation for each area. |
| |
| Then provide: |
| - A summary of strengths and improvement areas. |
| - One motivational line to end with. |
| |
| Transcript: |
| {transcript} |
| {history_section} |
| """.strip() |
|
|
| response = client.chat.completions.create( |
| model="gpt-4", |
| messages=[ |
| {"role": "system", "content": f"You are a warm and constructive communication coach responding in {language}."}, |
| {"role": "user", "content": prompt} |
| ], |
| temperature=0.7 |
| ) |
| return response.choices[0].message.content |
|
|
|
|
| def generate_progress_summary(current_feedback, previous_feedback): |
| current_scores = parse_scores_from_feedback(current_feedback) |
| previous_scores = parse_scores_from_feedback(previous_feedback) |
| if not current_scores or not previous_scores: |
| return "" |
|
|
| lines = [] |
| for cat in current_scores: |
| if cat in previous_scores: |
| diff = current_scores[cat] - previous_scores[cat] |
| if diff > 0: |
| lines.append(f"✅ **{cat}** improved by **+{diff}**") |
| elif diff < 0: |
| lines.append(f"⚠️ **{cat}** dropped by **{abs(diff)}**") |
| else: |
| lines.append(f"➖ **{cat}** stayed the same") |
|
|
| if lines: |
| return "\n\n**📈 Progress Tracker**\n" + "\n".join(lines) |
| return "" |
|
|
| |
| def build_score_comparison_data(current_feedback, previous_feedback): |
| current = parse_scores_from_feedback(current_feedback) |
| previous = parse_scores_from_feedback(previous_feedback) |
| categories = list(set(current.keys()).union(set(previous.keys()))) |
| return { |
| "Category": categories, |
| "Current Score": [current.get(cat, 0) for cat in categories], |
| "Previous Score": [previous.get(cat, 0) for cat in categories] |
| } |
|
|
| def render_score_chart(data): |
| fig, ax = plt.subplots() |
| x = range(len(data["Category"])) |
| ax.bar([i - 0.2 for i in x], data["Previous Score"], width=0.4, label='Previous') |
| ax.bar([i + 0.2 for i in x], data["Current Score"], width=0.4, label='Current') |
| ax.set_xticks(list(x)) |
| ax.set_xticklabels(data["Category"]) |
| ax.set_ylim(0, 10) |
| ax.legend() |
| ax.set_ylabel("Score") |
| ax.set_title("🎯 Score Comparison") |
| return fig |
|
|
| |
| def build_trend_data(sessions, category="Clarity"): |
| points = [] |
| timestamps = [] |
| for s in sessions: |
| score = parse_scores_from_feedback(s.feedback).get(category) |
| if score is not None: |
| points.append(score) |
| timestamps.append(s.timestamp.split()[0]) |
| return timestamps, points |
|
|
| def render_trend_chart(timestamps, points, category="Clarity"): |
| fig, ax = plt.subplots() |
| ax.plot(timestamps, points, marker="o", linestyle="-", color="blue") |
| ax.set_title(f"📈 {category} Progress Over Time") |
| ax.set_ylim(0, 10) |
| ax.set_ylabel("Score") |
| ax.set_xlabel("Date") |
| return fig |
|
|
