# -*- 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 hashlib import io import logging import os import re 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], } _STRICT_RESEARCH_TEMPLATE_EXTENSIONS = { "qualitative_materials": ".txt", "survey_quantitative_data": ".csv", } def _validate_qualitative_template_text(text: str) -> None: value = str(text or "").strip() if not value: raise FileParseError("template_content_empty") metadata_lines = re.findall(r"(?m)^\s*\[[^\]\n]{2,100}\]\s*[：:]\s*\S+", value) dialogue_lines = [ line.strip() for line in value.splitlines() if line.strip() and not line.lstrip().startswith("[") and re.search(r"[：:]", line) ] if not re.search(r"\bINT-\d+\b", value, flags=re.I): raise FileParseError("template_missing_interview_id") if len(metadata_lines) < 3 or len(dialogue_lines) < 2: raise FileParseError("template_structure_mismatch") # The downloaded template contains explicit placeholders. If any remain, # the file is not yet research material and must not enter generation. if re.search(r"\[(?:在这里输入|Enter\b|Введите\b|[^\]\n]*осында енгізіңіз)", value, flags=re.I): raise FileParseError("template_placeholders_not_replaced") def _validate_survey_template_text(text: str) -> None: value = str(text or "").strip() if not value: raise FileParseError("template_content_empty") lines = [line.strip() for line in value.splitlines() if line.strip()] section_markers = [line for line in lines if line.startswith("---") and line.endswith("---")] has_descriptive = any(",SD," in line and line.count(",") >= 4 for line in lines) has_reliability = any("Cronbach's Alpha" in line and ",CR," in line and ",AVE" in line for line in lines) has_hypothesis = any("Beta" in line and ("P-value" in line or "P-значение" in line or "P-мәні" in line) for line in lines) if len(section_markers) < 3 or not (has_descriptive and has_reliability and has_hypothesis): raise FileParseError("template_structure_mismatch") def _research_material_size_hint(file) -> int | None: if isinstance(file, (str, os.PathLike, Path)): try: return int(Path(file).stat().st_size) except OSError: return None value = getattr(file, "size", None) try: if value is not None: return int(value) except (TypeError, ValueError): pass try: return int(file.getbuffer().nbytes) except Exception: return None def research_material_text_fingerprint(text: str) -> str: """Stable fingerprint for detecting an unchanged downloadable template.""" normalized = str(text or "").replace("\r\n", "\n").replace("\r", "\n") normalized = "\n".join(line.rstrip() for line in normalized.splitlines()).strip() return hashlib.sha256(normalized.encode("utf-8")).hexdigest() def _strict_research_material_text(file, template_profile: str, *, max_bytes: int | None = None) -> tuple[str, int]: profile = str(template_profile or "").strip().casefold() expected_ext = _STRICT_RESEARCH_TEMPLATE_EXTENSIONS.get(profile) if not expected_ext: raise FileParseError("template_profile_unsupported") filename = _name(file) suffix = Path(filename).suffix.casefold() if suffix != expected_ext: raise FileParseError(f"template_extension_mismatch:expected={expected_ext}") byte_limit = int(max_bytes or 0) size_hint = _research_material_size_hint(file) if byte_limit > 0 and size_hint is not None and size_hint > byte_limit: raise FileParseError(f"research_material_file_too_large:{size_hint}:{byte_limit}") try: raw = _read_bytes(file) if byte_limit > 0 and len(raw) > byte_limit: raise FileParseError(f"research_material_file_too_large:{len(raw)}:{byte_limit}") value = _decode_text_bytes(raw).strip() finally: if not isinstance(file, (str, os.PathLike, Path)): try: file.seek(0) except Exception: pass if profile == "qualitative_materials": _validate_qualitative_template_text(value) elif profile == "survey_quantitative_data": _validate_survey_template_text(value) return value, len(raw) def extract_research_material_content(file, template_profile: str, max_chars=None, *, max_bytes: int | None = None) -> str: """Read one strict research-material file under the selected template contract.""" value, _byte_size = _strict_research_material_text(file, template_profile, max_bytes=max_bytes) if max_chars is None: max_chars = APP_CONFIG.get("max_extract_chars", 150000) max_chars = max(1, int(max_chars)) return value[:max_chars] def validate_research_material_template( file, template_profile: str, max_chars=None, *, max_bytes: int | None = None, disallowed_content_fingerprints: set[str] | None = None, ) -> dict: """Return an exact, UI-safe strict-template validation result. Validation counts the full decoded text before any bundle truncation, so the caller can reject oversized bundles before task creation/quota reservation. """ filename = os.path.basename(_name(file)) try: content, byte_size = _strict_research_material_text(file, template_profile, max_bytes=max_bytes) fingerprint = research_material_text_fingerprint(content) if disallowed_content_fingerprints and fingerprint in disallowed_content_fingerprints: raise FileParseError("template_example_not_replaced") shown = content if max_chars is None else content[: max(1, int(max_chars))] return { "ok": True, "filename": filename, "text_chars": len(content), "preview_chars": len(shown), "byte_size": byte_size, "fingerprint": fingerprint, "error": "", } except Exception as exc: return { "ok": False, "filename": filename, "text_chars": 0, "preview_chars": 0, "byte_size": int(_research_material_size_hint(file) or 0), "fingerprint": "", "error": str(exc)[:200], } 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 ""