Spaces:
Running
Running
File size: 4,237 Bytes
92038dd 51096c9 f9508a4 51096c9 178c618 f9508a4 178c618 f9508a4 51096c9 f9508a4 51096c9 f9508a4 51096c9 6a408fb 51096c9 0981a11 51096c9 0981a11 51096c9 0981a11 51096c9 f9508a4 51096c9 6a408fb 51096c9 4ce40cf 51096c9 4ce40cf 51096c9 f9508a4 51096c9 f9508a4 51096c9 f9508a4 51096c9 f9508a4 51096c9 f9508a4 51096c9 f9508a4 51096c9 f9508a4 51096c9 f9508a4 728426c 51096c9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | # -*- 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
|