Spaces:
Sleeping
Sleeping
File size: 7,236 Bytes
92038dd 1be38c4 51096c9 f9508a4 51096c9 178c618 f9508a4 178c618 f9508a4 51096c9 1be38c4 51096c9 1be38c4 51096c9 f9508a4 51096c9 1be38c4 51096c9 6a408fb 1be38c4 4ce40cf 51096c9 1be38c4 51096c9 1be38c4 f9508a4 51096c9 f9508a4 51096c9 f9508a4 1be38c4 51096c9 1be38c4 f9508a4 | 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 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 | # -*- coding: utf-8 -*-
"""Headless file parsing helpers used by both Streamlit and R3-Free executor.
P0-R7-B keeps parsing fail-soft for historical generation calls, while exposing
an explicit validator that the UI can run before task creation/quota reservation.
"""
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
class FileParseError(RuntimeError):
"""A user file cannot be parsed reliably by the current runtime."""
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 _decode_text_bytes(data: bytes) -> str:
if not data:
return ""
for encoding in ("utf-8-sig", "utf-8", "utf-16", "gb18030"):
try:
return data.decode(encoding)
except UnicodeDecodeError:
continue
raise FileParseError("text_encoding_unsupported")
def _read_table(file, lower_name: str):
if not PANDAS_AVAILABLE:
raise FileParseError("pandas_unavailable")
source = os.fspath(file) if isinstance(file, (str, os.PathLike, Path)) else file
if not isinstance(file, (str, os.PathLike, Path)):
try:
file.seek(0)
except Exception:
pass
if lower_name.endswith(".csv"):
last_exc = None
for encoding in ("utf-8-sig", "utf-8", "gb18030"):
try:
if not isinstance(file, (str, os.PathLike, Path)):
file.seek(0)
return pd.read_csv(source, encoding=encoding)
except UnicodeDecodeError as exc:
last_exc = exc
continue
if last_exc:
raise last_exc
return pd.read_csv(source)
return pd.read_excel(source)
def extract_file_content(file, max_chars=None) -> str:
"""Parse one supported user file and raise FileParseError on failure.
The returned text is suitable for either literature context or a bounded
research-material bundle. Structured files are converted to Markdown.
"""
if max_chars is None:
max_chars = APP_CONFIG.get("max_extract_chars", 150000)
max_chars = max(1, int(max_chars))
filename = _name(file)
lower_name = filename.lower()
try:
if lower_name.endswith(".pdf"):
if not PYMUPDF_AVAILABLE:
raise FileParseError("pdf_parser_unavailable")
pdf_bytes = _read_bytes(file)
parts = []
with pymupdf.open(stream=pdf_bytes, filetype="pdf") as pdf_doc:
for page in pdf_doc:
text = page.get_text()
if text:
parts.append(text)
if sum(len(part) for part in parts) >= max_chars:
break
result = "\n".join(parts)
elif lower_name.endswith((".docx", ".doc")):
stream = _binary_stream(file)
doc = Document(stream)
parts = []
total = 0
for para in doc.paragraphs:
value = str(para.text or "")
if value:
parts.append(value)
total += len(value)
if total >= max_chars:
break
result = "\n".join(parts)
elif lower_name.endswith((".csv", ".xls", ".xlsx")):
df = _read_table(file, lower_name)
if len(df) > 100:
logging.warning("Data file %s truncated safely to first 100 rows.", filename)
df = df.head(100)
result = df.to_markdown(index=False)
elif lower_name.endswith((".txt", ".md")):
result = _decode_text_bytes(_read_bytes(file))
else:
raise FileParseError("unsupported_file_extension")
except FileParseError:
raise
except Exception as exc:
raise FileParseError(f"parse_failed:{exc.__class__.__name__}") from exc
finally:
if not isinstance(file, (str, os.PathLike, Path)):
try:
file.seek(0)
except Exception:
pass
normalized = str(result or "").strip()
if not normalized:
raise FileParseError("parsed_content_empty")
return normalized[:max_chars]
def validate_file_parseable(file, max_chars=4000) -> dict:
"""Return an explicit, UI-safe parseability result for one file."""
filename = os.path.basename(_name(file))
try:
content = extract_file_content(file, max_chars=max_chars)
return {
"ok": True,
"filename": filename,
"text_chars": len(content),
"error": "",
}
except Exception as exc:
return {
"ok": False,
"filename": filename,
"text_chars": 0,
"error": str(exc)[:160],
}
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 = []
total = 0
for file in uploaded_files or []:
filename = _name(file)
try:
remaining = max(1, int(max_chars) - total)
text = extract_file_content(file, max_chars=remaining)
extracted.append(text)
total += len(text)
if total >= int(max_chars):
logging.warning("File %s caused aggregate text to reach %s characters.", filename, max_chars)
break
except Exception as exc:
# Historical generation calls remain fail-soft. P0-R7-B's UI gate
# uses validate_file_parseable() before task/quota creation.
logging.warning("File parse failed for %s: %s", filename, exc)
return "\n".join(extracted)[: int(max_chars)]
def extract_data_from_files(uploaded_file, max_chars=20000):
if not uploaded_file:
return ""
try:
# P0-R7-B research bundles are UTF-8 text. Existing CSV/XLS/XLSX and
# document inputs remain supported through the same parser.
return extract_file_content(uploaded_file, max_chars=max_chars)
except Exception as exc:
logging.warning("Data parse failed for %s: %s", _name(uploaded_file), exc)
return ""
|