Agent Zero
ec81940178
- 0.7: UI-Design-Richtlinien (docs/ui-design-guidelines.md, 535 lines) - 0.8: Theme-Customization Backend (4 theme fields, migration 0023) - 0.9: Theme-Customization Frontend (SettingsTheme.tsx, themeStore.ts, live preview) - 0.10: RBAC-Audit (4 plugins secured, 53 routes with require_permission) - 0.11: LiteLLM-Cleanup (llm_client.py migrated from httpx to litellm) - 0.12: KI-Agent-Framework docs (plugin-development-guide.md, agent_capabilities field) - 0.13: Heartbeat configurable (ProactiveSettings, migration 0024, frontend UI) - 0.14: Unified Search Field-Level RBAC (resolve_permissions + filter_fields_by_permission) - 0.15: Undo/History-System (EntityHistory model, service, routes, migration 0025, HistoryViewer) - 0.16: Storage Backend (LocalStorage + S3Storage, DMS/attachments/mail updated) - 0.17: Import/Export unified Contact fields (firstname, surname, email_1, phone_1) - 0.18: .gitignore & Config-Cleanup (webui→frontend, python-jose removed, .env untracked) - 0.19: Mail-Salt Security-Fix (per-account random salt, migration 0026) - 0.20: AGPL replaced (PyMuPDF→pypdf, OnlyOffice→Collabora, LICENSE + THIRD_PARTY_LICENSES.md)
128 lines
4.0 KiB
Python
128 lines
4.0 KiB
Python
"""Text extraction from various file formats for DMS indexing."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
MAX_CHARS = 50_000
|
|
|
|
|
|
async def extract_text_from_file(file_path: str, mime_type: str) -> str:
|
|
"""Extract text content from a file based on its MIME type.
|
|
|
|
Supports:
|
|
- PDF (via pypdf)
|
|
- DOCX (via python-docx)
|
|
- XLSX (via openpyxl)
|
|
- PPTX (via python-pptx)
|
|
- text/* (via aiofiles)
|
|
- Images: returns empty (OCR not implemented)
|
|
|
|
Args:
|
|
file_path: Absolute path to the file.
|
|
mime_type: MIME type of the file.
|
|
|
|
Returns:
|
|
Extracted text, truncated to MAX_CHARS. Empty string on error.
|
|
"""
|
|
try:
|
|
if mime_type == "application/pdf":
|
|
return await _extract_pdf(file_path)
|
|
elif mime_type == "application/vnd.openxmlformats-officedocument.wordprocessingml.document":
|
|
return await _extract_docx(file_path)
|
|
elif mime_type == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet":
|
|
return await _extract_xlsx(file_path)
|
|
elif mime_type == "application/vnd.openxmlformats-officedocument.presentationml.presentation":
|
|
return await _extract_pptx(file_path)
|
|
elif mime_type.startswith("text/"):
|
|
return await _extract_text(file_path)
|
|
elif mime_type.startswith("image/"):
|
|
return ""
|
|
else:
|
|
logger.debug("Unsupported mime_type=%s for text extraction", mime_type)
|
|
return ""
|
|
except Exception:
|
|
logger.warning("Failed to extract text from %s (mime=%s)", file_path, mime_type)
|
|
return ""
|
|
|
|
|
|
def _truncate(text: str) -> str:
|
|
"""Truncate text to MAX_CHARS with indicator."""
|
|
if len(text) > MAX_CHARS:
|
|
return text[:MAX_CHARS] + "... [truncated]"
|
|
return text
|
|
|
|
|
|
async def _extract_pdf(file_path: str) -> str:
|
|
"""Extract text from PDF using pypdf (BSD-licensed)."""
|
|
from pypdf import PdfReader
|
|
|
|
reader = PdfReader(file_path)
|
|
parts: list[str] = []
|
|
for page in reader.pages:
|
|
text = page.extract_text()
|
|
if text:
|
|
parts.append(text)
|
|
return _truncate("\n".join(parts))
|
|
|
|
|
|
async def _extract_docx(file_path: str) -> str:
|
|
"""Extract text from DOCX using python-docx."""
|
|
from docx import Document
|
|
|
|
doc = Document(file_path)
|
|
parts: list[str] = []
|
|
for para in doc.paragraphs:
|
|
if para.text.strip():
|
|
parts.append(para.text)
|
|
# Also extract table text
|
|
for table in doc.tables:
|
|
for row in table.rows:
|
|
for cell in row.cells:
|
|
if cell.text.strip():
|
|
parts.append(cell.text)
|
|
return _truncate("\n".join(parts))
|
|
|
|
|
|
async def _extract_xlsx(file_path: str) -> str:
|
|
"""Extract text from XLSX using openpyxl."""
|
|
from openpyxl import load_workbook
|
|
|
|
wb = load_workbook(file_path, read_only=True, data_only=True)
|
|
parts: list[str] = []
|
|
for ws in wb.worksheets:
|
|
for row in ws.iter_rows(values_only=True):
|
|
row_text = " ".join(str(c) for c in row if c is not None)
|
|
if row_text.strip():
|
|
parts.append(row_text)
|
|
wb.close()
|
|
return _truncate("\n".join(parts))
|
|
|
|
|
|
async def _extract_pptx(file_path: str) -> str:
|
|
"""Extract text from PPTX using python-pptx."""
|
|
from pptx import Presentation
|
|
|
|
prs = Presentation(file_path)
|
|
parts: list[str] = []
|
|
for slide in prs.slides:
|
|
for shape in slide.shapes:
|
|
if shape.has_text_frame:
|
|
for para in shape.text_frame.paragraphs:
|
|
text = para.text.strip()
|
|
if text:
|
|
parts.append(text)
|
|
return _truncate("\n".join(parts))
|
|
|
|
|
|
async def _extract_text(file_path: str) -> str:
|
|
"""Read plain text file using aiofiles."""
|
|
import aiofiles
|
|
|
|
async with aiofiles.open(file_path, mode="r", encoding="utf-8", errors="replace") as f:
|
|
content = await f.read()
|
|
return _truncate(content)
|