performance-reviewer

name	performance-reviewer
description	Match posted variations to actual LinkedIn performance from Plugin 1. Extract engagement metrics and feed learnings into the hook_performance learning loop.
version	1.0.0

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