# -*- 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 ""