Agent Zero
abbe7a18fc
- P0: hooks.py 3-tuple fix, trigger_dispatcher Contract, contacts/plugin unregister_actions_by_owner - P0: 5 test files — check_permission mocks removed, hardcoded DB credential → env var - P1: attachment_service DmsFile via Contract helper, restore_registry/history_hooks dedup - P1: mail/plugin restore unregister, mcp_client datetime.now(UTC), saved_views/filters patterns - P1: ProtectedRoute fail-closed, 13 test assertion fixes (bcrypt, DB-URLs, SECRET_KEYs) - P2: deprecated notifications → post_system_message (3 files), forgejo Base, report_generator lazy import - P2: webhooks permissions, deps.py/roles.py plugin perms removed, import_export default - P2: address/tags/entity_links patterns removed, worker.py Contract-Umgehungen fixed - P2: 28 frontend TODOs (hardcoded constants, deprecated notification API) - P3: dead code, duplicates, deprecated imports, private attr, __import__ inline - P3: 8 frontend TODOs (LucideIcons, inline styles, XSS, i18n) - ruff: 838 → 0 (612 auto-fix + 246 manual + 27 F821 regression fix) - F821: 30 → 0 (AutomationDefinition, DmsFile, user_id, Path, Any, String) - Contract-Umgehungen: 2 neue gefunden (worker.py:169, worker.py:280) und gefixt
127 lines
3.9 KiB
Python
127 lines
3.9 KiB
Python
"""Text extraction from various file formats for DMS indexing."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
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, encoding="utf-8", errors="replace") as f:
|
|
content = await f.read()
|
|
return _truncate(content)
|