#!/usr/bin/env python3 """ Persona Research Virtual Survey — Analysis & Visualization Pipeline Usage: python analyze_results.py --input results.json --survey-type concept-test python analyze_results.py --input results.json --survey-type brand-map python analyze_results.py --input results.json --survey-type price-test python analyze_results.py --input results.json --survey-type usage-habits python analyze_results.py --input results.json --survey-type survey python analyze_results.py --input results.json --survey-type ask python analyze_results.py --input results.json --survey-type concept-test --report-only python analyze_results.py --input results.json --survey-type concept-test --report-llm --backend codex-cli Outputs (in same directory as input): - results.csv All responses as flat table - summary.json Aggregate statistics - cross_tabs.csv Segment × response cross-tabulation (concept-test) - persona_comparison.csv Persona comparison table (degenerate segment case) - chart_*.png Survey-specific charts (skipped with --report-only) - report.md Markdown one-pager report """ import argparse import json import re import sys from collections import Counter, defaultdict from datetime import date from pathlib import Path import pandas as pd SCRIPT_DIR = Path(__file__).resolve().parent if str(SCRIPT_DIR) not in sys.path: sys.path.insert(0, str(SCRIPT_DIR)) SKILL_DIR = SCRIPT_DIR.parent from llm_backends import ( BACKEND_CHOICES, REPORT_BACKEND_CHOICES, format_model_label, resolve_backend, resolve_model, resolve_report_backend, run_text_completion, ) # ─── Lazy chart imports ──────────────────────────────────────────────────── _chart_libs_loaded = False def _load_chart_libs(): """Load matplotlib/seaborn on first use (skipped in --report-only mode).""" global _chart_libs_loaded, matplotlib, plt, sns, COLORS, OPTION_COLORS, FIG_SIZE, DPI if _chart_libs_loaded: return import matplotlib as _matplotlib _matplotlib.use("Agg") import matplotlib.pyplot as _plt import seaborn as _sns matplotlib = _matplotlib plt = _plt sns = _sns sns.set_theme(style="whitegrid", palette="Set2", font_scale=1.05) COLORS = sns.color_palette("Set2") OPTION_COLORS = {"A": "#66c2a5", "B": "#fc8d62", "C": "#8da0cb", "D": "#e78ac3"} FIG_SIZE = (10, 6) DPI = 150 _chart_libs_loaded = True # ─── Core: Load, Normalize, Flatten ──────────────────────────────────────── META_KEYS = {"name", "segment", "age", "gender", "occupation", "responses"} FREQUENCY_ORDER = {"daily": 5, "weekly": 4, "monthly": 3, "rarely": 2, "never": 1} def load_results(path: Path) -> list[dict]: """Load results JSON (array of persona response objects).""" with open(path) as f: data = json.load(f) if not isinstance(data, list): raise ValueError("Expected a JSON array of response objects") return data def normalize_result_entry(result: dict) -> dict: """Normalize legacy and canonical result rows to a single contract.""" if not isinstance(result, dict): raise ValueError("Each result entry must be a JSON object") responses = {} if isinstance(result.get("responses"), dict): responses.update(result["responses"]) for key, value in result.items(): if key in META_KEYS: continue responses.setdefault(key, value) return { "name": result.get("name", ""), "segment": result.get("segment", "Unknown") or "Unknown", "age": result.get("age"), "gender": result.get("gender", ""), "occupation": result.get("occupation", ""), "responses": responses, } def normalize_results(results: list[dict]) -> list[dict]: """Normalize all result entries.""" return [normalize_result_entry(result) for result in results] def _normalize_key(value: str) -> str: return re.sub(r"[^a-z0-9]+", "_", str(value).lower()).strip("_") def _display_label(value: str) -> str: text = str(value).replace("_", " ").strip() return re.sub(r"\s+", " ", text).title() def _truncate(text: object, limit: int = 140, sentence_aware: bool = False) -> str: text = str(text or "").strip() if len(text) <= limit: return text if sentence_aware: window = text[:limit] for marker in (". ", "! ", "? ", "。", "！", "？"): idx = window.rfind(marker) if idx >= limit // 2: return text[: idx + len(marker)].rstrip() return text[: limit - 3].rstrip() + "..." def _short_text_label(text: object, max_words: int = 6) -> str: value = str(text or "").strip() if not value: return "" line = value.splitlines()[0].strip() for separator in [" — ", " – ", " - ", ",", ";", "."]: if separator in line: candidate = line.split(separator, 1)[0].strip() if 1 <= len(candidate) <= 48: return candidate words = line.split() if len(words) <= max_words: return line return " ".join(words[:max_words]) def _format_value(value: object, limit: int = 140) -> str: if isinstance(value, list): return _truncate(", ".join(str(item) for item in value), limit=limit) if isinstance(value, dict): parts = [] for key, item in list(value.items())[:4]: if isinstance(item, list): item_text = ", ".join(str(v) for v in item[:3]) else: item_text = str(item) parts.append(f"{key}: {item_text}") return _truncate("; ".join(parts), limit=limit) return _truncate(value, limit=limit) def _normalize_usage_label(value: object) -> str: text = str(value or "").lower() text = re.sub(r"\([^)]*\)", "", text) text = text.replace("/", " ") text = re.sub(r"[_-]+", " ", text) text = re.sub(r"\s+", " ", text).strip() return text def flatten_responses(results: list[dict]) -> pd.DataFrame: """Flatten canonical response objects into a flat DataFrame.""" rows = [] for result in results: row = { "name": result.get("name", ""), "segment": result.get("segment", "Unknown"), "age": result.get("age"), "gender": result.get("gender", ""), "occupation": result.get("occupation", ""), } for key, value in result.get("responses", {}).items(): if isinstance(value, (str, int, float, bool)) or value is None: row[key] = value elif isinstance(value, list): row[key] = "; ".join(str(item) for item in value) elif isinstance(value, dict): row[key] = json.dumps(value, ensure_ascii=False) for subkey, subvalue in value.items(): flat_key = f"{key}_{_normalize_key(subkey)}" if isinstance(subvalue, list): row[flat_key] = "; ".join(str(item) for item in subvalue) elif isinstance(subvalue, (str, int, float, bool)) or subvalue is None: row[flat_key] = subvalue else: row[flat_key] = json.dumps(subvalue, ensure_ascii=False) else: row[key] = str(value) rows.append(row) return pd.DataFrame(rows) def _is_degenerate_crosstab(df: pd.DataFrame) -> bool: """Check if segment values are all unique (topic-only mode with archetype labels).""" return not df.empty and df["segment"].nunique() == len(df) def _generate_persona_comparison(df: pd.DataFrame, output_dir: Path): """Generate persona comparison table instead of cross-tab when segments are unique.""" cols = ["name", "segment", "age", "gender", "occupation"] preferred_cols = [ "preferred_option", "purchase_likelihood", "max_wtp", "current_product", "purchase_channel", "reasoning", "pain_points", ] for col in preferred_cols: if col in df.columns: cols.append(col) comparison = df[[col for col in cols if col in df.columns]].copy() comparison.to_csv(output_dir / "persona_comparison.csv", index=False) return comparison # ─── Utilities ───────────────────────────────────────────────────────────── def _parse_price_label(label: str) -> tuple[float, str]: """Parse a price label for sorting while preserving the original label.""" match = re.search(r"([0-9]+(?:\.[0-9]+)?)", str(label)) numeric = float(match.group(1)) if match else float("inf") cleaned = str(label) if cleaned and not cleaned.startswith("$") and match: cleaned = f"${cleaned}" return numeric, cleaned def _numeric_series(df: pd.DataFrame, column: str) -> pd.Series: """Return a numeric series with NaNs dropped.""" if column not in df.columns: return pd.Series(dtype=float) return pd.to_numeric(df[column], errors="coerce").dropna() def _summarize_numeric(series: pd.Series) -> dict | None: if series.empty: return None return { "mean": round(float(series.mean()), 2), "median": round(float(series.median()), 2), "min": round(float(series.min()), 2), "max": round(float(series.max()), 2), } def _sorted_counter(counter: Counter, limit: int | None = None) -> dict: items = counter.most_common(limit) return {key: value for key, value in items} def _top_ranked_factor(factor_ranking: dict) -> str: if not isinstance(factor_ranking, dict) or not factor_ranking: return "" best = None for factor, rank in factor_ranking.items(): try: numeric = float(rank) except (TypeError, ValueError): continue if best is None or numeric < best[1]: best = (str(factor), numeric) return best[0] if best else "" def _favorite_price_point(intent_by_price: dict) -> tuple[str, float] | tuple[None, None]: if not isinstance(intent_by_price, dict) or not intent_by_price: return (None, None) best = None for label, intent in intent_by_price.items(): try: numeric_intent = float(intent) except (TypeError, ValueError): continue price_value, cleaned = _parse_price_label(label) candidate = (numeric_intent, -price_value, cleaned) if best is None or candidate > best: best = candidate if best is None: return (None, None) return best[2], best[0] def _summarize_usage_frequency(usage_frequency: dict) -> str: if not isinstance(usage_frequency, dict) or not usage_frequency: return "" ranked = [] for occasion, frequency in usage_frequency.items(): score = FREQUENCY_ORDER.get(str(frequency).lower(), 0) if score > 1: ranked.append((score, str(occasion), str(frequency))) ranked.sort(key=lambda item: (-item[0], item[1])) if not ranked: return "No active usage occasions mentioned" parts = [f"{occasion} ({frequency})" for _, occasion, frequency in ranked[:3]] return ", ".join(parts) def _collect_verbatims(results: list[dict], survey_type: str) -> list[tuple[str, str, object, str]]: """Collect candidate verbatims for the report. Max 1 per persona.""" candidates = [] seen = set() for result in results: responses = result.get("responses", {}) texts = [] if survey_type == "concept-test": texts = [responses.get("reasoning", "")] elif survey_type == "brand-map": associations = responses.get("brand_associations", {}) if isinstance(associations, dict): texts = list(associations.values()) elif survey_type == "price-test": texts = [responses.get("reasoning", ""), responses.get("value_perception", "")] elif survey_type == "usage-habits": texts = [responses.get("pain_points", ""), responses.get("current_product", "")] elif survey_type == "ask": texts = [responses.get("reasoning", ""), responses.get("short_answer", "")] else: for value in responses.values(): if isinstance(value, str): texts.append(value) # Pick the single best (longest non-duplicate) text for this persona best = None for text in texts: cleaned = str(text).strip() if len(cleaned) < 20 or cleaned in seen: continue if best is None or len(cleaned) > len(best): best = cleaned if best is not None: seen.add(best) candidates.append( (best, result.get("name", "Unknown"), result.get("age", "?"), result.get("occupation", "")) ) # Sort by length descending so the most articulate quotes appear first candidates.sort(key=lambda c: len(c[0]), reverse=True) return candidates[:3] # ─── Analysis: Concept Test ─────────────────────────────────────────────── def analyze_concept_test(results: list[dict], df: pd.DataFrame, output_dir: Path, report_only: bool = False): """Analyze concept test results: cross-tabs, charts, summary.""" if df.empty: print("WARNING: No valid responses to analyze") return {"total_respondents": 0} if "preferred_option" not in df.columns: print("WARNING: No 'preferred_option' column found, skipping concept analysis") return {} degenerate = _is_degenerate_crosstab(df) if degenerate: _generate_persona_comparison(df, output_dir) else: cross_tab = pd.crosstab(df["segment"], df["preferred_option"], margins=True) cross_tab.to_csv(output_dir / "cross_tabs.csv") cross_pct = pd.crosstab(df["segment"], df["preferred_option"], normalize="index") * 100 cross_pct.to_csv(output_dir / "cross_tabs_pct.csv", float_format="%.1f") if not report_only: _load_chart_libs() options = sorted(df["preferred_option"].dropna().unique()) colors = [OPTION_COLORS.get(opt, COLORS[i % len(COLORS)]) for i, opt in enumerate(options)] if not degenerate: cross_tab_local = pd.crosstab(df["segment"], df["preferred_option"], margins=True) fig, ax = plt.subplots(figsize=FIG_SIZE) cross_tab_no_margin = cross_tab_local.drop("All", errors="ignore") cross_tab_no_margin = cross_tab_no_margin[[c for c in options if c in cross_tab_no_margin.columns]] cross_tab_no_margin.plot(kind="bar", stacked=True, color=colors, ax=ax, edgecolor="white") ax.set_title("Concept Preference by Segment", fontsize=14, fontweight="bold") ax.set_ylabel("Count") ax.set_xlabel("") ax.legend(title="Option", bbox_to_anchor=(1.02, 1), loc="upper left") plt.xticks(rotation=30, ha="right") plt.tight_layout() fig.savefig(output_dir / "chart_preference.png", dpi=DPI, bbox_inches="tight") plt.close(fig) fig, ax = plt.subplots(figsize=(8, 5)) counts = df["preferred_option"].value_counts().reindex(options, fill_value=0) bars = ax.bar( counts.index, counts.values, color=[OPTION_COLORS.get(option, "#999") for option in counts.index], edgecolor="white", linewidth=1.5, ) for bar, value in zip(bars, counts.values): ax.text(bar.get_x() + bar.get_width() / 2, bar.get_height() + 0.3, str(value), ha="center", va="bottom", fontweight="bold") ax.set_ylim(0, (max(counts.values) * 1.35 + 1) if len(counts) > 0 else 1) ax.set_title("Overall Concept Preference", fontsize=14, fontweight="bold") ax.set_ylabel("Count") ax.set_xlabel("Option") plt.tight_layout() fig.savefig(output_dir / "chart_overall.png", dpi=DPI, bbox_inches="tight") plt.close(fig) if "purchase_likelihood" in df.columns: purchase = pd.to_numeric(df["purchase_likelihood"], errors="coerce").dropna() if not purchase.empty: fig, ax = plt.subplots(figsize=(8, 5)) ax.hist(purchase, bins=[0.5, 1.5, 2.5, 3.5, 4.5, 5.5], color=COLORS[0], edgecolor="white") ax.set_title("Purchase Likelihood Distribution", fontsize=14, fontweight="bold") ax.set_xlabel("Likelihood (1-5)") ax.set_ylabel("Count") ax.set_xticks([1, 2, 3, 4, 5]) plt.tight_layout() fig.savefig(output_dir / "chart_purchase_likelihood.png", dpi=DPI, bbox_inches="tight") plt.close(fig) summary = { "total_respondents": len(df), "segments": df["segment"].value_counts().to_dict(), "overall_preference": df["preferred_option"].value_counts().to_dict(), "preference_by_segment": {}, } for segment in df["segment"].unique(): seg_df = df[df["segment"] == segment] summary["preference_by_segment"][segment] = { "count": len(seg_df), "preferences": seg_df["preferred_option"].value_counts().to_dict(), } purchase = _numeric_series(df, "purchase_likelihood") purchase_summary = _summarize_numeric(purchase) if purchase_summary: summary["purchase_likelihood"] = purchase_summary return summary # ─── Analysis: Brand Map ────────────────────────────────────────────────── def analyze_brand_map(results: list[dict], df: pd.DataFrame, output_dir: Path, report_only: bool = False): """Analyze brand perception results.""" if df.empty: print("WARNING: No valid responses to analyze") return {"total_respondents": 0} summary = { "total_respondents": len(df), "segments": df["segment"].value_counts().to_dict(), } if _is_degenerate_crosstab(df): _generate_persona_comparison(df, output_dir) awareness_counts = Counter() consideration_counts = Counter() sentiment_counts = defaultdict(Counter) familiarity_counts = defaultdict(Counter) for result in results: responses = result.get("responses", {}) awareness = responses.get("unaided_awareness", []) if isinstance(awareness, str): awareness = [item.strip() for item in awareness.split(";") if item.strip()] for brand in awareness: awareness_counts[str(brand).strip()] += 1 consideration = responses.get("consideration_set", []) if isinstance(consideration, str): consideration = [item.strip() for item in consideration.split(";") if item.strip()] for brand in consideration: consideration_counts[str(brand).strip()] += 1 familiarity = responses.get("aided_familiarity", {}) if isinstance(familiarity, dict): for brand, status in familiarity.items(): familiarity_counts[str(brand).strip()][str(status).strip()] += 1 buckets = responses.get("brand_buckets", {}) if isinstance(buckets, dict): for bucket_name in ("like", "dislike", "neutral"): brands = buckets.get(bucket_name, []) if isinstance(brands, str): brands = [item.strip() for item in brands.split(";") if item.strip()] for brand in brands: sentiment_counts[str(brand).strip()][bucket_name] += 1 summary["unaided_awareness_top10"] = _sorted_counter(awareness_counts, 10) summary["consideration_top10"] = _sorted_counter(consideration_counts, 10) summary["brand_sentiment"] = { brand: dict(counts) for brand, counts in sorted( sentiment_counts.items(), key=lambda item: sum(item[1].values()), reverse=True, )[:10] } summary["aided_familiarity"] = { brand: dict(counts) for brand, counts in sorted( familiarity_counts.items(), key=lambda item: sum(item[1].values()), reverse=True, )[:10] } if awareness_counts and not report_only: _load_chart_libs() fig, ax = plt.subplots(figsize=FIG_SIZE) brands, counts = zip(*awareness_counts.most_common(10)) ax.barh(list(reversed(brands)), list(reversed(counts)), color=COLORS[0], edgecolor="white") ax.set_title("Top-of-Mind Brand Awareness", fontsize=14, fontweight="bold") ax.set_xlabel("Mentions") plt.tight_layout() fig.savefig(output_dir / "chart_awareness.png", dpi=DPI, bbox_inches="tight") plt.close(fig) if consideration_counts and not report_only: _load_chart_libs() fig, ax = plt.subplots(figsize=FIG_SIZE) brands, counts = zip(*consideration_counts.most_common(10)) ax.barh(list(reversed(brands)), list(reversed(counts)), color=COLORS[1], edgecolor="white") ax.set_title("Brand Consideration Set Mentions", fontsize=14, fontweight="bold") ax.set_xlabel("Mentions") plt.tight_layout() fig.savefig(output_dir / "chart_consideration.png", dpi=DPI, bbox_inches="tight") plt.close(fig) return summary # ─── Analysis: Price Sensitivity ────────────────────────────────────────── def analyze_price_test(results: list[dict], df: pd.DataFrame, output_dir: Path, report_only: bool = False): """Analyze price sensitivity results.""" if df.empty: print("WARNING: No valid responses to analyze") return {"total_respondents": 0} summary = { "total_respondents": len(df), "segments": df["segment"].value_counts().to_dict(), } if _is_degenerate_crosstab(df): _generate_persona_comparison(df, output_dir) wtp = _numeric_series(df, "max_wtp") wtp_summary = _summarize_numeric(wtp) if wtp_summary: summary["willingness_to_pay"] = wtp_summary if not report_only: _load_chart_libs() fig, ax = plt.subplots(figsize=FIG_SIZE) ax.hist(wtp, bins=min(10, max(4, len(wtp))), color=COLORS[0], edgecolor="white") ax.set_title("Willingness to Pay Distribution", fontsize=14, fontweight="bold") ax.set_xlabel("Maximum WTP ($)") ax.set_ylabel("Count") plt.tight_layout() fig.savefig(output_dir / "chart_wtp.png", dpi=DPI, bbox_inches="tight") plt.close(fig) intent_by_price = defaultdict(list) for result in results: price_map = result.get("responses", {}).get("intent_by_price", {}) if not isinstance(price_map, dict): continue for label, intent in price_map.items(): try: numeric_intent = float(intent) except (TypeError, ValueError): continue _, cleaned_label = _parse_price_label(str(label)) intent_by_price[cleaned_label].append(numeric_intent) if intent_by_price: summary["mean_intent_by_price"] = { label: round(sum(values) / len(values), 2) for label, values in sorted(intent_by_price.items(), key=lambda item: _parse_price_label(item[0])[0]) } if not report_only: _load_chart_libs() fig, ax = plt.subplots(figsize=FIG_SIZE) prices = list(summary["mean_intent_by_price"].keys()) intents = list(summary["mean_intent_by_price"].values()) ax.plot(prices, intents, "o-", color=COLORS[0], linewidth=2, markersize=8) ax.set_title("Purchase Intent by Price Point", fontsize=14, fontweight="bold") ax.set_xlabel("Price") ax.set_ylabel("Mean Purchase Intent (1-5)") ax.set_ylim(0.5, 5.5) plt.xticks(rotation=30, ha="right") plt.tight_layout() fig.savefig(output_dir / "chart_demand_curve.png", dpi=DPI, bbox_inches="tight") plt.close(fig) if "price_quality_preference" in df.columns: summary["price_quality_preference"] = df["price_quality_preference"].dropna().value_counts().to_dict() competitive = _numeric_series(df, "competitive_reference") competitive_summary = _summarize_numeric(competitive) if competitive_summary: summary["competitive_reference"] = competitive_summary return summary # ─── Analysis: Usage Habits ─────────────────────────────────────────────── def analyze_usage_habits(results: list[dict], df: pd.DataFrame, output_dir: Path, report_only: bool = False): """Analyze usage-habits responses with dedicated summary logic.""" if df.empty: print("WARNING: No valid responses to analyze") return {"total_respondents": 0} summary = { "total_respondents": len(df), "segments": df["segment"].value_counts().to_dict(), } if _is_degenerate_crosstab(df): _generate_persona_comparison(df, output_dir) usage_counts = defaultdict(Counter) usage_labels = {} factor_scores = defaultdict(list) factor_labels = {} purchase_channels = Counter() current_products = Counter() info_sources = Counter() for result in results: responses = result.get("responses", {}) usage_frequency = responses.get("usage_frequency", {}) if isinstance(usage_frequency, dict): for occasion, frequency in usage_frequency.items(): key = _normalize_usage_label(occasion) usage_labels.setdefault(key, _display_label(key)) usage_counts[key][str(frequency).lower()] += 1 factor_ranking = responses.get("factor_ranking", {}) if isinstance(factor_ranking, dict): for factor, rank in factor_ranking.items(): try: numeric_rank = float(rank) except (TypeError, ValueError): continue key = _normalize_key(factor) factor_labels.setdefault(key, str(factor)) factor_scores[key].append(numeric_rank) channel = _short_text_label(responses.get("purchase_channel", "")) if channel: purchase_channels[channel] += 1 product = _short_text_label(responses.get("current_product", "")) if product: current_products[product] += 1 source_list = responses.get("info_sources", []) if isinstance(source_list, str): source_list = [item.strip() for item in source_list.split(";") if item.strip()] for source in source_list: label = _short_text_label(source) if label: info_sources[label] += 1 summary["usage_frequency_by_occasion"] = { usage_labels[key]: dict(counts) for key, counts in sorted(usage_counts.items(), key=lambda item: usage_labels.get(item[0], item[0])) } summary["factor_importance"] = { factor_labels[key]: round(sum(values) / len(values), 2) for key, values in sorted(factor_scores.items(), key=lambda item: sum(item[1]) / len(item[1])) } summary["top_purchase_channels"] = _sorted_counter(purchase_channels, 5) summary["top_current_products"] = _sorted_counter(current_products, 5) summary["top_info_sources"] = _sorted_counter(info_sources, 8) if factor_scores and not report_only: _load_chart_libs() fig, ax = plt.subplots(figsize=FIG_SIZE) ordered = list(summary["factor_importance"].items())[:7] labels = [label for label, _ in ordered] values = [value for _, value in ordered] ax.barh(list(reversed(labels)), list(reversed(values)), color=COLORS[0], edgecolor="white") ax.set_title("Average Factor Importance (Lower = More Important)", fontsize=14, fontweight="bold") ax.set_xlabel("Average Rank") plt.tight_layout() fig.savefig(output_dir / "chart_factor_importance.png", dpi=DPI, bbox_inches="tight") plt.close(fig) return summary # ─── Analysis: Generic Survey ───────────────────────────────────────────── def analyze_generic(results: list[dict], df: pd.DataFrame, output_dir: Path, report_only: bool = False): """Basic analysis for custom surveys.""" if df.empty: print("WARNING: No valid responses to analyze") return {"total_respondents": 0} summary = { "total_respondents": len(df), "segments": df["segment"].value_counts().to_dict(), "response_keys": sorted( { key for result in results for key in result.get("responses", {}).keys() } ), } if _is_degenerate_crosstab(df): _generate_persona_comparison(df, output_dir) numeric_summary = {} numeric_cols = df.select_dtypes(include=["number"]).columns.tolist() numeric_cols = [column for column in numeric_cols if column != "age"] for column in numeric_cols: series = _numeric_series(df, column) stats = _summarize_numeric(series) if stats: numeric_summary[column] = stats if numeric_summary: summary["numeric_summary"] = numeric_summary categorical_summary = {} for column in df.columns: if column in {"name", "segment", "age", "gender", "occupation"}: continue series = df[column].dropna() if series.empty: continue if pd.api.types.is_numeric_dtype(series): continue unique_values = series.astype(str).nunique() if unique_values <= 8: categorical_summary[column] = series.astype(str).value_counts().head(8).to_dict() if categorical_summary: summary["categorical_summary"] = categorical_summary text_columns = [] for column in df.columns: if column in {"name", "segment", "age", "gender", "occupation"}: continue series = df[column].dropna().astype(str) if series.empty: continue avg_length = series.map(len).mean() if avg_length >= 25: text_columns.append(column) return summary # ─── Analysis: Ask ───────────────────────────────────────────────────────── def analyze_ask(results: list[dict], df: pd.DataFrame, output_dir: Path, report_only: bool = False): """Analyze open-question ask results: themes, emotions.""" if df.empty: print("WARNING: No valid responses to analyze") return {"total_respondents": 0} all_themes: list[str] = [] all_emotions: list[str] = [] for result in results: responses = result.get("responses", {}) themes = responses.get("themes", []) if isinstance(themes, list): all_themes.extend(t.strip().lower() for t in themes if isinstance(t, str) and t.strip()) emotion = responses.get("emotion", "") if isinstance(emotion, str) and emotion.strip(): all_emotions.append(emotion.strip().lower()) theme_counts = Counter(all_themes) emotion_counts = Counter(all_emotions) summary = { "total_respondents": len(results), "segments": df["segment"].value_counts().to_dict(), "top_signals": [{"theme": t, "count": c} for t, c in theme_counts.most_common(10)], "emotion_distribution": dict(emotion_counts.most_common()), } if _is_degenerate_crosstab(df): _generate_persona_comparison(df, output_dir) return summary # ─── Report Generation ───────────────────────────────────────────────────── def _build_panel_overview(results: list[dict], survey_type: str) -> list[str]: headers = ["#", "Name", "Age", "Occupation", "Profile"] rows = [] for index, result in enumerate(results, 1): responses = result.get("responses", {}) row = [ str(index), result.get("name", ""), str(int(result["age"])) if pd.notna(result.get("age")) else "", result.get("occupation", ""), result.get("segment", ""), ] if survey_type == "concept-test": headers.append("Preferred") if "Preferred" not in headers else None row.append(str(responses.get("preferred_option", ""))) elif survey_type == "brand-map": headers.append("Consideration") if "Consideration" not in headers else None row.append(_truncate(", ".join(responses.get("consideration_set", [])[:3]), 40)) elif survey_type == "price-test": headers.extend(["Max WTP"]) if "Max WTP" not in headers else None row.append(str(responses.get("max_wtp", ""))) elif survey_type == "usage-habits": headers.append("Current Product") if "Current Product" not in headers else None row.append(_truncate(_short_text_label(responses.get("current_product", "")), 30)) elif survey_type == "ask": headers.append("Emotion") if "Emotion" not in headers else None row.append(str(responses.get("emotion", ""))) else: first_items = [] for key, value in list(responses.items())[:2]: first_items.append(f"{key}: {_format_value(value, limit=30)}") headers.append("Key Response") if "Key Response" not in headers else None row.append(_truncate("; ".join(first_items), 40)) rows.append(row) lines = ["## Panel Overview", "| " + " | ".join(headers) + " |", "| " + " | ".join(["---"] * len(headers)) + " |"] for row in rows: lines.append("| " + " | ".join(row) + " |") lines.append("") return lines def _render_concept_analysis(results: list[dict]) -> list[str]: """Concept-centric section: support reasons, who passed, and improvement themes.""" lines = ["## Concept Analysis", ""] total = len(results) by_concept: dict[str, list[dict]] = {} for r in results: opt = r.get("responses", {}).get("preferred_option", "") if opt: by_concept.setdefault(opt, []).append(r) all_improvements: list[tuple[str, str, str]] = [] for opt in sorted(by_concept.keys()): choosers = by_concept[opt] passers = [r for r in results if r.get("responses", {}).get("preferred_option", "") != opt] lines.append(f"### Concept {opt} — {len(choosers)}/{total} chose this") lines.append("**Support reasons:**") for r in choosers: name = r.get("name", "") reasoning = _truncate(r.get("responses", {}).get("reasoning", ""), 600, sentence_aware=True) if reasoning: lines.append(f"- {name}: {reasoning}") lines.append("") if passers: passer_desc = ", ".join( f"{r['name']} (chose {r.get('responses', {}).get('preferred_option', '')})" for r in passers ) lines.append(f"**Who passed:** {passer_desc}") lines.append("") for r in choosers: imp = r.get("responses", {}).get("improvement", "") if imp: all_improvements.append((opt, r.get("name", ""), imp)) if all_improvements: lines.append("### Improvement Themes") for opt, name, text in all_improvements: lines.append(f"- **Concept {opt}** ({name}): {_truncate(text, 400, sentence_aware=True)}") lines.append("") return lines def _render_concept_profiles(results: list[dict]) -> list[str]: lines = ["## Profile Analysis"] for result in results: responses = result.get("responses", {}) preferred = responses.get("preferred_option", "") purchase = responses.get("purchase_likelihood") improvement = responses.get("improvement", "") detail = _truncate(responses.get("reasoning", ""), 600, sentence_aware=True) prefix = f"**{result.get('segment', '')}** ({result.get('name', '')}, {result.get('age', '?')}):" if preferred: prefix += f" Chose option {preferred}" if purchase not in (None, ""): prefix += f" with purchase likelihood {purchase}/5." else: prefix += "." lines.append(prefix) if detail: lines.append(detail) if improvement: lines.append(f"Suggested improvement: {_truncate(improvement, 300, sentence_aware=True)}") lines.append("") return lines def _render_brand_profiles(results: list[dict]) -> list[str]: lines = ["## Brand Profiles"] for result in results: responses = result.get("responses", {}) awareness = ", ".join(responses.get("unaided_awareness", [])[:3]) or "no clear top-of-mind brands" consideration = ", ".join(responses.get("consideration_set", [])[:3]) or "no active consideration set" buckets = responses.get("brand_buckets", {}) if isinstance(responses.get("brand_buckets"), dict) else {} likes = ", ".join(buckets.get("like", [])[:3]) associations = responses.get("brand_associations", {}) if isinstance(responses.get("brand_associations"), dict) else {} association_line = "" if associations: brand, text = next(iter(associations.items())) association_line = f"{brand}: {_truncate(text, 220)}" lines.append( f"**{result.get('segment', '')}** ({result.get('name', '')}, {result.get('age', '?')}): " f"Top-of-mind brands are {awareness}. Consideration set includes {consideration}." ) if likes: lines.append(f"Positive leaning: {likes}.") if association_line: lines.append(association_line) lines.append("") return lines def _render_price_profiles(results: list[dict]) -> list[str]: lines = ["## Price Profiles"] for result in results: responses = result.get("responses", {}) favorite_price, favorite_intent = _favorite_price_point(responses.get("intent_by_price", {})) preference = responses.get("price_quality_preference", "") reasoning = _truncate(responses.get("reasoning", ""), 260) value_perception = _truncate(responses.get("value_perception", ""), 220) lead = ( f"**{result.get('segment', '')}** ({result.get('name', '')}, {result.get('age', '?')}): " f"Max WTP is ${responses.get('max_wtp', 'n/a')}." ) if favorite_price: lead += f" Highest stated intent is at {favorite_price} ({favorite_intent}/5)." if preference: lead += f" Leans {preference} on price-quality tradeoffs." lines.append(lead) if reasoning: lines.append(reasoning) if value_perception: lines.append(f"Value perception: {value_perception}") lines.append("") return lines def _render_usage_profiles(results: list[dict]) -> list[str]: lines = ["## Usage Profiles"] for result in results: responses = result.get("responses", {}) usage_summary = _summarize_usage_frequency(responses.get("usage_frequency", {})) top_factor = _top_ranked_factor(responses.get("factor_ranking", {})) current_product = _truncate(responses.get("current_product", ""), 180) purchase_channel = _truncate(responses.get("purchase_channel", ""), 160) pain_points = _truncate(responses.get("pain_points", ""), 240) lines.append( f"**{result.get('segment', '')}** ({result.get('name', '')}, {result.get('age', '?')}): " f"Current product is {_short_text_label(current_product) or 'not specified'}. " f"Most active occasions: {usage_summary}." ) if top_factor: lines.append(f"Top decision factor: {top_factor}.") if purchase_channel: lines.append(f"Purchase channel: {purchase_channel}") if pain_points: lines.append(f"Pain points: {pain_points}") lines.append("") return lines def _render_ask_synthesis(results: list[dict], summary: dict) -> list[str]: """Render signal and emotion synthesis section for ask reports.""" lines = ["## Signal Analysis", ""] top_signals = summary.get("top_signals", []) if top_signals: lines.append("### Top Signals") for item in top_signals[:6]: count = item["count"] total = summary.get("total_respondents", len(results)) lines.append(f"- **{item['theme']}** — {count}/{total} personas") lines.append("") emotions = summary.get("emotion_distribution", {}) if emotions: lines.append("### Emotion Landscape") for emotion, count in list(emotions.items())[:6]: lines.append(f"- {emotion}: {count}") lines.append("") return lines def _render_ask_profiles(results: list[dict]) -> list[str]: """Render per-persona answer profiles for ask reports.""" lines = ["## Response Profiles"] for result in results: responses = result.get("responses", {}) short_answer = _truncate(responses.get("short_answer", ""), 400, sentence_aware=True) emotion = responses.get("emotion", "") themes = responses.get("themes", []) theme_str = ", ".join(themes[:3]) if isinstance(themes, list) else "" profile_label = result.get("segment", "") name = result.get("name", "") age = result.get("age", "?") suffix = f" [{emotion}]" if emotion else "" lines.append( f"**{profile_label}** ({name}, {age}){suffix}: {short_answer}" ) if theme_str: lines.append(f" *Themes*: {theme_str}") lines.append("") return lines def _render_generic_profiles(results: list[dict]) -> list[str]: lines = ["## Response Profiles"] for result in results: responses = result.get("responses", {}) parts = [] for key, value in list(responses.items())[:4]: parts.append(f"{key}: {_format_value(value, limit=140)}") lines.append( f"**{result.get('segment', '')}** ({result.get('name', '')}, {result.get('age', '?')}): " + "; ".join(parts) ) lines.append("") return lines def _extract_findings(df: pd.DataFrame, summary: dict, survey_type: str) -> list[str]: """Extract 3-5 key findings from the summary data.""" findings = [] n = len(df) if survey_type == "concept-test" and "overall_preference" in summary: preferences = summary["overall_preference"] if preferences: winner = max(preferences, key=preferences.get) winner_pct = round(preferences[winner] / n * 100) if winner_pct > 60: findings.append(f"Concept {winner} is the clear winner with {winner_pct}% preference.") elif winner_pct > 40: findings.append(f"Concept {winner} leads with {winner_pct}%, but the panel is still split.") else: findings.append(f"No clear winner emerged; the leading option is {winner} at {winner_pct}%.") if "purchase_likelihood" in summary: findings.append(f"Average purchase likelihood is {summary['purchase_likelihood']['mean']}/5.") elif survey_type == "brand-map": awareness = summary.get("unaided_awareness_top10", {}) if awareness: findings.append( "Top-of-mind awareness is led by " + ", ".join(list(awareness.keys())[:3]) + "." ) consideration = summary.get("consideration_top10", {}) if consideration: findings.append( "Active consideration clusters around " + ", ".join(list(consideration.keys())[:3]) + "." ) for brand, counts in summary.get("brand_sentiment", {}).items(): if counts.get("like", 0) and counts.get("dislike", 0): findings.append( f"{brand} is polarizing: {counts.get('like', 0)} positive mentions vs {counts.get('dislike', 0)} negative mentions." ) break elif survey_type == "price-test": wtp = summary.get("willingness_to_pay") if wtp: findings.append(f"Mean WTP is ${wtp['mean']} with a median of ${wtp['median']}.") intent = summary.get("mean_intent_by_price", {}) if intent: best_price = max(intent, key=intent.get) worst_price = min(intent, key=intent.get) findings.append(f"Purchase intent peaks at {best_price} and is weakest at {worst_price}.") preference = summary.get("price_quality_preference", {}) if preference: leader = max(preference, key=preference.get) findings.append(f"The dominant price-quality stance is `{leader}`.") elif survey_type == "usage-habits": factor_importance = summary.get("factor_importance", {}) if factor_importance: top_factors = list(factor_importance.keys())[:2] findings.append("Top decision factors are " + " and ".join(top_factors) + ".") channels = summary.get("top_purchase_channels", {}) if channels: findings.append( "Purchase happens primarily through " + ", ".join(list(channels.keys())[:3]) + "." ) occasions = summary.get("usage_frequency_by_occasion", {}) if occasions: best = max( occasions.items(), key=lambda item: item[1].get("daily", 0), ) findings.append(f"The strongest daily usage occasion is {best[0]}.") elif survey_type == "survey": numeric_summary = summary.get("numeric_summary", {}) if numeric_summary: field, stats = next(iter(numeric_summary.items())) findings.append(f"{field} averages {stats['mean']} across the panel.") categorical = summary.get("categorical_summary", {}) if categorical: field, counts = next(iter(categorical.items())) leading = max(counts, key=counts.get) findings.append(f"The most common response on {field} is `{leading}`.") elif survey_type == "ask": top_signals = summary.get("top_signals", []) if top_signals: signal_labels = ", ".join(item["theme"] for item in top_signals[:3]) findings.append(f"Recurring signals: {signal_labels}.") emotions = summary.get("emotion_distribution", {}) if emotions: dominant = max(emotions, key=emotions.get) findings.append(f"Dominant emotional tone is `{dominant}` ({emotions[dominant]}/{n} personas).") return findings[:5] def resolve_report_date(output_dir: Path) -> str: """Prefer the original run timestamp when metadata is available.""" metadata = _load_run_metadata(output_dir) if metadata: timestamp = metadata.get("timestamp") if timestamp: return str(timestamp).split("T", 1)[0] return date.today().isoformat() def _load_run_metadata(output_dir: Path) -> dict | None: metadata_path = output_dir / "run_metadata.json" if not metadata_path.exists(): return None try: return json.loads(metadata_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None def _read_topic_from_config(config_path: Path) -> str | None: if not config_path.exists(): return None try: config = json.loads(config_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None topic = config.get("topic") return str(topic).strip() if topic else None def _read_user_question_from_config(config_path: Path) -> str | None: if not config_path.exists(): return None try: config = json.loads(config_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None uq = config.get("variables", {}).get("user_question") return str(uq).strip() if uq else None def _resolve_topic(output_dir: Path, df: pd.DataFrame, survey_type: str | None = None) -> str: """Resolve report topic from metadata, config, or segment labels.""" metadata = _load_run_metadata(output_dir) if metadata: if survey_type == "ask": user_question = metadata.get("user_question") if user_question: return str(user_question).strip() # Fallback: read user_question from config file before using generic topic config_file = metadata.get("config_file") if config_file: config_path = Path(str(config_file)) if not config_path.is_absolute(): config_path = Path.cwd() / config_path uq = _read_user_question_from_config(config_path) if uq: return uq topic = metadata.get("topic") if topic: return str(topic) config_file = metadata.get("config_file") if config_file: config_path = Path(str(config_file)) if not config_path.is_absolute(): config_path = Path.cwd() / config_path topic = _read_topic_from_config(config_path) if topic: return topic for candidate in [output_dir / "config.json", output_dir.parent / "config.json", output_dir / "ask-config.json"]: if survey_type == "ask": uq = _read_user_question_from_config(candidate) if uq: return uq topic = _read_topic_from_config(candidate) if topic: return topic segments = list(df["segment"].unique())[:3] if not df.empty else [] return ", ".join(segments) if segments else "Survey" def generate_markdown_report( results: list[dict], df: pd.DataFrame, summary: dict, survey_type: str, output_dir: Path, report_date: str | None = None, topic: str | None = None, ) -> str: """Generate a one-pager markdown report from survey results.""" today = report_date or resolve_report_date(output_dir) survey_label = survey_type.replace("-", " ").title() if not topic: topic = _resolve_topic(output_dir, df, survey_type=survey_type) if survey_type == "ask": lines = [ f"# Ask: {topic}", "", f"**Date**: {today} | **Panel**: {len(results)} personas", "", ] else: lines = [ f"# {survey_label}: {topic} — Virtual Research Report", "", f"**Date**: {today} | **Panel**: {len(results)} personas", "", ] lines.extend(_build_panel_overview(results, survey_type)) lines.append("## Key Findings") for index, finding in enumerate(_extract_findings(df, summary, survey_type), 1): lines.append(f"{index}. {finding}") lines.append("") if survey_type == "concept-test": lines.extend(_render_concept_analysis(results)) lines.extend(_render_concept_profiles(results)) elif survey_type == "brand-map": lines.extend(_render_brand_profiles(results)) elif survey_type == "price-test": lines.extend(_render_price_profiles(results)) elif survey_type == "usage-habits": lines.extend(_render_usage_profiles(results)) elif survey_type == "ask": lines.extend(_render_ask_synthesis(results, summary)) lines.extend(_render_ask_profiles(results)) else: lines.extend(_render_generic_profiles(results)) lines.append("## Notable Verbatims") verbatims = _collect_verbatims(results, survey_type) if verbatims: for text, name, age, occupation in verbatims: lines.append(f'> "{_truncate(text, 600, sentence_aware=True)}" — {name}, {age}, {occupation}') lines.append("") else: lines.append("No verbatim text was captured in this run.") lines.append("") lines.append("## Caveats") lines.append(f"- Virtual panel of {len(results)} personas; directional only") lines.append("- Not statistically representative; use for hypothesis generation") lines.append("- AI-generated responses may exhibit positivity bias") lines.append("") report = "\n".join(lines) report_path = output_dir / "report.md" report_path.write_text(report, encoding="utf-8") print(f"Saved report: {report_path}") return report # ─── LLM Report Generation ──────────────────────────────────────────────── GENERIC_REPORT_SYSTEM_PROMPT = """\ You are a consumer research analyst writing a one-pager report from virtual persona survey results. ## Report Structure Write a markdown report with these sections: # [Survey Type]: [Topic] — Virtual Research Report **Date**: {date} | **Panel**: {n} personas ## Panel Overview [Table: #, Name, Age, Occupation, Profile, and any survey-specific columns like Preferred Option or Max WTP] ## Key Findings [3-5 insights focusing on PATTERNS, SPLITS, and SURPRISES across the panel. DO NOT just summarize each persona — synthesize across them. Highlight: which personas agree/disagree, what drives the split, any unexpected choices.] ## Profile Analysis [One paragraph per persona. For each, explain: - What they chose and why (grounded in their personality/background) - How their Big Five traits or life circumstances influenced the response - What makes their perspective unique vs. the rest of the panel Write in third person, not persona voice.] ## Notable Verbatims [Pick 2-3 direct quotes from the most relevant free-text fields. Include full quotes, do not truncate. Format: > "quote" — Name, Age, Occupation] ## Caveats - Virtual panel of {n} personas; directional only - Not statistically representative; use for hypothesis generation - AI-generated responses may exhibit positivity bias ## Rules - Write for a marketing manager audience — insightful but accessible - Use data from the summary statistics to support findings with numbers - DO NOT invent data or statistics not present in the results - Total length: 400-800 words (excluding panel table and verbatims) """ ASK_REPORT_SYSTEM_PROMPT = """\ You are a consumer research analyst writing a qualitative synthesis report from an open-ended "ask" study where a virtual persona panel answered a single research question. Your job is cross-persona SYNTHESIS, not per-persona summarization. A reader should be able to skim this and understand what the panel collectively said, where they agreed, and where the interesting splits live. ## Report Structure Use this exact markdown skeleton. All sections are required. # Ask: [Topic] **Date**: {date} | **Panel**: {n} personas ## Panel Overview [Markdown table: #, Name, Age, Occupation, Profile, Emotion] ## Direct Answer [2-3 sentences. Summarize what the panel said overall — the dominant response pattern — WITHOUT attributing to any single persona. Lead with the most important takeaway.] ## Key Findings 1. [First synthesized insight — a pattern, split, or surprise] 2. [Second insight] 3. [Third insight] [Include 3-5 findings. Every finding must be cross-persona; no per-persona summary sentences. No-occurrence observations are valuable ("0 positive emotions across 10 responses"). Absolutely NO filler like "Panel size is N personas" or "Panel covers N segments".] ## Where They Agreed [What most or all personas said in common. 1-3 sentences. Focus on the shared belief/complaint/ experience, not on listing names.] ## Where They Differed [Notable splits — by age, segment, personality, circumstance, emotional register, engagement level. 2-4 sentences. Name the axis of difference explicitly.] ## Notable Verbatims [3-4 direct quotes from `reasoning` fields. Use FULL SENTENCES — do NOT truncate mid-sentence. Pick quotes that deliver the panel's feeling, not generic statements. Format: > "quote" — Name, Age, Occupation] ## Top Signals [Cluster semantically-similar themes before counting. For example, "greenwashing and false claims", "greenwashing with no accountability", and "greenwashing by brands" are ONE cluster mentioned by 3 personas — not three separate signals of 1 each. Output as: - **cluster label** (N) — short gloss List 3-7 clusters ordered by count.] ## Emotion Distribution [Raw emotion counts with a one-line observation. Call out striking patterns: three-way ties, skewed-negative panels, absent positive emotions, single outlier emotions. Format: emotion (N), emotion (N), ... followed by a 1-sentence takeaway.] ## Caveats - Virtual panel of {n} personas; directional only - Not statistically representative; use for hypothesis generation - AI-generated responses may exhibit positivity bias ## Rules - Write for a marketing / consumer-insights audience. - Synthesis over summary: no "Profile Analysis" section with one paragraph per persona. - Do NOT invent data not present in results. - Do NOT use mechanical filler findings. - Do NOT truncate verbatim quotes mid-sentence; prefer shorter quotes to mid-sentence cuts. - When in doubt, ground every claim in specific personas or `reasoning` snippets. """ CONCEPT_TEST_REPORT_SYSTEM_PROMPT = """\ You are a consumer research analyst writing a concept-test synthesis report from a virtual persona panel that evaluated multiple product concepts (typically A/B/C). Your job is cross-persona, cross-concept SYNTHESIS. A reader should be able to decide: (1) which concept leads, (2) for whom, (3) what to fix, and (4) whether to move forward. ## Report Structure Use this exact markdown skeleton. All sections are required. # Concept Test: [Topic] — Virtual Research Report **Date**: {date} | **Panel**: {n} personas ## Panel Overview [Markdown table: #, Name, Age, Occupation, Profile, Preferred, Purchase Likelihood] ## Preference Verdict [2-3 sentences. State the leader, the margin, and the confidence. Use precise language: "Clear winner" (>60% preference), "Narrow lead" (40-60%), "No winner; tied" (ties), "Fragmented" (every concept below 40% with multiple near-ties). Include raw counts.] ## Key Findings 1. [Cross-persona insight — who preferred what and why; not a per-persona summary] 2. [Second insight] 3. [Third insight] [3-5 findings. Absolutely NO filler like "Panel size is N personas" or "Panel covers N segments".] ## Segment / Profile Splits [Who picked what. Organize by segment/profile, not by persona. For each segment with a notable pattern, state: segment → preference → the shared reason. 3-6 lines.] ## Purchase-Intent Drivers [What moved `purchase_likelihood` up or down across the panel. Look for cross-concept patterns: ingredient transparency, price, claim credibility, packaging, etc. Cite specific factors that personas explicitly credited. 2-4 sentences.] ## Per-Concept Strengths and Weaknesses [For each concept that received at least one vote or meaningful reaction: ### Concept [X] — N/total chose this **Strengths (cross-persona):** What multiple personas liked. Synthesize, do not per-chooser-dump. **Weaknesses (cross-persona):** What passers (and sometimes choosers) flagged as issues. Write in third person. Reference 2-3 personas by name inline where it adds credibility, but do NOT list every chooser's verbatim reasoning — that is noise, not analysis.] ## Improvement Theme Clusters [Cluster semantically-similar `improvement` suggestions. Example: "disclose ceramide percentages", "list niacinamide %", "show active concentrations" are ONE cluster about ingredient transparency. Output as: - **Cluster label** (N mentions across Concepts X, Y) — one-line gloss with 1-2 persona attributions. List 3-6 clusters ordered by cross-concept relevance.] ## Notable Verbatims [3-4 direct quotes from `reasoning` fields. FULL SENTENCES — do NOT truncate mid-sentence. Pick quotes that illuminate the split or the verdict, not generic praise. Format: > "quote" — Name, Age, Occupation] ## Caveats - Virtual panel of {n} personas; directional only - Not statistically representative; use for hypothesis generation - AI-generated responses may exhibit positivity bias ## Rules - Synthesis over summary. NO "Profile Analysis" section with one paragraph per persona. - NO per-concept "support reasons" list that just repeats each chooser's reasoning verbatim. - Cluster improvement suggestions; do not dump them. - Do NOT invent data not present in results. - Do NOT truncate verbatim quotes mid-sentence. - When segments are small (N=1 per segment), state that explicitly rather than over-generalize. """ def _select_report_system_prompt(survey_type: str) -> str: """Pick the survey-type-specific LLM report prompt; fall back to generic.""" return { "ask": ASK_REPORT_SYSTEM_PROMPT, "concept-test": CONCEPT_TEST_REPORT_SYSTEM_PROMPT, }.get(survey_type, GENERIC_REPORT_SYSTEM_PROMPT) REPORT_SYSTEM_PROMPT = GENERIC_REPORT_SYSTEM_PROMPT def generate_llm_report( results: list[dict], df: pd.DataFrame, summary: dict, survey_type: str, output_dir: Path, *, backend: str, model: str | None = None, report_date: str | None = None, topic: str | None = None, isolation: bool = True, effort: str | None = None, ) -> str: """Generate a narrative one-pager report using the resolved LLM backend.""" system_prompt = _select_report_system_prompt(survey_type).format( date=report_date or resolve_report_date(output_dir), n=len(results), ) resolved_topic = topic or _resolve_topic(output_dir, df, survey_type=survey_type) user_message = ( f"Survey type: {survey_type}\n" f"Topic: {resolved_topic}\n\n" f"## Results Data ({len(results)} personas)\n\n" f"{json.dumps(results, indent=2, ensure_ascii=False)}\n\n" f"## Summary Statistics\n\n" f"{json.dumps(summary, indent=2, ensure_ascii=False)}\n\n" f"Generate the one-pager report now. Return only the markdown report content." ) try: completion = run_text_completion( backend=backend, system_prompt=system_prompt, user_message=user_message, model=model, cwd=output_dir, isolation=isolation, effort=effort, ) except (FileNotFoundError, RuntimeError, TimeoutError, json.JSONDecodeError) as error: print(f"WARNING: LLM report generation failed: {error}", file=sys.stderr) print("Falling back to Python report generation...", file=sys.stderr) fallback_report = generate_markdown_report( results, df, summary, survey_type, output_dir, report_date=report_date, topic=resolved_topic, ) fallback_note = ( "> **Note**: This report was generated using the Python template engine " f"(LLM report generation failed: {error}). " "Re-run with `--report-llm` to retry.\n\n" ) return fallback_note + fallback_report report_text = completion["text"] report_text = re.sub(r"^```(?:markdown)?\s*\n", "", report_text) report_text = re.sub(r"\n```\s*$", "", report_text) report_path = output_dir / "report.md" report_path.write_text(report_text, encoding="utf-8") print(f"Saved LLM report: {report_path}") return report_text # ─── Main ────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser(description="Persona Research Survey Analysis") parser.add_argument("--input", required=True, help="Path to results.json") parser.add_argument( "--survey-type", default="concept-test", choices=["concept-test", "brand-map", "price-test", "usage-habits", "survey", "ask"], help="Type of survey to analyze", ) parser.add_argument( "--report-only", action="store_true", help="Skip chart generation, produce report.md only", ) parser.add_argument( "--topic", help="Research topic for report title (auto-detected from config/manifest if omitted)", ) parser.add_argument( "--report-llm", action="store_true", help="Generate report using the selected LLM backend for narrative synthesis", ) parser.add_argument( "--model", help="Model override for the selected report backend", ) parser.add_argument( "--backend", choices=BACKEND_CHOICES, default="auto", help="Default execution backend (`auto` infers from runtime markers/CLI availability)", ) parser.add_argument( "--report-backend", choices=REPORT_BACKEND_CHOICES, default="same", help="Backend for LLM report generation (`same` uses --backend)", ) parser.add_argument( "--no-isolation", action="store_true", help="Disable claude CLI --safe-mode context isolation for the report call", ) parser.add_argument( "--effort", choices=["low", "medium", "high", "xhigh", "max"], help="claude CLI effort level for the report call (not supported by haiku)", ) args = parser.parse_args() input_path = Path(args.input).resolve() if not input_path.exists(): print(f"ERROR: Input file not found: {input_path}") sys.exit(1) output_dir = input_path.parent resolved_backend = None resolved_report_backend = "python" resolved_report_model = None if args.report_llm: resolved_backend = resolve_backend(args.backend) resolved_report_backend = resolve_report_backend( args.report_backend, resolved_backend, ) if resolved_report_backend != "python": resolved_report_model = resolve_model(args.model, resolved_report_backend) print(f"Loading results from: {input_path}") raw_results = load_results(input_path) results = normalize_results(raw_results) print(f"Loaded {len(results)} responses") if args.report_llm: print( "LLM report backend: " f"{resolved_report_backend} | model: {format_model_label(resolved_report_model)}" ) df = flatten_responses(results) csv_path = output_dir / "results.csv" df.to_csv(csv_path, index=False) print(f"Saved flat CSV: {csv_path}") analyzers = { "concept-test": analyze_concept_test, "brand-map": analyze_brand_map, "price-test": analyze_price_test, "usage-habits": analyze_usage_habits, "survey": analyze_generic, "ask": analyze_ask, } analyzer = analyzers[args.survey_type] summary = analyzer(results, df, output_dir, report_only=args.report_only) summary_path = output_dir / "summary.json" with open(summary_path, "w") as f: json.dump(summary, f, indent=2, ensure_ascii=False) print(f"Saved summary: {summary_path}") report_date = resolve_report_date(output_dir) if args.report_llm and resolved_report_backend != "python": generate_llm_report( results, df, summary, args.survey_type, output_dir, backend=resolved_report_backend, model=resolved_report_model, report_date=report_date, topic=args.topic, isolation=not args.no_isolation, effort=args.effort, ) else: generate_markdown_report( results, df, summary, args.survey_type, output_dir, report_date=report_date, topic=args.topic, ) print("\n" + "=" * 60) print("ANALYSIS SUMMARY") print("=" * 60) print(json.dumps(summary, indent=2, ensure_ascii=False)) print("=" * 60) print(f"\nOutput files in: {output_dir}") for path in sorted(output_dir.glob("*")): if path.name != "results.json": print(f" {path.name}") if __name__ == "__main__": main()