python -m venv venv
source venv/bin/activate
pip install openai tweepy schedule pandas httpx
export OPENAI_API_KEY="sk-your-key"

import json
import logging
from typing import Any, Dict, List, Optional

from openai import AsyncOpenAI

logger = logging.getLogger(__name__)


class ContentGenerator:
    """Generate platform-specific social media content with brand voice."""

    CHAR_LIMITS: Dict[str, int] = {
        "twitter": 280,
        "linkedin": 3000,
        "instagram": 2200,
    }

    def __init__(self, model: str = "gpt-4o", api_key: Optional[str] = None):
        self.client = AsyncOpenAI(api_key=api_key)
        self.model = model

    async def generate_post(
        self,
        topic: str,
        platform: str,
        brand_voice: str,
        tone: str = "professional",
        include_hashtags: bool = True,
    ) -> Dict[str, Any]:
        """Generate a single platform-specific post."""
        limit = self.CHAR_LIMITS.get(platform, 280)
        prompt = f"""Create a {platform} post about: {topic}

Brand voice: {brand_voice}
Tone: {tone}
Character limit: {limit}
Include hashtags: {include_hashtags}

Platform-specific rules:
- Twitter: Concise, punchy, use threads for longer content
- LinkedIn: Professional, thought leadership, include CTA
- Instagram: Visual-focused, emoji-friendly, strong hashtag game

Return JSON:
{{
    "content": "the post text",
    "hashtags": ["list of hashtags"],
    "best_posting_time": "suggested time",
    "engagement_prediction": "high|medium|low"
}}"""

        try:
            response = await self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "You are an expert social media copywriter."},
                    {"role": "user", "content": prompt},
                ],
                temperature=0.7,
            )
            return json.loads(response.choices[0].message.content)
        except (json.JSONDecodeError, IndexError) as e:
            logger.warning(f"Failed to parse response: {e}")
            return {
                "content": response.choices[0].message.content,
                "hashtags": [],
                "best_posting_time": "09:00",
                "engagement_prediction": "medium",
            }
        except Exception as e:
            logger.error(f"Content generation failed: {e}")
            raise

    async def generate_content_calendar(
        self,
        topics: List[str],
        brand_voice: str,
        days: int = 7,
    ) -> List[Dict[str, Any]]:
        """Generate a content calendar spanning multiple days."""
        calendar = []
        platforms = ["twitter", "linkedin", "instagram"]
        for day in range(days):
            for i, topic in enumerate(topics[: len(platforms)]):
                platform = platforms[i % len(platforms)]
                post = await self.generate_post(topic, platform, brand_voice)
                calendar.append({"day": day + 1, "platform": platform, "topic": topic, **post})
        return calendar

from typing import Dict, List, Optional


class BrandVoiceManager:
    """Manage brand voice profiles for consistent content generation."""

    def __init__(self):
        self.voices: Dict[str, str] = {}

    def add_voice(self, brand_name: str, description: str) -> None:
        self.voices[brand_name] = description

    def get_voice(self, brand_name: str) -> str:
        return self.voices.get(brand_name, "Professional and helpful")

    def create_voice_profile(
        self,
        brand_name: str,
        values: List[str],
        tone: str,
        do_s: List[str],
        dont_s: List[str],
    ) -> str:
        profile = f"""Brand: {brand_name}
Tone: {tone}
Values: {', '.join(values)}
Do: {'; '.join(do_s)}
Don't: {'; '.join(dont_s)}"""
        self.voices[brand_name] = profile
        return profile

from typing import Dict, Optional
import asyncio

import tweepy


class TwitterClient:
    """Async Twitter/X API client for posting and engagement."""

    def __init__(
        self,
        api_key: str,
        api_secret: str,
        access_token: str,
        access_secret: str,
    ):
        self.client = tweepy.Client(
            consumer_key=api_key,
            consumer_secret=api_secret,
            access_token=access_token,
            access_token_secret=access_secret,
        )

    async def post_tweet(self, content: str, hashtags: Optional[List[str]] = None) -> Dict:
        text = content
        if hashtags:
            tag_str = " ".join(f"#{t.strip('#')}" for t in hashtags[:5])
            if len(text) + len(tag_str) + 1 <= 280:
                text = f"{text}\n\n{tag_str}"
        try:
            response = self.client.create_tweet(text=text)
            return {"success": True, "tweet_id": response.data["id"]}
        except Exception as e:
            logger.error(f"Tweet failed: {e}")
            return {"success": False, "error": str(e)}

    async def get_mentions(self, count: int = 10) -> list:
        try:
            me = self.client.get_me()
            response = self.client.get_users_mentions(
                me.data.id,
                max_results=count,
                tweet_fields=["text", "created_at"],
            )
            return [
                {"id": m.id, "text": m.text, "created_at": m.created_at}
                for m in (response.data or [])
            ]
        except Exception as e:
            logger.error(f"Mentions fetch failed: {e}")
            return []


import schedule
import time
from typing import Callable
from datetime import datetime


class PostScheduler:
    """Schedule and manage social media posts."""

    def __init__(self):
        self.scheduled_posts: list = []

    def schedule_post(self, post: Dict, post_time: str, callback: Callable) -> None:
        self.scheduled_posts.append({
            "post": post,
            "time": post_time,
            "callback": callback,
            "status": "scheduled",
        })
        schedule.every().day.at(post_time).do(self._execute_post, post, callback)

    def _execute_post(self, post: Dict, callback: Callable) -> None:
        result = callback(post.get("content", ""), post.get("hashtags", []))
        post["status"] = "posted" if result.get("success") else "failed"
        post["result"] = result

    def get_optimal_times(self) -> list:
        return ["09:00", "12:00", "17:00", "20:00"]

    def run_pending(self) -> None:
        schedule.run_pending()

    def get_scheduled(self) -> list:
        return self.scheduled_posts

from typing import Any, Dict, List, Optional
from datetime import datetime

import pandas as pd


class AnalyticsTracker:
    """Track and analyze social media post performance."""

    def __init__(self):
        self.posts: List[Dict] = []

    def record_post(self, post: Dict, platform: str) -> None:
        post["platform"] = platform
        post["posted_at"] = datetime.now().isoformat()
        post["metrics"] = {"likes": 0, "shares": 0, "comments": 0, "impressions": 0}
        self.posts.append(post)

    def update_metrics(self, post_id: str, metrics: Dict) -> None:
        for post in self.posts:
            if post.get("id") == post_id:
                post["metrics"].update(metrics)
                break

    def get_performance(self, platform: Optional[str] = None) -> Dict:
        posts = [p for p in self.posts if not platform or p.get("platform") == platform]
        if not posts:
            return {"total_posts": 0}
        total_metrics = {"likes": 0, "shares": 0, "comments": 0, "impressions": 0}
        for post in posts:
            for key in total_metrics:
                total_metrics[key] += post.get("metrics", {}).get(key, 0)
        avg_metrics = {k: v / len(posts) for k, v in total_metrics.items()}
        engagement_rate = (
            (total_metrics["likes"] + total_metrics["shares"] + total_metrics["comments"])
            / max(total_metrics["impressions"], 1)
            * 100
        )
        return {
            "total_posts": len(posts),
            "total_metrics": total_metrics,
            "avg_metrics": avg_metrics,
            "engagement_rate": engagement_rate,
        }


class SocialMediaAgent:
    """Orchestrate social media operations across platforms."""

    def __init__(self, model: str = "gpt-4o", api_key: Optional[str] = None):
        self.generator = ContentGenerator(model, api_key)
        self.brand_voice = BrandVoiceManager()
        self.analytics = AnalyticsTracker()
        self.scheduler = PostScheduler()

    async def create_post(
        self,
        topic: str,
        platform: str,
        brand: str = "default",
        tone: str = "professional",
    ) -> Dict:
        voice = self.brand_voice.get_voice(brand)
        return await self.generator.generate_post(topic, platform, voice, tone)

    async def schedule_post(self, post: Dict, platform: str, time: str) -> None:
        self.scheduler.schedule_post(post, time, lambda c, h: {"success": True})
        self.analytics.record_post(post, platform)

    async def generate_calendar(
        self,
        topics: List[str],
        brand: str = "default",
        days: int = 7,
    ) -> List[Dict]:
        voice = self.brand_voice.get_voice(brand)
        return await self.generator.generate_content_calendar(topics, voice, days)

    def get_analytics(self, platform: Optional[str] = None) -> Dict:
        return self.analytics.get_performance(platform)