324 lines
10 KiB
Python
324 lines
10 KiB
Python
"""
|
|
Document Processor - Estrae testo da vari formati di documento.
|
|
Supporta: PDF, Immagini, XLSX, DOCX, PPTX, TXT, MD
|
|
"""
|
|
|
|
import pytesseract
|
|
from pdf2image import convert_from_path
|
|
from PIL import Image
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
import tempfile
|
|
|
|
from config import settings
|
|
from logger import get_logger
|
|
|
|
log = get_logger("document_processor")
|
|
|
|
|
|
class DocumentProcessor:
|
|
SUPPORTED_EXTENSIONS = {
|
|
".pdf": "pdf",
|
|
".png": "image",
|
|
".jpg": "image",
|
|
".jpeg": "image",
|
|
".tiff": "image",
|
|
".bmp": "image",
|
|
".xlsx": "excel",
|
|
".xls": "excel",
|
|
".docx": "word",
|
|
".doc": "word",
|
|
".pptx": "powerpoint",
|
|
".ppt": "powerpoint",
|
|
".txt": "text",
|
|
".md": "text",
|
|
".csv": "csv",
|
|
}
|
|
|
|
def __init__(self, lang: str = None):
|
|
self.lang = lang or settings.TESSERACT_LANG
|
|
log.info(f"DocumentProcessor initialized: lang={self.lang}")
|
|
|
|
def get_file_type(self, suffix: str) -> Optional[str]:
|
|
return self.SUPPORTED_EXTENSIONS.get(suffix.lower())
|
|
|
|
def extract_from_image(self, image_path: Path | str) -> str:
|
|
log.debug(f"Extracting from image: {image_path}")
|
|
image = Image.open(image_path)
|
|
log.debug(f"Image size: {image.size}")
|
|
|
|
text = pytesseract.image_to_string(image, lang=self.lang)
|
|
log.info(f"Image OCR: {len(text)} chars extracted")
|
|
|
|
return text.strip()
|
|
|
|
def extract_from_pdf(self, pdf_path: Path | str) -> list[dict]:
|
|
log.debug(f"Converting PDF to images: {pdf_path}")
|
|
|
|
pages = convert_from_path(pdf_path)
|
|
log.debug(f"PDF has {len(pages)} pages")
|
|
|
|
results = []
|
|
|
|
for i, page in enumerate(pages, 1):
|
|
log.debug(f"Processing page {i}/{len(pages)}")
|
|
text = pytesseract.image_to_string(page, lang=self.lang)
|
|
if text.strip():
|
|
results.append({
|
|
"page": i,
|
|
"text": text.strip()
|
|
})
|
|
log.debug(f"Page {i}: {len(text)} chars")
|
|
else:
|
|
log.debug(f"Page {i}: empty (skipped)")
|
|
|
|
log.info(f"PDF OCR complete: {len(results)} pages with text")
|
|
return results
|
|
|
|
def extract_from_excel(self, excel_path: Path | str) -> list[dict]:
|
|
log.debug(f"Extracting from Excel: {excel_path}")
|
|
|
|
try:
|
|
from openpyxl import load_workbook
|
|
except ImportError:
|
|
log.error("openpyxl not installed. Run: pip install openpyxl")
|
|
raise ImportError("openpyxl required for Excel files. Run: pip install openpyxl")
|
|
|
|
wb = load_workbook(excel_path, data_only=True)
|
|
results = []
|
|
|
|
for sheet_name in wb.sheetnames:
|
|
log.debug(f"Processing sheet: {sheet_name}")
|
|
sheet = wb[sheet_name]
|
|
|
|
rows_data = []
|
|
for row in sheet.iter_rows(values_only=True):
|
|
row_text = " | ".join(
|
|
str(cell) if cell is not None else ""
|
|
for cell in row
|
|
)
|
|
if row_text.strip(" |"):
|
|
rows_data.append(row_text)
|
|
|
|
if rows_data:
|
|
sheet_text = "\n".join(rows_data)
|
|
results.append({
|
|
"page": None,
|
|
"sheet": sheet_name,
|
|
"text": sheet_text
|
|
})
|
|
log.debug(f"Sheet '{sheet_name}': {len(sheet_text)} chars, {len(rows_data)} rows")
|
|
|
|
log.info(f"Excel extraction complete: {len(results)} sheets")
|
|
return results
|
|
|
|
def extract_from_word(self, word_path: Path | str) -> list[dict]:
|
|
log.debug(f"Extracting from Word: {word_path}")
|
|
|
|
try:
|
|
from docx import Document
|
|
except ImportError:
|
|
log.error("python-docx not installed. Run: pip install python-docx")
|
|
raise ImportError("python-docx required for Word files. Run: pip install python-docx")
|
|
|
|
doc = Document(word_path)
|
|
|
|
paragraphs = []
|
|
for para in doc.paragraphs:
|
|
if para.text.strip():
|
|
paragraphs.append(para.text)
|
|
|
|
tables_text = []
|
|
for table in doc.tables:
|
|
table_data = []
|
|
for row in table.rows:
|
|
row_text = " | ".join(
|
|
cell.text.strip() for cell in row.cells
|
|
)
|
|
table_data.append(row_text)
|
|
if table_data:
|
|
tables_text.append("\n".join(table_data))
|
|
|
|
full_text = "\n\n".join(paragraphs)
|
|
if tables_text:
|
|
full_text += "\n\n--- TABELLE ---\n\n" + "\n\n".join(tables_text)
|
|
|
|
log.info(f"Word extraction: {len(full_text)} chars")
|
|
|
|
return [{
|
|
"page": 1,
|
|
"text": full_text.strip()
|
|
}]
|
|
|
|
def extract_from_powerpoint(self, ppt_path: Path | str) -> list[dict]:
|
|
log.debug(f"Extracting from PowerPoint: {ppt_path}")
|
|
|
|
try:
|
|
from pptx import Presentation
|
|
except ImportError:
|
|
log.error("python-pptx not installed. Run: pip install python-pptx")
|
|
raise ImportError("python-pptx required for PowerPoint files. Run: pip install python-pptx")
|
|
|
|
prs = Presentation(ppt_path)
|
|
results = []
|
|
|
|
for i, slide in enumerate(prs.slides, 1):
|
|
slide_text = []
|
|
|
|
for shape in slide.shapes:
|
|
if hasattr(shape, "text") and shape.text.strip():
|
|
slide_text.append(shape.text)
|
|
|
|
if slide_text:
|
|
results.append({
|
|
"page": i,
|
|
"text": "\n".join(slide_text)
|
|
})
|
|
log.debug(f"Slide {i}: {len(slide_text)} shapes")
|
|
|
|
log.info(f"PowerPoint extraction: {len(results)} slides")
|
|
return results
|
|
|
|
def extract_from_text(self, text_path: Path | str) -> list[dict]:
|
|
log.debug(f"Reading text file: {text_path}")
|
|
|
|
try:
|
|
content = Path(text_path).read_text(encoding="utf-8")
|
|
except UnicodeDecodeError:
|
|
content = Path(text_path).read_text(encoding="latin-1")
|
|
|
|
log.info(f"Text file: {len(content)} chars")
|
|
|
|
return [{
|
|
"page": 1,
|
|
"text": content.strip()
|
|
}]
|
|
|
|
def extract_from_csv(self, csv_path: Path | str) -> list[dict]:
|
|
log.debug(f"Extracting from CSV: {csv_path}")
|
|
|
|
try:
|
|
import csv
|
|
except ImportError:
|
|
log.error("csv module not available")
|
|
raise
|
|
|
|
rows = []
|
|
with open(csv_path, "r", encoding="utf-8") as f:
|
|
reader = csv.reader(f)
|
|
for row in reader:
|
|
row_text = " | ".join(row)
|
|
rows.append(row_text)
|
|
|
|
content = "\n".join(rows)
|
|
log.info(f"CSV extraction: {len(rows)} rows, {len(content)} chars")
|
|
|
|
return [{
|
|
"page": 1,
|
|
"text": content
|
|
}]
|
|
|
|
def extract(self, file_path: Path | str) -> dict:
|
|
path = Path(file_path)
|
|
log.info(f"Extracting: {path.name}")
|
|
|
|
if not path.exists():
|
|
log.error(f"File not found: {path}")
|
|
raise FileNotFoundError(f"File not found: {path}")
|
|
|
|
suffix = path.suffix.lower()
|
|
file_type = self.get_file_type(suffix)
|
|
|
|
if not file_type:
|
|
log.error(f"Unsupported file type: {suffix}")
|
|
raise ValueError(f"Unsupported file type: {suffix}")
|
|
|
|
log.debug(f"File type: {file_type}")
|
|
|
|
if file_type == "pdf":
|
|
pages = self.extract_from_pdf(path)
|
|
full_text = "\n\n".join(p["text"] for p in pages)
|
|
return {
|
|
"source": str(path),
|
|
"type": "pdf",
|
|
"pages": pages,
|
|
"full_text": full_text,
|
|
"total_pages": len(pages)
|
|
}
|
|
|
|
elif file_type == "image":
|
|
text = self.extract_from_image(path)
|
|
return {
|
|
"source": str(path),
|
|
"type": "image",
|
|
"full_text": text,
|
|
"total_pages": 1
|
|
}
|
|
|
|
elif file_type == "excel":
|
|
sheets = self.extract_from_excel(path)
|
|
full_text = "\n\n--- SHEET ---\n\n".join(
|
|
f"[{s['sheet']}]\n{s['text']}" for s in sheets
|
|
)
|
|
return {
|
|
"source": str(path),
|
|
"type": "excel",
|
|
"pages": sheets,
|
|
"full_text": full_text,
|
|
"total_pages": len(sheets)
|
|
}
|
|
|
|
elif file_type == "word":
|
|
pages = self.extract_from_word(path)
|
|
return {
|
|
"source": str(path),
|
|
"type": "word",
|
|
"pages": pages,
|
|
"full_text": pages[0]["text"] if pages else "",
|
|
"total_pages": 1
|
|
}
|
|
|
|
elif file_type == "powerpoint":
|
|
slides = self.extract_from_powerpoint(path)
|
|
full_text = "\n\n".join(
|
|
f"[Slide {s['page']}]\n{s['text']}" for s in slides
|
|
)
|
|
return {
|
|
"source": str(path),
|
|
"type": "powerpoint",
|
|
"pages": slides,
|
|
"full_text": full_text,
|
|
"total_pages": len(slides)
|
|
}
|
|
|
|
elif file_type == "text":
|
|
pages = self.extract_from_text(path)
|
|
return {
|
|
"source": str(path),
|
|
"type": "text",
|
|
"pages": pages,
|
|
"full_text": pages[0]["text"] if pages else "",
|
|
"total_pages": 1
|
|
}
|
|
|
|
elif file_type == "csv":
|
|
pages = self.extract_from_csv(path)
|
|
return {
|
|
"source": str(path),
|
|
"type": "csv",
|
|
"pages": pages,
|
|
"full_text": pages[0]["text"] if pages else "",
|
|
"total_pages": 1
|
|
}
|
|
|
|
else:
|
|
raise ValueError(f"Unsupported file type: {suffix}")
|
|
|
|
|
|
def extract_text(file_path: Path | str, lang: str = None) -> dict:
|
|
processor = DocumentProcessor(lang=lang)
|
|
return processor.extract(file_path)
|
|
|
|
|
|
OCRExtractor = DocumentProcessor
|