app.py

# espeak.py
import os
import json
import time
import gradio as gr
import speech_recognition as sr
import pyttsx3
import threading

from typing import Tuple

# Try importing OpenAI; if not present app will use local model fallback
USE_OPENAI = bool(os.getenv("OPENAI_API_KEY", "").strip())
if USE_OPENAI:
    import openai

# Local model fallback (T5-based)
try:
    import torch
    from transformers import AutoTokenizer, AutoModelForSeq2SeqLM
    from happytransformer import HappyTextToText, TTSettings
    LOCAL_MODEL_AVAILABLE = True
except Exception:
    LOCAL_MODEL_AVAILABLE = False

# Optional: Levenshtein for better scoring
try:
    import Levenshtein
    _have_lev = True
except Exception:
    _have_lev = False

APP_TITLE = "ESPeak — AI Grammar & Speech Assistant"

# ----------------------
# Utilities
# ----------------------
def levenshtein_distance(a: str, b: str) -> int:
    if _have_lev:
        return Levenshtein.distance(a, b)
    # fallback simple DP (O(len(a)*len(b))) — fine for short sentences
    la, lb = len(a), len(b)
    if la == 0: return lb
    if lb == 0: return la
    dp = [[0]*(lb+1) for _ in range(la+1)]
    for i in range(la+1):
        dp[i][0] = i
    for j in range(lb+1):
        dp[0][j] = j
    for i in range(1, la+1):
        for j in range(1, lb+1):
            cost = 0 if a[i-1]==b[j-1] else 1
            dp[i][j] = min(dp[i-1][j]+1, dp[i][j-1]+1, dp[i-1][j-1]+cost)
    return dp[la][lb]

def score_from_edit(orig: str, corrected: str) -> int:
    # Compute a simple score: smaller edit distance -> higher score
    if not orig.strip():
        return 0
    dist = levenshtein_distance(orig, corrected)
    # Normalize by length, clamp to [0,100]
    norm = max(len(orig), 1)
    ratio = max(0.0, 1.0 - dist / norm)
    score = int(round(ratio * 100))
    return score

# ----------------------
# Model loading
# ----------------------
tokenizer = model = happy_tt = None
if not USE_OPENAI and LOCAL_MODEL_AVAILABLE:
    def load_local_models():
        global tokenizer, model, happy_tt
        model_name = "prithivida/grammar_error_correcter_v1"
        try:
            tokenizer = AutoTokenizer.from_pretrained(model_name)
            model = AutoModelForSeq2SeqLM.from_pretrained(model_name)
            happy_tt = HappyTextToText("T5", model_name)
        except Exception as e:
            print("Local model load failed:", e)
            raise
    load_local_models()

# ----------------------
# Speech transcription helper (speech_recognition)
# ----------------------
def transcribe_audio_file(audio_filepath: str) -> str:
    r = sr.Recognizer()
    try:
        with sr.AudioFile(audio_filepath) as source:
            audio_data = r.record(source)
        text = r.recognize_google(audio_data)
        return text
    except sr.UnknownValueError:
        return ""
    except Exception as e:
        return f"[transcription_error]: {str(e)}"

# ----------------------
# LLM connectors
# ----------------------
OPENAI_PROMPT_SYSTEM = (
    "You are ESPeak Assistant — expert grammar corrector. "
    "Return JSON only with keys: corrected_text (string), score (0-100 integer), explanation (short string)."
)

OPENAI_USER_TEMPLATE = (
    "Correct this sentence for grammar, punctuation, and clarity while preserving tone:\n\n"
    "### INPUT\n{input_text}\n\n"
    "Return only JSON with corrected_text, score, and explanation."
)

def call_openai_correct(text: str) -> Tuple[str,int,str]:
    messages = [
        {"role":"system", "content": OPENAI_PROMPT_SYSTEM},
        {"role":"user", "content": OPENAI_USER_TEMPLATE.format(input_text=text)}
    ]
    resp = openai.ChatCompletion.create(
        model="gpt-4o-mini" if "gpt-4o-mini" in openai.Model.list() else "gpt-4",
        messages=messages,
        temperature=0.0,
        max_tokens=300
    )
    content = resp["choices"][0]["message"]["content"].strip()
    # Try to parse JSON from response
    try:
        parsed = json.loads(content)
        corrected = parsed.get("corrected_text", "")
        score = int(parsed.get("score", score_from_edit(text, corrected)))
        explanation = parsed.get("explanation", "")
        return corrected, score, explanation
    except Exception:
        # fallback: use raw content — try to extract a JSON substring
        try:
            start = content.index("{")
            end = content.rindex("}")+1
            data = json.loads(content[start:end])
            corrected = data.get("corrected_text","")
            score = int(data.get("score", score_from_edit(text, corrected)))
            explanation = data.get("explanation","")
            return corrected, score, explanation
        except Exception:
            # Last resort: return plain corrected via model-less heuristic
            corrected = content
            score = score_from_edit(text, corrected)
            explanation = "Auto-correction from OpenAI; parsing fallback used."
            return corrected, score, explanation

def call_local_correct(text: str) -> Tuple[str,int,str]:
    # Using prithivida T5 model and HappyTransformer to generate correction
    prefix = "gec: " + text
    # generate with transformers (fast)
    try:
        inputs = tokenizer.encode(prefix, return_tensors="pt", max_length=256, truncation=True)
        with torch.no_grad():
            outputs = model.generate(inputs, max_length=256, num_beams=4)
        corrected = tokenizer.decode(outputs[0], skip_special_tokens=True)
    except Exception:
        corrected = text

    # use happy transformer to also generate explanation-like correction (best-effort)
    try:
        args = TTSettings(num_beams=4, min_length=1)
        happy_out = happy_tt.generate_text(prefix, args=args).text
        # If happy returns something meaningful, prefer it to compute score
        alt_correction = happy_out or corrected
    except Exception:
        alt_correction = corrected

    score = score_from_edit(text, alt_correction)
    # Basic explanation: detect what changed (very short)
    explanation = []
    if text.strip() == alt_correction.strip():
        explanation = ["No change needed."]
    else:
        explanation = ["Adjusted grammar/punctuation; minor wording edits to improve clarity."]
    return alt_correction, score, "; ".join(explanation)

# ----------------------
# Main processing function
# ----------------------
def process_input(audio, typed_text, use_tts=False, prefer_openai=False):
    """
    audio: filepath from Gradio (or None)
    typed_text: str
    use_tts: bool -> read corrected text with local pyttsx3
    prefer_openai: triage flag to prefer OpenAI (if key available)
    """
    source_text = ""
    # 1) Transcribe audio if present
    if audio:
        transcribed = transcribe_audio_file(audio)
        if transcribed.startswith("[transcription_error]"):
            source_text = typed_text or ""
            trans_msg = transcribed
        else:
            source_text = transcribed
            trans_msg = f"Transcribed: {transcribed}"
    else:
        source_text = typed_text or ""
        trans_msg = "Typed input"

    if not source_text.strip():
        return "No input detected.", 0, "No correction (empty input).", trans_msg, json.dumps({})

    # 2) Choose backend
    use_openai_backend = False
    if USE_OPENAI and prefer_openai:
        use_openai_backend = True
    elif USE_OPENAI and not LOCAL_MODEL_AVAILABLE:
        use_openai_backend = True
    elif not USE_OPENAI and LOCAL_MODEL_AVAILABLE:
        use_openai_backend = False
    elif USE_OPENAI and LOCAL_MODEL_AVAILABLE:
        # default: prefer OpenAI if available (more robust), unless user opts out
        use_openai_backend = prefer_openai or True

    try:
        if use_openai_backend:
            corrected, score, explanation = call_openai_correct(source_text)
        else:
            corrected, score, explanation = call_local_correct(source_text)
    except Exception as e:
        # fallback to local heuristic if something fails
        corrected = source_text
        score = 0
        explanation = f"Model error: {e}"

    # 3) Optionally speak corrected text (pyttsx3)
    tts_msg = ""
    if use_tts:
        try:
            def speak(text):
                engine = pyttsx3.init()
                engine.say(text)
                engine.runAndWait()
            threading.Thread(target=speak, args=(corrected,), daemon=True).start()
            tts_msg = "Speaking corrected text..."
        except Exception as e:
            tts_msg = f"TTS failed: {e}"

    # 4) Build JSON metadata
    meta = {
        "original": source_text,
        "corrected": corrected,
        "score": score,
        "explanation": explanation,
        "backend": "openai" if use_openai_backend else "local",
        "transcription_note": trans_msg,
        "timestamp": int(time.time())
    }

    return corrected, score, explanation, trans_msg + (" • " + tts_msg if tts_msg else ""), json.dumps(meta, ensure_ascii=False, indent=2)

# ----------------------
# Gradio UI
# ----------------------
def build_ui():
    with gr.Blocks(title=APP_TITLE, css="""
        .header {background: linear-gradient(90deg,#ff8fa3,#ff6aa3); padding: 18px; border-radius: 12px; color:white}
        .muted {color: #6b7280}
    """) as demo:
        # Header
        with gr.Row(elem_id="top-row"):
            with gr.Column(scale=3):
                gr.Markdown(f"## <div class='header'>ESPeak — AI Grammar & Speech Assistant</div>")
                gr.Markdown("Speak or type a sentence — ESPeak will correct grammar, score it, and explain changes. Use OpenAI backend if you set `OPENAI_API_KEY` in environment.")
            with gr.Column(scale=1):
                gr.Markdown("**Quick tips**\n- Speak clearly (short sentences work best)\n- Toggle TTS to hear the corrected sentence\n- Use `Prefer OpenAI` to route to ChatGPT if available")
        gr.Markdown("---")

        with gr.Row():
            with gr.Column(scale=1):
                audio = gr.Audio(sources="microphone", type="filepath", label="Record (microphone)")
                typed = gr.Textbox(lines=3, placeholder="Or type your sentence here...", label="Text input")
                with gr.Row():
                    tts_checkbox = gr.Checkbox(label="Play corrected (TTS)", value=False)
                    prefer_openai = gr.Checkbox(label="Prefer OpenAI backend (if available)", value=True)
                run_btn = gr.Button("Check Grammar", variant="primary")
            with gr.Column(scale=2):
                corrected_out = gr.Textbox(label="Corrected Text", interactive=False)
                score_out = gr.Number(label="Grammar Score (0-100)", interactive=False)
                explanation_out = gr.Textbox(label="Explanation (what I changed)", interactive=False)
                trans_note = gr.Textbox(label="Transcription / Info", interactive=False)
                meta_out = gr.Code(label="JSON metadata (copyable)", language="json")

        def on_submit(audio_file, typed_text, use_tts, use_openai):
            return process_input(audio_file, typed_text, use_tts, use_openai)

        run_btn.click(on_submit, inputs=[audio, typed, tts_checkbox, prefer_openai],
                      outputs=[corrected_out, score_out, explanation_out, trans_note, meta_out])

        gr.Markdown("---")
        gr.Markdown("**ESPeak** · Built for quick grammar checking of spoken and typed English. Designed for demos and interview projects.")
    return demo

if __name__ == "__main__":
    demo = build_ui()
    demo.launch(share=False, inbrowser=True)