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)