Blame for utils/file_parser.py

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