Spaces:
Running
Running
| # -*- coding: utf-8 -*- | |
| """Qualitative evidence preparation for SmartPaper's ASQ-oriented track.""" | |
| from __future__ import annotations | |
| import hashlib | |
| import json | |
| import logging | |
| import re | |
| from config import APP_CONFIG | |
| class QualitativeEngine: | |
| QUALITATIVE_APPENDIX_MANIFEST_VERSION = "qualitative-appendix-manifest-v1" | |
| TARGET_APPENDIX_BINDING_VERSION = "target-qualitative-appendix-binding-v2" | |
| SEMANTIC_CLAIM_FRAME_VERSION = "smartpaper-core-semantic-claim-frame-v1" | |
| APPENDIX_FIELDS = ("theme", "concept", "stage", "case_label", "evidence_role") | |
| APPENDIX_CANONICAL_LABEL_FIELDS = ("theme", "concept") | |
| APPENDIX_HEADERS = ( | |
| "Higher-order Theme", "Empirical Concept", "Process Stage", | |
| "Case/Actor", "Evidence Role", "Representative Quote", | |
| ) | |
| _TARGET_APPENDIX_LOCALIZATION = { | |
| "en": { | |
| "heading": "Appendix: Data Structure and Evidence", | |
| "headers": APPENDIX_HEADERS, | |
| }, | |
| "zh": { | |
| "heading": "附录：数据结构与证据", | |
| "headers": ("高阶主题", "经验概念", "过程阶段", "案例/参与者", "证据角色", "代表性引文"), | |
| }, | |
| "ru": { | |
| "heading": "Приложение: Структура данных и доказательства", | |
| "headers": ("Тема высшего порядка", "Эмпирическое понятие", "Этап процесса", "Случай/участник", "Роль доказательства", "Репрезентативная цитата"), | |
| }, | |
| "kk": { | |
| "heading": "Қосымша: Деректер құрылымы және дәлелдер", | |
| "headers": ("Жоғары деңгейлі тақырып", "Эмпирикалық ұғым", "Процесс кезеңі", "Кейс/қатысушы", "Дәлел рөлі", "Репрезентативті дәйексөз"), | |
| }, | |
| } | |
| ALLOWED_ANALYTIC_DESIGNS = { | |
| "Process analysis", | |
| "Comparative case analysis", | |
| "Grounded theory", | |
| "Gioia-style inductive coding", | |
| "Longitudinal qualitative analysis", | |
| "Iterative thematic/process analysis", | |
| } | |
| def __init__(self, agent): | |
| self.agent = agent | |
| def _clean_cell(value, default="Not specified"): | |
| cleaned = str(value or default).replace("\n", " ").strip() | |
| return cleaned or default | |
| def _normalize_analytic_design(cls, value): | |
| cleaned = re.sub(r"\s+", " ", str(value or "")).strip() | |
| if not cleaned: | |
| return "Iterative thematic/process analysis" | |
| lookup = {item.lower(): item for item in cls.ALLOWED_ANALYTIC_DESIGNS} | |
| return lookup.get(cleaned.lower(), "Iterative thematic/process analysis") | |
| def _normalize_string_list(cls, values): | |
| if values is None: | |
| return [] | |
| if not isinstance(values, list): | |
| values = [values] | |
| out = [] | |
| for value in values: | |
| cleaned = cls._clean_cell(value, "") | |
| if cleaned and cleaned.casefold() != "not specified" and cleaned not in out: | |
| out.append(cleaned) | |
| return out | |
| def _normalize_data_source_roles(cls, values): | |
| rows = [] | |
| for item in values or []: | |
| if not isinstance(item, dict): | |
| continue | |
| actor_role = cls._clean_cell(item.get("actor_role"), "") | |
| data_source = cls._clean_cell(item.get("data_source"), "") | |
| count = item.get("count") | |
| if not actor_role or not data_source: | |
| continue | |
| if count is not None and (not isinstance(count, int) or count <= 0): | |
| continue | |
| row = {"actor_role": actor_role, "data_source": data_source, "count": count} | |
| if row not in rows: | |
| rows.append(row) | |
| return rows | |
| def build_qualitative_semantic_claim_frame(cls, metadata): | |
| """Freeze actor/source, process-evidence, and quote identities in one frame. | |
| The frame is server-side authority. It contains no model-inferred prose | |
| meaning and no verbatim quotation payload, only typed identities/hashes. | |
| """ | |
| meta = dict(metadata or {}) | |
| facts = dict(meta.get("study_facts") or {}) | |
| relations = [] | |
| for row in cls._normalize_data_source_roles(facts.get("data_source_roles") or []): | |
| actor = cls._clean_cell(row.get("actor_role"), "") | |
| source = cls._clean_cell(row.get("data_source"), "") | |
| folded = source.casefold() | |
| if "interview" in folded: | |
| family = "interview" | |
| elif "focus" in folded and "group" in folded: | |
| family = "focus_group" | |
| elif "observ" in folded: | |
| family = "observation" | |
| elif "log" in folded: | |
| family = "log" | |
| elif "document" in folded or "archiv" in folded or "record" in folded: | |
| family = "document" | |
| else: | |
| family = "other" | |
| relation = { | |
| "actor_role": actor, | |
| "data_source": source, | |
| "source_family": family, | |
| "count": row.get("count"), | |
| "relation": "EVIDENCED_BY", | |
| } | |
| relation_id_payload = json.dumps(relation, ensure_ascii=False, sort_keys=True, separators=(",", ":")) | |
| relation["relation_id"] = hashlib.sha256(relation_id_payload.encode("utf-8")).hexdigest()[:16] | |
| relations.append(relation) | |
| quotes = [] | |
| for item in meta.get("evidence") or []: | |
| if not isinstance(item, dict): | |
| continue | |
| quote_id = str(item.get("tag") or "").strip() | |
| if not re.fullmatch(r"QUOTE-\d+", quote_id): | |
| continue | |
| quote_hash = str(item.get("quote_hash") or "").strip() | |
| if not quote_hash and str(item.get("quote") or "").strip(): | |
| normalized = re.sub(r"\s+", " ", str(item.get("quote") or "")).strip() | |
| quote_hash = hashlib.sha256(normalized.encode("utf-8")).hexdigest() | |
| quotes.append({ | |
| "quote_id": quote_id, | |
| "quote_hash": quote_hash, | |
| "theme_id": str(item.get("theme_id") or "").strip(), | |
| "stage": cls._clean_cell(item.get("stage")), | |
| "case_label": cls._clean_cell(item.get("case_label")), | |
| "evidence_role": str(item.get("evidence_role") or "typical").strip().lower(), | |
| }) | |
| process_evidence = [] | |
| for theme in meta.get("themes") or []: | |
| if not isinstance(theme, dict): | |
| continue | |
| theme_id = str(theme.get("theme_id") or "").strip() | |
| if not theme_id: | |
| continue | |
| process_evidence.append({ | |
| "theme_id": theme_id, | |
| "stages": [cls._clean_cell(x, "") for x in (theme.get("stages") or []) if cls._clean_cell(x, "")], | |
| "has_deviant_evidence": bool(theme.get("has_deviant_evidence")), | |
| "quote_ids": [ | |
| str(row.get("tag") or "").strip() | |
| for row in (theme.get("evidence") or []) if isinstance(row, dict) and str(row.get("tag") or "").strip() | |
| ], | |
| }) | |
| frame = { | |
| "version": cls.SEMANTIC_CLAIM_FRAME_VERSION, | |
| "mode": "qualitative", | |
| "actor_data_source_relations": sorted(relations, key=lambda row: (row["actor_role"].casefold(), row["data_source"].casefold())), | |
| "process_evidence": sorted(process_evidence, key=lambda row: row["theme_id"]), | |
| "quote_identities": sorted(quotes, key=lambda row: row["quote_id"]), | |
| } | |
| canonical = json.dumps(frame, ensure_ascii=False, sort_keys=True, separators=(",", ":")) | |
| frame["frame_hash"] = hashlib.sha256(canonical.encode("utf-8")).hexdigest() | |
| return frame | |
| def build_scoped_evidence_md( | |
| cls, | |
| metadata, | |
| *, | |
| include_themes=True, | |
| theme_titles=None, | |
| assigned_evidence=None, | |
| include_quotes=False, | |
| include_quote_ids=True, | |
| ): | |
| """Render the minimum qualitative evidence needed by one drafting unit. | |
| Study facts/theme summaries are an internal empirical channel. Verbatim | |
| quotations are a separate channel and are exposed only to their owning | |
| Findings unit (or the final appendix when explicitly requested by the | |
| caller). This prevents a quotation visible in one unit from being | |
| opportunistically reused elsewhere and keeps external literature from | |
| becoming evidence for this study's own facts. | |
| """ | |
| metadata = metadata or {} | |
| facts = metadata.get("study_facts") or {} | |
| sources = facts.get("data_sources") or ["Not specified"] | |
| if not isinstance(sources, list): | |
| sources = [sources] | |
| clean_sources = [cls._clean_cell(item) for item in sources] | |
| participants = facts.get("participant_count") | |
| interviews = facts.get("interview_count") | |
| analytic_design = cls._normalize_analytic_design( | |
| facts.get("analytic_design") or facts.get("analysis_method") | |
| ) | |
| analytic_logic = cls._clean_cell( | |
| facts.get("analytic_logic"), | |
| "Iteratively compare empirical observations to refine a process/mechanism explanation.", | |
| ) | |
| temporal_structure = cls._clean_cell(facts.get("temporal_structure")) | |
| case_structure = cls._clean_cell(facts.get("case_structure")) | |
| setting = cls._clean_cell(facts.get("research_setting")) | |
| data_source_roles = cls._normalize_data_source_roles(facts.get("data_source_roles") or []) | |
| trustworthiness = cls._normalize_string_list(facts.get("trustworthiness_procedures")) | |
| ethics = cls._normalize_string_list(facts.get("ethics_procedures")) | |
| reflexivity = cls._normalize_string_list(facts.get("reflexivity_procedures")) | |
| lines = [ | |
| "### Frozen Qualitative Study Facts", | |
| "| Field | Frozen value |", | |
| "| :--- | :--- |", | |
| f"| Research setting | {setting} |", | |
| f"| Data sources | {', '.join(clean_sources)} |", | |
| f"| Participants | {participants if participants is not None else 'Not specified'} |", | |
| f"| Interviews | {interviews if interviews is not None else 'Not specified'} |", | |
| f"| Analytic design | {analytic_design} |", | |
| f"| Analytic logic | {analytic_logic} |", | |
| f"| Temporal structure | {temporal_structure} |", | |
| f"| Case structure | {case_structure} |", | |
| f"| Trustworthiness procedures | {'; '.join(trustworthiness) if trustworthiness else 'Not specified'} |", | |
| f"| Ethics procedures | {'; '.join(ethics) if ethics else 'Not specified'} |", | |
| f"| Reflexivity procedures | {'; '.join(reflexivity) if reflexivity else 'Not specified'} |", | |
| ] | |
| if data_source_roles: | |
| lines.extend([ | |
| "", | |
| "### Frozen Actor / Data-Source Relations", | |
| "| Actor role | Data source | Count |", | |
| "| :--- | :--- | :--- |", | |
| ]) | |
| lines.extend( | |
| f"| {row['actor_role']} | {row['data_source']} | {row['count'] if row['count'] is not None else 'Not specified'} |" | |
| for row in data_source_roles | |
| ) | |
| lines.extend([ | |
| "", | |
| "[QUALITATIVE EVIDENCE CHANNELS]", | |
| "1. STUDY FACTS and THEME SUMMARIES below are this study's internal empirical evidence. Never attach an external [CITE-XX] tag to a sentence merely because it reports or interprets these frozen facts.", | |
| "2. External literature may support only prior-theory, prior-evidence, or methodological-rationale claims. Keep those claims in distinct sentences from this study's setting, sample, data, events, quotations, and findings.", | |
| "3. ASSIGNED QUOTATIONS, when present, are analysis-only verbatim evidence owned by the current stable theme. Do not reproduce the verbatim wording or any internal quotation identifier in generated prose; the Executor renders frozen quotations deterministically after narrative QA.", | |
| "4. Study-owned trustworthiness, ethics, reflexivity, recruitment, consent, observation, or audit procedures may be stated as performed only when explicitly frozen above. Never convert general methodological best practice into a fact about this study.", | |
| ]) | |
| allowed_theme_titles = None | |
| if theme_titles: | |
| allowed_theme_titles = { | |
| re.sub(r"\s+", " ", str(item or "")).strip().casefold() | |
| for item in theme_titles | |
| if str(item or "").strip() | |
| } | |
| if include_themes: | |
| theme_rows = [] | |
| for theme in metadata.get("themes") or []: | |
| if not isinstance(theme, dict): | |
| continue | |
| title = cls._clean_cell(theme.get("title"), "Emergent pattern") | |
| if allowed_theme_titles is not None and title.casefold() not in allowed_theme_titles: | |
| continue | |
| concepts = ", ".join( | |
| cls._clean_cell(item, "") | |
| for item in (theme.get("concepts") or []) | |
| if cls._clean_cell(item, "") | |
| ) or "Not specified" | |
| stages = ", ".join( | |
| cls._clean_cell(item, "") | |
| for item in (theme.get("stages") or []) | |
| if cls._clean_cell(item, "") and cls._clean_cell(item, "") != "Not specified" | |
| ) or "Not specified" | |
| deviant = "yes" if theme.get("has_deviant_evidence") else "no" | |
| theme_rows.append((title, concepts, stages, deviant)) | |
| if theme_rows: | |
| lines.extend([ | |
| "", | |
| "### Frozen Qualitative Theme Summary", | |
| "| Higher-order Theme | Empirical Concepts | Process Stages | Deviant/Boundary Evidence Present |", | |
| "| :--- | :--- | :--- | :--- |", | |
| ]) | |
| lines.extend( | |
| f"| {title} | {concepts} | {stages} | {deviant} |" | |
| for title, concepts, stages, deviant in theme_rows | |
| ) | |
| if include_quotes: | |
| evidence_items = assigned_evidence | |
| if evidence_items is None: | |
| evidence_items = metadata.get("evidence") or [] | |
| quote_rows = [] | |
| for item in evidence_items or []: | |
| if not isinstance(item, dict): | |
| continue | |
| tag = cls._clean_cell(item.get("tag"), "") | |
| quote = cls._clean_cell(item.get("quote"), "").replace('"', "'") | |
| if not tag or not quote: | |
| continue | |
| stage = cls._clean_cell(item.get("stage")) | |
| case_label = cls._clean_cell(item.get("case_label")) | |
| evidence_role = str(item.get("evidence_role") or "typical").strip().lower() | |
| if evidence_role not in {"typical", "deviant"}: | |
| evidence_role = "typical" | |
| quote_rows.append((tag, stage, case_label, evidence_role, quote)) | |
| if quote_rows: | |
| lines.extend(["", "### Assigned Qualitative Quotations"]) | |
| if include_quote_ids: | |
| lines.extend([ | |
| "| Tag | Process Stage | Case/Actor | Evidence Role | Verbatim Quote |", | |
| "| :--- | :--- | :--- | :--- | :--- |", | |
| ]) | |
| lines.extend( | |
| f'| {tag} | {stage} | {case_label} | {role} | "{quote}" |' | |
| for tag, stage, case_label, role, quote in quote_rows | |
| ) | |
| else: | |
| lines.extend([ | |
| "| Process Stage | Case/Actor | Evidence Role | Verbatim Quote |", | |
| "| :--- | :--- | :--- | :--- |", | |
| ]) | |
| lines.extend( | |
| f'| {stage} | {case_label} | {role} | "{quote}" |' | |
| for _tag, stage, case_label, role, quote in quote_rows | |
| ) | |
| return "\n".join(lines).strip() | |
| def _appendix_target_language_code(target_lang): | |
| value = str(target_lang or "").strip().casefold().replace("_", "-") | |
| if value.startswith("zh") or "chinese" in value or "中文" in value or "简体" in value: | |
| return "zh" | |
| if value.startswith("ru") or "russian" in value or "рус" in value: | |
| return "ru" | |
| if value.startswith("kk") or value.startswith("kz") or "kazakh" in value or "қазақ" in value: | |
| return "kk" | |
| return "en" | |
| def build_appendix_manifest(cls, metadata): | |
| """Freeze Appendix row/cell identity before any target-language operation.""" | |
| rows = [] | |
| seen = set() | |
| for index, item in enumerate((metadata or {}).get("evidence") or [], start=1): | |
| if not isinstance(item, dict): | |
| continue | |
| quote = cls._clean_cell(item.get("quote"), "") | |
| if not quote: | |
| continue | |
| quote_id = str(item.get("tag") or "").strip().upper() | |
| if not re.fullmatch(r"QUOTE-\d+", quote_id) or quote_id in seen: | |
| return {} | |
| seen.add(quote_id) | |
| row = { | |
| "row_id": quote_id, | |
| "quote_id": quote_id, | |
| "quote_hash": hashlib.sha256(re.sub(r"\s+", " ", quote).strip().encode("utf-8")).hexdigest(), | |
| "theme": cls._clean_cell(item.get("theme"), "Emergent pattern"), | |
| "concept": cls._clean_cell(item.get("concept"), "Empirical observation"), | |
| "stage": cls._clean_cell(item.get("stage")), | |
| "case_label": cls._clean_cell(item.get("case_label")), | |
| "evidence_role": cls._clean_cell( | |
| str(item.get("evidence_role") or "typical").strip().lower(), "typical" | |
| ), | |
| "quote": quote, | |
| } | |
| rows.append(row) | |
| if not rows: | |
| return {} | |
| manifest = { | |
| "version": cls.QUALITATIVE_APPENDIX_MANIFEST_VERSION, | |
| "columns": list(cls.APPENDIX_HEADERS), | |
| "rows": rows, | |
| } | |
| manifest["manifest_hash"] = cls.appendix_manifest_hash(manifest) | |
| return manifest | |
| def appendix_manifest_hash(cls, manifest): | |
| record = dict(manifest or {}) | |
| record.pop("manifest_hash", None) | |
| payload = json.dumps(record, ensure_ascii=False, sort_keys=True, separators=(",", ":")) | |
| return hashlib.sha256(payload.encode("utf-8")).hexdigest() | |
| def appendix_translation_cells(cls, manifest): | |
| """Return typed non-quote Appendix cells; IDs stay server-side.""" | |
| cells = {} | |
| for row in (manifest or {}).get("rows") or []: | |
| if not isinstance(row, dict): | |
| continue | |
| row_id = str(row.get("row_id") or "").strip() | |
| if not row_id: | |
| continue | |
| for field in cls.APPENDIX_FIELDS: | |
| value = cls._clean_cell(row.get(field), "Not specified") | |
| cells[f"{row_id}:{field}"] = value | |
| return cells | |
| def appendix_translation_plan(cls, manifest, glossary=None): | |
| """Split Appendix cells into server-bound labels and model-owned surfaces. | |
| Theme/concept cells are frozen research labels. When the task-level | |
| canonical glossary already owns their target identity, the translation | |
| model must not rewrite those labels or their internal anchors. Remaining | |
| ordinary cells (stage/case/evidence-role) continue through the typed | |
| translation compiler and are rebound by server-side cell ID. | |
| """ | |
| expected = cls.appendix_translation_cells(manifest) | |
| glossary_value = dict(glossary or {}) | |
| folded_glossary = { | |
| str(key or '').strip().casefold(): str(value or '').strip() | |
| for key, value in glossary_value.items() if str(key or '').strip() and str(value or '').strip() | |
| } | |
| prebound = {} | |
| model_cells = {} | |
| for cell_id, source in expected.items(): | |
| field = str(cell_id or '').rsplit(':', 1)[-1] | |
| canonical_target = str(glossary_value.get(source) or '').strip() | |
| if not canonical_target: | |
| canonical_target = folded_glossary.get(str(source or '').strip().casefold(), '') | |
| if field in cls.APPENDIX_CANONICAL_LABEL_FIELDS and canonical_target: | |
| prebound[cell_id] = canonical_target | |
| else: | |
| model_cells[cell_id] = source | |
| if set(prebound) & set(model_cells) or set(prebound) | set(model_cells) != set(expected): | |
| raise RuntimeError("qualitative_appendix_translation_plan_identity_mismatch") | |
| return prebound, model_cells | |
| def _escape_appendix_markdown_cell(value): | |
| return re.sub(r"\s+", " ", str(value or "").strip()).replace("|", r"\|") | |
| def _validate_appendix_manifest(cls, manifest): | |
| record = dict(manifest or {}) | |
| if record.get("version") != cls.QUALITATIVE_APPENDIX_MANIFEST_VERSION: | |
| return False | |
| if list(record.get("columns") or []) != list(cls.APPENDIX_HEADERS): | |
| return False | |
| rows = [row for row in (record.get("rows") or []) if isinstance(row, dict)] | |
| ids = [str(row.get("row_id") or "") for row in rows] | |
| quote_ids = [str(row.get("quote_id") or "") for row in rows] | |
| if not rows or len(ids) != len(set(ids)) or ids != quote_ids: | |
| return False | |
| if any(not re.fullmatch(r"QUOTE-\d+", item) for item in ids): | |
| return False | |
| expected_hash = str(record.get("manifest_hash") or "") | |
| return bool(expected_hash) and expected_hash == cls.appendix_manifest_hash(record) | |
| def bind_target_language_appendix(cls, manifest, target_lang, cell_translations, quote_translations): | |
| """Render one six-column target Appendix from frozen semantic objects. | |
| Markdown structure, row identity, column identity, and quote identity are | |
| program-owned. The translation model may supply only ordinary cell | |
| surfaces through ``cell_translations``. | |
| """ | |
| if not cls._validate_appendix_manifest(manifest): | |
| raise RuntimeError("target_qualitative_appendix_manifest_invalid") | |
| expected_cells = cls.appendix_translation_cells(manifest) | |
| translations = dict(cell_translations or {}) | |
| if set(translations) != set(expected_cells): | |
| missing = sorted(set(expected_cells) - set(translations)) | |
| extra = sorted(set(translations) - set(expected_cells)) | |
| raise RuntimeError( | |
| f"target_qualitative_appendix_cell_identity_mismatch:missing={missing[:4]} extra={extra[:4]}" | |
| ) | |
| quote_map = dict(quote_translations or {}) | |
| required_quotes = {str(row.get("quote_id") or "") for row in manifest.get("rows") or []} | |
| missing_quotes = sorted(tag for tag in required_quotes if not str(quote_map.get(tag) or "").strip()) | |
| if missing_quotes: | |
| raise RuntimeError(f"target_qualitative_appendix_quote_identity_missing:{missing_quotes[:6]}") | |
| code = cls._appendix_target_language_code(target_lang) | |
| profile = dict(cls._TARGET_APPENDIX_LOCALIZATION.get(code) or cls._TARGET_APPENDIX_LOCALIZATION["en"]) | |
| headers = list(profile.get("headers") or cls.APPENDIX_HEADERS) | |
| heading = str(profile.get("heading") or cls._TARGET_APPENDIX_LOCALIZATION["en"]["heading"]) | |
| lines = [ | |
| "| " + " | ".join(cls._escape_appendix_markdown_cell(item) for item in headers) + " |", | |
| "| :--- | :--- | :--- | :--- | :--- | :--- |", | |
| ] | |
| for row in manifest.get("rows") or []: | |
| row_id = str(row.get("row_id") or "") | |
| values = [translations[f"{row_id}:{field}"] for field in cls.APPENDIX_FIELDS] | |
| values.append(quote_map[str(row.get("quote_id") or "")]) | |
| lines.append( | |
| "| " + " | ".join(cls._escape_appendix_markdown_cell(value) for value in values) + " |" | |
| ) | |
| rendered = f"# {heading}\n\n" + "\n".join(lines) | |
| cls.validate_bound_target_appendix(rendered, manifest, target_lang) | |
| return rendered.strip() | |
| def validate_bound_target_appendix(cls, text, manifest, target_lang): | |
| """Fail before DOCX packaging if semantic Appendix identity drifted.""" | |
| if not cls._validate_appendix_manifest(manifest): | |
| raise RuntimeError("target_qualitative_appendix_manifest_invalid") | |
| code = cls._appendix_target_language_code(target_lang) | |
| profile = dict(cls._TARGET_APPENDIX_LOCALIZATION.get(code) or cls._TARGET_APPENDIX_LOCALIZATION["en"]) | |
| heading = str(profile.get("heading") or "") | |
| headers = list(profile.get("headers") or cls.APPENDIX_HEADERS) | |
| value = str(text or "") | |
| heading_pattern = re.compile(rf"(?m)^#\s+{re.escape(heading)}\s*$") | |
| matches = list(heading_pattern.finditer(value)) | |
| if len(matches) != 1: | |
| raise RuntimeError("target_qualitative_appendix_heading_identity_missing_or_ambiguous") | |
| tail = value[matches[0].end():] | |
| table_blocks = re.findall(r"(?m)(?:^[ \t]*\|[^\n]+\|[ \t]*$\n?)+", tail) | |
| valid_blocks = [] | |
| for block in table_blocks: | |
| lines = [line.strip() for line in block.splitlines() if line.strip().startswith("|")] | |
| if not lines: | |
| continue | |
| if all(len(re.findall(r"(?<!\\)\|", line)) == 7 for line in lines): | |
| valid_blocks.append(lines) | |
| if len(valid_blocks) != 1: | |
| raise RuntimeError("target_qualitative_appendix_semantic_table_identity_missing_or_ambiguous") | |
| lines = valid_blocks[0] | |
| expected_rows = len(manifest.get("rows") or []) + 2 | |
| if len(lines) != expected_rows: | |
| raise RuntimeError( | |
| f"target_qualitative_appendix_row_identity_mismatch:expected={expected_rows} got={len(lines)}" | |
| ) | |
| expected_header = "| " + " | ".join(cls._escape_appendix_markdown_cell(item) for item in headers) + " |" | |
| if lines[0] != expected_header or not re.fullmatch( | |
| r"\|\s*:---\s*\|\s*:---\s*\|\s*:---\s*\|\s*:---\s*\|\s*:---\s*\|\s*:---\s*\|", | |
| lines[1], | |
| ): | |
| raise RuntimeError("target_qualitative_appendix_column_identity_mismatch") | |
| if re.search(r"\bQUOTE-\d+\b", "\n".join(lines), flags=re.I): | |
| raise RuntimeError("target_qualitative_appendix_internal_quote_id_leak") | |
| return True | |
| def build_appendix_evidence_md(cls, metadata): | |
| """Render the reader-facing English Appendix from the same frozen manifest.""" | |
| manifest = cls.build_appendix_manifest(metadata) | |
| if not cls._validate_appendix_manifest(manifest): | |
| raise RuntimeError("qualitative_appendix_manifest_invalid") | |
| lines = [ | |
| "| " + " | ".join(cls.APPENDIX_HEADERS) + " |", | |
| "| :--- | :--- | :--- | :--- | :--- | :--- |", | |
| ] | |
| for row in manifest.get("rows") or []: | |
| values = [row.get(field) for field in cls.APPENDIX_FIELDS] + [f'"{row.get("quote")}"'] | |
| lines.append( | |
| "| " + " | ".join(cls._escape_appendix_markdown_cell(value) for value in values) + " |" | |
| ) | |
| return "\n".join(lines).strip() | |
| def _build_request(self, topic, keywords, raw_interviews): | |
| schema = """Return ONLY one valid JSON object with this schema: | |
| { | |
| "study_facts": { | |
| "research_setting": "string or Not specified", | |
| "data_sources": ["string"], | |
| "data_source_roles": [{"actor_role":"string", "data_source":"string", "count":integer or null}], | |
| "participant_count": integer or null, | |
| "interview_count": integer or null, | |
| "trustworthiness_procedures": ["explicitly supported/frozen procedure"], | |
| "ethics_procedures": ["explicitly supported/frozen procedure"], | |
| "reflexivity_procedures": ["explicitly supported/frozen procedure"], | |
| "analytic_design": "one allowed design", | |
| "analytic_logic": "one concise sentence", | |
| "temporal_structure": "string or Not specified", | |
| "case_structure": "string or Not specified" | |
| }, | |
| "evidence": [ | |
| { | |
| "theme": "higher-order pattern/theme", | |
| "concept": "empirical concept", | |
| "quote": "verbatim quotation", | |
| "stage": "temporal/process stage or Not specified", | |
| "case_label": "generic case/actor label or Not specified", | |
| "evidence_role": "typical or deviant" | |
| } | |
| ] | |
| } | |
| Allowed analytic designs: Process analysis; Comparative case analysis; Grounded theory; Gioia-style inductive coding; Longitudinal qualitative analysis; Iterative thematic/process analysis. | |
| """ | |
| if raw_interviews and len(raw_interviews) > 100: | |
| return f"""Topic: {topic} | |
| Keywords: {keywords} | |
| Analyze the supplied interview or field material for an ASQ-oriented qualitative manuscript. | |
| Rules: | |
| 1. Treat the supplied material as the only source of study facts. Use null, empty arrays, or "Not specified" for any setting, sample, count, date, case, actor/data-source relation, trustworthiness procedure, ethics procedure, reflexivity procedure, or data source not explicitly supported by the material. | |
| 2. Extract 8-15 representative quotations VERBATIM. Never repair, paraphrase, translate, combine, or invent a quotation. | |
| 3. Choose the analytic design that best fits the actual material. Do NOT default to Gioia. Use Gioia-style coding only if the evidence structure genuinely warrants it. | |
| 4. Organize evidence around a central organizational puzzle and, when the material supports it, preserve temporal stages, actor/case contrasts, process transitions, and at least one deviant or boundary observation. | |
| 5. Do not manufacture a longitudinal sequence when the material is cross-sectional, and do not manufacture multiple cases when only one is present. | |
| 6. Freeze actor-to-data-source relations explicitly in data_source_roles whenever the material supports them (for example extension officers -> interviews; farmers -> focus groups). Never infer that an actor participated in a data source not supported by the material. The study_facts object and quotations will become the frozen single source of truth for the entire manuscript. | |
| {schema} | |
| <QUALITATIVE_MATERIAL> | |
| {raw_interviews[:30000]} | |
| </QUALITATIVE_MATERIAL> | |
| """ | |
| return f"""Topic: {topic} | |
| Keywords: {keywords} | |
| Construct one internally coherent qualitative evidence packet for a complete ASQ-oriented manuscript because no interview material was supplied. | |
| Rules: | |
| 1. Choose one generic organizational research setting directly relevant to the topic and keep it fixed. Do not name a real organization, institution, or identifiable person. | |
| 2. Freeze one participant count, interview count, data-source set, actor-to-data-source relation map, analytic design, temporal/case structure, and analytic logic. Keep them internally consistent. Every actor/data-source relationship later stated in the paper must be represented in data_source_roles. | |
| 3. Produce exactly 12 natural representative quotations distributed across at least three higher-order themes. Quotations must sound like distinct participants rather than repeated template variants. | |
| 4. Choose the analytic design that best fits the constructed evidence; do NOT default to Gioia. Use process, comparative-case, grounded-theory, longitudinal, or Gioia-style logic only when the packet supports it. | |
| 5. Include at least one clearly labelled deviant/boundary observation so the eventual theory is not built only from confirming examples. | |
| 6. The evidence should support a process/mechanism explanation, not merely descriptive topic categories. | |
| 7. Freeze concise, internally coherent trustworthiness, ethics, and reflexivity procedures in their typed arrays so later Methods prose has an explicit study-owned source. Do not name a real ethics board, institution, vendor, approval number, or identifiable organization. | |
| 8. The study_facts object and evidence packet will become the frozen single source of truth for the entire manuscript. | |
| 9. Avoid decorative language, generic AI phrasing, and repetitive quotation patterns. | |
| {schema} | |
| """ | |
| def _qualitative_packet_validation_error(self, data, *, has_user_data): | |
| """Return a bounded local validation reason, or an empty string when safe. | |
| This pre-freeze validator mirrors the existing fatal packet gates so a | |
| malformed/incomplete model response can be regenerated once before any | |
| study fact or quotation becomes authoritative. It never repairs JSON or | |
| mutates model output. | |
| """ | |
| if not isinstance(data, dict): | |
| return "packet_not_json_object" | |
| facts = data.get("study_facts") or {} | |
| evidence = data.get("evidence") or data.get("quotes") or [] | |
| if not isinstance(facts, dict) or not isinstance(evidence, list): | |
| return "invalid_packet_schema" | |
| for key in ("participant_count", "interview_count"): | |
| value = facts.get(key) | |
| if value is not None and (not isinstance(value, int) or value <= 0): | |
| return f"invalid_{key}" | |
| raw_relations = facts.get("data_source_roles") or [] | |
| if raw_relations and not isinstance(raw_relations, list): | |
| return "invalid_data_source_roles" | |
| declared_sources = [ | |
| re.sub(r"\s+", " ", str(item or "")).strip().casefold() | |
| for item in (facts.get("data_sources") or []) | |
| if str(item or "").strip() | |
| ] | |
| for item in raw_relations: | |
| if not isinstance(item, dict) or not str(item.get("actor_role") or "").strip() or not str(item.get("data_source") or "").strip(): | |
| return "invalid_data_source_role_item" | |
| count = item.get("count") | |
| if count is not None and (not isinstance(count, int) or count <= 0): | |
| return "invalid_data_source_role_count" | |
| relation_source = re.sub(r"\s+", " ", str(item.get("data_source") or "")).strip().casefold() | |
| if declared_sources and not any( | |
| relation_source == source or relation_source in source or source in relation_source | |
| for source in declared_sources | |
| ): | |
| return "data_source_role_not_in_declared_sources" | |
| for key in ("trustworthiness_procedures", "ethics_procedures", "reflexivity_procedures"): | |
| value = facts.get(key) or [] | |
| if not isinstance(value, list): | |
| return f"invalid_{key}" | |
| if any(not str(item or "").strip() for item in value): | |
| return f"invalid_{key}" | |
| if not has_user_data: | |
| if not raw_relations: | |
| return "generated_packet_missing_data_source_roles" | |
| for key in ("trustworthiness_procedures", "ethics_procedures", "reflexivity_procedures"): | |
| if not (facts.get(key) or []): | |
| return f"generated_packet_missing_{key}" | |
| seen_quotes = set() | |
| theme_keys = set() | |
| retained = 0 | |
| has_deviant = False | |
| for item in evidence[:15]: | |
| if not isinstance(item, dict): | |
| continue | |
| quote = self._clean_cell(item.get("quote"), "").replace('"', "'") | |
| normalized_quote = re.sub(r"\s+", " ", quote).strip().casefold() | |
| if not quote or len(quote.split()) < 4 or normalized_quote in seen_quotes: | |
| continue | |
| seen_quotes.add(normalized_quote) | |
| retained += 1 | |
| theme = self._clean_cell(item.get("theme"), "Emergent pattern") | |
| theme_keys.add(re.sub(r"\s+", " ", theme).strip().casefold()) | |
| if str(item.get("evidence_role") or "typical").strip().lower() == "deviant": | |
| has_deviant = True | |
| validation = APP_CONFIG.get("manuscript_validation", {}) | |
| minimum_themes = int(validation.get("minimum_qualitative_themes", 3)) | |
| minimum_quotes = int(validation.get("minimum_qualitative_quotes", 8)) | |
| if len(theme_keys) < minimum_themes or retained < minimum_quotes: | |
| return f"incomplete_packet_themes_{len(theme_keys)}_quotes_{retained}" | |
| if not has_user_data and not has_deviant: | |
| return "generated_packet_missing_deviant_evidence" | |
| return "" | |
| def _generate_qualitative_packet(self, topic, keywords, raw_interviews, *, has_user_data): | |
| """Generate and validate one evidence packet with one bounded fresh retry.""" | |
| system_prompt = ( | |
| "You are a rigorous qualitative research analyst. " | |
| "Evidence fidelity has priority over narrative completeness." | |
| ) | |
| base_prompt = self._build_request(topic, keywords, raw_interviews) | |
| last_reason = "packet_not_json_object" | |
| for local_attempt in range(2): | |
| # API/auth/balance/cancellation/transport failures remain owned by | |
| # the model client and are deliberately not converted into local | |
| # content retries here. Only a completed but locally invalid packet | |
| # reaches the second attempt. | |
| if local_attempt == 0: | |
| response = self.agent.call_deepseek( | |
| system_prompt + "\n\nDIRECT OUTPUT MODE: Produce the complete evidence packet directly without extended hidden reasoning.", | |
| base_prompt, | |
| max_tokens=8000, temp=0.0, model_role="reasoning", | |
| thinking=False, allow_fallback=False, response_format={"type": "json_object"}, | |
| logical_node='structured_json', | |
| ) | |
| else: | |
| retry_prompt = base_prompt + ( | |
| "\n\nLOCAL VALIDATION RETRY: The previous completed response failed local " | |
| "JSON/schema/content validation. Generate a fresh complete qualitative evidence " | |
| "packet from the ORIGINAL inputs above. Do not repair, continue, quote, or rely on " | |
| "the previous response. Return exactly one complete JSON object matching the required " | |
| "schema, with no Markdown or explanatory prose. Preserve every evidence-fidelity rule " | |
| "from the original request, including verbatim quotation requirements when user material " | |
| "was supplied." | |
| ) | |
| response = self.agent.call_deepseek( | |
| system_prompt, | |
| retry_prompt, | |
| max_tokens=8000, | |
| temp=0.0, | |
| model_role="reasoning", | |
| thinking=False, | |
| allow_fallback=False, | |
| response_format={"type": "json_object"}, | |
| logical_node='structured_json', | |
| ) | |
| response = re.sub( | |
| r"<think>.*?</think>", | |
| "", | |
| str(response or ""), | |
| flags=re.DOTALL, | |
| ).strip() | |
| data = self.agent._extract_json_from_text(response) | |
| last_reason = self._qualitative_packet_validation_error( | |
| data, | |
| has_user_data=has_user_data, | |
| ) | |
| if not last_reason: | |
| return data | |
| if local_attempt == 0: | |
| logging.warning( | |
| "Qualitative evidence structured-output local retry | attempt=1/2 | reason=%s", | |
| last_reason, | |
| ) | |
| if last_reason == "packet_not_json_object": | |
| raise RuntimeError("Qualitative evidence packet could not be parsed safely.") | |
| if last_reason == "invalid_packet_schema": | |
| raise RuntimeError("Qualitative evidence packet has an invalid schema.") | |
| raise RuntimeError( | |
| "Qualitative evidence packet failed local validation after one bounded regeneration: " | |
| f"{last_reason}." | |
| ) | |
| def resolve_qualitative_data(self, topic, keywords, raw_interviews=""): | |
| has_user_data = bool(raw_interviews and len(raw_interviews) > 100) | |
| # R3-Free engine code is headless. Streamlit rendering belongs to app.py; | |
| # this method only produces frozen evidence + metadata. | |
| data = self._generate_qualitative_packet( | |
| topic, keywords, raw_interviews, has_user_data=has_user_data | |
| ) | |
| facts = data.get("study_facts") or {} | |
| evidence = data.get("evidence") or data.get("quotes") or [] | |
| data_sources = facts.get("data_sources") or ["Not specified"] | |
| if not isinstance(data_sources, list): | |
| data_sources = [data_sources] | |
| setting = self._clean_cell(facts.get("research_setting")) | |
| clean_sources = [self._clean_cell(item) for item in data_sources] | |
| sources = ", ".join(clean_sources) | |
| data_source_roles = self._normalize_data_source_roles(facts.get("data_source_roles") or []) | |
| trustworthiness = self._normalize_string_list(facts.get("trustworthiness_procedures")) | |
| ethics = self._normalize_string_list(facts.get("ethics_procedures")) | |
| reflexivity = self._normalize_string_list(facts.get("reflexivity_procedures")) | |
| participants = facts.get("participant_count") | |
| interviews = facts.get("interview_count") | |
| analytic_design = self._normalize_analytic_design( | |
| facts.get("analytic_design") or facts.get("analysis_method") | |
| ) | |
| analytic_logic = self._clean_cell( | |
| facts.get("analytic_logic"), | |
| "Iteratively compare empirical observations to refine a process/mechanism explanation.", | |
| ) | |
| temporal_structure = self._clean_cell(facts.get("temporal_structure")) | |
| case_structure = self._clean_cell(facts.get("case_structure")) | |
| for value, label in ((participants, "participant_count"), (interviews, "interview_count")): | |
| if value is not None and (not isinstance(value, int) or value <= 0): | |
| raise RuntimeError(f"Qualitative evidence packet has invalid {label}.") | |
| md = ( | |
| "### Frozen Qualitative Study Design\n" | |
| "| Field | Frozen value |\n" | |
| "| :--- | :--- |\n" | |
| f"| Research setting | {setting} |\n" | |
| f"| Data sources | {sources} |\n" | |
| f"| Participants | {participants if participants is not None else 'Not specified'} |\n" | |
| f"| Interviews | {interviews if interviews is not None else 'Not specified'} |\n" | |
| f"| Analytic design | {analytic_design} |\n" | |
| f"| Analytic logic | {analytic_logic} |\n" | |
| f"| Temporal structure | {temporal_structure} |\n" | |
| f"| Case structure | {case_structure} |\n" | |
| f"| Trustworthiness procedures | {'; '.join(trustworthiness) if trustworthiness else 'Not specified'} |\n" | |
| f"| Ethics procedures | {'; '.join(ethics) if ethics else 'Not specified'} |\n" | |
| f"| Reflexivity procedures | {'; '.join(reflexivity) if reflexivity else 'Not specified'} |\n\n" | |
| "[QUALITATIVE EVIDENCE CONTRACT]\n" | |
| "Use the frozen design summary and tagged evidence below as the only source of this study's empirical facts. " | |
| "Do not change the setting, sample, interview count, data sources, actor/data-source relations, analytic design, temporal structure, cases, procedures, or quotations. " | |
| "Do not infer an event sequence, actor comparison, trustworthiness/ethics/reflexivity procedure, negative case, or quotation that is absent from this packet.\n\n" | |
| "### Table: Qualitative Evidence Structure\n" | |
| "| Higher-order Theme | Empirical Concept | Process Stage | Case/Actor | Evidence Role | Representative Quote |\n" | |
| "| :--- | :--- | :--- | :--- | :--- | :--- |\n" | |
| ) | |
| if data_source_roles: | |
| md += ( | |
| "### Frozen Actor / Data-Source Relations\n" | |
| "| Actor role | Data source | Count |\n" | |
| "| :--- | :--- | :--- |\n" | |
| ) | |
| for row in data_source_roles: | |
| md += f"| {row['actor_role']} | {row['data_source']} | {row['count'] if row['count'] is not None else 'Not specified'} |\n" | |
| md += "\n" | |
| clean_evidence = [] | |
| seen_quotes = set() | |
| for item in evidence[:15]: | |
| if not isinstance(item, dict): | |
| continue | |
| quote = self._clean_cell(item.get("quote"), "").replace('"', "'") | |
| concept = self._clean_cell(item.get("concept"), "Empirical observation") | |
| theme = self._clean_cell(item.get("theme"), "Emergent pattern") | |
| stage = self._clean_cell(item.get("stage")) | |
| case_label = self._clean_cell(item.get("case_label")) | |
| evidence_role = str(item.get("evidence_role") or "typical").strip().lower() | |
| if evidence_role not in {"typical", "deviant"}: | |
| evidence_role = "typical" | |
| normalized_quote = re.sub(r"\s+", " ", quote).strip().lower() | |
| if not quote or len(quote.split()) < 4 or normalized_quote in seen_quotes: | |
| continue | |
| seen_quotes.add(normalized_quote) | |
| tag = f"QUOTE-{len(clean_evidence) + 1:02d}" | |
| md += ( | |
| f'| {theme} | {concept} | {stage} | {case_label} | {evidence_role} | ' | |
| f'"{quote}" [{tag}] |\n' | |
| ) | |
| clean_evidence.append( | |
| { | |
| "theme": theme, | |
| "concept": concept, | |
| "quote": quote, | |
| "quote_hash": hashlib.sha256(re.sub(r"\s+", " ", quote).strip().encode("utf-8")).hexdigest(), | |
| "tag": tag, | |
| "stage": stage, | |
| "case_label": case_label, | |
| "evidence_role": evidence_role, | |
| } | |
| ) | |
| theme_map = {} | |
| theme_ids = {} | |
| for item in clean_evidence: | |
| theme_key = re.sub(r"\s+", " ", item["theme"]).strip().casefold() | |
| theme_id = theme_ids.setdefault(theme_key, f"THEME-{len(theme_ids) + 1:02d}") | |
| item["theme_id"] = theme_id | |
| theme_entry = theme_map.setdefault( | |
| item["theme"], | |
| { | |
| "theme_id": theme_id, | |
| "title": item["theme"], | |
| "concepts": [], | |
| "evidence": [], | |
| "stages": [], | |
| "has_deviant_evidence": False, | |
| }, | |
| ) | |
| if item["concept"] not in theme_entry["concepts"]: | |
| theme_entry["concepts"].append(item["concept"]) | |
| if item["stage"] != "Not specified" and item["stage"] not in theme_entry["stages"]: | |
| theme_entry["stages"].append(item["stage"]) | |
| if item["evidence_role"] == "deviant": | |
| theme_entry["has_deviant_evidence"] = True | |
| theme_entry["evidence"].append( | |
| { | |
| "theme_id": item["theme_id"], | |
| "tag": item["tag"], | |
| "quote": item["quote"], | |
| "quote_hash": item["quote_hash"], | |
| "stage": item["stage"], | |
| "case_label": item["case_label"], | |
| "evidence_role": item["evidence_role"], | |
| } | |
| ) | |
| themes = list(theme_map.values()) | |
| validation = APP_CONFIG.get("manuscript_validation", {}) | |
| minimum_themes = int(validation.get("minimum_qualitative_themes", 3)) | |
| minimum_quotes = int(validation.get("minimum_qualitative_quotes", 8)) | |
| if len(themes) < minimum_themes or len(clean_evidence) < minimum_quotes: | |
| raise RuntimeError( | |
| "Qualitative evidence packet is incomplete: " | |
| f"themes={len(themes)}, quotes={len(clean_evidence)}; " | |
| f"required themes>={minimum_themes}, quotes>={minimum_quotes}." | |
| ) | |
| if not has_user_data and not any(item["evidence_role"] == "deviant" for item in clean_evidence): | |
| raise RuntimeError("Generated qualitative evidence packet lacks a deviant/boundary observation.") | |
| metadata = { | |
| "mode": "qualitative", | |
| "source_type": ( | |
| "user_provided" if has_user_data else "system_generated" | |
| ), | |
| "disclosure_required": not has_user_data, | |
| "study_facts": { | |
| "research_setting": setting, | |
| "data_sources": clean_sources, | |
| "data_source_roles": data_source_roles, | |
| "participant_count": participants, | |
| "interview_count": interviews, | |
| "trustworthiness_procedures": trustworthiness, | |
| "ethics_procedures": ethics, | |
| "reflexivity_procedures": reflexivity, | |
| "analytic_design": analytic_design, | |
| "analysis_method": analytic_design, | |
| "analytic_logic": analytic_logic, | |
| "temporal_structure": temporal_structure, | |
| "case_structure": case_structure, | |
| }, | |
| "evidence_count": len(clean_evidence), | |
| "generation_parameters": { | |
| "analytic_design": analytic_design, | |
| "requested_quote_count_when_generated": 12, | |
| "maximum_retained_quotes": 15, | |
| "minimum_themes": minimum_themes, | |
| "minimum_quotes": minimum_quotes, | |
| "requires_generated_deviant_evidence": not has_user_data, | |
| "consistency_constraints": [ | |
| "one frozen research setting", | |
| "one frozen participant count", | |
| "one frozen interview count", | |
| "one frozen data-source set", | |
| "one frozen analytic design", | |
| "no unsupported temporal or case structure", | |
| "quotation tags must remain unique", | |
| ], | |
| }, | |
| "evidence": clean_evidence, | |
| "themes": themes, | |
| } | |
| metadata["semantic_claim_frame"] = self.build_qualitative_semantic_claim_frame(metadata) | |
| status_message = "✅ 定性研究事实、分析设计与证据结构已冻结。" | |
| return md, status_message, metadata | |
| def get_qualitative_engine(agent): | |
| return QualitativeEngine(agent) | |