Wispr Analytics

{
  "name": "wispr-analytics",
  "description": "This skill should be used when analyzing Wispr Flow voice dictation history for self-reflection, work patterns, mental h",
  "author": {
    "name": "Gleb Kalinin"
  },
  "repository": "https://github.com/glebis/claude-skills",
  "license": "MIT"
}

#!/usr/bin/env python3
"""
Extract prosodic (audio-based) features from Wispr Flow dictation history.

This is the "prosody" analysis mode -- a peer of technical/soft/trends/mental,
but it reads the recorded WAV audio (History.audio BLOB) instead of text, and
uses Praat (via parselmouth) to extract pitch, intensity, and voice-quality
features as gentle affect/energy proxies for self-reflection.

Usage:
    python3 extract_prosody.py [--period today|yesterday|week|month|YYYY-MM-DD|YYYY-MM-DD:YYYY-MM-DD]
                               [--format text|json] [--limit N] [--output PATH]

AUDIO RETENTION CAVEAT:
    Wispr keeps the recorded audio only for recent dictations (~900 of 16,000+
    rows). Older rows have audio pruned after upload. Prosody is therefore
    available ONLY for recent dictations -- the report surfaces this honestly
    with a coverage line (X of Y dictations in period had audio).

Requires: praat-parselmouth (pip install praat-parselmouth)

The database is opened strictly READ-ONLY; this script never writes to it.
"""

import sqlite3
import json
import argparse
import os
import sys
import tempfile
import statistics
import math
from datetime import datetime, timedelta
from collections import defaultdict
from pathlib import Path

try:
    import parselmouth
    from parselmouth.praat import call
except ImportError:
    print("ERROR: praat-parselmouth is required. Install with: pip install praat-parselmouth",
          file=sys.stderr)
    sys.exit(1)

DB_PATH = os.path.expanduser("~/Library/Application Support/Wispr Flow/flow.sqlite")

# Default cap on clips processed to bound runtime (audio analysis is slow vs SQL).
DEFAULT_LIMIT = 300


def parse_period(period_str):
    """Return (start_datetime_str, end_datetime_str) for SQL WHERE clause.

    Mirrors extract_wispr.py period semantics exactly.
    """
    now = datetime.now()
    today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)

    if period_str == "today":
        start = today_start
        end = now
    elif period_str == "yesterday":
        start = today_start - timedelta(days=1)
        end = today_start
    elif period_str == "week":
        start = today_start - timedelta(days=7)
        end = now
    elif period_str == "month":
        start = today_start - timedelta(days=30)
        end = now
    elif ":" in period_str:
        parts = period_str.split(":")
        start = datetime.strptime(parts[0], "%Y-%m-%d")
        end = datetime.strptime(parts[1], "%Y-%m-%d").replace(hour=23, minute=59, second=59)
    else:
        start = datetime.strptime(period_str, "%Y-%m-%d")
        end = start.replace(hour=23, minute=59, second=59)

    return start.strftime("%Y-%m-%d %H:%M:%S"), end.strftime("%Y-%m-%d %H:%M:%S")


def _finite(x):
    """Return float(x) if finite, else None (filters Praat inf/nan sentinels)."""
    try:
        f = float(x)
    except (TypeError, ValueError):
        return None
    return f if math.isfinite(f) else None


def open_readonly(db_path):
    """Open the SQLite DB strictly read-only (immutable URI)."""
    uri = f"file:{db_path}?mode=ro&immutable=1"
    conn = sqlite3.connect(uri, uri=True)
    conn.row_factory = sqlite3.Row
    return conn


def count_period(conn, start, end):
    """Return (total_in_period, with_audio_in_period)."""
    cur = conn.cursor()
    cur.execute("""
        SELECT COUNT(*) AS total,
               SUM(CASE WHEN audio IS NOT NULL THEN 1 ELSE 0 END) AS with_audio
        FROM History
        WHERE isArchived = 0
          AND timestamp >= ? AND timestamp <= ?
    """, (start, end))
    r = cur.fetchone()
    return r["total"] or 0, r["with_audio"] or 0


def fetch_audio_rows(conn, start, end, limit):
    """Fetch period rows that have audio, newest first, capped at `limit`."""
    cur = conn.cursor()
    cur.execute("""
        SELECT timestamp, app, numWords, duration, speechDuration,
               detectedLanguage, language, audio
        FROM History
        WHERE isArchived = 0
          AND timestamp >= ? AND timestamp <= ?
          AND audio IS NOT NULL
        ORDER BY timestamp DESC
        LIMIT ?
    """, (start, end, limit))
    return cur.fetchall()


def normalize_language(row):
    lang = row["detectedLanguage"] or row["language"] or "unknown"
    # Collapse locale variants (en-US -> en, ru-RU -> ru)
    if lang and "-" in lang:
        lang = lang.split("-")[0]
    return lang or "unknown"


def analyze_clip(wav_bytes):
    """Extract prosodic features from a single WAV blob using Praat.

    Returns a dict of features, or None if the clip has no voiced frames /
    cannot be analyzed at all. Individual fragile features (jitter/shimmer/HNR)
    are wrapped so they degrade to None instead of failing the whole clip.
    """
    tmp_path = None
    try:
        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tf:
            tf.write(wav_bytes)
            tmp_path = tf.name

        snd = parselmouth.Sound(tmp_path)

        # --- Pitch (F0) ---
        try:
            pitch = snd.to_pitch()
            f0 = pitch.selected_array["frequency"]
            voiced = f0[f0 > 0]  # drop unvoiced frames
        except Exception:
            voiced = []

        if len(voiced) < 3:
            # No usable voiced signal -> skip this clip
            return None

        f0_mean = float(statistics.mean(voiced))
        f0_std = float(statistics.pstdev(voiced)) if len(voiced) > 1 else 0.0
        feat = {
            "f0_mean": f0_mean,
            "f0_median": float(statistics.median(voiced)),
            "f0_min": float(min(voiced)),
            "f0_max": float(max(voiced)),
            "f0_range": float(max(voiced) - min(voiced)),
            "f0_std": f0_std,
            "f0_cv": float(f0_std / f0_mean) if f0_mean else None,  # monotone<->expressive
        }

        # --- Intensity (dB) ---
        try:
            intensity = snd.to_intensity()
            ivals = intensity.values[0]
            ivals = ivals[~(ivals != ivals)]  # drop NaN
            ivals = [float(v) for v in ivals if v > 0]
            if ivals:
                feat["intensity_mean"] = float(statistics.mean(ivals))
                feat["intensity_range"] = float(max(ivals) - min(ivals))
                feat["intensity_std"] = float(statistics.pstdev(ivals)) if len(ivals) > 1 else 0.0
        except Exception:
            pass

        # --- Voice quality: jitter, shimmer, HNR ---
        # These are fragile on short/noisy clips -- degrade gracefully.
        try:
            point_process = call(snd, "To PointProcess (periodic, cc)", 75, 500)
            jitter = _finite(call(point_process, "Get jitter (local)",
                                  0, 0, 0.0001, 0.02, 1.3))
            if jitter is not None:
                feat["jitter_local"] = jitter
            shimmer = _finite(call([snd, point_process], "Get shimmer (local)",
                                   0, 0, 0.0001, 0.02, 1.3, 1.6))
            if shimmer is not None:
                feat["shimmer_local"] = shimmer
        except Exception:
            pass

        try:
            harmonicity = snd.to_harmonicity_cc()
            hnr = _finite(call(harmonicity, "Get mean", 0, 0))
            if hnr is not None and hnr > -200:  # filter undefined sentinel
                feat["hnr"] = hnr
        except Exception:
            pass

        return feat
    except Exception:
        return None
    finally:
        if tmp_path and os.path.exists(tmp_path):
            try:
                os.remove(tmp_path)
            except OSError:
                pass


def compute_tempo(row):
    """Speaking rate (WPM) and pause ratio from DB columns (not audio)."""
    numWords = row["numWords"] or 0
    speech = row["speechDuration"] or 0
    total = row["duration"] or 0
    wpm = None
    if speech and speech > 0:
        wpm = numWords / (speech / 60.0)
    pause_ratio = None
    if total and total > 0:
        pause_ratio = max(0.0, (total - speech) / total)
    return wpm, pause_ratio


def _agg(values):
    """Mean/median/min/max/std for a list of numbers, ignoring None."""
    vals = [v for v in values if v is not None and math.isfinite(v)]
    if not vals:
        return None
    return {
        "n": len(vals),
        "mean": round(float(statistics.mean(vals)), 3),
        "median": round(float(statistics.median(vals)), 3),
        "min": round(float(min(vals)), 3),
        "max": round(float(max(vals)), 3),
        "std": round(float(statistics.pstdev(vals)), 3) if len(vals) > 1 else 0.0,
    }


def _wmean(pairs):
    """Weighted mean of (value, weight) pairs, ignoring None values."""
    num = den = 0.0
    for v, w in pairs:
        if v is None or w is None or w <= 0:
            continue
        num += v * w
        den += w
    return round(num / den, 3) if den else None


def aggregate(clips):
    """Aggregate per-clip feature dicts into period-level summaries.

    `clips` is a list of dicts: each has 'feat' (or None), 'lang', 'date',
    'wpm', 'pause_ratio', 'speech'.
    """
    metrics = ["f0_mean", "f0_median", "f0_range", "f0_std", "f0_cv",
               "intensity_mean", "intensity_range", "intensity_std",
               "jitter_local", "shimmer_local", "hnr"]

    overall = {}
    for m in metrics:
        overall[m] = _agg([c["feat"].get(m) for c in clips if c["feat"]])

    # Tempo (from DB, present even when audio feature extraction fails)
    overall["wpm"] = _agg([c["wpm"] for c in clips])
    overall["pause_ratio"] = _agg([c["pause_ratio"] for c in clips])

    # Speech-duration-weighted F0 mean and WPM (longer clips count more)
    overall["f0_mean_weighted"] = _wmean(
        [(c["feat"].get("f0_mean") if c["feat"] else None, c["speech"]) for c in clips]
    )
    overall["wpm_weighted"] = _wmean([(c["wpm"], c["speech"]) for c in clips])

    return overall


def by_language(clips):
    groups = defaultdict(list)
    for c in clips:
        groups[c["lang"]].append(c)
    out = {}
    for lang, cs in sorted(groups.items(), key=lambda x: -len(x[1])):
        out[lang] = {
            "n_clips": len(cs),
            "n_voiced": sum(1 for c in cs if c["feat"]),
            "f0_mean": _agg([c["feat"].get("f0_mean") for c in cs if c["feat"]]),
            "f0_cv": _agg([c["feat"].get("f0_cv") for c in cs if c["feat"]]),
            "wpm": _agg([c["wpm"] for c in cs]),
            "hnr": _agg([c["feat"].get("hnr") for c in cs if c["feat"]]),
        }
    return out


def daily_trend(clips):
    days = defaultdict(list)
    for c in clips:
        days[c["date"]].append(c)
    rows = []
    for day in sorted(days.keys()):
        cs = days[day]
        feats = [c["feat"] for c in cs if c["feat"]]

        def m(key, source=None):
            src = source if source is not None else feats
            vals = [f.get(key) for f in src if f and f.get(key) is not None] if source is None \
                else [x for x in src if x is not None]
            return round(statistics.mean(vals), 1) if vals else None

        rows.append({
            "date": day,
            "n_clips": len(cs),
            "n_voiced": len(feats),
            "f0_mean": m("f0_mean"),
            "f0_cv": (round(statistics.mean([f["f0_cv"] for f in feats if f.get("f0_cv") is not None]), 3)
                      if any(f.get("f0_cv") is not None for f in feats) else None),
            "intensity_mean": m("intensity_mean"),
            "wpm": (round(statistics.mean([c["wpm"] for c in cs if c["wpm"] is not None]), 1)
                    if any(c["wpm"] is not None for c in cs) else None),
            "hnr": m("hnr"),
        })
    return rows


def build_report(period_label, total, with_audio, processed, skipped_no_voice,
                 truncated, limit, clips):
    overall = aggregate(clips)
    langs = by_language(clips)
    trend = daily_trend(clips)
    feature_failures = {
        "jitter": sum(1 for c in clips if c["feat"] and "jitter_local" not in c["feat"]),
        "shimmer": sum(1 for c in clips if c["feat"] and "shimmer_local" not in c["feat"]),
        "hnr": sum(1 for c in clips if c["feat"] and "hnr" not in c["feat"]),
        "intensity": sum(1 for c in clips if c["feat"] and "intensity_mean" not in c["feat"]),
    }
    return {
        "period": period_label,
        "coverage": {
            "total_dictations_in_period": total,
            "dictations_with_audio": with_audio,
            "clips_processed": processed,
            "clips_skipped_no_voice": skipped_no_voice,
            "truncated_by_limit": truncated,
            "limit": limit,
        },
        "overall": overall,
        "by_language": langs,
        "daily_trend": trend,
        "feature_failures": feature_failures,
    }


def _fmt(agg, key="mean", unit=""):
    if not agg or agg.get(key) is None:
        return "n/a"
    return f"{agg[key]}{unit}"


def format_text(report):
    cov = report["coverage"]
    ov = report["overall"]
    lines = []
    lines.append(f"## Wispr Prosody Analysis: {report['period']}")
    lines.append("")
    lines.append(f"**Coverage**: {cov['dictations_with_audio']} of "
                 f"{cov['total_dictations_in_period']} dictations in this period had "
                 f"retained audio. Processed {cov['clips_processed']} clips "
                 f"({cov['clips_skipped_no_voice']} skipped: no voiced frames).")
    if cov["truncated_by_limit"]:
        lines.append(f"> NOTE: more clips had audio than the --limit of {cov['limit']}; "
                     f"coverage is capped. Raise --limit to process all.")
    lines.append("")
    lines.append("_Audio is retained only for recent dictations; older rows are "
                 "timing-only. Treat these as gentle reflection proxies, not measures._")
    lines.append("")

    lines.append("### Pitch (F0) -- monotone <-> expressive")
    f0m = ov.get("f0_mean") or {}
    lines.append(f"- Mean F0: {_fmt(f0m, 'mean', ' Hz')} "
                 f"(median {_fmt(f0m, 'median', ' Hz')}, "
                 f"range {_fmt(f0m, 'min')}-{_fmt(f0m, 'max')} Hz)")
    if ov.get("f0_mean_weighted") is not None:
        lines.append(f"- Mean F0 (speech-duration weighted): {ov['f0_mean_weighted']} Hz")
    lines.append(f"- F0 CV (variability proxy): {_fmt(ov.get('f0_cv'))} "
                 f"(higher = more expressive/varied intonation)")
    lines.append(f"- F0 std: {_fmt(ov.get('f0_std'), 'mean', ' Hz')}, "
                 f"range {_fmt(ov.get('f0_range'), 'mean', ' Hz')}")
    lines.append("")

    lines.append("### Intensity (loudness dynamics)")
    lines.append(f"- Mean intensity: {_fmt(ov.get('intensity_mean'), 'mean', ' dB')}")
    lines.append(f"- Intensity range: {_fmt(ov.get('intensity_range'), 'mean', ' dB')}, "
                 f"std {_fmt(ov.get('intensity_std'), 'mean', ' dB')}")
    lines.append("")

    lines.append("### Voice quality")
    lines.append(f"- Jitter (local): {_fmt(ov.get('jitter_local'))}")
    lines.append(f"- Shimmer (local): {_fmt(ov.get('shimmer_local'))}")
    lines.append(f"- HNR: {_fmt(ov.get('hnr'), 'mean', ' dB')} "
                 f"(lower can track fatigue/vocal strain)")
    lines.append("")

    lines.append("### Tempo (from timing columns)")
    lines.append(f"- Speaking rate: {_fmt(ov.get('wpm'), 'mean', ' WPM')}")
    if ov.get("wpm_weighted") is not None:
        lines.append(f"- Speaking rate (weighted): {ov['wpm_weighted']} WPM")
    pr = ov.get("pause_ratio") or {}
    lines.append(f"- Pause ratio: {_fmt(pr, 'mean')} "
                 f"(fraction of total time not actively speaking)")
    lines.append("")

    if report["by_language"]:
        lines.append("### By Language")
        lines.append("| Lang | Clips | Voiced | Mean F0 | F0 CV | WPM | HNR |")
        lines.append("|------|-------|--------|---------|-------|-----|-----|")
        for lang, d in report["by_language"].items():
            def cell(a, k="mean"):
                return a[k] if a and a.get(k) is not None else "n/a"
            lines.append(f"| {lang} | {d['n_clips']} | {d['n_voiced']} | "
                         f"{cell(d['f0_mean'])} | {cell(d['f0_cv'])} | "
                         f"{cell(d['wpm'])} | {cell(d['hnr'])} |")
        lines.append("")

    if len(report["daily_trend"]) > 1:
        lines.append("### Daily Trend")
        lines.append("| Date | Clips | Voiced | Mean F0 | F0 CV | Intensity | WPM | HNR |")
        lines.append("|------|-------|--------|---------|-------|-----------|-----|-----|")
        for t in report["daily_trend"]:
            def c(v):
                return v if v is not None else "n/a"
            lines.append(f"| {t['date']} | {t['n_clips']} | {t['n_voiced']} | "
                         f"{c(t['f0_mean'])} | {c(t['f0_cv'])} | "
                         f"{c(t['intensity_mean'])} | {c(t['wpm'])} | {c(t['hnr'])} |")
        lines.append("")

    ff = report["feature_failures"]
    if any(ff.values()):
        lines.append("### Feature extraction notes")
        lines.append(f"- Voiced clips where a feature could not be computed: "
                     f"jitter {ff['jitter']}, shimmer {ff['shimmer']}, "
                     f"HNR {ff['hnr']}, intensity {ff['intensity']}.")
        lines.append("")

    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Extract prosodic (audio-based) features from Wispr Flow history")
    parser.add_argument("--period", default="today",
                        help="today, yesterday, week, month, YYYY-MM-DD, or "
                             "YYYY-MM-DD:YYYY-MM-DD")
    parser.add_argument("--format", default="text", choices=["text", "json"],
                        help="Output format")
    parser.add_argument("--limit", type=int, default=DEFAULT_LIMIT,
                        help=f"Max clips to process (default {DEFAULT_LIMIT}). "
                             "Logs when coverage is truncated.")
    parser.add_argument("--output", default=None,
                        help="Output file path (default: stdout)")
    args = parser.parse_args()

    if not os.path.exists(DB_PATH):
        print(f"ERROR: database not found at {DB_PATH}", file=sys.stderr)
        sys.exit(1)

    start, end = parse_period(args.period)
    period_label = (f"{start} to {end}"
                    if args.period not in ("today", "yesterday", "week", "month")
                    else args.period)

    conn = open_readonly(DB_PATH)
    total, with_audio = count_period(conn, start, end)
    rows = fetch_audio_rows(conn, start, end, args.limit)
    conn.close()

    truncated = with_audio > args.limit
    if truncated:
        print(f"[prosody] NOTE: {with_audio} clips have audio but --limit={args.limit}; "
              f"processing newest {args.limit}. Coverage truncated.", file=sys.stderr)

    print(f"[prosody] Processing {len(rows)} clips for period '{args.period}'...",
          file=sys.stderr)

    clips = []
    skipped_no_voice = 0
    for i, row in enumerate(rows):
        if i and i % 50 == 0:
            print(f"[prosody]   {i}/{len(rows)} clips analyzed...", file=sys.stderr)
        feat = analyze_clip(row["audio"])
        if feat is None:
            skipped_no_voice += 1
        wpm, pause_ratio = compute_tempo(row)
        date = row["timestamp"].split(" ")[0] if row["timestamp"] else "unknown"
        clips.append({
            "feat": feat,
            "lang": normalize_language(row),
            "date": date,
            "wpm": wpm,
            "pause_ratio": pause_ratio,
            "speech": row["speechDuration"] or 0,
        })

    report = build_report(period_label, total, with_audio, len(rows),
                          skipped_no_voice, truncated, args.limit, clips)

    if args.format == "json":
        result = json.dumps(report, ensure_ascii=False, indent=2)
    else:
        result = format_text(report)

    if args.output:
        Path(args.output).parent.mkdir(parents=True, exist_ok=True)
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(result)
        print(f"Output written to {args.output}", file=sys.stderr)
    else:
        print(result)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Extract Wispr Flow dictation data from local SQLite database.

Usage:
    python3 extract_wispr.py [--period today|week|month|YYYY-MM-DD|YYYY-MM-DD:YYYY-MM-DD] [--mode all|technical|soft|trends|mental] [--format json|markdown] [--output PATH]

Modes:
    all       - Full analysis (default)
    technical - Coding/work patterns, app usage, productivity metrics
    soft      - Communication patterns, language use, interpersonal context
    trends    - Dictation volume, frequency changes, time-of-day patterns
    mental    - Sentiment indicators, energy proxies, activity pattern changes

Period shortcuts:
    today     - Current day
    yesterday - Previous day
    week      - Last 7 days
    month     - Last 30 days
    YYYY-MM-DD           - Specific date
    YYYY-MM-DD:YYYY-MM-DD - Date range
"""

import sqlite3
import json
import argparse
import os
import sys
from datetime import datetime, timedelta
from collections import Counter, defaultdict
from pathlib import Path

DB_PATH = os.path.expanduser("~/Library/Application Support/Wispr Flow/flow.sqlite")

APP_CATEGORIES = {
    "coding": [
        "com.googlecode.iterm2", "com.microsoft.VSCode",
        "com.exafunction.windsurf", "dev.zed.Zed",
        "com.cursor.Cursor", "com.apple.Terminal"
    ],
    "ai_tools": [
        "com.openai.chat", "com.anthropic.claudefordesktop",
        "ai.perplexity.comet", "com.openai.atlas"
    ],
    "communication": [
        "ru.keepcoder.Telegram", "com.apple.MobileSMS",
        "com.tinyspeck.slackmacgap", "us.zoom.xos"
    ],
    "writing": [
        "md.obsidian", "com.apple.Notes",
        "com.google.Chrome", "company.thebrowser.Browser"
    ],
}

def get_category(app_id):
    for cat, apps in APP_CATEGORIES.items():
        if app_id in apps:
            return cat
    return "other"

def parse_period(period_str):
    """Return (start_datetime_str, end_datetime_str) for SQL WHERE clause."""
    now = datetime.now()
    today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)

    if period_str == "today":
        start = today_start
        end = now
    elif period_str == "yesterday":
        start = today_start - timedelta(days=1)
        end = today_start
    elif period_str == "week":
        start = today_start - timedelta(days=7)
        end = now
    elif period_str == "month":
        start = today_start - timedelta(days=30)
        end = now
    elif ":" in period_str:
        parts = period_str.split(":")
        start = datetime.strptime(parts[0], "%Y-%m-%d")
        end = datetime.strptime(parts[1], "%Y-%m-%d").replace(hour=23, minute=59, second=59)
    else:
        start = datetime.strptime(period_str, "%Y-%m-%d")
        end = start.replace(hour=23, minute=59, second=59)

    return start.strftime("%Y-%m-%d %H:%M:%S"), end.strftime("%Y-%m-%d %H:%M:%S")

def extract_data(period="today"):
    """Extract dictation data for the given period."""
    start, end = parse_period(period)

    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()

    cursor.execute("""
        SELECT transcriptEntityId, formattedText, asrText, timestamp,
               app, url, numWords, duration, language, detectedLanguage,
               speechDuration
        FROM History
        WHERE isArchived = 0
          AND timestamp >= ? AND timestamp <= ?
          AND formattedText IS NOT NULL AND formattedText != ''
        ORDER BY timestamp ASC
    """, (start, end))

    rows = [dict(r) for r in cursor.fetchall()]
    conn.close()
    return rows, start, end

def compute_stats(rows):
    """Compute quantitative statistics from dictation data."""
    if not rows:
        return {"total_dictations": 0, "message": "No dictations found for this period."}

    total_words = sum(r["numWords"] or 0 for r in rows)
    total_duration = sum(r["duration"] or 0 for r in rows)
    total_speech = sum(r["speechDuration"] or 0 for r in rows)

    # App distribution
    app_counts = Counter(r["app"] for r in rows if r["app"])
    app_words = defaultdict(int)
    for r in rows:
        if r["app"]:
            app_words[r["app"]] += r["numWords"] or 0

    # Category distribution
    cat_counts = Counter(get_category(r["app"]) for r in rows if r["app"])
    cat_words = defaultdict(int)
    for r in rows:
        if r["app"]:
            cat_words[get_category(r["app"])] += r["numWords"] or 0

    # Language distribution
    lang_counts = Counter(r["detectedLanguage"] or r["language"] or "unknown" for r in rows)

    # Hourly distribution
    hourly = Counter()
    for r in rows:
        if r["timestamp"]:
            try:
                ts = r["timestamp"].split(" ")[1].split(":")[0]
                hourly[int(ts)] += 1
            except (IndexError, ValueError):
                pass

    # Average words per dictation
    avg_words = total_words / len(rows) if rows else 0

    # Longest dictations
    longest = sorted(rows, key=lambda r: r["numWords"] or 0, reverse=True)[:5]

    return {
        "total_dictations": len(rows),
        "total_words": total_words,
        "total_duration_seconds": round(total_duration, 1),
        "total_speech_seconds": round(total_speech or 0, 1),
        "avg_words_per_dictation": round(avg_words, 1),
        "app_distribution": dict(app_counts.most_common(15)),
        "app_words": dict(sorted(app_words.items(), key=lambda x: x[1], reverse=True)[:15]),
        "category_distribution": dict(cat_counts.most_common()),
        "category_words": dict(sorted(cat_words.items(), key=lambda x: x[1], reverse=True)),
        "language_distribution": dict(lang_counts.most_common()),
        "hourly_distribution": dict(sorted(hourly.items())),
        "longest_dictations": [
            {"text": d["formattedText"][:200], "words": d["numWords"], "app": d["app"]}
            for d in longest
        ],
    }

def compute_trends(rows):
    """Compute daily trends for multi-day periods."""
    daily = defaultdict(lambda: {"count": 0, "words": 0, "duration": 0, "apps": Counter()})
    for r in rows:
        if r["timestamp"]:
            day = r["timestamp"].split(" ")[0]
            daily[day]["count"] += 1
            daily[day]["words"] += r["numWords"] or 0
            daily[day]["duration"] += r["duration"] or 0
            if r["app"]:
                daily[day]["apps"][get_category(r["app"])] += 1

    trend_data = []
    for day in sorted(daily.keys()):
        d = daily[day]
        trend_data.append({
            "date": day,
            "dictations": d["count"],
            "words": d["words"],
            "duration_min": round(d["duration"] / 60, 1),
            "top_category": d["apps"].most_common(1)[0][0] if d["apps"] else "none",
        })
    return trend_data

def extract_texts_by_mode(rows, mode):
    """Extract relevant text samples for LLM analysis based on mode."""
    if mode == "technical":
        filtered = [r for r in rows if get_category(r["app"]) in ("coding", "ai_tools")]
    elif mode == "soft":
        filtered = [r for r in rows if get_category(r["app"]) in ("communication", "writing")]
    elif mode == "mental":
        filtered = rows  # all text relevant for mental health
    else:
        filtered = rows

    # Sample strategy: take all if <100, otherwise proportional sample
    if len(filtered) <= 100:
        sample = filtered
    else:
        step = len(filtered) / 100
        sample = [filtered[int(i * step)] for i in range(100)]

    return [
        {
            "text": r["formattedText"],
            "timestamp": r["timestamp"],
            "app": r["app"],
            "words": r["numWords"],
            "category": get_category(r["app"]) if r["app"] else "unknown",
            "language": r["detectedLanguage"] or r["language"] or "unknown",
        }
        for r in sample if r["formattedText"]
    ]

def format_markdown_stats(stats, period, trends=None):
    """Format statistics as markdown."""
    lines = []
    lines.append(f"## Wispr Flow Analytics: {period}")
    lines.append("")
    lines.append(f"- **Total dictations**: {stats['total_dictations']}")
    lines.append(f"- **Total words**: {stats['total_words']:,}")
    lines.append(f"- **Total speech time**: {stats.get('total_speech_seconds', 0) / 60:.1f} min")
    lines.append(f"- **Avg words/dictation**: {stats['avg_words_per_dictation']}")
    lines.append("")

    # Category breakdown
    lines.append("### Activity by Category")
    for cat, count in stats.get("category_distribution", {}).items():
        words = stats.get("category_words", {}).get(cat, 0)
        lines.append(f"- **{cat}**: {count} dictations, {words:,} words")
    lines.append("")

    # Language
    lines.append("### Language Distribution")
    for lang, count in stats.get("language_distribution", {}).items():
        lines.append(f"- {lang}: {count}")
    lines.append("")

    # Hourly
    if stats.get("hourly_distribution"):
        lines.append("### Hourly Activity")
        for hour, count in stats["hourly_distribution"].items():
            bar = "#" * min(count, 40)
            lines.append(f"- {hour:02d}:00 {bar} ({count})")
        lines.append("")

    # Trends
    if trends:
        lines.append("### Daily Trends")
        lines.append("| Date | Dictations | Words | Duration (min) | Top Category |")
        lines.append("|------|-----------|-------|----------------|--------------|")
        for t in trends:
            lines.append(f"| {t['date']} | {t['dictations']} | {t['words']:,} | {t['duration_min']} | {t['top_category']} |")
        lines.append("")

    # Top apps
    lines.append("### Top Apps")
    for app, count in list(stats.get("app_distribution", {}).items())[:10]:
        words = stats.get("app_words", {}).get(app, 0)
        short_name = app.split(".")[-1] if app else "unknown"
        lines.append(f"- **{short_name}**: {count} dictations, {words:,} words")

    return "\n".join(lines)

def main():
    parser = argparse.ArgumentParser(description="Extract Wispr Flow dictation data")
    parser.add_argument("--period", default="today",
                        help="today, yesterday, week, month, YYYY-MM-DD, or YYYY-MM-DD:YYYY-MM-DD")
    parser.add_argument("--mode", default="all",
                        choices=["all", "technical", "soft", "trends", "mental"],
                        help="Analysis mode")
    parser.add_argument("--format", default="json", choices=["json", "markdown"],
                        help="Output format")
    parser.add_argument("--output", default=None,
                        help="Output file path (default: stdout)")
    parser.add_argument("--texts-only", action="store_true",
                        help="Output only text samples for LLM analysis")

    args = parser.parse_args()

    rows, start, end = extract_data(args.period)
    period_label = f"{start} to {end}" if args.period not in ("today", "yesterday", "week", "month") else args.period

    if args.texts_only:
        texts = extract_texts_by_mode(rows, args.mode)
        result = json.dumps(texts, ensure_ascii=False, indent=2)
    elif args.format == "json":
        stats = compute_stats(rows)
        trends = compute_trends(rows) if len(set(r["timestamp"].split(" ")[0] for r in rows if r["timestamp"])) > 1 else None
        texts = extract_texts_by_mode(rows, args.mode)
        result = json.dumps({
            "period": period_label,
            "mode": args.mode,
            "stats": stats,
            "trends": trends,
            "text_samples": texts[:50],  # limit for JSON output
        }, ensure_ascii=False, indent=2)
    else:
        stats = compute_stats(rows)
        trends = compute_trends(rows) if len(set(r["timestamp"].split(" ")[0] for r in rows if r["timestamp"])) > 1 else None
        result = format_markdown_stats(stats, period_label, trends)

    if args.output:
        Path(args.output).parent.mkdir(parents=True, exist_ok=True)
        with open(args.output, "w") as f:
            f.write(result)
        print(f"Output written to {args.output}", file=sys.stderr)
    else:
        print(result)

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Wispr Flow Dictionary Manager.

Manages dictionary entries (recognition terms and replacement rules)
in the Wispr Flow SQLite database. Supports export/import via JSON
for version control and cross-machine sync.

IMPORTANT: Wispr Flow must be quit before any write operations.
Read operations (export, list, suggest) are safe while running.

Usage:
    python3 wispr_dictionary.py export [--output PATH]
    python3 wispr_dictionary.py import [--input PATH] [--dry-run]
    python3 wispr_dictionary.py add "phrase" ["replacement"]
    python3 wispr_dictionary.py remove "phrase"
    python3 wispr_dictionary.py list [--filter PATTERN]
    python3 wispr_dictionary.py suggest [--days 30] [--min-freq 3]
    python3 wispr_dictionary.py propose [--days 30] [--min-freq 3] [--format text]
    python3 wispr_dictionary.py check
"""

import sqlite3
import json
import argparse
import os
import re
import sys
import subprocess
import uuid
import difflib
from collections import Counter
from datetime import datetime, timedelta, timezone
from pathlib import Path

DB_PATH = os.path.expanduser("~/Library/Application Support/Wispr Flow/flow.sqlite")
DEFAULT_DICT_PATH = os.path.expanduser("~/ai_projects/claude-skills/wispr-analytics/data/dictionary.json")


def get_db(readonly=True):
    if not os.path.exists(DB_PATH):
        print("Error: Wispr Flow database not found", file=sys.stderr)
        sys.exit(1)
    uri = f"file:{DB_PATH}?mode=ro" if readonly else DB_PATH
    if readonly:
        conn = sqlite3.connect(uri, uri=True)
    else:
        conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


def is_wispr_running():
    try:
        result = subprocess.run(["pgrep", "-f", "Wispr Flow"], capture_output=True, text=True)
        return result.returncode == 0
    except Exception:
        return False


def require_wispr_stopped():
    if is_wispr_running():
        print("Error: Wispr Flow is running. Quit it first (Cmd+Q) to avoid database corruption.", file=sys.stderr)
        sys.exit(1)


def cmd_export(args):
    conn = get_db(readonly=True)
    rows = conn.execute(
        "SELECT id, phrase, replacement, manualEntry, frequencyUsed, "
        "createdAt, modifiedAt, isDeleted, isSnippet, isStarred "
        "FROM Dictionary WHERE isDeleted = 0 ORDER BY phrase"
    ).fetchall()
    conn.close()

    entries = []
    for r in rows:
        entry = {
            "phrase": r["phrase"],
            "replacement": r["replacement"],
            "is_snippet": bool(r["isSnippet"]),
            "is_starred": bool(r["isStarred"]),
            "manual": bool(r["manualEntry"]),
            "frequency": r["frequencyUsed"],
        }
        if entry["replacement"] is None:
            del entry["replacement"]
        if not entry["is_snippet"]:
            del entry["is_snippet"]
        if not entry["is_starred"]:
            del entry["is_starred"]
        if not entry["manual"]:
            del entry["manual"]
        if entry["frequency"] == 0:
            del entry["frequency"]
        entries.append(entry)

    output = {
        "version": 1,
        "exported_at": datetime.now(timezone.utc).isoformat() + "Z",
        "count": len(entries),
        "entries": entries,
    }

    out_path = args.output or DEFAULT_DICT_PATH
    os.makedirs(os.path.dirname(out_path), exist_ok=True)
    with open(out_path, "w", encoding="utf-8") as f:
        json.dump(output, f, indent=2, ensure_ascii=False)

    print(f"Exported {len(entries)} entries to {out_path}")


def cmd_import(args):
    require_wispr_stopped()

    in_path = args.input or DEFAULT_DICT_PATH
    if not os.path.exists(in_path):
        print(f"Error: {in_path} not found", file=sys.stderr)
        sys.exit(1)

    with open(in_path, "r", encoding="utf-8") as f:
        data = json.load(f)

    entries = data.get("entries", [])
    conn = get_db(readonly=False)

    existing = {
        row[0]
        for row in conn.execute(
            "SELECT phrase FROM Dictionary WHERE isDeleted = 0"
        ).fetchall()
    }

    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
    added = 0
    skipped = 0

    for entry in entries:
        phrase = entry["phrase"]
        if phrase in existing:
            skipped += 1
            continue

        if args.dry_run:
            replacement = entry.get("replacement", "")
            label = f"{phrase} → {replacement}" if replacement else phrase
            print(f"  Would add: {label}")
            added += 1
            continue

        entry_id = str(uuid.uuid4())
        replacement = entry.get("replacement") or phrase
        is_snippet = 1 if entry.get("is_snippet") else 0
        is_starred = 1 if entry.get("is_starred") else 0
        manual = 1 if entry.get("manual", True) else 0

        conn.execute(
            "INSERT INTO Dictionary (id, phrase, replacement, teamDictionaryId, "
            "lastUsed, frequencyUsed, remoteFrequencyUsed, manualEntry, "
            "createdAt, modifiedAt, isDeleted, source, isSnippet, observedSource, isStarred) "
            "VALUES (?, ?, ?, '00000000-0000-0000-0000-000000000000', "
            "NULL, 0, 0, ?, ?, ?, 0, 'manual', ?, NULL, ?)",
            (entry_id, phrase, replacement, manual, now, now, is_snippet, is_starred),
        )
        added += 1

    if not args.dry_run:
        conn.commit()

    conn.close()

    prefix = "[DRY RUN] " if args.dry_run else ""
    print(f"{prefix}Added: {added}, Skipped (existing): {skipped}")

    if not args.dry_run and added > 0:
        restart_wispr()


def restart_wispr():
    print("Starting Wispr Flow...")
    try:
        subprocess.Popen(["open", "-a", "Wispr Flow"],
                         stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        print("Wispr Flow restarted.")
    except Exception:
        print("Warning: Could not restart Wispr Flow. Start it manually.", file=sys.stderr)


def cmd_add(args):
    require_wispr_stopped()
    conn = get_db(readonly=False)

    existing = conn.execute(
        "SELECT COUNT(*) FROM Dictionary WHERE phrase = ? AND isDeleted = 0",
        (args.phrase,),
    ).fetchone()[0]

    if existing > 0:
        print(f"'{args.phrase}' already exists in dictionary")
        conn.close()
        return

    entry_id = str(uuid.uuid4())
    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
    replacement = args.replacement if args.replacement else args.phrase

    conn.execute(
        "INSERT INTO Dictionary (id, phrase, replacement, teamDictionaryId, "
        "lastUsed, frequencyUsed, remoteFrequencyUsed, manualEntry, "
        "createdAt, modifiedAt, isDeleted, source, isSnippet, observedSource, isStarred) "
        "VALUES (?, ?, ?, '00000000-0000-0000-0000-000000000000', "
        "NULL, 0, 0, 1, ?, ?, 0, 'manual', 0, NULL, 0)",
        (entry_id, args.phrase, replacement, now, now),
    )
    conn.commit()
    conn.close()

    label = f"'{args.phrase}' → '{replacement}'" if replacement else f"'{args.phrase}'"
    print(f"Added: {label}")


def cmd_remove(args):
    require_wispr_stopped()
    conn = get_db(readonly=False)
    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

    result = conn.execute(
        "UPDATE Dictionary SET isDeleted = 1, modifiedAt = ? WHERE phrase = ? AND isDeleted = 0",
        (now, args.phrase),
    )
    conn.commit()

    if result.rowcount > 0:
        print(f"Removed: '{args.phrase}'")
    else:
        print(f"Not found: '{args.phrase}'")
    conn.close()


def cmd_list(args):
    conn = get_db(readonly=True)

    query = "SELECT phrase, replacement, frequencyUsed, isSnippet FROM Dictionary WHERE isDeleted = 0"
    params = []
    if args.filter:
        query += " AND (phrase LIKE ? OR replacement LIKE ?)"
        params = [f"%{args.filter}%", f"%{args.filter}%"]
    query += " ORDER BY phrase"

    rows = conn.execute(query, params).fetchall()
    conn.close()

    for r in rows:
        phrase = r["phrase"]
        replacement = r["replacement"] or ""
        freq = r["frequencyUsed"]
        snippet = " [snippet]" if r["isSnippet"] else ""
        freq_str = f" (used {freq}x)" if freq > 0 else ""

        if replacement and replacement != phrase:
            print(f"  {phrase} → {replacement}{freq_str}{snippet}")
        else:
            print(f"  {phrase} [vocab]{freq_str}{snippet}")

    print(f"\n{len(rows)} entries")


def get_existing_phrases(conn):
    """Return set of lowercased existing dictionary phrases (and replacements)."""
    existing = set()
    for row in conn.execute(
        "SELECT phrase, replacement FROM Dictionary WHERE isDeleted = 0"
    ).fetchall():
        if row[0]:
            existing.add(row[0].strip().lower())
        if row[1]:
            existing.add(row[1].strip().lower())
    return existing


def find_mishears(conn, cutoff, existing):
    """Find recurring ASR mishears (asrText vs formattedText diffs).

    Returns a list of (asr, fmt, freq) tuples sorted by frequency desc.
    Shared by both `suggest` and `propose`.
    """
    rows = conn.execute(
        "SELECT asrText, formattedText FROM History "
        "WHERE timestamp > ? AND asrText IS NOT NULL AND formattedText IS NOT NULL "
        "AND asrText != formattedText",
        (cutoff,),
    ).fetchall()

    corrections = {}
    for r in rows:
        asr = r["asrText"].strip()
        fmt = r["formattedText"].strip()

        if len(asr) > 100 or len(fmt) > 100:
            continue
        if asr.lower() == fmt.lower():
            continue
        if asr.lower() in existing:
            continue

        ratio = difflib.SequenceMatcher(None, asr.lower(), fmt.lower()).ratio()
        if ratio > 0.6 and ratio < 1.0:
            key = (asr, fmt)
            corrections[key] = corrections.get(key, 0) + 1

    return sorted(
        ((asr, fmt, freq) for (asr, fmt), freq in corrections.items()),
        key=lambda x: -x[2],
    )


def cmd_suggest(args):
    """Analyze dictation history for potential dictionary additions (mishears only)."""
    conn = get_db(readonly=True)

    days = args.days or 30
    min_freq = args.min_freq or 3
    cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).strftime("%Y-%m-%d")

    existing = get_existing_phrases(conn)
    mishears = find_mishears(conn, cutoff, existing)
    conn.close()

    print(f"Suggested dictionary additions (last {days} days, min {min_freq} occurrences):\n")

    count = 0
    for asr, fmt, freq in mishears:
        if freq >= min_freq:
            print(f"  {asr} → {fmt}  ({freq}x)")
            count += 1

    if count == 0:
        print("  No suggestions found. Try lowering --min-freq or increasing --days.")
    else:
        print(f"\n{count} suggestions. Add with: python3 wispr_dictionary.py add \"phrase\" \"replacement\"")


# ---------------------------------------------------------------------------
# propose: human-style review that proposes snippets, replacement rules, vocab
# ---------------------------------------------------------------------------

URL_RE = re.compile(r"https?://[^\s<>\"')]+", re.IGNORECASE)
EMAIL_RE = re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b")
PHONE_RE = re.compile(r"(?<!\w)\+?\d[\d\s().-]{7,}\d(?!\w)")
# Capitalized / technical tokens (incl. CamelCase, ALLCAPS acronyms). Latin only
# to avoid splitting Cyrillic sentence-initial words.
TECH_RE = re.compile(r"\b(?:[A-Z][a-z]+(?:[A-Z][a-z]+)+|[A-Z]{2,}[a-z]*|[A-Z][a-z]+\.[a-z]+)\b")


_TRIGGER_SKIP = {
    "https", "http", "www", "com", "org", "net", "io", "co", "in", "of",
    "the", "me", "my", "a", "an", "to", "is", "it", "do", "we",
}


def _slug_trigger(text):
    """Build a short, memorable `My X` trigger phrase for a snippet expansion.

    For URLs/emails the service/domain name is the distinctive part; for
    boilerplate sentences use the first couple of content words.
    """
    tokens = re.findall(r"[A-Za-z0-9]+", text)
    content = [t for t in tokens if t.lower() not in _TRIGGER_SKIP and len(t) > 1]
    if not content:
        content = tokens[:2]
    if not content:
        return "My snippet"
    return "My " + " ".join(content[:2]).lower()


def _normalize_boiler(text):
    """Normalize text for boilerplate frequency counting (whitespace/case)."""
    return re.sub(r"\s+", " ", text.strip()).lower()


def find_snippet_candidates(rows, existing, min_freq):
    """Detect recurring URLs/emails/phones and repeated boilerplate sentences."""
    contacts = Counter()       # URL/email/phone -> count (dedup per dictation)
    boiler = Counter()         # normalized full short dictation -> count
    boiler_display = {}        # normalized -> first-seen original text

    for r in rows:
        fmt = (r["formattedText"] or "").strip()
        if not fmt:
            continue

        seen = set()
        for rx in (URL_RE, EMAIL_RE, PHONE_RE):
            for m in rx.findall(fmt):
                m = m.strip().rstrip(".,);:")
                if len(m) < 6:
                    continue
                if m.lower() in existing or m in seen:
                    continue
                seen.add(m)
                contacts[m] += 1

        # Boilerplate: whole dictations repeated verbatim (intros, sign-offs,
        # bios, standard prompts). Require >= 8 words so trivial fillers like
        # "Okay let's do it." don't surface; cap length to avoid huge one-offs.
        words = fmt.split()
        if 8 <= len(words) <= 60:
            norm = _normalize_boiler(fmt)
            if norm in existing:
                continue
            boiler[norm] += 1
            boiler_display.setdefault(norm, fmt)

    candidates = []
    for value, freq in contacts.most_common():
        if freq >= min_freq:
            candidates.append({
                "kind": "contact",
                "trigger": _slug_trigger(value),
                "expansion": value,
                "freq": freq,
            })
    for norm, freq in boiler.most_common():
        if freq >= max(min_freq, 2):
            text = boiler_display[norm]
            candidates.append({
                "kind": "boilerplate",
                "trigger": _slug_trigger(text),
                "expansion": text,
                "freq": freq,
            })
    return candidates


def find_vocab_candidates(rows, existing, min_freq):
    """Frequent capitalized/technical terms that may be mis-recognized."""
    counts = Counter()
    STOP = {
        "I", "The", "This", "That", "And", "But", "For", "You", "We", "It",
        "So", "If", "Or", "My", "No", "Yes", "OK", "Okay", "Also", "Then",
    }
    for r in rows:
        fmt = (r["formattedText"] or "")
        for tok in TECH_RE.findall(fmt):
            if tok in STOP or len(tok) < 3:
                continue
            if tok.lower() in existing:
                continue
            counts[tok] += 1

    return [
        {"term": term, "freq": freq}
        for term, freq in counts.most_common(40)
        if freq >= min_freq
    ]


def _add_cmd(phrase, replacement=None):
    """Render a ready-to-run add command line for a proposal."""
    p = phrase.replace('"', '\\"')
    if replacement is None:
        return f'python3 wispr_dictionary.py add "{p}"'
    r = replacement.replace('"', '\\"')
    return f'python3 wispr_dictionary.py add "{p}" "{r}"'


def cmd_propose(args):
    """Review recent dictation logs and propose dictionary additions.

    Three categories (read-only, safe while Wispr runs):
      1. Snippet candidates    -- recurring URLs/emails/phones + boilerplate
      2. Replacement-rule cand -- recurring ASR mishears (shared with suggest)
      3. Vocab candidates      -- frequent capitalized/technical terms
    """
    conn = get_db(readonly=True)

    days = args.days or 30
    min_freq = args.min_freq or 3
    cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).strftime("%Y-%m-%d")

    existing = get_existing_phrases(conn)

    rows = conn.execute(
        "SELECT formattedText FROM History "
        "WHERE timestamp > ? AND formattedText IS NOT NULL AND formattedText != ''",
        (cutoff,),
    ).fetchall()

    snippets = find_snippet_candidates(rows, existing, min_freq)
    mishears = [
        {"asr": asr, "fmt": fmt, "freq": freq}
        for asr, fmt, freq in find_mishears(conn, cutoff, existing)
        if freq >= min_freq
    ]
    vocab = find_vocab_candidates(rows, existing, min_freq)
    conn.close()

    if args.format == "json":
        out = {
            "days": days,
            "min_freq": min_freq,
            "snippet_candidates": snippets,
            "replacement_candidates": mishears,
            "vocab_candidates": vocab,
        }
        print(json.dumps(out, indent=2, ensure_ascii=False))
        return

    total = len(snippets) + len(mishears) + len(vocab)
    print(f"Dictionary proposals from the last {days} days (min {min_freq} occurrences)")
    print(f"Analyzed {len(rows)} dictations. {total} proposals.\n")

    # --- Snippets (highest leverage) ---
    print("=" * 70)
    print("SNIPPET CANDIDATES  (text expansion -- the highest-leverage lever)")
    print("=" * 70)
    if not snippets:
        print("  none found.\n")
    else:
        for c in snippets:
            tag = "URL/contact" if c["kind"] == "contact" else "boilerplate"
            exp = c["expansion"]
            preview = exp if len(exp) <= 80 else exp[:77] + "..."
            print(f"  [{tag}] used {c['freq']}x")
            print(f"    trigger:   {c['trigger']}")
            print(f"    expansion: {preview}")
            print(f"    add:       {_add_cmd(c['trigger'], exp)}")
            print()

    # --- Replacement rules ---
    print("=" * 70)
    print("REPLACEMENT-RULE CANDIDATES  (recurring ASR mishears)")
    print("=" * 70)
    if not mishears:
        print("  none found.\n")
    else:
        for m in mishears:
            print(f"  {m['asr']} → {m['fmt']}  ({m['freq']}x)")
            print(f"    add: {_add_cmd(m['asr'], m['fmt'])}")
            print()

    # --- Vocab ---
    print("=" * 70)
    print("VOCAB CANDIDATES  (frequent technical terms -- teach recognition)")
    print("=" * 70)
    if not vocab:
        print("  none found.\n")
    else:
        for v in vocab:
            print(f"  {v['term']}  ({v['freq']}x)")
            print(f"    add: {_add_cmd(v['term'])}")
            print()

    if total == 0:
        print("No proposals. Try lowering --min-freq or increasing --days.")
    else:
        print("Review, then (after quitting Wispr Flow) run the `add` lines you approve,")
        print("followed by `python3 wispr_dictionary.py check` and restart Wispr.")


def cmd_check(args):
    """Check database health and dictionary stats."""
    conn = get_db(readonly=True)

    integrity = conn.execute("PRAGMA integrity_check").fetchone()[0]

    total = conn.execute("SELECT COUNT(*) FROM Dictionary").fetchone()[0]
    active = conn.execute("SELECT COUNT(*) FROM Dictionary WHERE isDeleted = 0").fetchone()[0]
    manual = conn.execute("SELECT COUNT(*) FROM Dictionary WHERE isDeleted = 0 AND manualEntry = 1").fetchone()[0]
    snippets = conn.execute("SELECT COUNT(*) FROM Dictionary WHERE isDeleted = 0 AND isSnippet = 1").fetchone()[0]
    with_replacement = conn.execute(
        "SELECT COUNT(*) FROM Dictionary WHERE isDeleted = 0 AND replacement IS NOT NULL AND replacement != ''"
    ).fetchone()[0]

    conn.close()

    running = "YES ⚠️  (quit before writing)" if is_wispr_running() else "no"

    print(f"Database: {DB_PATH}")
    print(f"Integrity: {integrity}")
    print(f"Wispr running: {running}")
    print(f"Dictionary: {active} active / {total} total")
    print(f"  Manual entries: {manual}")
    print(f"  With replacements: {with_replacement}")
    print(f"  Snippets: {snippets}")
    print(f"  Auto-learned: {active - manual}")


def main():
    parser = argparse.ArgumentParser(description="Wispr Flow Dictionary Manager")
    sub = parser.add_subparsers(dest="command", required=True)

    p_export = sub.add_parser("export", help="Export dictionary to JSON")
    p_export.add_argument("--output", "-o", help=f"Output path (default: {DEFAULT_DICT_PATH})")

    p_import = sub.add_parser("import", help="Import dictionary from JSON")
    p_import.add_argument("--input", "-i", help=f"Input path (default: {DEFAULT_DICT_PATH})")
    p_import.add_argument("--dry-run", action="store_true", help="Show what would be added")

    p_add = sub.add_parser("add", help="Add a single entry")
    p_add.add_argument("phrase", help="The phrase to recognize")
    p_add.add_argument("replacement", nargs="?", help="Optional replacement text")

    p_remove = sub.add_parser("remove", help="Remove an entry (soft delete)")
    p_remove.add_argument("phrase", help="The phrase to remove")

    p_list = sub.add_parser("list", help="List dictionary entries")
    p_list.add_argument("--filter", "-f", help="Filter by phrase or replacement")

    p_suggest = sub.add_parser("suggest", help="Suggest new entries from dictation history")
    p_suggest.add_argument("--days", type=int, default=30, help="Days of history to analyze")
    p_suggest.add_argument("--min-freq", type=int, default=3, help="Minimum correction frequency")

    p_propose = sub.add_parser(
        "propose",
        help="Propose snippets, replacement rules, and vocab from dictation logs",
    )
    p_propose.add_argument("--days", type=int, default=30, help="Days of history to analyze")
    p_propose.add_argument("--min-freq", type=int, default=3, help="Minimum occurrence frequency")
    p_propose.add_argument("--format", choices=["text", "json"], default="text", help="Output format")

    p_check = sub.add_parser("check", help="Check database health")

    args = parser.parse_args()

    commands = {
        "export": cmd_export,
        "import": cmd_import,
        "add": cmd_add,
        "remove": cmd_remove,
        "list": cmd_list,
        "suggest": cmd_suggest,
        "propose": cmd_propose,
        "check": cmd_check,
    }
    commands[args.command](args)


if __name__ == "__main__":
    main()