Spaces:
Running
Running
"""import gradio as gr | |
from transformers import pipeline | |
import fitz # PyMuPDF | |
import docx | |
import pptx | |
import openpyxl | |
import os | |
from fastapi import FastAPI | |
from fastapi.responses import RedirectResponse | |
# Load your custom summarization model | |
pipe = pipeline("summarization", model="facebook/bart-large-cnn", tokenizer="facebook/bart-large-cnn") | |
# Document text extraction function | |
def extract_text(file): | |
ext = file.name.split(".")[-1].lower() | |
path = file.name | |
if ext == "pdf": | |
try: | |
with fitz.open(path) as doc: | |
return "\n".join([page.get_text("text") for page in doc]) | |
except Exception as e: | |
return f"Error reading PDF: {e}" | |
elif ext == "docx": | |
try: | |
doc = docx.Document(path) | |
return "\n".join([p.text for p in doc.paragraphs]) | |
except Exception as e: | |
return f"Error reading DOCX: {e}" | |
elif ext == "pptx": | |
try: | |
prs = pptx.Presentation(path) | |
text = "" | |
for slide in prs.slides: | |
for shape in slide.shapes: | |
if hasattr(shape, "text"): | |
text += shape.text + "\n" | |
return text | |
except Exception as e: | |
return f"Error reading PPTX: {e}" | |
elif ext == "xlsx": | |
try: | |
wb = openpyxl.load_workbook(path) | |
text = "" | |
for sheet in wb.sheetnames: | |
for row in wb[sheet].iter_rows(values_only=True): | |
text += " ".join([str(cell) for cell in row if cell]) + "\n" | |
return text | |
except Exception as e: | |
return f"Error reading XLSX: {e}" | |
else: | |
return "Unsupported file format" | |
# Summarization logic | |
def summarize_document(file): | |
text = extract_text(file) | |
if "Error" in text or "Unsupported" in text: | |
return text | |
word_count = len(text.split()) | |
max_summary_len = max(20, int(word_count * 0.2)) | |
try: | |
summary = pipe(text, max_length=max_summary_len, min_length=int(max_summary_len * 0.6), do_sample=False) | |
# Print the summary to debug its structure | |
print(summary) | |
return summary[0]['summary_text'] # Access the correct key for the output | |
except Exception as e: | |
return f"Error during summarization: {e}" | |
# Gradio Interface | |
demo = gr.Interface( | |
fn=summarize_document, | |
inputs=gr.File(label="Upload a document (PDF, DOCX, PPTX, XLSX)", file_types=[".pdf", ".docx", ".pptx", ".xlsx"]), | |
outputs=gr.Textbox(label="20% Summary"), | |
title="π Document Summarizer (20% Length)", | |
description="Upload a document and get a concise summary generated by your custom Hugging Face model." | |
) | |
# FastAPI setup | |
app = FastAPI() | |
# Mount Gradio at "/" | |
app = gr.mount_gradio_app(app, demo, path="/") | |
# Optional root redirect | |
@app.get("/") | |
def redirect_to_interface(): | |
return RedirectResponse(url="/")""" | |
import gradio as gr | |
from transformers import pipeline, AutoTokenizer, AutoModelForSeq2SeqLM | |
import fitz # PyMuPDF | |
import docx | |
import pptx | |
import openpyxl | |
import re | |
import nltk | |
from nltk.tokenize import sent_tokenize | |
import torch | |
from fastapi import FastAPI | |
from fastapi.responses import RedirectResponse | |
# Download required NLTK data | |
nltk.download('punkt', quiet=True) | |
# Initialize components | |
app = FastAPI() | |
# Load summarization model (CPU optimized) | |
MODEL_NAME = "facebook/bart-large-cnn" | |
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME) | |
model = AutoModelForSeq2SeqLM.from_pretrained(MODEL_NAME) | |
summarizer = pipeline( | |
"summarization", | |
model=model, | |
tokenizer=tokenizer, | |
device=-1, # Force CPU usage | |
torch_dtype=torch.float32 | |
) | |
def clean_text(text: str) -> str: | |
"""Clean and normalize document text""" | |
text = re.sub(r'\s+', ' ', text) # Normalize whitespace | |
text = re.sub(r'β’\s*|\d\.\s+', '', text) # Remove bullets and numbering | |
text = re.sub(r'\[.*?\]|\(.*?\)', '', text) # Remove brackets/parentheses | |
text = re.sub(r'\bPage\s*\d+\b', '', text, flags=re.IGNORECASE) # Remove page numbers | |
return text.strip() | |
def extract_text(file_path: str, file_extension: str) -> tuple[str, str]: | |
"""Extract text from various document formats""" | |
try: | |
if file_extension == "pdf": | |
with fitz.open(file_path) as doc: | |
return clean_text("\n".join(page.get_text("text") for page in doc)), "" | |
elif file_extension == "docx": | |
doc = docx.Document(file_path) | |
return clean_text("\n".join(p.text for p in doc.paragraphs)), "" | |
elif file_extension == "pptx": | |
prs = pptx.Presentation(file_path) | |
text = [] | |
for slide in prs.slides: | |
for shape in slide.shapes: | |
if hasattr(shape, "text"): | |
text.append(shape.text) | |
return clean_text("\n".join(text)), "" | |
elif file_extension == "xlsx": | |
wb = openpyxl.load_workbook(file_path, read_only=True) | |
text = [] | |
for sheet in wb.sheetnames: | |
for row in wb[sheet].iter_rows(values_only=True): | |
text.append(" ".join(str(cell) for cell in row if cell)) | |
return clean_text("\n".join(text)), "" | |
return "", "Unsupported file format" | |
except Exception as e: | |
return "", f"Error reading {file_extension.upper()} file: {str(e)}" | |
def chunk_text(text: str, max_tokens: int = 768) -> list[str]: | |
"""Split text into manageable chunks for summarization""" | |
try: | |
sentences = sent_tokenize(text) | |
except: | |
# Fallback if sentence tokenization fails | |
words = text.split() | |
sentences = [' '.join(words[i:i+20]) for i in range(0, len(words), 20)] | |
chunks = [] | |
current_chunk = "" | |
for sentence in sentences: | |
if len(current_chunk.split()) + len(sentence.split()) <= max_tokens: | |
current_chunk += " " + sentence | |
else: | |
chunks.append(current_chunk.strip()) | |
current_chunk = sentence | |
if current_chunk: | |
chunks.append(current_chunk.strip()) | |
return chunks | |
def generate_summary(text: str, length: str = "medium") -> str: | |
"""Generate summary with appropriate length parameters""" | |
length_params = { | |
"short": {"max_length": 80, "min_length": 30}, | |
"medium": {"max_length": 150, "min_length": 60}, | |
"long": {"max_length": 200, "min_length": 80} | |
} | |
chunks = chunk_text(text) | |
summaries = [] | |
for chunk in chunks: | |
try: | |
summary = summarizer( | |
chunk, | |
max_length=length_params[length]["max_length"], | |
min_length=length_params[length]["min_length"], | |
do_sample=False, | |
truncation=True, | |
no_repeat_ngram_size=2, | |
num_beams=2, | |
early_stopping=True | |
) | |
summaries.append(summary[0]['summary_text']) | |
except Exception as e: | |
summaries.append(f"[Chunk error: {str(e)}]") | |
# Combine and format the final summary | |
final_summary = " ".join(summaries) | |
final_summary = ". ".join(s.strip().capitalize() for s in final_summary.split(". ") if s.strip()) | |
return final_summary if len(final_summary) > 25 else "Summary too short - document may be too brief" | |
def summarize_document(file, summary_length: str): | |
"""Main processing function for Gradio interface""" | |
if file is None: | |
return "Please upload a document first", "Ready" | |
file_path = file.name | |
file_extension = file_path.split(".")[-1].lower() | |
text, error = extract_text(file_path, file_extension) | |
if error: | |
return error, "Error" | |
if not text or len(text.split()) < 30: | |
return "Document is too short or contains too little text to summarize", "Ready" | |
try: | |
summary = generate_summary(text, summary_length) | |
return summary, "Summary complete" | |
except Exception as e: | |
return f"Summarization error: {str(e)}", "Error" | |
# Gradio Interface | |
with gr.Blocks(title="Document Summarizer", theme=gr.themes.Soft()) as demo: | |
gr.Markdown("# π Document Summarizer") | |
gr.Markdown("Upload a document to generate a concise summary") | |
with gr.Row(): | |
with gr.Column(): | |
file_input = gr.File( | |
label="Upload Document", | |
file_types=[".pdf", ".docx", ".pptx", ".xlsx"], | |
type="filepath" | |
) | |
length_radio = gr.Radio( | |
["short", "medium", "long"], | |
value="medium", | |
label="Summary Length" | |
) | |
submit_btn = gr.Button("Generate Summary", variant="primary") | |
with gr.Column(): | |
output = gr.Textbox(label="Summary", lines=10) | |
status = gr.Textbox(label="Status", interactive=False) | |
submit_btn.click( | |
fn=summarize_document, | |
inputs=[file_input, length_radio], | |
outputs=[output, status], | |
api_name="summarize" | |
) | |
# Mount Gradio app to FastAPI | |
app = gr.mount_gradio_app(app, demo, path="/") | |
def redirect_to_interface(): | |
return RedirectResponse(url="/") |