""" ============================================================================= FastAPI Production Server for Bilingual Summarization NLP Suite ============================================================================= """ import os import sys import time from io import BytesIO from typing import Optional, Dict, Any # Ensure project root is in Python path PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) if PROJECT_ROOT not in sys.path: sys.path.insert(0, PROJECT_ROOT) import torch from fastapi import FastAPI, UploadFile, File, Form, HTTPException, status from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, RedirectResponse from nlp_core.language_detector import LanguageDetector from nlp_core.tokenizer import BilingualTokenizer from models.extractive.textrank import TextRankSummarizer from models.extractive.lsa import LSASummarizer from models.extractive.hybrid_scorer import HybridSummarizer from models.abstractive.seq2seq_model import Seq2SeqSummarizer from models.abstractive.transformer_wrap import TransformerSummarizer from evaluation.metrics_manager import MetricsManager from api.schemas import ( HealthResponse, ModelsListResponse, ModelDetails, LanguageDetectionRequest, LanguageDetectionResponse, TokenizeRequest, TokenizeResponse, SummarizeRequest, SummarizeResponse, SummaryMetrics, EvaluationRequest, EvaluationResponse, ) # --- App Initialization --- app = FastAPI( title="Bilingual Text Summarization NLP API", description=""" ## High-Performance Arabic & English Text Summarization API This production REST API provides complete NLP capabilities for automatic text summarization: * **Multilingual NLP Core**: Language detection, Arabic normalizer, sentence segmenter, bilingual tokenization. * **Extractive Summarization**: Graph-based TextRank, Latent Semantic Analysis (LSA), and Hybrid Multi-feature Scorer. * **Abstractive Deep Learning**: PyTorch Seq2Seq with Bahdanau Attention, Bi-GRU, and Beam Search decoding. * **Evaluation Suite**: ROUGE-1, ROUGE-2, ROUGE-L, BLEU-1..4, Cumulative BLEU, TTR, and Compression Analytics. * **Document Parsing**: Automatic text extraction from `.txt`, `.pdf`, and `.docx` files. """, version="1.0.0", docs_url="/docs", redoc_url="/redoc", ) # --- CORS Middleware (Allows Frontend / Cross-Origin requests) --- app.add_middleware( CORSMiddleware, allow_origins=["*"], # Adjust for production domains if necessary allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # --- Singleton Engines & Model Cache --- lang_detector = LanguageDetector() tokenizer = BilingualTokenizer() metrics_manager = MetricsManager() # Checkpoint paths CKPT_DIR = os.path.join(PROJECT_ROOT, "checkpoints") CKPT_AR = os.path.join(CKPT_DIR, "seq2seq_arabic.pt") CKPT_EN = os.path.join(CKPT_DIR, "seq2seq_english.pt") # Lazy-loaded / Cached Seq2Seq models loaded_models: Dict[str, Any] = {} def get_seq2seq_model(lang: str) -> Optional[Seq2SeqSummarizer]: ckpt = CKPT_AR if lang == "ar" else CKPT_EN if lang not in loaded_models: if os.path.exists(ckpt): try: device = "cuda" if torch.cuda.is_available() else "cpu" loaded_models[lang] = Seq2SeqSummarizer.load_checkpoint(ckpt, device=device) except Exception as e: print(f"[Warning] Failed to load Seq2Seq model for {lang}: {e}") return None else: return None return loaded_models.get(lang) # --- Helper: Extract text from uploaded file --- def extract_text_from_upload(filename: str, content: bytes) -> str: ext = filename.lower().split(".")[-1] if ext == "txt": return content.decode("utf-8", errors="ignore") elif ext == "docx": try: import docx doc = docx.Document(BytesIO(content)) return "\n".join([p.text for p in doc.paragraphs if p.text.strip()]) except Exception as e: raise HTTPException(status_code=400, detail=f"Failed to extract DOCX: {e}") elif ext == "pdf": try: import pypdf reader = pypdf.PdfReader(BytesIO(content)) text_parts = [] for page in reader.pages: t = page.extract_text() if t: text_parts.append(t) return "\n".join(text_parts) except Exception as e: raise HTTPException(status_code=400, detail=f"Failed to extract PDF: {e}") else: raise HTTPException( status_code=400, detail=f"Unsupported file format: .{ext}. Supported formats: .txt, .pdf, .docx" ) # ============================================================================= # API ENDPOINTS # ============================================================================= @app.get("/", include_in_schema=False) def root(): return RedirectResponse(url="/docs") @app.get("/api/v1/health", response_model=HealthResponse, tags=["System"]) def health_check(): """Returns system status, PyTorch hardware acceleration, and available checkpoints.""" ckpts = { "seq2seq_arabic": os.path.exists(CKPT_AR), "seq2seq_english": os.path.exists(CKPT_EN), "rich_seq2seq_ar": os.path.exists(os.path.join(CKPT_DIR, "rich_seq2seq_ar.pt")), "rich_seq2seq_en": os.path.exists(os.path.join(CKPT_DIR, "rich_seq2seq_en.pt")), } return HealthResponse( status="healthy", version="1.0.0", device="cuda" if torch.cuda.is_available() else "cpu", cuda_available=torch.cuda.is_available(), loaded_checkpoints=ckpts, ) @app.get("/api/v1/models", response_model=ModelsListResponse, tags=["System"]) def list_models(): """Lists all available summarization algorithms, paradigms, and checkpoint states.""" models_info = [ ModelDetails( id="textrank", name="TextRank", paradigm="extractive", supported_languages=["ar", "en"], description="Graph-based PageRank sentence centrality algorithm computing lexical and TF-IDF similarity graphs." ), ModelDetails( id="lsa", name="Latent Semantic Analysis (LSA)", paradigm="extractive", supported_languages=["ar", "en"], description="Singular Value Decomposition (SVD) identifying latent semantic concepts across sentence-term matrices." ), ModelDetails( id="hybrid", name="Hybrid Multi-Feature Scorer", paradigm="extractive", supported_languages=["ar", "en"], description="Composite scorer linearly weighting graph centrality, positional bias, and sentence length penalties." ), ModelDetails( id="seq2seq", name="Seq2Seq with Bahdanau Attention", paradigm="abstractive", supported_languages=["ar", "en"], description="Deep Bidirectional GRU Encoder-Decoder network with additive attention and Beam Search decoding.", checkpoint_available=os.path.exists(CKPT_AR) and os.path.exists(CKPT_EN) ), ModelDetails( id="transformer", name="Pretrained Transformer Wrapper", paradigm="abstractive", supported_languages=["ar", "en"], description="HuggingFace Transformers pipeline wrapper (AraBART / BART-large-CNN) with local fallback." ) ] return ModelsListResponse(total_models=len(models_info), models=models_info) @app.post("/api/v1/detect-language", response_model=LanguageDetectionResponse, tags=["NLP Core"]) def detect_language(req: LanguageDetectionRequest): """Detects if text is Arabic or English with script breakdown and confidence score.""" detected = lang_detector.detect_language(req.text) conf = lang_detector.get_language_confidence(req.text) lang_name = "Arabic" if detected == "ar" else ("English" if detected == "en" else "Unknown") return LanguageDetectionResponse( detected_language=detected, language_name=lang_name, confidence=conf.get("confidence", 0.95), script_breakdown=conf.get("breakdown", {"arabic": 0.0, "latin": 0.0, "other": 0.0}) ) @app.post("/api/v1/tokenize", response_model=TokenizeResponse, tags=["NLP Core"]) def tokenize_text(req: TokenizeRequest): """Normalizes text, segments sentences, and extracts linguistic tokens.""" detected = lang_detector.detect_language(req.text) if req.lang == "auto" else req.lang norm = tokenizer.preprocess_sentence(req.text, lang=detected, remove_stopwords=req.remove_stopwords, stem=req.stem) sentences = tokenizer.split_sentences(req.text, lang=detected) tokens = tokenizer.tokenize_words(req.text, lang=detected, remove_stopwords=req.remove_stopwords, stem=req.stem) return TokenizeResponse( detected_language=detected, sentence_count=len(sentences), word_count=len(tokens), sentences=sentences, tokens=tokens, normalized_text=norm ) @app.post("/api/v1/summarize", response_model=SummarizeResponse, tags=["Summarization"]) def summarize(req: SummarizeRequest): """Summarizes text using Extractive or Abstractive models with automatic evaluation.""" t0 = time.time() detected_lang = lang_detector.detect_language(req.text) if req.lang == "auto" else req.lang summary_text = "" selected_indices = None scores_data = None note = None # Extractive Routing if req.mode == "extractive" or req.method in ["textrank", "lsa", "hybrid"]: if req.method == "lsa": model = LSASummarizer() elif req.method == "hybrid": model = HybridSummarizer() else: model = TextRankSummarizer() res = model.summarize( req.text, num_sentences=req.sentences, ratio=req.ratio, lang=detected_lang ) summary_text = res["summary"] selected_indices = res.get("selected_indices") raw_scores = res.get("sentence_scores", []) if raw_scores: scores_data = [{"sentence": s, "score": float(sc)} for s, sc in raw_scores] # Abstractive Routing elif req.mode == "abstractive" or req.method in ["seq2seq", "transformer"]: if req.method == "transformer": tr_wrap = TransformerSummarizer(lang=detected_lang) gen = tr_wrap.summarize(req.text) if gen: summary_text = gen else: fallback = HybridSummarizer().summarize(req.text, num_sentences=req.sentences, lang=detected_lang) summary_text = fallback["summary"] note = "Transformer model weights not available locally; used Hybrid fallback." else: # Seq2Seq GRU with Attention seq_model = get_seq2seq_model(detected_lang) if seq_model: toks = tokenizer.tokenize_words(req.text, lang=detected_lang) summary_toks = seq_model.summarize_beam(toks, beam_width=req.beam_width, max_len=60) summary_text = " ".join(summary_toks) else: fallback = HybridSummarizer().summarize(req.text, num_sentences=req.sentences, lang=detected_lang) summary_text = fallback["summary"] selected_indices = fallback.get("selected_indices") note = f"Seq2Seq weights for [{detected_lang.upper()}] not found; used Hybrid fallback." latency = round((time.time() - t0) * 1000.0, 2) # Compute Metrics eval_metrics = metrics_manager.evaluate_summary( original_text=req.text, generated_summary=summary_text, reference_summary=req.reference_summary, lang=detected_lang ) r = eval_metrics.get("rouge", {}) b = eval_metrics.get("bleu", {}) comp = eval_metrics.get("compression", {}) metrics_obj = SummaryMetrics( original_words=comp.get("original_words", len(req.text.split())), summary_words=comp.get("summary_words", len(summary_text.split())), compression_ratio=comp.get("compression_ratio", 0.0), reduction_percentage=comp.get("reduction_percentage", 0.0), type_token_ratio=eval_metrics.get("type_token_ratio", 0.0), rouge_1_f1=r.get("rouge-1", {}).get("f1"), rouge_2_f1=r.get("rouge-2", {}).get("f1"), rouge_l_f1=r.get("rouge-l", {}).get("f1"), bleu_1=b.get("bleu-1"), bleu_2=b.get("bleu-2"), bleu_cumulative=b.get("bleu_cumulative"), ) return SummarizeResponse( status="success", detected_language=detected_lang, mode=req.mode, method=req.method, summary=summary_text, latency_ms=latency, metrics=metrics_obj, selected_indices=selected_indices, sentence_scores=scores_data, note=note ) @app.post("/api/v1/summarize/file", response_model=SummarizeResponse, tags=["Summarization"]) async def summarize_file( file: UploadFile = File(..., description="Document file (.txt, .pdf, .docx)"), mode: str = Form("extractive", description="'extractive', 'abstractive', 'hybrid'"), method: str = Form("textrank", description="'textrank', 'lsa', 'hybrid', 'seq2seq'"), lang: str = Form("auto", description="'auto', 'ar', 'en'"), sentences: int = Form(3), ratio: Optional[float] = Form(None), beam_width: int = Form(3) ): """Parses an uploaded document (.txt, .pdf, .docx) and generates a condensed summary.""" content = await file.read() raw_text = extract_text_from_upload(file.filename, content) if not raw_text.strip(): raise HTTPException(status_code=400, detail="Uploaded file is empty or contains no readable text.") req = SummarizeRequest( text=raw_text, mode=mode, method=method, lang=lang, sentences=sentences, ratio=ratio, beam_width=beam_width ) return summarize(req) @app.post("/api/v1/evaluate", response_model=EvaluationResponse, tags=["Evaluation"]) def evaluate_metrics(req: EvaluationRequest): """Calculates comprehensive ROUGE-1/2/L, BLEU 1..4, and Lexical Diversity scores.""" detected = lang_detector.detect_language(req.original_text) if req.lang == "auto" else req.lang res = metrics_manager.evaluate_summary( original_text=req.original_text, generated_summary=req.generated_summary, reference_summary=req.reference_summary, lang=detected ) r = res.get("rouge", {}) b = res.get("bleu", {}) comp = res.get("compression", {}) return EvaluationResponse( language=detected, compression_ratio=comp.get("compression_ratio", 0.0), reduction_percentage=comp.get("reduction_percentage", 0.0), original_words=comp.get("original_words", len(req.original_text.split())), summary_words=comp.get("summary_words", len(req.generated_summary.split())), type_token_ratio=res.get("type_token_ratio", 0.0), rouge_1=r.get("rouge-1"), rouge_2=r.get("rouge-2"), rouge_l=r.get("rouge-l"), bleu_1=b.get("bleu-1"), bleu_2=b.get("bleu-2"), bleu_3=b.get("bleu-3"), bleu_4=b.get("bleu-4"), bleu_cumulative=b.get("bleu_cumulative"), ) # --- Entry Point for direct execution --- if __name__ == "__main__": import uvicorn uvicorn.run("api.server:app", host="0.0.0.0", port=8000, reload=True)