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