# -*- 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
@staticmethod
def _clean_cell(value, default="Not specified"):
cleaned = str(value or default).replace("\n", " ").strip()
return cleaned or default
@classmethod
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")
@classmethod
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
@classmethod
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
@classmethod
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
@classmethod
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()
@staticmethod
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"
@classmethod
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
@classmethod
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()
@classmethod
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
@classmethod
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
@staticmethod
def _escape_appendix_markdown_cell(value):
return re.sub(r"\s+", " ", str(value or "").strip()).replace("|", r"\|")
@classmethod
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)
@classmethod
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()
@classmethod
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"(? 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}
{raw_interviews[:30000]}
"""
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".*?",
"",
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)