9fd748154d
- auth: self-registration (role=user); first-created-user becomes super_admin - roles: super_admin may promote others; regular admin cannot see super_admin accounts - products: user-created private groups; admin 'สินค้าขององค์กร' (shared) with hidden/public; users can create groups - analytics: team + per-user weak areas, close-rate-by-difficulty buckets, 30-day default, weekly trend, trainee table, active users; dashboard redesigned - files: docx + xlsx upload support (python-docx + openpyxl) - ui: tabs การฝึก→ผลการฝึก→ภาพรวม; admin lands on ภาพรวม / user on การฝึก; guide in topbar - consolidate: weak-areas merged into Results (10/page), my-personas merged into Training - copy: บุคคลต้นแบบ→persona everywhere; clearer add-product form (A/B/C, upload-or-fill) - report: remove ดูรายงาน UI entry (endpoint kept) Backend 348 tests pass; frontend build + vitest clean.
239 lines
8.4 KiB
Python
239 lines
8.4 KiB
Python
"""File parsing for uploaded documents (pdf / markdown / txt)."""
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
from pathlib import Path
|
|
|
|
from ..config import Config
|
|
|
|
|
|
class ParseError(Exception):
|
|
pass
|
|
|
|
|
|
def _iter_bounded_page_text(page, fitz):
|
|
"""Extract a page in clipped bands so one page cannot materialize whole text."""
|
|
rect = getattr(page, "rect", None)
|
|
try:
|
|
x0 = float(getattr(rect, "x0"))
|
|
y0 = float(getattr(rect, "y0"))
|
|
x1 = float(getattr(rect, "x1"))
|
|
y1 = float(getattr(rect, "y1"))
|
|
except (AttributeError, TypeError, ValueError, OverflowError) as exc:
|
|
raise ParseError("PDF page geometry unavailable") from exc
|
|
height = y1 - y0
|
|
if (
|
|
not all(math.isfinite(value) for value in (x0, y0, x1, y1))
|
|
or x1 <= x0
|
|
or height <= 0
|
|
):
|
|
raise ParseError("PDF page geometry unavailable")
|
|
band_count = max(1, min(128, int(height / 32) + 1))
|
|
for index in range(band_count):
|
|
band_y0 = y0 + height * index / band_count
|
|
band_y1 = y0 + height * (index + 1) / band_count
|
|
clip = fitz.Rect(x0, band_y0, x1, band_y1)
|
|
chunk = page.get_text("text", clip=clip)
|
|
text = chunk if isinstance(chunk, str) else str(chunk)
|
|
max_chunk_chars = min(
|
|
Config.UPLOAD_MAX_EXTRACTED_CHARS,
|
|
Config.UPLOAD_MAX_PDF_CHUNK_CHARS,
|
|
)
|
|
if len(text) > max_chunk_chars:
|
|
raise ParseError("extracted text too large")
|
|
yield text
|
|
|
|
|
|
def parse_pdf(path: Path) -> str:
|
|
try:
|
|
import fitz # PyMuPDF
|
|
except ImportError as exc:
|
|
raise ParseError("PDF parser is unavailable") from exc
|
|
|
|
try:
|
|
too_large = path.exists() and path.stat().st_size > Config.UPLOAD_MAX_MB * 1024 * 1024
|
|
except OSError as exc:
|
|
raise ParseError("cannot inspect PDF") from exc
|
|
if too_large:
|
|
raise ParseError("PDF file too large")
|
|
try:
|
|
doc = fitz.open(path)
|
|
except Exception as exc:
|
|
raise ParseError("cannot open PDF") from exc
|
|
try:
|
|
page_count = getattr(doc, "page_count", None)
|
|
if page_count is None:
|
|
page_count = len(doc)
|
|
if int(page_count) > Config.UPLOAD_MAX_PDF_PAGES:
|
|
raise ParseError("PDF has too many pages")
|
|
parts = []
|
|
extracted_chars = 0
|
|
for page in doc:
|
|
for chunk in _iter_bounded_page_text(page, fitz):
|
|
text = chunk if isinstance(chunk, str) else str(chunk)
|
|
separator_chars = 1 if parts else 0
|
|
extracted_chars += separator_chars + len(text)
|
|
if extracted_chars > Config.UPLOAD_MAX_EXTRACTED_CHARS:
|
|
raise ParseError("extracted text too large")
|
|
parts.append(text)
|
|
return "\n".join(parts)
|
|
except ParseError:
|
|
raise
|
|
except Exception as exc:
|
|
raise ParseError("could not extract PDF text") from exc
|
|
finally:
|
|
doc.close()
|
|
|
|
|
|
def parse_text(path: Path) -> str:
|
|
try:
|
|
if path.stat().st_size > Config.UPLOAD_TEXT_MAX_BYTES:
|
|
raise ParseError("text document too large")
|
|
except FileNotFoundError as exc:
|
|
raise ParseError("text document is unavailable") from exc
|
|
raw = path.read_bytes()
|
|
if len(raw) > Config.UPLOAD_TEXT_MAX_BYTES:
|
|
raise ParseError("text document too large")
|
|
# Try utf-8 first, else detect encoding
|
|
try:
|
|
return raw.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
pass
|
|
try:
|
|
from charset_normalizer import from_bytes
|
|
match = from_bytes(raw).best()
|
|
enc = getattr(match, "encoding", None) if match is not None else None
|
|
except ImportError as exc:
|
|
raise ParseError("text encoding detector is unavailable") from exc
|
|
if not enc:
|
|
raise ParseError("could not detect text encoding")
|
|
try:
|
|
return raw.decode(enc, errors="replace")
|
|
except (LookupError, UnicodeError) as exc:
|
|
raise ParseError("could not decode text document") from exc
|
|
|
|
|
|
def _check_size(path: Path, *, label: str) -> None:
|
|
"""Fail closed when the file exceeds the configured upload size cap."""
|
|
try:
|
|
if path.exists() and path.stat().st_size > Config.UPLOAD_MAX_MB * 1024 * 1024:
|
|
raise ParseError(f"{label} file too large")
|
|
except OSError as exc:
|
|
raise ParseError(f"cannot inspect {label}") from exc
|
|
|
|
|
|
def parse_docx(path: Path) -> str:
|
|
"""Extract paragraph text from a .docx (python-docx) with size caps.
|
|
|
|
Only document text is read; embedded OLE/macros are never touched. The
|
|
result is capped at UPLOAD_MAX_EXTRACTED_CHARS.
|
|
"""
|
|
_check_size(path, label="docx")
|
|
try:
|
|
import docx # python-docx
|
|
except ImportError as exc:
|
|
raise ParseError("docx parser is unavailable") from exc
|
|
try:
|
|
document = docx.Document(str(path))
|
|
except Exception as exc:
|
|
raise ParseError("cannot open docx") from exc
|
|
parts = []
|
|
chars = 0
|
|
try:
|
|
for para in document.paragraphs:
|
|
text = (para.text or "") if isinstance(para.text, str) else str(para.text or "")
|
|
separator_chars = 1 if parts else 0
|
|
chars += separator_chars + len(text)
|
|
if chars > Config.UPLOAD_MAX_EXTRACTED_CHARS:
|
|
raise ParseError("extracted text too large")
|
|
parts.append(text)
|
|
# Tables carry structured data too; append cell text bounded.
|
|
for table in document.tables:
|
|
for row in table.rows:
|
|
for cell in row.cells:
|
|
text = (cell.text or "") if isinstance(cell.text, str) else str(cell.text or "")
|
|
separator_chars = 1 if parts else 0
|
|
chunks = text.split("\n")
|
|
for chunk in chunks:
|
|
if not chunk:
|
|
continue
|
|
chars += separator_chars + len(chunk)
|
|
if chars > Config.UPLOAD_MAX_EXTRACTED_CHARS:
|
|
raise ParseError("extracted text too large")
|
|
parts.append(chunk)
|
|
return "\n".join(parts)
|
|
except ParseError:
|
|
raise
|
|
except Exception as exc:
|
|
raise ParseError("could not extract docx text") from exc
|
|
|
|
|
|
def parse_xlsx(path: Path) -> str:
|
|
"""Extract cell text from a .xlsx (openpyxl) with size caps.
|
|
|
|
Data-only mode for safety (no formulas evaluated); caps on rows and chars.
|
|
"""
|
|
_check_size(path, label="xlsx")
|
|
try:
|
|
import openpyxl
|
|
except ImportError as exc:
|
|
raise ParseError("xlsx parser is unavailable") from exc
|
|
try:
|
|
workbook = openpyxl.load_workbook(
|
|
str(path), read_only=True, data_only=True, keep_links=False
|
|
)
|
|
except Exception as exc:
|
|
raise ParseError("cannot open xlsx") from exc
|
|
try:
|
|
parts = []
|
|
chars = 0
|
|
sheet_count = 0
|
|
for worksheet in workbook.worksheets:
|
|
sheet_count += 1
|
|
if sheet_count > Config.UPLOAD_MAX_XLSX_SHEETS:
|
|
raise ParseError("xlsx has too many sheets")
|
|
row_count = 0
|
|
for row in worksheet.iter_rows(values_only=True):
|
|
row_count += 1
|
|
if row_count > Config.UPLOAD_MAX_XLSX_ROWS:
|
|
raise ParseError("xlsx has too many rows")
|
|
cells = []
|
|
for value in row:
|
|
if value is None:
|
|
continue
|
|
text = str(value)
|
|
if not text:
|
|
continue
|
|
cells.append(text)
|
|
if not cells:
|
|
continue
|
|
line = " | ".join(cells)
|
|
separator_chars = 1 if parts else 0
|
|
chars += separator_chars + len(line)
|
|
if chars > Config.UPLOAD_MAX_EXTRACTED_CHARS:
|
|
raise ParseError("extracted text too large")
|
|
parts.append(line)
|
|
return "\n".join(parts)
|
|
except ParseError:
|
|
raise
|
|
except Exception as exc:
|
|
raise ParseError("could not extract xlsx text") from exc
|
|
finally:
|
|
try:
|
|
workbook.close()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def parse_document(path: Path) -> str:
|
|
ext = path.suffix.lower().lstrip(".")
|
|
if ext == "pdf":
|
|
return parse_pdf(path)
|
|
if ext == "docx":
|
|
return parse_docx(path)
|
|
if ext == "xlsx":
|
|
return parse_xlsx(path)
|
|
if ext not in {"md", "txt"}:
|
|
raise ParseError("unsupported document type")
|
|
return parse_text(path)
|