score-draft

Score Draft — LinkedIn Audience Simulator Skill

Core skill: Tag a draft with topics, analyze hook strength per segment, predict per-segment engagement, calculate overall 0-100 score, and provide detailed reasoning.

Input

{
  "draft_text": "string (1-500 words of a LinkedIn draft post)",
  "track_for_calibration": "boolean (optional, default false)"
}

Output

{
  "overall_score": 78,
  "confidence": "medium",
  "predicted_metrics": {
    "reactions": 32,
    "comments": 8,
    "reach": 2100,
    "follower_growth": 2
  },
  "per_segment": [
    {
      "segment": "Technical Leaders",
      "size": 420,
      "predicted_reactions": 22,
      "predicted_comments": 6,
      "hook_strength": 85,
      "topic_affinity": 92,
      "reasoning": "Strong technical topic match..."
    }
  ],
  "topics": ["AI/ML", "Infrastructure"],
  "hook_type": "question",
  "top_recommendations": [
    "Add a business outcome sentence...",
    "Consider leading with the result..."
  ],
  "cold_start_note": "You have 3 posts scored. After 10+, calibration will kick in...",
  "reasoning_summary": "78/100 — Technical Leaders' Territory...",
  "scoring_metadata": {
    "hook_strength_method": "regex",
    "calibration_applied": false,
    "baseline_engagement": 28,
    "segment_sizes": {...}
  }
}

Implementation

Step 1: Database Connection & Validation

import os
import glob
import sqlite3
import json
import re
import math
from datetime import datetime, timedelta

def find_feeds_db():
    """See _shared-preamble.md for full implementation."""
    workspace_dbs = glob.glob('/sessions/*/mnt/CoWork Os/LinkedIn Feed Tracker/feeds.db')
    if workspace_dbs:
        return workspace_dbs[0]

    data_dir_file = os.path.expanduser('~/.linkedin-feed-tracker/data-dir.txt')
    if os.path.exists(data_dir_file):
        with open(data_dir_file, 'r') as f:
            data_dir = f.read().strip()
        db_path = os.path.join(data_dir, 'feeds.db')
        if os.path.exists(db_path):
            return db_path

    raise FileNotFoundError("Cannot find feeds.db")

def get_db_connection():
    db_path = find_feeds_db()
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    return conn

# Initialize database
conn = get_db_connection()
try:
    data = verify_plugin1_data(conn)
except Exception as e:
    return {"error": f"Plugin 1 error: {e}"}

# Check if setup has been run
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM audience_segments")
segment_count = cursor.fetchone()[0]
if segment_count == 0:
    return {
        "error": "Audience Simulator not set up. Run /setup-audience-simulator first.",
        "status": "not_initialized"
    }

Step 2: Topic Assignment

def assign_topics(draft_text):
    """
    Assign 1-3 topics to a draft using regex patterns + Claude for ambiguity.

    Returns: list of (topic_name, confidence, method)
    """
    topic_patterns = {
        'AI/ML': r'(ai\b|artificial intelligence|machine learning|neural|llm|gpt|transformer|nlp|deep learning|algorithm)',
        'Leadership': r'(leadership|leading|team building|culture|management|hiring|mentorship|organizational)',
        'Startup': r'(startup|founder|founding|fundraising|series a|seed|scaling|early stage|mvp)',
        'Remote Work': r'(remote|distributed|async|work from home|wfh|async first)',
        'Product': r'(product|roadmap|feature|user|ux|usability|product-market fit)',
        'Growth': r'(growth|metrics|kpi|acquisition|scaling revenue|expansion|growth hacking)',
        'Hiring': r'(hiring|recruiting|recruitment|talent|candidate|hiring process)',
        'Strategy': r'(strategy|strategic|planning|vision|direction|roadmap)',
        'Cloud': r'(cloud|kubernetes|serverless|docker|aws|gcp|azure)',
        'DevOps': r'(devops|infrastructure|deployment|ci/cd|monitoring|observability)',
        'Open Source': r'(open source|github|repo|library|package)',
        'Thought Leadership': r'(insights?|perspective|take|thoughts?|opinion\b)',
        'Fundraising': r'(fundraising|vc|venture capital|investor|funding|capital raise)',
        'Revenue': r'(revenue|pricing|price point|monetization|revenue model)',
        'Ethics': r'(ethics|privacy|security|responsible|bias|compliance)',
        'Founder Stories': r'(founder journey|founder story|started|bootstrapped)',
        'Team Culture': r'(culture|team values|company values|diversity|inclusion)',
    }

    draft_lower = draft_text.lower()
    found_topics = []

    for topic, pattern in topic_patterns.items():
        if re.search(pattern, draft_lower):
            found_topics.append((topic, 1.0, 'regex'))

    # If no matches or too many, use Claude for clarification
    if len(found_topics) == 0:
        # Claude: "What are the main themes in this post? Pick 1-3."
        # For MVP, default to 'Thought Leadership'
        found_topics = [('Thought Leadership', 0.8, 'claude')]

    elif len(found_topics) > 3:
        # Keep top 3 by salience (topic patterns later in dict have higher salience)
        found_topics = found_topics[-3:]

    return found_topics[:3]

Step 3: Hook Detection

def detect_hook_type(draft_text):
    """
    Identify the primary hook type.

    Returns: (hook_type, confidence)
    """
    draft_stripped = draft_text.strip()

    # Question hook
    if draft_stripped.endswith('?') and draft_text.count('?') >= 1:
        return ('question', 1.0)

    # Listicle hook
    if re.match(r'^\s*\d+\.', draft_text) or re.search(r'\n\s*\d+\.', draft_text):
        return ('listicle', 1.0)

    # Data-driven hook
    if re.search(r'\d+%|\d+ (million|thousand|hundred|people|users|companies|employees)', draft_text, re.IGNORECASE):
        return ('data-driven', 1.0)

    # Contrarian hook
    if re.search(r'\b(but|however|actually|wrong|myth|misconception|rethinking)\b', draft_text.lower()):
        return ('contrarian', 0.9)

    # Personal hook
    if re.search(r'\b(learned|failed|realized|mistake|regret|admit|confess)\b', draft_text.lower()):
        return ('personal', 0.9)

    # Call-to-action hook
    if re.search(r'\b(join|sign up|subscribe|register|apply|download|get started|learn more)\b', draft_text.lower()):
        return ('call-to-action', 0.9)

    # Default to story
    return ('story', 0.6)

def get_hook_strength_by_segment(hook_type, segment_name):
    """
    Get hook appeal strength for a segment.

    Returns: 0-100 score
    """
    hook_affinities = {
        'Executives': {
            'story': 85, 'contrarian': 80, 'data-driven': 75,
            'question': 60, 'personal': 75, 'listicle': 40, 'call-to-action': 20
        },
        'Technical Leaders': {
            'data-driven': 90, 'question': 85, 'contrarian': 60,
            'story': 65, 'personal': 50, 'listicle': 70, 'call-to-action': 30
        },
        'Growth & Sales': {
            'data-driven': 85, 'story': 82, 'contrarian': 70,
            'question': 65, 'listicle': 75, 'personal': 60, 'call-to-action': 70
        },
        'General Audience': {
            'story': 70, 'question': 65, 'data-driven': 60,
            'personal': 65, 'listicle': 60, 'contrarian': 50, 'call-to-action': 40
        },
    }

    return hook_affinities.get(segment_name, {}).get(hook_type, 50)

Step 4: Affinity Lookup

def get_topic_affinities_for_segment(conn, segment_id, topics):
    """
    Get affinity scores for topics in this segment.

    Returns: dict {topic: affinity_score}
    """
    cursor = conn.cursor()

    topic_names = [t[0] for t in topics]  # Extract topic names
    placeholders = ','.join(['?' ] * len(topic_names))

    cursor.execute(f'''
        SELECT topic_name, affinity_score
        FROM segment_topic_affinity
        WHERE segment_id = ? AND topic_name IN ({placeholders})
    ''', [segment_id] + topic_names)

    affinities = {}
    for row in cursor.fetchall():
        affinities[row['topic_name']] = row['affinity_score']

    # If no affinity data, use heuristic
    for topic in topic_names:
        if topic not in affinities:
            affinities[topic] = 0.5  # Default 50% affinity

    return affinities

Step 5: Calibration Lookup

def get_calibration_coefficients(conn, segment_name, topic_name):
    """
    Get calibration slope and intercept for segment × topic.

    Returns: dict {slope, intercept, prediction_count}
    """
    cursor = conn.cursor()

    cursor.execute('''
        SELECT calibration_slope, calibration_intercept, prediction_count
        FROM prediction_accuracy
        WHERE segment_name = ? AND topic_name = ?
    ''', (segment_name, topic_name))

    row = cursor.fetchone()
    if row:
        return {
            'slope': row['calibration_slope'],
            'intercept': row['calibration_intercept'],
            'prediction_count': row['prediction_count']
        }
    else:
        return {
            'slope': 1.0,
            'intercept': 0.0,
            'prediction_count': 0
        }

def apply_calibration(predicted_score, calibration):
    """Apply linear calibration: calibrated = (predicted × slope) + intercept."""
    return max(0, (predicted_score * calibration['slope']) + calibration['intercept'])

Step 6: Baseline Engagement Calculation

def get_baseline_engagement(conn):
    """
    Calculate average reactions from own_posts.

    Returns: float
    """
    cursor = conn.cursor()

    cursor.execute('SELECT AVG(likes) as avg_likes FROM own_posts')
    row = cursor.fetchone()

    return float(row['avg_likes']) if row['avg_likes'] else 20.0  # Default fallback

Step 7: Per-Segment Prediction

def predict_per_segment(conn, draft_text, segment, hook_type, hook_strength, topics, affinities):
    """
    Predict engagement for a specific segment.

    Uses Claude reasoning + calibration.

    Returns: dict {segment, predicted_reactions, predicted_comments, predicted_reach, reasoning, ...}
    """
    cursor = conn.cursor()

    # Get segment size
    cursor.execute('SELECT estimated_size FROM audience_segments WHERE name = ?', (segment,))
    seg_row = cursor.fetchone()
    segment_size = seg_row['estimated_size'] if seg_row else 100

    # Get activity rate
    cursor.execute('''
        SELECT COUNT(*) as active_count
        FROM connection_profiles
        WHERE segment_id = (SELECT id FROM audience_segments WHERE name = ?)
        AND activity_status = 'active'
    ''', (segment,))
    active_row = cursor.fetchone()
    active_count = active_row['active_count'] if active_row else 0
    activity_rate = (active_count / segment_size) if segment_size > 0 else 0.5

    # Baseline
    baseline = get_baseline_engagement(conn)

    # Calculate topic affinity average
    avg_affinity = sum(affinities.values()) / len(affinities) if affinities else 0.5

    # Claude reasoning (simplified for MVP)
    prompt = f"""You are a LinkedIn engagement prediction AI.

Predict how many reactions this post will get from {segment}:

Draft: "{draft_text[:200]}..."

Segment: {segment} (~{segment_size} people, {activity_rate*100:.0f}% active)
Topics: {list(affinities.keys())} (avg affinity: {avg_affinity*100:.0f}%)
Hook type: {hook_type} (strength: {hook_strength}/100 for this segment)
Baseline reactions: {baseline:.0f} (your average)

Reason about:
1. Topic relevance to this segment
2. Hook strength for this segment
3. Overall engagement likelihood

Return JSON:
{{
    "predicted_reactions": <int>,
    "predicted_comments": <int>,
    "predicted_reach": <int>,
    "hook_analysis": "<1 sentence on hook fit>",
    "affinity_analysis": "<1 sentence on topic relevance>"
}}"""

    # For MVP: use heuristic instead of Claude (no API call)
    predicted_reactions = calculate_predicted_reactions(
        baseline, hook_strength, avg_affinity, segment_size, activity_rate
    )
    predicted_comments = int(predicted_reactions * 0.25)
    predicted_reach = int(predicted_reactions * 70)

    # Apply calibration
    primary_topic = topics[0][0] if topics else 'Thought Leadership'
    calib = get_calibration_coefficients(conn, segment, primary_topic)
    if calib['prediction_count'] >= 10:
        predicted_reactions = apply_calibration(predicted_reactions, calib)
        predicted_comments = apply_calibration(predicted_comments, calib)

    reasoning = f"Topic affinity is {avg_affinity*100:.0f}% with {segment}. Hook strength: {hook_strength}/100."

    return {
        'segment': segment,
        'size': segment_size,
        'predicted_reactions': int(predicted_reactions),
        'predicted_comments': int(predicted_comments),
        'predicted_reach': int(predicted_reach),
        'hook_strength': hook_strength,
        'topic_affinity': int(avg_affinity * 100),
        'activity_rate': activity_rate,
        'reasoning': reasoning
    }

def calculate_predicted_reactions(baseline, hook_strength, affinity, segment_size, activity_rate):
    """
    Heuristic prediction formula.

    predicted_reactions = baseline × (hook_strength/100) × affinity × activity_rate × (segment_size / total_size)
    """
    segment_portion = segment_size / 2000.0  # Assume ~2000 total connections
    multiplier = (hook_strength / 100.0) * affinity * activity_rate
    return baseline * multiplier * segment_portion

Step 8: Overall Score Calculation

def calculate_overall_score(per_segment_predictions):
    """
    Aggregate per-segment predictions to overall 0-100 score.

    Formula:
    overall_score = sum(segment_score × weight) / sum(weights)

    where:
    - segment_score = (hook_strength + topic_affinity) / 2
    - weight = segment_size × activity_rate
    """
    weighted_sum = 0.0
    weight_sum = 0.0

    for pred in per_segment_predictions:
        # Combine hook and topic into segment score
        segment_score = (pred['hook_strength'] + pred['topic_affinity']) / 2.0

        # Weight by segment size and activity
        weight = pred['size'] * pred['activity_rate']

        weighted_sum += segment_score * weight
        weight_sum += weight

    if weight_sum == 0:
        return 50  # Default

    overall_score = weighted_sum / weight_sum
    return int(min(100, max(0, overall_score)))

Step 9: Confidence Determination

def determine_confidence(conn, topics):
    """
    Determine confidence level based on calibration data.

    Low: < 10 predictions per segment/topic
    Medium: 10-49 predictions
    High: 50+ predictions
    """
    cursor = conn.cursor()

    primary_topic = topics[0][0] if topics else 'Thought Leadership'

    cursor.execute('''
        SELECT AVG(prediction_count) as avg_count
        FROM prediction_accuracy
        WHERE topic_name = ?
    ''', (primary_topic,))

    row = cursor.fetchone()
    avg_count = row['avg_count'] if row and row['avg_count'] else 0

    if avg_count < 10:
        return 'low'
    elif avg_count < 50:
        return 'medium'
    else:
        return 'high'

Step 10: Recommendations Generation

def generate_recommendations(per_segment_predictions, topics, hook_type, draft_text):
    """
    Generate actionable recommendations for improving the draft.

    Returns: list of strings
    """
    recommendations = []

    # Check for weak segments
    weak_segments = [p for p in per_segment_predictions if p['topic_affinity'] < 30]
    if weak_segments:
        segment_names = ', '.join([p['segment'] for p in weak_segments])
        recommendations.append(
            f"Add business context. {segment_names} don't engage with pure technical content. "
            "Include 'why this matters' or 'business outcome'."
        )

    # Check for missing hook
    if hook_type == 'story' and '?' not in draft_text:
        recommendations.append(
            "End with a question to drive comments. Strongest hooks for your audience: "
            "questions (especially for Technical Leaders) or personal stories (for Executives)."
        )

    # Check for topic breadth
    if len(topics) == 1:
        recommendations.append(
            "Consider mentioning a secondary topic (e.g., product + growth, or culture + hiring). "
            "This broadens appeal to more segments."
        )

    # Check for call-to-action
    if 'sign up' not in draft_text.lower() and 'join' not in draft_text.lower():
        recommendations.append(
            "Strong draft! Consider adding a soft CTA if you want audience interaction. "
            "For now, the question hook will drive engagement naturally."
        )

    return recommendations[:3]  # Top 3 recommendations

Step 11: Store Prediction (Optional)

def store_prediction_for_calibration(conn, draft_text, overall_score, confidence, per_segment_predictions):
    """
    Optionally store the prediction for later calibration.

    Called when track_for_calibration = True.
    """
    cursor = conn.cursor()

    top_segments = [
        {
            'segment': p['segment'],
            'hook_strength': p['hook_strength'],
            'reasoning': p['reasoning']
        }
        for p in per_segment_predictions[:3]
    ]

    reasoning_json = json.dumps(top_segments)

    cursor.execute('''
        INSERT INTO predictions (
            draft_text, overall_score, confidence, predicted_reactions,
            predicted_comments, predicted_reach, top_segments_json, scored_at
        )
        VALUES (?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
    ''', (
        draft_text,
        overall_score,
        confidence,
        sum(p['predicted_reactions'] for p in per_segment_predictions),
        sum(p['predicted_comments'] for p in per_segment_predictions),
        sum(p['predicted_reach'] for p in per_segment_predictions),
        reasoning_json
    ))

    conn.commit()

Step 12: Build Final Response

def build_response(overall_score, confidence, per_segment_predictions, topics, hook_type, recommendations):
    """Assemble the final JSON response."""

    total_reactions = sum(p['predicted_reactions'] for p in per_segment_predictions)
    total_comments = sum(p['predicted_comments'] for p in per_segment_predictions)
    total_reach = sum(p['predicted_reach'] for p in per_segment_predictions)

    # Determine primary segment (highest predicted reactions)
    primary_segment = max(per_segment_predictions, key=lambda p: p['predicted_reactions'])['segment']

    # Build reasoning summary
    if overall_score >= 80:
        strength = "strong"
    elif overall_score >= 60:
        strength = "solid"
    else:
        strength = "weak"

    reasoning_summary = f"{overall_score}/100 — {primary_segment} Sweet Spot\n\n"
    reasoning_summary += f"This is a {strength} post for your audience. "
    reasoning_summary += f"It performs best with {primary_segment} "
    reasoning_summary += f"({per_segment_predictions[0]['hook_strength']}/100 hook strength). "

    return {
        'overall_score': overall_score,
        'confidence': confidence,
        'predicted_metrics': {
            'reactions': int(total_reactions),
            'comments': int(total_comments),
            'reach': int(total_reach),
            'follower_growth': int(total_reactions / 10)
        },
        'per_segment': per_segment_predictions,
        'topics': [t[0] for t in topics],
        'hook_type': hook_type,
        'top_recommendations': recommendations,
        'reasoning_summary': reasoning_summary,
        'cold_start_note': (
            "You have < 10 scored posts. After 10-15, calibration kicks in and predictions become much more accurate."
            if confidence == 'low' else
            "Predictions are becoming more accurate as you build calibration data."
        )
    }

Full Workflow

def score_draft(draft_text, track_for_calibration=False):
    """Main entry point for scoring a draft."""

    # Step 1: Validate
    if not draft_text or len(draft_text) < 10:
        return {'error': 'Draft too short (min 10 characters)'}

    conn = get_db_connection()

    try:
        # Step 2: Get segments
        cursor = conn.cursor()
        cursor.execute('SELECT id, name FROM audience_segments')
        segments = cursor.fetchall()

        if not segments:
            return {'error': 'No segments configured. Run /setup-audience-simulator first.'}

        # Step 3: Assign topics
        topics = assign_topics(draft_text)

        # Step 4: Detect hook
        hook_type, hook_confidence = detect_hook_type(draft_text)

        # Step 5: Per-segment prediction
        per_segment = []
        for segment_row in segments:
            segment_name = segment_row['name']
            segment_id = segment_row['id']

            hook_strength = get_hook_strength_by_segment(hook_type, segment_name)
            affinities = get_topic_affinities_for_segment(conn, segment_id, topics)

            pred = predict_per_segment(
                conn, draft_text, segment_name, hook_type, hook_strength, topics, affinities
            )
            per_segment.append(pred)

        # Step 6: Overall score
        overall_score = calculate_overall_score(per_segment)

        # Step 7: Confidence
        confidence = determine_confidence(conn, topics)

        # Step 8: Recommendations
        recommendations = generate_recommendations(per_segment, topics, hook_type, draft_text)

        # Step 9: Store if needed
        if track_for_calibration:
            store_prediction_for_calibration(conn, draft_text, overall_score, confidence, per_segment)

        # Step 10: Build response
        response = build_response(overall_score, confidence, per_segment, topics, hook_type, recommendations)

        return response

    except Exception as e:
        return {'error': f'Scoring error: {str(e)}'}

    finally:
        conn.close()

Example Output

{
  "overall_score": 78,
  "confidence": "medium",
  "predicted_metrics": {
    "reactions": 32,
    "comments": 8,
    "reach": 2240,
    "follower_growth": 3
  },
  "per_segment": [
    {
      "segment": "Technical Leaders",
      "size": 420,
      "predicted_reactions": 22,
      "predicted_comments": 6,
      "predicted_reach": 1540,
      "hook_strength": 85,
      "topic_affinity": 92,
      "activity_rate": 0.68,
      "reasoning": "Topic affinity is 92% with Technical Leaders. Hook strength: 85/100."
    },
    {
      "segment": "Growth & Sales",
      "size": 380,
      "predicted_reactions": 6,
      "predicted_comments": 1,
      "predicted_reach": 420,
      "hook_strength": 35,
      "topic_affinity": 18,
      "activity_rate": 0.75,
      "reasoning": "Topic affinity is 18% with Growth & Sales. Hook strength: 35/100."
    },
    {
      "segment": "Executives",
      "size": 150,
      "predicted_reactions": 3,
      "predicted_comments": 0,
      "predicted_reach": 210,
      "hook_strength": 28,
      "topic_affinity": 5,
      "activity_rate": 0.72,
      "reasoning": "Topic affinity is 5% with Executives. Hook strength: 28/100."
    },
    {
      "segment": "General Audience",
      "size": 1016,
      "predicted_reactions": 1,
      "predicted_comments": 1,
      "predicted_reach": 70,
      "hook_strength": 15,
      "topic_affinity": 25,
      "activity_rate": 0.45,
      "reasoning": "Topic affinity is 25% with General Audience. Hook strength: 15/100."
    }
  ],
  "topics": ["AI/ML", "Cloud"],
  "hook_type": "question",
  "top_recommendations": [
    "Add a business outcome sentence (e.g., 'This enables us to serve 50% more users on the same hardware'). Executives need the 'why'.",
    "Consider the audience: This is hyper-technical. If you want broader reach, add a 1-2 sentence business context."
  ],
  "reasoning_summary": "78/100 — Technical Leaders Sweet Spot\n\nThis is a strong post for your audience. It performs best with Technical Leaders (85/100 hook strength).",
  "cold_start_note": "You have 3 scored posts. After 10-15, calibration kicks in and predictions become much more accurate."
}