Automate Brand Storytelling 🚀

import datetime
from typing import List, Dict

def prepare_weighted_training_data(examples: List[Dict]) -> List[Dict]:
    """Weight recent examples higher to prevent brand drift"""
    now = datetime.datetime.now()
    weighted_examples = []
    
    for ex in examples:
        # Calculate age in months
        age_months = (now - ex['date']).days / 30
        
        # Exponential decay: recent = 1.0, 12 months old = 0.5, 24+ months = 0.25
        weight = max(0.25, 1.0 * (0.5 ** (age_months / 12)))
        
        # Duplicate recent examples based on weight
        num_copies = int(weight * 4)  # 4x for recent, 1x for old
        weighted_examples.extend([ex] * num_copies)
    
    return weighted_examples

# Usage
training_data = prepare_weighted_training_data(brand_examples)
# Recent content appears 4x more often in training

interface ChannelConfig {
  platform: string;
  max_length: number;
  tone_shift: string;
  formatting_rules: string[];
}

const CHANNEL_CONFIGS: Record<string, ChannelConfig> = {
  linkedin: {
    platform: 'LinkedIn',
    max_length: 3000,
    tone_shift: 'slightly more professional, use data/stats',
    formatting_rules: ['Use line breaks for readability', 'Bold key points', 'Add relevant hashtags (3-5)']
  },
  twitter: {
    platform: 'Twitter',
    max_length: 280,
    tone_shift: 'more casual, punchy, use thread format for longer ideas',
    formatting_rules: ['One idea per tweet', 'Use emojis sparingly', 'End with CTA or question']
  },
  blog: {
    platform: 'Blog',
    max_length: 10000,
    tone_shift: 'more depth, storytelling, educational',
    formatting_rules: ['H2/H3 headers', 'Code examples', 'Internal links']
  }
};

async function generateForChannel(
  brief: string,
  channel: string,
  brandDNA: any
): Promise<string> {
  const config = CHANNEL_CONFIGS[channel];
  
  const response = await openai.chat.completions.create({
    model: 'gpt-4',
    messages: [
      {
        role: 'system',
        content: `Generate content for ${config.platform}. Brand DNA: ${JSON.stringify(brandDNA)}. Tone shift: ${config.tone_shift}. Max length: ${config.max_length} chars.`
      },
      { role: 'user', content: brief }
    ]
  });
  
  return response.choices[0].message.content || '';
}

// Generate same brief for different channels
const linkedinPost = await generateForChannel(brief, 'linkedin', brandDNA);
const twitterThread = await generateForChannel(brief, 'twitter', brandDNA);

from typing import List, Dict
import re

class SensitiveTopicFilter:
    def __init__(self):
        # Keywords that trigger mandatory human review
        self.sensitive_keywords = [
            'layoff', 'firing', 'lawsuit', 'investigation',
            'controversy', 'scandal', 'apology', 'mistake',
            'data breach', 'security', 'privacy violation',
            'discrimination', 'harassment', 'complaint'
        ]
        
        # Competitors (never mention without approval)
        self.competitor_names = [
            'competitor_a', 'competitor_b', 'competitor_c'
        ]
    
    def requires_review(self, content: str) -> Dict:
        """Check if content needs human review before publishing"""
        content_lower = content.lower()
        
        # Check sensitive keywords
        found_sensitive = [
            kw for kw in self.sensitive_keywords 
            if kw in content_lower
        ]
        
        # Check competitor mentions
        found_competitors = [
            comp for comp in self.competitor_names
            if comp in content_lower
        ]
        
        needs_review = bool(found_sensitive or found_competitors)
        
        return {
            'needs_review': needs_review,
            'reason': 'sensitive_topic' if found_sensitive else 'competitor_mention' if found_competitors else None,
            'flagged_terms': found_sensitive + found_competitors,
            'action': 'BLOCK_AUTO_PUBLISH' if needs_review else 'ALLOW'
        }

# Usage
filter = SensitiveTopicFilter()
check = filter.requires_review(generated_content)

if check['needs_review']:
    # Send to review queue, don't auto-publish
    send_to_slack_review_channel(
        content=generated_content,
        reason=check['reason'],
        flagged_terms=check['flagged_terms']
    )
else:
    # Safe to auto-publish
    publish_to_cms(generated_content)

interface ComplianceCheck {
  needs_legal_review: boolean;
  flagged_claims: string[];
  risk_level: 'low' | 'medium' | 'high';
  review_type: string[];
}

class ComplianceValidator {
  private readonly CLAIM_PATTERNS = [
    /\d+%\s+(increase|decrease|improvement|growth)/gi,  // Performance claims
    /guaranteed|promise|ensure|100%/gi,  // Absolute claims
    /\$[\d,]+\s+(saved|earned|profit|return)/gi,  // Financial claims
    /(cure|treat|diagnose|prevent)\s+\w+/gi,  // Medical claims
  ];

  async validateContent(content: string): Promise<ComplianceCheck> {
    const flaggedClaims: string[] = [];
    const reviewTypes: Set<string> = new Set();

    // Check for unsubstantiated claims
    for (const pattern of this.CLAIM_PATTERNS) {
      const matches = content.match(pattern);
      if (matches) {
        flaggedClaims.push(...matches);
        
        if (pattern.source.includes('saved|earned')) {
          reviewTypes.add('financial_claims');
        }
        if (pattern.source.includes('cure|treat')) {
          reviewTypes.add('medical_claims');
        }
        if (pattern.source.includes('guaranteed')) {
          reviewTypes.add('absolute_claims');
        }
      }
    }

    // Risk assessment
    let riskLevel: 'low' | 'medium' | 'high' = 'low';
    if (flaggedClaims.length > 3) riskLevel = 'high';
    else if (flaggedClaims.length > 0) riskLevel = 'medium';

    return {
      needs_legal_review: flaggedClaims.length > 0,
      flagged_claims: flaggedClaims,
      risk_level: riskLevel,
      review_type: Array.from(reviewTypes),
    };
  }
}

// Usage
const validator = new ComplianceValidator();
const check = await validator.validateContent(generatedContent);

if (check.needs_legal_review) {
  // Route to appropriate review queue
  await sendToLegalReview({
    content: generatedContent,
    risk_level: check.risk_level,
    flagged_claims: check.flagged_claims,
    review_types: check.review_type,
  });
} else {
  // Safe to publish
  await publishContent(generatedContent);
}

import re
from collections import Counter
from typing import Dict, List

class SEOValidator:
    def __init__(self, target_keywords: List[str]):
        self.target_keywords = [kw.lower() for kw in target_keywords]
        self.max_keyword_density = 0.03  # 3% max density
    
    def calculate_keyword_density(self, content: str) -> Dict:
        """Check if keywords are used naturally, not stuffed"""
        words = re.findall(r'\w+', content.lower())
        total_words = len(words)
        
        keyword_counts = {}
        for keyword in self.target_keywords:
            # Count exact matches and variations
            keyword_words = keyword.split()
            count = 0
            
            # Sliding window to find multi-word keywords
            for i in range(len(words) - len(keyword_words) + 1):
                window = ' '.join(words[i:i+len(keyword_words)])
                if window == keyword:
                    count += 1
            
            density = count / total_words if total_words > 0 else 0
            keyword_counts[keyword] = {
                'count': count,
                'density': density,
                'is_stuffed': density > self.max_keyword_density
            }
        
        # Overall assessment
        is_keyword_stuffed = any(kw['is_stuffed'] for kw in keyword_counts.values())
        has_all_keywords = all(kw['count'] > 0 for kw in keyword_counts.values())
        
        return {
            'keyword_analysis': keyword_counts,
            'is_keyword_stuffed': is_keyword_stuffed,
            'has_all_keywords': has_all_keywords,
            'needs_revision': is_keyword_stuffed or not has_all_keywords,
            'total_words': total_words
        }
    
    def generate_seo_feedback(self, analysis: Dict) -> str:
        """Generate feedback for regeneration if needed"""
        feedback = []
        
        for keyword, data in analysis['keyword_analysis'].items():
            if data['is_stuffed']:
                feedback.append(
                    f"Keyword '{keyword}' appears {data['count']} times ({data['density']:.1%} density). "
                    f"Reduce to max 3% density. Use synonyms and natural variations."
                )
            elif data['count'] == 0:
                feedback.append(
                    f"Missing target keyword '{keyword}'. Include naturally in content."
                )
        
        return ' '.join(feedback) if feedback else 'SEO optimization looks good.'

# Usage
seo_validator = SEOValidator(
    target_keywords=['brand voice automation', 'AI content generation', 'marketing automation']
)

analysis = seo_validator.calculate_keyword_density(generated_content)

if analysis['needs_revision']:
    feedback = seo_validator.generate_seo_feedback(analysis)
    # Regenerate with SEO feedback
    regenerated = generate_content_with_feedback(brief, feedback)
else:
    # SEO looks good, proceed
    publish_content(generated_content)