206 lines
7.9 KiB
Python
206 lines
7.9 KiB
Python
from dataclasses import dataclass, field
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
import re
|
|
|
|
|
|
class ImportFailure(ValueError):
|
|
pass
|
|
|
|
|
|
@dataclass
|
|
class ImportResult:
|
|
markdown: str
|
|
warnings: list[str] = field(default_factory=list)
|
|
|
|
|
|
@dataclass
|
|
class ImportedImage:
|
|
name: str
|
|
content: bytes
|
|
content_type: str
|
|
cell: str = ""
|
|
|
|
|
|
@dataclass
|
|
class ExcelSheetResult:
|
|
title: str
|
|
markdown: str
|
|
images: list[ImportedImage] = field(default_factory=list)
|
|
|
|
|
|
def import_document(upload, flatten_excel=False) -> ImportResult:
|
|
suffix = Path(upload.name).suffix.lower()
|
|
content = upload.read()
|
|
upload.seek(0)
|
|
if suffix == ".docx":
|
|
return _docx(content)
|
|
if suffix in {".xlsx", ".xlsm"}:
|
|
return _xlsx(content, flatten=flatten_excel)
|
|
if suffix == ".pdf":
|
|
return _pdf(content)
|
|
if suffix in {".md", ".txt"}:
|
|
return ImportResult(content.decode("utf-8-sig"))
|
|
if suffix in {".doc", ".xls"}:
|
|
raise ImportFailure("Alte .doc/.xls-Dateien bitte zuerst als .docx/.xlsx speichern.")
|
|
raise ImportFailure("Unterstützt werden DOCX, XLSX, XLSM, PDF, Markdown und Text.")
|
|
|
|
|
|
def _docx(content):
|
|
import mammoth
|
|
result = mammoth.convert_to_markdown(BytesIO(content))
|
|
warnings = [message.message for message in result.messages]
|
|
return ImportResult(result.value.strip(), warnings)
|
|
|
|
|
|
def _render_excel_rows(rows, flatten=False):
|
|
from tabulate import tabulate
|
|
width = max(len(row) for row in rows)
|
|
rows = [row + [""] * (width - len(row)) for row in rows]
|
|
header, body = rows[0], rows[1:]
|
|
if not flatten:
|
|
return tabulate(body, headers=header, tablefmt="github")
|
|
# Document-style flattening: retain every cell in reading order without
|
|
# inventing record headings or field/value labels. Cells in one Excel row
|
|
# become lines in one paragraph; Excel rows are separated by a blank line.
|
|
blocks = []
|
|
for row in rows:
|
|
values = [_escape_markdown_text(value) for value in row if value != ""]
|
|
if values:
|
|
blocks.append(" \n".join(values))
|
|
return "\n\n".join(blocks) or "_Keine Inhalte_"
|
|
|
|
|
|
def _escape_markdown_text(value):
|
|
value = str(value).replace("\\", "\\\\")
|
|
for character in ("*", "_", "[", "]", "<", ">", "#", "|"):
|
|
value = value.replace(character, f"\\{character}")
|
|
return value.replace("\r\n", " \n").replace("\r", " \n").replace("\n", " \n")
|
|
|
|
|
|
def _xlsx(content, flatten=False):
|
|
from openpyxl import load_workbook
|
|
workbook = load_workbook(BytesIO(content), read_only=not flatten, data_only=True)
|
|
sections = []
|
|
for sheet in workbook.worksheets:
|
|
rows = [["" if cell is None else str(cell) for cell in row] for row in sheet.iter_rows(values_only=True)]
|
|
while rows and not any(value for value in rows[-1]):
|
|
rows.pop()
|
|
if not rows:
|
|
continue
|
|
rendered = _render_flattened_worksheet(sheet) if flatten else _render_excel_rows(rows)
|
|
sections.append(f"## {sheet.title}\n\n{rendered}")
|
|
if not sections:
|
|
raise ImportFailure("Die Arbeitsmappe enthält keine Daten.")
|
|
return ImportResult("\n\n".join(sections))
|
|
|
|
|
|
def import_excel_sheets(upload, flatten=False) -> list[ExcelSheetResult]:
|
|
"""Convert each non-empty worksheet into an individual document payload."""
|
|
from openpyxl import load_workbook
|
|
from openpyxl.utils import get_column_letter
|
|
|
|
suffix = Path(upload.name).suffix.lower()
|
|
if suffix not in {".xlsx", ".xlsm"}:
|
|
raise ImportFailure("Der Mehrblattimport unterstützt XLSX- und XLSM-Dateien.")
|
|
content = upload.read()
|
|
upload.seek(0)
|
|
workbook = load_workbook(BytesIO(content), read_only=False, data_only=True)
|
|
results = []
|
|
for sheet in workbook.worksheets:
|
|
rows = [["" if cell is None else str(cell) for cell in row] for row in sheet.iter_rows(values_only=True)]
|
|
while rows and not any(value for value in rows[-1]):
|
|
rows.pop()
|
|
images = []
|
|
for number, image in enumerate(getattr(sheet, "_images", ()), 1):
|
|
image_format = (getattr(image, "format", None) or "png").lower()
|
|
if image_format == "jpg":
|
|
image_format = "jpeg"
|
|
if image_format not in {"png", "jpeg", "gif", "webp"}:
|
|
continue
|
|
anchor = getattr(image, "anchor", None)
|
|
marker = getattr(anchor, "_from", None)
|
|
cell = f"{get_column_letter(marker.col + 1)}{marker.row + 1}" if marker else ""
|
|
try:
|
|
image_content = image._data()
|
|
except (AttributeError, OSError, ValueError):
|
|
continue
|
|
images.append(ImportedImage(
|
|
name=f"{slugify_filename(sheet.title)}-{number}.{image_format if image_format != 'jpeg' else 'jpg'}",
|
|
content=image_content, content_type=f"image/{image_format}", cell=cell,
|
|
))
|
|
if not rows and not images:
|
|
continue
|
|
if rows:
|
|
markdown = _render_flattened_worksheet(sheet) if flatten else _render_excel_rows(rows)
|
|
else:
|
|
markdown = ""
|
|
results.append(ExcelSheetResult(title=sheet.title, markdown=markdown, images=images))
|
|
if not results:
|
|
raise ImportFailure("Die Arbeitsmappe enthält keine Daten oder unterstützten Bilder.")
|
|
return results
|
|
|
|
|
|
def slugify_filename(value):
|
|
value = re.sub(r"[^A-Za-z0-9._-]+", "-", value).strip("-.")
|
|
return value[:80] or "arbeitsblatt"
|
|
|
|
|
|
def _render_flattened_worksheet(sheet):
|
|
"""Flatten ordinary cells while preserving explicitly defined Excel tables."""
|
|
from openpyxl.utils.cell import range_boundaries
|
|
|
|
table_ranges = []
|
|
for table in sheet.tables.values():
|
|
min_col, min_row, max_col, max_row = range_boundaries(table.ref)
|
|
table_ranges.append((min_col, min_row, max_col, max_row))
|
|
table_ranges.sort(key=lambda item: (item[1], item[0]))
|
|
|
|
def containing_table(row, column):
|
|
for bounds in table_ranges:
|
|
min_col, min_row, max_col, max_row = bounds
|
|
if min_row <= row <= max_row and min_col <= column <= max_col:
|
|
return bounds
|
|
return None
|
|
|
|
blocks = []
|
|
rendered_tables = set()
|
|
for row_number in range(1, sheet.max_row + 1):
|
|
ordinary_values = []
|
|
tables_starting_here = []
|
|
for column_number in range(1, sheet.max_column + 1):
|
|
bounds = containing_table(row_number, column_number)
|
|
if bounds:
|
|
if bounds[1] == row_number and bounds not in rendered_tables:
|
|
tables_starting_here.append(bounds)
|
|
rendered_tables.add(bounds)
|
|
continue
|
|
value = sheet.cell(row=row_number, column=column_number).value
|
|
if value not in (None, ""):
|
|
ordinary_values.append(_escape_markdown_text(value))
|
|
if ordinary_values:
|
|
blocks.append(" \n".join(ordinary_values))
|
|
for min_col, min_row, max_col, max_row in tables_starting_here:
|
|
rows = []
|
|
for table_row in sheet.iter_rows(
|
|
min_row=min_row, max_row=max_row, min_col=min_col, max_col=max_col, values_only=True
|
|
):
|
|
rows.append(["" if value is None else str(value) for value in table_row])
|
|
if rows:
|
|
blocks.append(_render_excel_rows(rows, flatten=False))
|
|
return "\n\n".join(blocks) or "_Keine Inhalte_"
|
|
|
|
|
|
def _pdf(content):
|
|
from pypdf import PdfReader
|
|
reader = PdfReader(BytesIO(content))
|
|
pages = []
|
|
for number, page in enumerate(reader.pages, 1):
|
|
text = (page.extract_text() or "").strip()
|
|
if text:
|
|
text = re.sub(r"[ \t]+\n", "\n", text)
|
|
pages.append(f"## Seite {number}\n\n{text}")
|
|
if not pages:
|
|
raise ImportFailure("Das PDF enthält keinen extrahierbaren Text. Für Scans ist OCR erforderlich.")
|
|
return ImportResult("\n\n".join(pages), ["PDF-Layout und Bilder können nicht vollständig übernommen werden."])
|