# -*- coding: utf-8 -*- """Qualitative evidence preparation for SmartPaper's ASQ track.""" from __future__ import annotations import logging import re from config import APP_CONFIG class QualitativeEngine: 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 def _build_request(self, topic, keywords, raw_interviews): schema = """Return ONLY one valid JSON object with this schema: { "study_facts": { "research_setting": "string", "data_sources": ["string"], "participant_count": integer or null, "interview_count": integer or null, "analysis_method": "string" }, "evidence": [ {"theme": "2nd-order theme", "concept": "1st-order concept", "quote": "verbatim quote"} ] } """ if raw_interviews and len(raw_interviews) > 100: return f"""Topic: {topic} Keywords: {keywords} Analyze the supplied interview or field material. Rules: 1. Use only setting, sample, source, and count information supported by the supplied material. Use null or "Not specified" when absent. 2. Extract up to 15 representative quotations verbatim; do not rewrite quotations. 3. Build a coherent Gioia-style coding structure. 4. The study_facts object will become the single source of truth for the entire manuscript. {schema} {raw_interviews[:30000]} """ return f"""Topic: {topic} Keywords: {keywords} Construct one internally coherent qualitative study design and evidence packet for a complete ASQ-style manuscript when no interview material is supplied. Rules: 1. Choose one research setting and keep it fixed. 2. Choose one participant count, interview count, data-source set, and analysis method; do not create conflicting counts. 3. Produce exactly 12 natural representative quotations organized into a Gioia-style coding structure. 4. The study_facts object will become the single source of truth for the entire manuscript. 5. 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 deterministic evidence + metadata. response = "".join( str(chunk or "") for chunk in self.agent.stream_deepseek( "You are a rigorous qualitative research analyst.", 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): logging.error("Qualitative JSON parsing failed; using safe fallback.") data = { "study_facts": { "research_setting": "Not specified", "data_sources": ["Interview or field material"], "participant_count": None, "interview_count": None, "analysis_method": "Gioia-style inductive coding", }, "evidence": [ { "theme": "Sensemaking", "concept": "Initial uncertainty", "quote": ( "We did not yet have a shared understanding " "of how to respond." ), } ], } facts = data.get("study_facts") or {} evidence = data.get("evidence") or data.get("quotes") or [] if not isinstance(evidence, list): evidence = [] 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")) sources = ", ".join(self._clean_cell(item) for item in data_sources) participants = facts.get("participant_count") interviews = facts.get("interview_count") method = self._clean_cell( facts.get("analysis_method"), "Gioia-style inductive coding", ) md = ( "### Qualitative Study Design Summary\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"| Analysis method | {method} |\n\n" "[CONSISTENCY MANDATE]\n" "Use the study-design summary above as the single source of truth. " "Do not change the setting, sample, interview count, data sources, " "or analysis method in any section.\n\n" "### Table: Qualitative Coding Structure (Data Evidence)\n" "| 2nd-order Theme | 1st-order Concept | Representative Quote |\n" "| :--- | :--- | :--- |\n" ) clean_evidence = [] for index, item in enumerate(evidence[:15]): if not isinstance(item, dict): continue quote = self._clean_cell(item.get("quote"), "").replace('"', "'") concept = self._clean_cell(item.get("concept"), "Analysis") theme = self._clean_cell(item.get("theme"), "Observation") if not quote: continue tag = f"QUOTE-{index + 1:02d}" md += f'| {theme} | {concept} | "{quote}" [{tag}] |\n' clean_evidence.append( { "theme": theme, "concept": concept, "quote": quote, "tag": tag, } ) theme_map = {} for item in clean_evidence: theme_entry = theme_map.setdefault( item["theme"], {"title": item["theme"], "concepts": [], "evidence": []}, ) if item["concept"] not in theme_entry["concepts"]: theme_entry["concepts"].append(item["concept"]) theme_entry["evidence"].append( {"tag": item["tag"], "quote": item["quote"]} ) 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}." ) 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": data_sources, "participant_count": participants, "interview_count": interviews, "analysis_method": method, }, "evidence_count": len(clean_evidence), "generation_parameters": { "coding_approach": "Gioia-style inductive coding", "requested_quote_count_when_generated": 12, "maximum_retained_quotes": 15, "minimum_themes": minimum_themes, "minimum_quotes": minimum_quotes, "consistency_constraints": [ "one fixed research setting", "one fixed participant count", "one fixed interview count", "one fixed data-source set", "one fixed analysis method", "quotation tags must remain unique", ], }, "evidence": clean_evidence, "themes": themes, } status_message = "✅ 定性研究事实、摘要锚点输入和Gioia编码证据已准备完成。" return md, status_message, metadata def get_qualitative_engine(agent): return QualitativeEngine(agent)