performance-reviewer

Performance Reviewer Skill

Analyzes actual performance of posted variations. Links generated content to real engagement metrics and updates the learning loop.

Triggers

This skill runs when:

• User runs /review command
• Automatic review after user posts (hourly checks)
• Scheduled weekly review

How It Works

Step 1: Fetch Recent Own Posts

import sqlite3
import json
import difflib
from datetime import datetime, timedelta

def get_recent_own_posts(db_path, days=30):
    """Fetch user's own posts from Plugin 1 (last N days)"""
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()

    cursor.execute('''
        SELECT id, content_short, likes, comments, reposts, impressions,
               post_type, posted_at, content_hash
        FROM own_posts
        WHERE posted_at >= datetime('now', ? || ' days')
        ORDER BY posted_at DESC
    ''', (f'-{days}',))

    posts = [dict(row) for row in cursor.fetchall()]
    conn.close()
    return posts

Step 2: Match Variations to Posted Posts

Try multiple matching strategies:

def match_variations_to_posts(db_path, own_posts):
    """
    Link content_variations to actual own_posts.

    Strategies:
    1. Exact content_hash match
    2. Fuzzy text similarity (>85%)
    3. Manual review fallback
    """
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()

    # Get all variations with status='presented'
    cursor.execute('''
        SELECT id, idea_id, full_text, hook_type, self_score
        FROM content_variations
        WHERE posted_at IS NULL
        ORDER BY created_at DESC
    ''')
    variations = [dict(row) for row in cursor.fetchall()]
    conn.close()

    matches = []

    for var in variations:
        # Strategy 1: Try to find by content similarity
        var_text = var['full_text'][:200]  # First 200 chars

        for own_post in own_posts:
            own_text = own_post['content_short'][:200]

            # Similarity ratio
            ratio = difflib.SequenceMatcher(None, var_text, own_text).ratio()

            if ratio > 0.85:  # High confidence match
                matches.append({
                    'variation_id': var['id'],
                    'own_post_id': own_post['id'],
                    'confidence': ratio,
                    'own_post': own_post,
                    'variation': var
                })
                break

    return matches

Step 3: Extract Performance Data

def extract_performance(match):
    """Extract engagement metrics from matched posts"""
    own_post = match['own_post']

    return {
        'likes': own_post['likes'],
        'comments': own_post['comments'],
        'reposts': own_post['reposts'],
        'impressions': own_post['impressions'],
        'engagement': own_post['likes'] + own_post['comments'],
        'engagement_rate': (own_post['likes'] + own_post['comments']) / max(own_post['impressions'], 1),
        'posted_at': own_post['posted_at'],
        'post_type': own_post['post_type']
    }

def store_variation_performance(db_path, match, performance):
    """Store actual performance in content_variations"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    cursor.execute('''
        UPDATE content_variations
        SET actual_performance_json = ?, posted_at = ?
        WHERE id = ?
    ''', (
        json.dumps(performance),
        performance['posted_at'],
        match['variation_id']
    ))

    conn.commit()
    conn.close()

Step 4: Aggregate by Hook Type

def aggregate_hook_performance(db_path):
    """
    For each hook_type + topic combo, calculate aggregates.
    Update hook_performance table for next generation's learning.
    """
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()

    # Get all variations with actual performance
    cursor.execute('''
        SELECT cv.id, cv.hook_type, ci.topic, cv.actual_performance_json
        FROM content_variations cv
        JOIN content_ideas ci ON cv.idea_id = ci.id
        WHERE cv.actual_performance_json IS NOT NULL
    ''')

    variations = [dict(row) for row in cursor.fetchall()]

    # Group by hook_type + topic
    aggregates = {}
    for var in variations:
        key = (var['hook_type'], var['topic'])

        if key not in aggregates:
            aggregates[key] = {
                'hook_type': var['hook_type'],
                'topic': var['topic'],
                'total_likes': 0,
                'total_comments': 0,
                'count': 0
            }

        perf = json.loads(var['actual_performance_json'])
        aggregates[key]['total_likes'] += perf['likes']
        aggregates[key]['total_comments'] += perf['comments']
        aggregates[key]['count'] += 1

    # Update hook_performance table
    for (hook_type, topic), data in aggregates.items():
        avg_likes = data['total_likes'] / data['count']
        avg_comments = data['total_comments'] / data['count']
        confidence = min(1.0, data['count'] / 10.0)  # Max confidence at 10+ samples

        cursor.execute('''
            SELECT id FROM hook_performance
            WHERE hook_type = ? AND topic = ?
        ''', (hook_type, topic))
        exists = cursor.fetchone()

        if exists:
            # Update
            cursor.execute('''
                UPDATE hook_performance
                SET avg_likes = ?, avg_comments = ?, sample_count = ?, confidence = ?,
                    last_updated = CURRENT_TIMESTAMP
                WHERE hook_type = ? AND topic = ?
            ''', (avg_likes, avg_comments, data['count'], confidence, hook_type, topic))
        else:
            # Insert
            cursor.execute('''
                INSERT INTO hook_performance
                (hook_type, topic, avg_likes, avg_comments, sample_count, confidence)
                VALUES (?, ?, ?, ?, ?, ?)
            ''', (hook_type, topic, avg_likes, avg_comments, data['count'], confidence))

    conn.commit()
    conn.close()

Step 5: Generate Performance Summary

def generate_performance_summary(db_path, days=30):
    """
    Create a user-friendly summary of what worked and what didn't.
    """
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()

    # Overall stats
    cursor.execute('''
        SELECT COUNT(*) as posted_count,
               ROUND(AVG(CAST(json_extract(actual_performance_json, '$.likes') AS REAL)), 1) as avg_likes,
               ROUND(AVG(CAST(json_extract(actual_performance_json, '$.comments') AS REAL)), 1) as avg_comments
        FROM content_variations
        WHERE actual_performance_json IS NOT NULL
        AND posted_at >= datetime('now', ? || ' days')
    ''', (f'-{days}',))

    overall = dict(cursor.fetchone())

    # Top performers
    cursor.execute('''
        SELECT cv.id, ci.topic, cv.hook_type,
               CAST(json_extract(cv.actual_performance_json, '$.likes') AS INTEGER) as likes,
               CAST(json_extract(cv.actual_performance_json, '$.comments') AS INTEGER) as comments,
               cv.self_score
        FROM content_variations cv
        JOIN content_ideas ci ON cv.idea_id = ci.id
        WHERE cv.actual_performance_json IS NOT NULL
        AND cv.posted_at >= datetime('now', ? || ' days')
        ORDER BY (
            CAST(json_extract(cv.actual_performance_json, '$.likes') AS INTEGER) +
            CAST(json_extract(cv.actual_performance_json, '$.comments') AS INTEGER) * 3
        ) DESC
        LIMIT 3
    ''', (f'-{days}',))

    top_performers = [dict(row) for row in cursor.fetchall()]

    # Hook performance
    cursor.execute('''
        SELECT hook_type, topic, avg_likes, avg_comments, sample_count
        FROM hook_performance
        ORDER BY avg_likes DESC
    ''')

    hook_perf = [dict(row) for row in cursor.fetchall()]

    conn.close()

    summary = f"""
PERFORMANCE REVIEW — Last {days} Days

Variations posted: {overall['posted_count']}
Avg engagement: {overall['avg_likes']:.0f} likes, {overall['avg_comments']:.0f} comments

TOP PERFORMERS:
"""
    for i, post in enumerate(top_performers, 1):
        engagement = post['likes'] + post['comments']
        summary += f"""
{i}. {post['topic']} ({post['hook_type']} hook)
   {post['likes']} likes, {post['comments']} comments
   Self score: {post['self_score']:.2f}
"""

    summary += "\n\nHOOK PERFORMANCE (your audience):\n"
    for perf in hook_perf[:5]:
        summary += f"""
{perf['hook_type'].upper()}: {perf['avg_likes']:.0f} likes, {perf['avg_comments']:.0f} comments ({perf['sample_count']} posts)
"""

    return summary

Input

Called by /review command. Takes:

• days_back (default 30) — How many days to review

Output

Returns JSON:

{
  "status": "ok",
  "period": "30 days",
  "posted_count": 8,
  "avg_engagement": {"likes": 45, "comments": 12},
  "top_performers": [
    {
      "topic": "Leadership",
      "hook_type": "story",
      "likes": 89,
      "comments": 31
    }
  ],
  "hook_performance": {
    "story": {"avg_likes": 67, "avg_comments": 19, "samples": 3},
    "question": {"avg_likes": 45, "avg_comments": 11, "samples": 2}
  },
  "recommendations": "Keep writing stories about leadership..."
}

Database Schema

Reads:

• own_posts (Plugin 1)
• content_variations (to match and store performance)
• content_ideas (to get topic context)
• hook_performance (aggregates)

Writes:

• content_variations.actual_performance_json
• content_variations.posted_at
• hook_performance (aggregates)

Performance Matching

Fuzzy matching at >85% confidence is conservative. User can manually link posts if auto-matching fails.

Learning Loop Integration

Updated hook_performance is read by idea-generator and variation-writer to bias future generation toward what works.

Next Steps

User can:

• Run /generate to create new ideas (benefiting from learned patterns)
• Run /configure to adjust voice based on learnings
• Run /calendar to see performance timeline