Menú
Core skill for scoring a draft post with per-segment predictions and reasoning
Deep analysis of a specific segment including roles, topics, and past performance
Display high-level summary of audience segments, sizes, topics, and activity
Score 3 content variations and rank them with comparative analysis
Review calibration metrics, accuracy trends, and model confidence levels
| name | score-draft |
| description | Core skill for scoring a draft post with per-segment predictions and reasoning |
| version | 1.0.0 |
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.
{
"draft_text": "string (1-500 words of a LinkedIn draft post)",
"track_for_calibration": "boolean (optional, default false)"
}
{
"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": {...}
}
}
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"
}
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]
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)
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
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'])
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
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
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)))
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'
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
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()
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."
)
}
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()
{
"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."
}