Spaces:
Running
Running
| # -*- 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 "" | |