# -*- coding: utf-8 -*-
"""Qualitative evidence preparation for SmartPaper's ASQ-oriented track."""
from __future__ import annotations
import logging
import re
from config import APP_CONFIG
class QualitativeEngine:
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")
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"],
"participant_count": integer or null,
"interview_count": integer or null,
"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 or "Not specified" for any setting, sample, count, date, case, or data source not 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. 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, analytic design, temporal/case structure, and analytic logic. Keep them internally consistent.
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. The study_facts object and evidence packet will become the frozen single source of truth for the entire manuscript.
8. Avoid decorative language, generic AI phrasing, and repetitive quotation patterns.
{schema}
"""
def resolve_qualitative_data(self, topic, keywords, raw_interviews=""):
has_user_data = bool(raw_interviews and len(raw_interviews) > 100)
prompt = self._build_request(topic, keywords, raw_interviews)
# R3-Free engine code is headless. Streamlit rendering belongs to app.py;
# this method only produces frozen evidence + metadata.
response = "".join(
str(chunk or "")
for chunk in self.agent.stream_deepseek(
"You are a rigorous qualitative research analyst. Evidence fidelity has priority over narrative completeness.",
prompt,
model_role="reasoning",
thinking=True,
allow_fallback=False,
)
)
response = re.sub(
r".*?",
"",
response,
flags=re.DOTALL,
).strip()
data = self.agent._extract_json_from_text(response)
if not isinstance(data, dict):
raise RuntimeError("Qualitative evidence packet could not be parsed safely.")
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):
raise RuntimeError("Qualitative evidence packet has an invalid schema.")
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)
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\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, analytic design, temporal structure, cases, or quotations. "
"Do not infer an event sequence, actor comparison, 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"
)
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,
"tag": tag,
"stage": stage,
"case_label": case_label,
"evidence_role": evidence_role,
}
)
theme_map = {}
for item in clean_evidence:
theme_entry = theme_map.setdefault(
item["theme"],
{
"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(
{
"tag": item["tag"],
"quote": item["quote"],
"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,
"participant_count": participants,
"interview_count": interviews,
"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,
}
status_message = "✅ 定性研究事实、分析设计与证据结构已冻结。"
return md, status_message, metadata
def get_qualitative_engine(agent):
return QualitativeEngine(agent)