app.py

import json
import os
import spacy
from transformers import pipeline, AutoModelForSeq2SeqLM, AutoTokenizer
import gradio as gr
from huggingface_hub import Repository
from datetime import datetime 
from spacy.cli import download

# Load or download spaCy model
try:
    nlp = spacy.load("en_core_web_sm")
except OSError:
    print("Downloading 'en_core_web_sm' model...")
    download("en_core_web_sm")
    nlp = spacy.load("en_core_web_sm")

# Load Question Generation model
qg_model = AutoModelForSeq2SeqLM.from_pretrained("valhalla/t5-base-qa-qg-hl")
qg_tokenizer = AutoTokenizer.from_pretrained("valhalla/t5-base-qa-qg-hl", use_fast=True)
qg_pipeline = pipeline("text2text-generation", model=qg_model, tokenizer=qg_tokenizer)

# Global variable to accumulate Q&A
batch_data = []

# Utility functions
def extract_paragraph_facts(raw_text):
    return [p.strip() for p in raw_text.strip().split("\n\n") if p.strip()]

def extract_noun_phrases(text):
    doc = nlp(text)
    return [np.text for np in doc.noun_chunks]

def auto_highlight_noun_phrase(text):
    doc = nlp(text)
    noun_phrases = sorted(doc.noun_chunks, key=lambda np: len(np.text), reverse=True)
    for np in noun_phrases:
        if len(np.text.split()) > 1 or np.root.pos_ == "NOUN":
            return np.text
    return text

def highlight_selected_phrase(fact, selected_np):
    return fact.replace(selected_np, f"<hl>{selected_np}<hl>", 1)

def generate_single_qna(fact, noun_phrase, min_len, max_len, temperature, top_k, top_p):
    hl_fact = highlight_selected_phrase(fact, noun_phrase)
    try:
        prompt = f"generate question: {hl_fact}"
        output = qg_pipeline(
            prompt,
            min_length=min_len,
            max_length=max_len,
            temperature=temperature,
            top_k=top_k,
            top_p=top_p,
            do_sample=True
        )[0]
        question = output.get("generated_text", "").strip()
        if not question.endswith("?"):
            question += "?"
    except Exception as e:
        question = f"Error generating question: {str(e)}"

    return {"question": question, "answer": fact}

def generate_qna_all(input_text, selected_fact, selected_np, min_len, max_len, temperature, top_k, top_p):
    facts = extract_paragraph_facts(input_text)
    global batch_data

    if selected_fact:
        noun_phrase = selected_np if selected_np else auto_highlight_noun_phrase(selected_fact)
        result = generate_single_qna(selected_fact, noun_phrase, min_len, max_len, temperature, top_k, top_p)
        batch_data.append(result)
    else:
        for fact in facts:
            noun_phrase = auto_highlight_noun_phrase(fact)
            result = generate_single_qna(fact, noun_phrase, min_len, max_len, temperature, top_k, top_p)
            batch_data.append(result)

    return json.dumps(batch_data, indent=2, ensure_ascii=False), json.dumps(batch_data, indent=2, ensure_ascii=False)

def save_json_to_dataset():
    try:
        if not batch_data:
            return "❌ No data to save. Generate some Q&A first."

        hf_token = os.environ.get("QandA_Generator")
        if not hf_token:
            return "❌ HF_TOKEN not found in environment."

        repo_id = "UniversityAIChatbot/University_Inquiries_AI_Chatbot"
        target_file = "dataset.json"
        local_dir = "hf_repo"

        repo = Repository(
            local_dir=local_dir,
            clone_from=repo_id,
            use_auth_token=hf_token,
            repo_type="space"
        )

        repo.git_pull()

        full_path = os.path.join(local_dir, target_file)

        if os.path.exists(full_path):
            with open(full_path, "r", encoding="utf-8") as f:
                existing_data = json.load(f)
        else:
            existing_data = []

        now = datetime.now()
        for entry in batch_data:
            entry["month"] = now.strftime("%B")
            entry["year"] = now.year

        updated_data = existing_data + batch_data

        with open(full_path, "w", encoding="utf-8") as f:
            json.dump(updated_data, f, indent=2, ensure_ascii=False)

        repo.push_to_hub(commit_message="📥 Add new Q&A to database.json")

        batch_data.clear()

        return "✅ Data with timestamp successfully pushed to Space!"
    except Exception as e:
        return f"❌ Error: {str(e)}"

# New: Preview function
def preview_batch_data():
    return json.dumps(batch_data, indent=2, ensure_ascii=False)

# New: Append from manual JSON editor
def append_json_to_batch(json_text):
    global batch_data
    try:
        new_data = json.loads(json_text)
        if isinstance(new_data, dict):
            new_data = [new_data]
        if not isinstance(new_data, list):
            return "❌ Invalid format. Must be a list or object.", preview_batch_data()

        batch_data.extend(new_data)
        return "✅ Successfully appended to batch_data.", preview_batch_data()
    except Exception as e:
        return f"❌ Error: {str(e)}", preview_batch_data()

# Dropdown callbacks
def on_extract_facts(text):
    facts = extract_paragraph_facts(text)
    default_fact = facts[0] if facts else None
    return gr.update(choices=facts, value=default_fact), gr.update(choices=[], value=None)

def on_select_fact(fact):
    noun_phrases = extract_noun_phrases(fact)
    return gr.update(choices=noun_phrases, value=noun_phrases[0] if noun_phrases else None)

# UI
def main():
    with gr.Blocks() as demo:
        gr.Markdown("## Paragraph-to-Question Generator (Auto Q&A for HF Dataset)")

        input_text = gr.Textbox(lines=10, label="Enter Data (Seperated by paragraph per question)")

        with gr.Accordion("⚙️ Customize Question Generation", open=False):
            extract_btn = gr.Button("Extract & Customize")
            fact_dropdown = gr.Dropdown(label="Select a Fact", interactive=True)
            np_dropdown = gr.Dropdown(label="Select Noun Phrase to Highlight (optional)", interactive=True)

            extract_btn.click(fn=on_extract_facts, inputs=input_text, outputs=[fact_dropdown, np_dropdown])
            fact_dropdown.change(fn=on_select_fact, inputs=fact_dropdown, outputs=np_dropdown)

            gr.Markdown("🔽 **Min Length**: Minimum number of tokens in the generated question.")
            min_len = gr.Slider(5, 50, value=10, step=1, label="Min Length")

            gr.Markdown("🔼 **Max Length**: Maximum number of tokens in the generated question.")
            max_len = gr.Slider(20, 100, value=64, step=1, label="Max Length")

            gr.Markdown("🌡️ **Temperature**: Controls randomness. Lower = more predictable, higher = more creative.")
            temperature = gr.Slider(0.1, 1.5, value=1.0, step=0.1, label="Temperature")

            gr.Markdown("🎯 **Top-k Sampling**: Limits sampling to the top-k most likely words.")
            top_k = gr.Slider(0, 100, value=50, step=1, label="Top-k")

            gr.Markdown("🎲 **Top-p (Nucleus Sampling)**: Selects from the smallest set of words with a cumulative probability > p.")
            top_p = gr.Slider(0.1, 1.0, value=0.95, step=0.05, label="Top-p")

        gr.Markdown("✏️ You can manually edit the generated JSON here or paste your own in the same format.")
        output_json = gr.Textbox(
            lines=14,
            label="Q&A JSON",
            interactive=True,
            placeholder='[\n{\n"question": "Your question?",\n"answer": "Your answer."\n}\n]'
        )

        preview_box = gr.Textbox(
            lines=14,
            label="📦 Preview",
            interactive=False
        )

        with gr.Row():
            generate_btn = gr.Button("Generate Q&A")
            append_btn = gr.Button("➕ Add to Dataset")
            send_btn = gr.Button("📤 Send to Dataset")

        send_status = gr.Textbox(label="Save Status", interactive=False)

        generate_btn.click(
            fn=generate_qna_all,
            inputs=[input_text, fact_dropdown, np_dropdown, min_len, max_len, temperature, top_k, top_p],
            outputs=[output_json, preview_box]
        )

        append_btn.click(
            fn=append_json_to_batch,
            inputs=output_json,
            outputs=[send_status, preview_box]
        )

        send_btn.click(fn=save_json_to_dataset, inputs=None, outputs=send_status)

    demo.launch()

if __name__ == "__main__":
    main()