P0-R7-D-R1 mixed reference/research materials + parse-first gate

app.py

@@ -39,6 +39,7 @@ from core.advanced_method_registry import (
     reconcile_method_authority, resolve_generation_preflight,
 )
 from smartpaper_restore_bridge import edge_bridge
+from utils.file_parser import extract_file_content, validate_file_parseable
 
 st.set_page_config(
     page_title="Article Agent",
@@ -309,7 +310,8 @@ for key, default in {
     "sp_method_selector_ui": "",
     "sp_research_design_ui_snapshot": "",
     "sp_research_intent_snapshot": "",
-    "sp_material_document_use": "reference",
+    "sp_material_document_use": "reference",  # legacy recovered-session compatibility only
+    "sp_material_parse_cache": {},
 }.items():
     if key not in st.session_state:
         st.session_state[key] = default
@@ -477,6 +479,7 @@ def _task_output_path(task: dict) -> Path | None:
 
 
 def _save_uploaded(items, inputs_dir: Path, prefix: str) -> list[str]:
+    """Persist uploads without copying the entire file into one Python bytes object."""
     saved = []
     for index, item in enumerate(items or []):
         if item is None:
@@ -485,8 +488,15 @@ def _save_uploaded(items, inputs_dir: Path, prefix: str) -> list[str]:
         safe_name = re.sub(r"[^A-Za-z0-9_.-]", "_", original)[:140] or f"input_{index}"
         filename = f"{prefix}_{index:02d}_{safe_name}"
         path = inputs_dir / filename
-        data = item.getvalue() if hasattr(item, "getvalue") else item.read()
-        path.write_bytes(bytes(data or b""))
+        try:
+            item.seek(0)
+        except Exception:
+            pass
+        if hasattr(item, "read"):
+            with path.open("wb") as target:
+                shutil.copyfileobj(item, target, length=1024 * 1024)
+        else:
+            path.write_bytes(bytes(item or b""))
         saved.append(str(Path("inputs") / filename))
         try:
             item.seek(0)
@@ -495,6 +505,176 @@ def _save_uploaded(items, inputs_dir: Path, prefix: str) -> list[str]:
     return saved
 
 
+def _research_material_upload_policy() -> dict:
+    policy = dict(APP_CONFIG.get("research_material_upload_policy") or {})
+    reference_limit = int(policy.get("reference_file_limit") or 0)
+    research_limit = int(policy.get("research_material_file_limit") or 0)
+    if reference_limit <= 0 or research_limit <= 0:
+        raise RuntimeError("research_material_upload_policy_invalid")
+    return {
+        **policy,
+        "reference_file_limit": reference_limit,
+        "research_material_file_limit": research_limit,
+        "qualitative_min_combined_chars": max(
+            1, int(policy.get("qualitative_min_combined_chars") or 1)
+        ),
+        "quantitative_bundle_max_chars": max(
+            1000, int(policy.get("quantitative_bundle_max_chars") or 20000)
+        ),
+        "qualitative_bundle_max_chars": max(
+            1000, int(policy.get("qualitative_bundle_max_chars") or APP_CONFIG.get("max_extract_chars", 150000))
+        ),
+    }
+
+
+def _uploaded_file_identity(item, occurrence: int = 0) -> str:
+    provider_id = str(getattr(item, "file_id", "") or "").strip()
+    name = os.path.basename(str(getattr(item, "name", "") or ""))
+    size = str(getattr(item, "size", "") or "")
+    mime = str(getattr(item, "type", "") or "")
+    source = provider_id or f"{name}|{size}|{mime}|{occurrence}"
+    return hashlib.sha256(source.encode("utf-8")).hexdigest()[:24]
+
+
+def _uploaded_material_entries() -> list[dict]:
+    uploads = list(st.session_state.get("up_research_materials") or [])
+    occurrences: dict[tuple[str, str, str], int] = {}
+    entries: list[dict] = []
+    for item in uploads:
+        name = os.path.basename(str(getattr(item, "name", "") or ""))
+        size = str(getattr(item, "size", "") or "")
+        mime = str(getattr(item, "type", "") or "")
+        signature = (name, size, mime)
+        occurrence = occurrences.get(signature, 0)
+        occurrences[signature] = occurrence + 1
+        identity = _uploaded_file_identity(item, occurrence)
+        key = f"sp_material_role_{identity}"
+        ext = Path(name).suffix.lower()
+        default_role = "research" if ext in {".txt", ".csv", ".xls", ".xlsx"} else "reference"
+        if key not in st.session_state:
+            st.session_state[key] = default_role
+        role = str(st.session_state.get(key) or default_role).strip().lower()
+        if role not in {"reference", "research"}:
+            role = default_role
+        entries.append(
+            {
+                "item": item,
+                "identity": identity,
+                "state_key": key,
+                "name": name,
+                "extension": ext,
+                "role": role,
+            }
+        )
+    return entries
+
+
+def _material_role_counts(entries: list[dict] | None = None) -> dict[str, int]:
+    entries = list(entries if entries is not None else _uploaded_material_entries())
+    return {
+        "reference": sum(1 for entry in entries if entry.get("role") == "reference"),
+        "research": sum(1 for entry in entries if entry.get("role") == "research"),
+    }
+
+
+def _material_limits_ok(entries: list[dict] | None = None) -> tuple[bool, dict[str, int], dict]:
+    entries = list(entries if entries is not None else _uploaded_material_entries())
+    counts = _material_role_counts(entries)
+    policy = _research_material_upload_policy()
+    ok = (
+        counts["reference"] <= policy["reference_file_limit"]
+        and counts["research"] <= policy["research_material_file_limit"]
+    )
+    return ok, counts, policy
+
+
+def _explicit_material_bucket(method_family: str, role: str) -> str:
+    capability = method_capability(method_family)
+    material_routing = dict(capability.get("material_routing") or {})
+    route_key = "user_reference" if role == "reference" else "user_research"
+    bucket = str(material_routing.get(route_key) or "").strip().casefold()
+    if bucket not in {"reference", "data", "interview"}:
+        raise RuntimeError("method_material_routing_unsupported")
+    return bucket
+
+
+def _material_preflight_report(method_family: str, entries: list[dict] | None = None) -> dict:
+    """Validate every selected file before task creation/quota reservation."""
+    entries = list(entries if entries is not None else _uploaded_material_entries())
+    limits_ok, counts, policy = _material_limits_ok(entries)
+    report = {
+        "ok": bool(limits_ok),
+        "counts": counts,
+        "policy": policy,
+        "failures": [],
+        "research_chars": 0,
+    }
+    if not limits_ok:
+        return report
+
+    cache = st.session_state.get("sp_material_parse_cache")
+    if not isinstance(cache, dict):
+        cache = {}
+        st.session_state["sp_material_parse_cache"] = cache
+
+    live_cache_keys = set()
+    for entry in entries:
+        item = entry["item"]
+        role = str(entry["role"])
+        cache_key = f"{entry['identity']}|{role}"
+        live_cache_keys.add(cache_key)
+        result = cache.get(cache_key)
+        if not isinstance(result, dict):
+            result = validate_file_parseable(item)
+            cache[cache_key] = dict(result)
+        if not bool(result.get("ok")):
+            report["failures"].append(
+                {
+                    "name": entry["name"],
+                    "role": role,
+                    "error": str(result.get("error") or "parse_failed"),
+                }
+            )
+        elif role == "research":
+            report["research_chars"] += int(result.get("text_chars") or 0)
+
+    # Drop stale cache entries when files are removed or roles change.
+    st.session_state["sp_material_parse_cache"] = {
+        key: value for key, value in cache.items() if key in live_cache_keys
+    }
+
+    if report["failures"]:
+        report["ok"] = False
+    research_bucket = _explicit_material_bucket(method_family, "research")
+    if (
+        report["ok"]
+        and counts["research"] > 0
+        and research_bucket == "interview"
+        and report["research_chars"] < policy["qualitative_min_combined_chars"]
+    ):
+        report["ok"] = False
+        report["content_insufficient"] = True
+    return report
+
+
+def _material_bundle_text(entries: list[dict], research_bucket: str) -> str:
+    research_entries = [entry for entry in entries if entry.get("role") == "research"]
+    if not research_entries:
+        return ""
+    policy = _research_material_upload_policy()
+    if research_bucket == "data":
+        total_budget = int(policy["quantitative_bundle_max_chars"])
+    elif research_bucket == "interview":
+        total_budget = int(policy["qualitative_bundle_max_chars"])
+    else:
+        raise RuntimeError("method_material_routing_unsupported")
+    per_file_budget = max(1, total_budget // len(research_entries))
+    sections = []
+    for entry in research_entries:
+        text = extract_file_content(entry["item"], max_chars=per_file_budget)
+        sections.append(f"### Source: {entry['name']}\n{text.strip()}")
+    return "\n\n".join(sections).strip()[:total_budget]
+
 def _uploaded_material_descriptors() -> list[dict[str, str]]:
     out: list[dict[str, str]] = []
     for item in st.session_state.get("up_research_materials") or []:
@@ -598,55 +778,70 @@ def _current_method_preflight(*, confirmed_method_family: str | None = None) ->
 
 def _classify_and_save_research_materials(
     method_family: str, inputs_dir: Path
-) -> tuple[list[str], list[str], list[str], str]:
-    """Map one visible uploader into legacy request buckets without changing engines."""
-    uploads = list(st.session_state.get("up_research_materials") or [])
-    if not uploads:
-        return [], [], [], "none"
+) -> tuple[list[str], list[str], list[str], str, list[dict]]:
+    """Persist all approved files while preserving the user's per-file role authority.
 
-    structured = []
-    text_materials = []
-    documents = []
-    for item in uploads:
-        name = os.path.basename(str(getattr(item, "name", "") or ""))
-        ext = Path(name).suffix.lower()
-        if ext in {".csv", ".xls", ".xlsx"}:
-            structured.append(item)
-        elif ext == ".txt":
-            text_materials.append(item)
-        else:
-            documents.append(item)
+    Research materials are combined into one deterministic text bundle so the
+    frozen GenerationRunner can continue using its proven single empirical-input
+    seam without silently dropping files 2..N.
+    """
+    entries = _uploaded_material_entries()
+    if not entries:
+        return [], [], [], "none", []
 
-    decision = str(st.session_state.get("sp_material_document_use") or "reference").strip().lower()
-    if decision not in {"reference", "research"}:
-        decision = "reference"
+    report = _material_preflight_report(method_family, entries)
+    if not report.get("ok"):
+        raise RuntimeError("research_material_preflight_failed")
 
-    buckets = {"reference": [], "data": [], "interview": []}
-    capability = method_capability(method_family)
-    material_routing = dict(capability.get("material_routing") or {})
+    reference_items = [entry["item"] for entry in entries if entry.get("role") == "reference"]
+    research_entries = [entry for entry in entries if entry.get("role") == "research"]
 
-    def _route(items: list, route_key: str) -> None:
-        bucket = str(material_routing.get(route_key) or "").strip().casefold()
-        if bucket not in buckets:
-            raise RuntimeError("method_material_routing_unsupported")
-        buckets[bucket].extend(items)
-
-    _route(structured, "structured")
-    _route(text_materials, "text")
-    _route(
-        documents,
-        "document_research" if decision == "research" else "document_reference",
+    refs = _save_uploaded(reference_items, inputs_dir, "ref")
+    original_research = _save_uploaded(
+        [entry["item"] for entry in research_entries], inputs_dir, "research"
     )
 
-    reference_items = buckets["reference"]
-    data_items = buckets["data"]
-    interview_items = buckets["interview"]
-
-    refs = _save_uploaded(reference_items, inputs_dir, "ref")
-    data = _save_uploaded(data_items[:1], inputs_dir, "data")
-    interviews = _save_uploaded(interview_items[:1], inputs_dir, "interview")
-    return refs, data, interviews, decision
+    data: list[str] = []
+    interviews: list[str] = []
+    capability = method_capability(method_family)
+    material_routing = dict(capability.get("material_routing") or {})
+    reference_bucket = str(material_routing.get("user_reference") or "").strip().casefold()
+    research_bucket = str(material_routing.get("user_research") or "").strip().casefold()
+    if reference_bucket != "reference" or research_bucket not in {"data", "interview"}:
+        raise RuntimeError("method_material_routing_unsupported")
+    if research_entries:
+        bundle_text = _material_bundle_text(research_entries, research_bucket)
+        if not bundle_text:
+            raise RuntimeError("research_material_bundle_empty")
+        bundle_name = "research_materials_bundle.txt"
+        bundle_path = inputs_dir / bundle_name
+        bundle_path.write_text(bundle_text, encoding="utf-8")
+        bundle_rel = str(Path("inputs") / bundle_name)
+        if research_bucket == "data":
+            data = [bundle_rel]
+        else:
+            interviews = [bundle_rel]
+
+    role_manifest = []
+    research_index = 0
+    reference_index = 0
+    for entry in entries:
+        role = str(entry.get("role") or "")
+        if role == "reference":
+            saved_path = refs[reference_index]
+            reference_index += 1
+        else:
+            saved_path = original_research[research_index]
+            research_index += 1
+        role_manifest.append(
+            {
+                "name": entry["name"],
+                "role": role,
+                "saved_path": saved_path,
+            }
+        )
 
+    return refs, data, interviews, "per_file_v1", role_manifest
 
 def _persist_task_request(
     task: dict,
@@ -669,7 +864,7 @@ def _persist_task_request(
     if not expected_lane or expected_lane != str(mode or "").strip().lower():
         raise RuntimeError("method_family_policy_lane_mismatch")
 
-    refs, data, interviews, material_use_decision = _classify_and_save_research_materials(
+    refs, data, interviews, material_use_decision, material_file_roles = _classify_and_save_research_materials(
         method_family, inputs_dir
     )
 
@@ -681,6 +876,7 @@ def _persist_task_request(
         "method_resolver_version": str(method_resolver_version or METHOD_RESOLVER_VERSION),
         "derived_policy_lane": expected_lane,
         "material_use_decision": material_use_decision,
+        "material_file_roles": material_file_roles,
         "interface_language": lang_code,
         "target_lang": get_target_language(lang_code),
         "output_suffix": get_output_suffix(lang_code),
@@ -889,6 +1085,14 @@ def _process_pending_launch(lang_code: str) -> None:
         st.warning(tr(lang_code, "method_resolution_invalid"))
         return
 
+    material_gate = _material_preflight_report(
+        pending_family, _uploaded_material_entries()
+    )
+    if not bool(material_gate.get("ok")):
+        st.session_state["sp_pending_launch"] = None
+        st.warning(tr(lang_code, "material_preflight_blocked"))
+        return
+
     projection = get_product_ui_projection()
     selected_product_tier = (
         str(st.session_state.get("sp_product_tier") or "").strip().upper()
@@ -1586,18 +1790,39 @@ def render_workspace_controls(user):
     )
 
     uploaded_materials = list(st.session_state.get("up_research_materials") or [])
-    has_document_material = any(
-        Path(str(getattr(item, "name", "") or "")).suffix.lower() in {".pdf", ".doc", ".docx"}
-        for item in uploaded_materials
-    )
-    if has_document_material and not active:
-        st.radio(
-            tr(lang_code, "material_document_use_label"),
-            ["reference", "research"],
-            format_func=lambda value: tr(lang_code, f"material_use_{value}"),
-            horizontal=True,
-            key="sp_material_document_use",
-        )
+    material_entries = _uploaded_material_entries()
+    if uploaded_materials and not active:
+        for entry in material_entries:
+            st.radio(
+                entry["name"],
+                ["reference", "research"],
+                format_func=lambda value: tr(lang_code, f"material_use_{value}"),
+                horizontal=True,
+                key=entry["state_key"],
+            )
+        # Re-read after widgets because Streamlit may have changed role state.
+        material_entries = _uploaded_material_entries()
+        limits_ok, material_counts, material_policy = _material_limits_ok(material_entries)
+        if material_counts["reference"] > material_policy["reference_file_limit"]:
+            st.error(
+                tr(
+                    lang_code,
+                    "material_reference_limit_exceeded",
+                    limit=material_policy["reference_file_limit"],
+                )
+            )
+        if material_counts["research"] > material_policy["research_material_file_limit"]:
+            st.error(
+                tr(
+                    lang_code,
+                    "material_research_limit_exceeded",
+                    limit=material_policy["research_material_file_limit"],
+                )
+            )
+    else:
+        limits_ok = True
+        material_counts = {"reference": 0, "research": 0}
+        material_policy = _research_material_upload_policy()
 
     if active:
         active_family = _task_method_family(task)
@@ -1636,6 +1861,30 @@ def render_workspace_controls(user):
     )
     if active:
         st.markdown(f"**{tr(lang_code, 'detected_method_label')}:** {method_label}")
+
+    material_gate = {"ok": True, "failures": []}
+    if not active and limits_ok:
+        material_gate = _material_preflight_report(method_family, material_entries)
+        for failure in material_gate.get("failures") or []:
+            role = str(failure.get("role") or "reference")
+            formats_key = (
+                "material_reference_convert_formats"
+                if role == "reference"
+                else "material_research_convert_formats"
+            )
+            st.error(
+                tr(
+                    lang_code,
+                    "material_parse_failed",
+                    filename=str(failure.get("name") or ""),
+                    formats=tr(lang_code, formats_key),
+                )
+            )
+        if material_gate.get("content_insufficient"):
+            st.error(tr(lang_code, "material_research_content_insufficient"))
+    elif not active:
+        material_gate = {"ok": False, "failures": []}
+
     st.caption(tr(lang_code, "method_detection_explanation"))
 
     block_reason = str(preflight.get("block_reason") or "")
@@ -1740,6 +1989,7 @@ def render_workspace_controls(user):
                 st.rerun()
         can_start = bool(
             preflight.get("can_start_generation")
+            and bool(material_gate.get("ok"))
             and product_ui_ready
             and selected_remaining is not None
             and selected_remaining > 0
@@ -1757,12 +2007,17 @@ def render_workspace_controls(user):
             click_preflight, click_signature = _current_method_preflight(
                 confirmed_method_family=launch_family
             )
+            click_material_gate = _material_preflight_report(
+                launch_family, _uploaded_material_entries()
+            )
             if (
                 not launch_family
                 or not bool(click_preflight.get("can_start_generation"))
                 or str(click_preflight.get("method_family") or "").strip().casefold() != launch_family
             ):
                 st.warning(tr(lang_code, "method_resolution_invalid"))
+            elif not bool(click_material_gate.get("ok")):
+                st.warning(tr(lang_code, "material_preflight_blocked"))
             else:
                 resolved_mode = str(click_preflight.get("legacy_policy_lane") or "").strip().lower()
                 capability = dict(click_preflight.get("capability") or {})

config.py

@@ -68,6 +68,37 @@ def load_config():
         "rag_chunk_size": int(os.environ.get("RAG_CHUNK_SIZE", "3000")),
         "rag_top_k": int(os.environ.get("RAG_TOP_K", "3")),
         "max_extract_chars": int(os.environ.get("MAX_EXTRACT_CHARS", "150000")),
+        # P0-R7-B: upload-role limits are policy-driven and centrally owned.
+        # app.py reads this policy; it must not embed business-count literals.
+        "research_material_upload_policy": {
+            "version": os.environ.get(
+                "SMARTPAPER_RESEARCH_MATERIAL_UPLOAD_POLICY_VERSION",
+                "research-material-upload-policy-v1",
+            ).strip() or "research-material-upload-policy-v1",
+            "reference_file_limit": max(
+                1,
+                min(100, int(os.environ.get("SMARTPAPER_REFERENCE_FILE_LIMIT", "10"))),
+            ),
+            "research_material_file_limit": max(
+                1,
+                min(100, int(os.environ.get("SMARTPAPER_RESEARCH_MATERIAL_FILE_LIMIT", "10"))),
+            ),
+            # Preserve the existing qualitative engine's >100-character
+            # user-evidence threshold without duplicating it in UI code.
+            "qualitative_min_combined_chars": max(
+                1,
+                int(os.environ.get("SMARTPAPER_QUALITATIVE_MIN_USER_MATERIAL_CHARS", "101")),
+            ),
+            "quantitative_bundle_max_chars": max(
+                1000,
+                int(os.environ.get("SMARTPAPER_QUANTITATIVE_RESEARCH_BUNDLE_MAX_CHARS", "20000")),
+            ),
+            "qualitative_bundle_max_chars": max(
+                1000,
+                int(os.environ.get("SMARTPAPER_QUALITATIVE_RESEARCH_BUNDLE_MAX_CHARS", os.environ.get("MAX_EXTRACT_CHARS", "150000"))),
+            ),
+            "accepted_extensions": ["pdf", "txt", "doc", "docx", "csv", "xls", "xlsx"],
+        },
         # Model requests may legitimately run longer than one Worker lease.
         # The independent Executor renews authority continuously; model-token
         # activity is observability only and never the liveness authority.

core/advanced_method_registry.py

@@ -39,6 +39,9 @@ _METHOD_CAPABILITIES: dict[str, dict[str, Any]] = {
             "text": "interview",
             "document_reference": "reference",
             "document_research": "interview",
+            # P0-R7-B: explicit per-file user roles remain registry-owned.
+            "user_reference": "reference",
+            "user_research": "interview",
         },
     },
     "survey_path": {
@@ -53,6 +56,9 @@ _METHOD_CAPABILITIES: dict[str, dict[str, Any]] = {
             "text": "reference",
             "document_reference": "reference",
             "document_research": "reference",
+            # P0-R7-B: user research material is empirical evidence for survey work.
+            "user_reference": "reference",
+            "user_research": "data",
         },
     },
 }
@@ -337,7 +343,10 @@ def resolve_generation_preflight(
     template_profile = str(capability.get("template_profile") or "").strip()
     material_profile = str(capability.get("material_profile") or "").strip()
     material_routing = dict(capability.get("material_routing") or {})
-    routing_keys = {"structured", "text", "document_reference", "document_research"}
+    routing_keys = {
+        "structured", "text", "document_reference", "document_research",
+        "user_reference", "user_research",
+    }
     routing_values = {str(value or "").strip() for value in material_routing.values()}
     if (
         family not in METHOD_FAMILIES

i18n.py

@@ -284,6 +284,13 @@ TEXTS: dict[str, dict[str, str]] = {
         "material_document_use_label": "如何使用上传的 PDF/DOC/DOCX 文件？",
         "material_use_reference": "作为参考文献资料",
         "material_use_research": "作为研究资料",
+        "material_reference_limit_exceeded": "参考文献最多可选择 {limit} 个，请调整文件用途或删除多余文件。",
+        "material_research_limit_exceeded": "研究资料最多可选择 {limit} 个，请调整文件用途或删除多余文件。",
+        "material_parse_failed": "文件“{filename}”无法被系统正确解析。请先转换为可识别格式后重新上传（建议：{formats}）。",
+        "material_reference_convert_formats": "PDF、DOCX 或 TXT",
+        "material_research_convert_formats": "PDF、DOCX、TXT、CSV 或 XLSX",
+        "material_research_content_insufficient": "已选择的研究资料可解析内容不足，系统无法可靠识别。请补充内容或转换为可正确识别的格式后重新上传。",
+        "material_preflight_blocked": "研究资料预检未通过，请先处理上方文件问题；系统尚未创建任务或预留额度。",
         "detected_method_label": "研究设计（可调整）",
         "method_detection_explanation": "默认为质性研究；生成前可根据需要调整为问卷量化研究。",
         "method_confirm_label": "调整研究设计",
@@ -532,6 +539,13 @@ TEXTS: dict[str, dict[str, str]] = {
         "material_document_use_label": "How should SmartPaper use the uploaded PDF/DOC/DOCX files?",
         "material_use_reference": "Use as reference literature",
         "material_use_research": "Use as research materials",
+        "material_reference_limit_exceeded": "You can select at most {limit} reference files. Change file roles or remove extra files.",
+        "material_research_limit_exceeded": "You can select at most {limit} research-material files. Change file roles or remove extra files.",
+        "material_parse_failed": "The system cannot reliably parse “{filename}”. Convert it to a supported readable format and upload it again (recommended: {formats}).",
+        "material_reference_convert_formats": "PDF, DOCX, or TXT",
+        "material_research_convert_formats": "PDF, DOCX, TXT, CSV, or XLSX",
+        "material_research_content_insufficient": "The selected research materials do not contain enough parseable content for reliable use. Add content or convert the files to a readable format and upload them again.",
+        "material_preflight_blocked": "Research-material preflight did not pass. Resolve the file issues above first; no task or quota reservation has been created.",
         "detected_method_label": "Research Design (Adjustable)",
         "method_detection_explanation": "Qualitative Research is selected by default. Before generation, you can adjust the Research Design to Survey-Based Quantitative Research.",
         "method_confirm_label": "Adjust Research Design",
@@ -780,6 +794,13 @@ TEXTS: dict[str, dict[str, str]] = {
         "material_document_use_label": "Как использовать загруженные файлы PDF/DOC/DOCX?",
         "material_use_reference": "Как научные источники",
         "material_use_research": "Как исследовательские материалы",
+        "material_reference_limit_exceeded": "Можно выбрать не более {limit} файлов в качестве научных источников. Измените назначение файлов или удалите лишние.",
+        "material_research_limit_exceeded": "Можно выбрать не более {limit} файлов в качестве исследовательских материалов. Измените назначение файлов или удалите лишние.",
+        "material_parse_failed": "Система не может корректно распознать файл «{filename}». Преобразуйте его в поддерживаемый читаемый формат и загрузите снова (рекомендуется: {formats}).",
+        "material_reference_convert_formats": "PDF, DOCX или TXT",
+        "material_research_convert_formats": "PDF, DOCX, TXT, CSV или XLSX",
+        "material_research_content_insufficient": "В выбранных исследовательских материалах недостаточно корректно распознаваемого содержимого. Дополните материалы или преобразуйте файлы в читаемый формат и загрузите их снова.",
+        "material_preflight_blocked": "Предварительная проверка исследовательских материалов не пройдена. Сначала устраните указанные выше проблемы; задача и резерв квоты ещё не созданы.",
         "detected_method_label": "Дизайн исследования (можно изменить)",
         "method_detection_explanation": "По умолчанию выбрано качественное исследование. До начала генерации дизайн можно изменить на количественное исследование на основе опроса.",
         "method_confirm_label": "Изменить дизайн исследования",
@@ -1028,6 +1049,13 @@ TEXTS: dict[str, dict[str, str]] = {
         "material_document_use_label": "Жүктелген PDF/DOC/DOCX файлдарын қалай пайдалану керек?",
         "material_use_reference": "Ғылыми дереккөз ретінде",
         "material_use_research": "Зерттеу материалы ретінде",
+        "material_reference_limit_exceeded": "Ғылыми дереккөз ретінде ең көбі {limit} файл таңдауға болады. Файлдардың мақсатын өзгертіңіз немесе артық файлдарды жойыңыз.",
+        "material_research_limit_exceeded": "Зерттеу материалы ретінде ең көбі {limit} файл таңдауға болады. Файлдардың мақсатын өзгертіңіз немесе артық файлдарды жойыңыз.",
+        "material_parse_failed": "Жүйе «{filename}» файлын сенімді түрде тани алмады. Оны оқылатын қолдау көрсетілетін пішімге түрлендіріп, қайта жүктеңіз (ұсынылады: {formats}).",
+        "material_reference_convert_formats": "PDF, DOCX немесе TXT",
+        "material_research_convert_formats": "PDF, DOCX, TXT, CSV немесе XLSX",
+        "material_research_content_insufficient": "Таңдалған зерттеу материалдарында сенімді пайдалануға жеткілікті танылатын мазмұн жоқ. Мазмұнды толықтырыңыз немесе файлдарды оқылатын пішімге түрлендіріп, қайта жүктеңіз.",
+        "material_preflight_blocked": "Зерттеу материалдарының алдын ала тексеруі өтпеді. Алдымен жоғарыдағы файл мәселелерін түзетіңіз; тапсырма да, квота резерві де әлі жасалған жоқ.",
         "detected_method_label": "Зерттеу дизайны (өзгертуге болады)",
         "method_detection_explanation": "Әдепкі бойынша сапалық зерттеу таңдалады. Генерацияны бастамас бұрын зерттеу дизайнын сауалнамаға негізделген сандық зерттеуге өзгертуге болады.",
         "method_confirm_label": "Зерттеу дизайнын өзгерту",

utils/file_parser.py

@@ -1,5 +1,9 @@
 # -*- coding: utf-8 -*-
-"""Headless file parsing helpers used by both Streamlit and R3-Free executor."""
+"""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
@@ -24,6 +28,10 @@ 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)
@@ -58,74 +66,157 @@ def _binary_stream(file) -> io.BytesIO | BinaryIO:
     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_text = ""
+    extracted = []
+    total = 0
     for file in uploaded_files or []:
         filename = _name(file)
-        lower_name = filename.lower()
         try:
-            if lower_name.endswith(".pdf") and PYMUPDF_AVAILABLE:
-                pdf_bytes = _read_bytes(file)
-                with pymupdf.open(stream=pdf_bytes, filetype="pdf") as pdf_doc:
-                    for page in pdf_doc:
-                        text = page.get_text()
-                        if text:
-                            extracted_text += text + "\n"
-                        if len(extracted_text) > max_chars:
-                            break
-            elif lower_name.endswith((".docx", ".doc")):
-                stream = _binary_stream(file)
-                doc = Document(stream)
-                for para in doc.paragraphs:
-                    extracted_text += para.text + "\n"
-                    if len(extracted_text) > max_chars:
-                        break
-            else:
-                extracted_text += _read_bytes(file).decode("utf-8", errors="ignore") + "\n"
-
-            if len(extracted_text) > max_chars:
-                logging.warning("File %s was truncated safely to %s characters.", filename, max_chars)
+            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)
-        finally:
-            if not isinstance(file, (str, os.PathLike, Path)):
-                try:
-                    file.seek(0)
-                except Exception:
-                    pass
-    return extracted_text[:max_chars]
+    return "\n".join(extracted)[: int(max_chars)]
 
 
 def extract_data_from_files(uploaded_file, max_chars=20000):
     if not uploaded_file:
         return ""
-
-    filename = _name(uploaded_file)
-    lower_name = filename.lower()
     try:
-        if PANDAS_AVAILABLE:
-            source = os.fspath(uploaded_file) if isinstance(uploaded_file, (str, os.PathLike, Path)) else uploaded_file
-            if not isinstance(uploaded_file, (str, os.PathLike, Path)):
-                try:
-                    uploaded_file.seek(0)
-                except Exception:
-                    pass
-            df = pd.read_csv(source) if lower_name.endswith(".csv") else pd.read_excel(source)
-            if len(df) > 100:
-                logging.warning("Data file %s truncated safely to first 100 rows.", filename)
-                df = df.head(100)
-            return df.to_markdown(index=False)
-        return _read_bytes(uploaded_file).decode("utf-8", errors="ignore")[:max_chars]
+        # 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", filename, exc)
+        logging.warning("Data parse failed for %s: %s", _name(uploaded_file), exc)
         return ""
-    finally:
-        if not isinstance(uploaded_file, (str, os.PathLike, Path)):
-            try:
-                uploaded_file.seek(0)
-            except Exception:
-                pass