app.py

import os
import logging
import sys
import argparse
from dotenv import load_dotenv
load_dotenv()
from fastapi import FastAPI, File, Form, UploadFile, HTTPException, BackgroundTasks
from fastapi.responses import JSONResponse, FileResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional, List, Dict, Any
import pandas as pd
import tempfile
import shutil
from models import Grammar, Error
from grammar_checker import (
    check_grammar,
    check_grammar_from_file,
    check_grammar_qa,
    display_results,
    DEFAULT_PROPER_NOUNS,
)

# Configure logging

# Create FastAPI app
app = FastAPI(
    title="Grammar Checker API",
    description="API for checking grammar in text, files, and quiz questions",
    docs_url="/",
)

# Configure CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Allows all origins
    allow_credentials=True,
    allow_methods=["*"],  # Allows all methods
    allow_headers=["*"],  # Allows all headers
)

# Define allowed file extensions
ALLOWED_EXTENSIONS = {".txt", ".docx", ".xlsx"}


class GrammarCheckResponse(BaseModel):
    output_file: str
    message: str
    records: List[Dict[str, Any]] = []


@app.post("/check-grammar-quiz", response_model=GrammarCheckResponse)
def check_grammar_quiz(
    background_tasks: BackgroundTasks,
    file: UploadFile = File(...),
    limit: Optional[int] = None,
):
    """
    Process an Excel file with questions and answers, check grammar, and return the corrected data.

    Args:
        file: The input Excel file
        limit: Limit the number of records to process. If None, process all records.

    Returns:
        JSON response with the path to the output file and processed records
    """
    # Create temp directory to store files
    temp_dir = tempfile.mkdtemp()
    input_path = os.path.join(temp_dir, file.filename)
    output_filename = f"corrected_{file.filename}"
    output_path = os.path.join(temp_dir, output_filename)

    # Save uploaded file
    # try:
    with open(input_path, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    # except Exception as e:
    #     raise HTTPException(status_code=500, detail=f"Error saving file: {str(e)}")

    # Process the file
    # try:
    result_file, processed_records = process_grammar_check(input_path, output_path, limit)
    background_tasks.add_task(cleanup_temp_files, temp_dir)
    return GrammarCheckResponse(
        output_file=result_file, 
        message="Grammar check completed successfully",
        records=processed_records
    )
    # except Exception as e:
    #     background_tasks.add_task(cleanup_temp_files, temp_dir)
    #     raise HTTPException(status_code=500, detail=f"Error processing file: {str(e)}")


def cleanup_temp_files(temp_dir: str):
    """Clean up temporary files"""
    shutil.rmtree(temp_dir)


def process_grammar_check(input_file, output_file, limit=None):
    """
    Process an Excel file with questions and answers, check grammar, and save the corrected data.

    Args:
        input_file (str): Path to the input Excel file
        output_file (str): Path to save the output Excel file
        limit (int, optional): Limit the number of records to process. If None, process all records.

    Returns:
        tuple: (Path to the output file, List of processed records)
    """
    # Read the input file
    df = pd.read_excel(input_file, sheet_name="Sheet1")
    records = df.to_dict(orient="records")

    # Process the records
    data_processed = []
    for i, record in enumerate(records):
        if limit is not None and i >= limit:
            break

        dict_result = check_grammar_qa(record)
        temp_dict = record.copy()
        temp_dict["Question"] = dict_result["output"]["Question"]
        temp_dict["Answer Option A"] = dict_result["output"].get(
            "Answer Option A", None
        )
        temp_dict["Answer Option B"] = dict_result["output"].get(
            "Answer Option B", None
        )
        temp_dict["Answer Option C"] = dict_result["output"].get(
            "Answer Option C", None
        )
        temp_dict["Answer Option D"] = dict_result["output"].get(
            "Answer Option D", None
        )
        temp_dict["wrong_locations"] = dict_result["wrong_locations"]

        data_processed.append(temp_dict)

    # Create a DataFrame from the processed data and write to Excel
    output_df = pd.DataFrame(data_processed)
    output_df.to_excel(output_file, index=False)

    return output_file, data_processed


def allowed_file(filename: str) -> bool:
    """
    Check if the uploaded file has an allowed extension.

    Args:
        filename: The name of the uploaded file

    Returns:
        True if the file extension is allowed, False otherwise
    """
    return os.path.splitext(filename)[1].lower() in ALLOWED_EXTENSIONS


class TextRequest(BaseModel):
    text: str
    proper_nouns: Optional[str] = DEFAULT_PROPER_NOUNS


@app.post("/check-text")
def check_text(body: TextRequest):
    """Process text input and check grammar."""
    try:
        if not body.text:
            raise HTTPException(status_code=400, detail="No text provided")

        # Check grammar using LangChain
        result = check_grammar(body.text, body.proper_nouns)

        # Convert Pydantic model to dict for JSON response
        return result.model_dump()

    except Exception as e:
        logging.error(f"Error processing text: {str(e)}")
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/check-file")
async def check_file(
    file: UploadFile, proper_nouns: Optional[str] = Form(DEFAULT_PROPER_NOUNS)
):
    """Process file upload and check grammar."""
    try:
        # Check if a valid file was uploaded
        if file.filename == "":
            raise HTTPException(status_code=400, detail="No file selected")

        # Check if the file has an allowed extension
        if not allowed_file(file.filename):
            raise HTTPException(
                status_code=400,
                detail=f"File type not supported. Allowed types: {', '.join(ALLOWED_EXTENSIONS)}",
            )

        # Process the file
        file_content = await file.read()
        result = check_grammar_from_file(file_content, file.filename, proper_nouns)

        # Convert Pydantic model to dict for JSON response
        return result.model_dump()

    except Exception as e:
        logging.error(f"Error processing file: {str(e)}")
        raise HTTPException(status_code=500, detail=str(e))


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8080)