Spaces:
Running
Running
| # -*- coding: utf-8 -*- | |
| """Headless file parsing helpers used by both Streamlit and R3-Free executor.""" | |
| from __future__ import annotations | |
| import io | |
| import logging | |
| import os | |
| from pathlib import Path | |
| from typing import BinaryIO, Iterable | |
| from docx import Document | |
| from config import APP_CONFIG | |
| try: | |
| import pandas as pd | |
| PANDAS_AVAILABLE = True | |
| except ImportError: | |
| PANDAS_AVAILABLE = False | |
| try: | |
| import pymupdf | |
| PYMUPDF_AVAILABLE = True | |
| except ImportError: | |
| PYMUPDF_AVAILABLE = False | |
| def _name(file) -> str: | |
| if isinstance(file, (str, os.PathLike, Path)): | |
| return os.fspath(file) | |
| return str(getattr(file, "name", "uploaded_file")) | |
| def _read_bytes(file) -> bytes: | |
| if isinstance(file, (str, os.PathLike, Path)): | |
| return Path(file).read_bytes() | |
| try: | |
| file.seek(0) | |
| except Exception: | |
| pass | |
| if hasattr(file, "getvalue"): | |
| data = file.getvalue() | |
| else: | |
| data = file.read() | |
| try: | |
| file.seek(0) | |
| except Exception: | |
| pass | |
| return bytes(data or b"") | |
| def _binary_stream(file) -> io.BytesIO | BinaryIO: | |
| if isinstance(file, (str, os.PathLike, Path)): | |
| return io.BytesIO(Path(file).read_bytes()) | |
| try: | |
| file.seek(0) | |
| except Exception: | |
| pass | |
| return file | |
| def extract_text_from_files(uploaded_files: Iterable, max_chars=None): | |
| if max_chars is None: | |
| max_chars = APP_CONFIG.get("max_extract_chars", 150000) | |
| extracted_text = "" | |
| for file in uploaded_files or []: | |
| filename = _name(file) | |
| lower_name = filename.lower() | |
| try: | |
| if lower_name.endswith(".pdf") and PYMUPDF_AVAILABLE: | |
| pdf_bytes = _read_bytes(file) | |
| with pymupdf.open(stream=pdf_bytes, filetype="pdf") as pdf_doc: | |
| for page in pdf_doc: | |
| text = page.get_text() | |
| if text: | |
| extracted_text += text + "\n" | |
| if len(extracted_text) > max_chars: | |
| break | |
| elif lower_name.endswith((".docx", ".doc")): | |
| stream = _binary_stream(file) | |
| doc = Document(stream) | |
| for para in doc.paragraphs: | |
| extracted_text += para.text + "\n" | |
| if len(extracted_text) > max_chars: | |
| break | |
| else: | |
| extracted_text += _read_bytes(file).decode("utf-8", errors="ignore") + "\n" | |
| if len(extracted_text) > max_chars: | |
| logging.warning("File %s was truncated safely to %s characters.", filename, max_chars) | |
| break | |
| except Exception as exc: | |
| logging.warning("File parse failed for %s: %s", filename, exc) | |
| finally: | |
| if not isinstance(file, (str, os.PathLike, Path)): | |
| try: | |
| file.seek(0) | |
| except Exception: | |
| pass | |
| return extracted_text[:max_chars] | |
| def extract_data_from_files(uploaded_file, max_chars=20000): | |
| if not uploaded_file: | |
| return "" | |
| filename = _name(uploaded_file) | |
| lower_name = filename.lower() | |
| try: | |
| if PANDAS_AVAILABLE: | |
| source = os.fspath(uploaded_file) if isinstance(uploaded_file, (str, os.PathLike, Path)) else uploaded_file | |
| if not isinstance(uploaded_file, (str, os.PathLike, Path)): | |
| try: | |
| uploaded_file.seek(0) | |
| except Exception: | |
| pass | |
| df = pd.read_csv(source) if lower_name.endswith(".csv") else pd.read_excel(source) | |
| if len(df) > 100: | |
| logging.warning("Data file %s truncated safely to first 100 rows.", filename) | |
| df = df.head(100) | |
| return df.to_markdown(index=False) | |
| return _read_bytes(uploaded_file).decode("utf-8", errors="ignore")[:max_chars] | |
| except Exception as exc: | |
| logging.warning("Data parse failed for %s: %s", filename, exc) | |
| return "" | |
| finally: | |
| if not isinstance(uploaded_file, (str, os.PathLike, Path)): | |
| try: | |
| uploaded_file.seek(0) | |
| except Exception: | |
| pass | |