Social media manager that integrates X, Instagram, LinkedIn, and uses local AI tools for content generation.

social.py

#open source code generated with Claude by Andy
#[email protected]
#The Postiz project is now AGPL.
#this code is public domain

import requests
import json
import time
from datetime import datetime
import schedule
from linkedin import linkedin  # python-linkedin package
from instagram_private_api import Client  # instagram-private-api
import ollama
import subprocess
from PIL import Image
import io
import os

class SocialMediaManager:
    def __init__(self, config_path="config.json"):
        with open(config_path) as f:
            self.config = json.load(f)
        
        self.setup_x()
        self.setup_instagram()
        self.setup_linkedin()
        self.monthly_post_count = {
            'x': 0,
            'instagram': 0,
            'linkedin': 0
        }
        self.reset_date = self._get_next_reset_date()

    def setup_x(self):
        """Initialize X API"""
        self.x_api_base = "https://api.twitter.com/2"
        self.x_headers = {
            "Authorization": f"Bearer {self.config['x']['bearer_token']}",
            "Content-Type": "application/json"
        }

    def setup_instagram(self):
        """Initialize Instagram API"""
        self.instagram_api = Client(
            self.config['instagram']['username'],
            self.config['instagram']['password']
        )

    def setup_linkedin(self):
        """Initialize LinkedIn API"""
        self.linkedin_api = linkedin.LinkedInApplication(
            token=self.config['linkedin']['access_token']
        )

    def _get_next_reset_date(self):
        """Get first day of next month for limit resets"""
        now = datetime.now()
        if now.month == 12:
            return datetime(now.year + 1, 1, 1)
        return datetime(now.year, now.month + 1, 1)

    def generate_text_content(self, prompt):
        """Generate content using Ollama with Llama 2"""
        try:
            # Using Ollama's API to generate text with llama2
            response = ollama.generate(
                model='llama2',
                prompt=prompt,
                temperature=0.7,
                max_tokens=200
            )
            return response['response']
        except Exception as e:
            print(f"Error generating text: {str(e)}")
            return None

    def generate_image(self, prompt):
        """Generate image using Flux ML"""
        try:
            # Note: This is a placeholder for Flux implementation
            # You would need to implement the specific Flux ML integration
            # based on their API or local setup
            image_path = "generated_image.png"
            
            # Placeholder for Flux image generation command
            subprocess.run([
                "flux", "generate",
                "--prompt", prompt,
                "--output", image_path
            ])
            
            return image_path
        except Exception as e:
            print(f"Error generating image: {str(e)}")
            return None

    def post_to_x(self, content, image_path=None):
        """Post to X with optional image"""
        if self.monthly_post_count['x'] >= 500:
            print("X monthly post limit reached")
            return None

        endpoint = f"{self.x_api_base}/tweets"
        payload = {"text": content}

        try:
            if image_path:
                # Upload media first
                media_id = self.upload_media_to_x(image_path)
                if media_id:
                    payload["media"] = {"media_ids": [media_id]}

            response = requests.post(
                endpoint,
                headers=self.x_headers,
                json=payload
            )

            if response.status_code == 201:
                self.monthly_post_count['x'] += 1
                print(f"Posted to X: {content}")
                return response.json()
            else:
                print(f"Error posting to X: {response.status_code}")
                return None

        except Exception as e:
            print(f"Exception posting to X: {str(e)}")
            return None

    def post_to_instagram(self, content, image_path):
        """Post to Instagram"""
        try:
            # Upload photo with caption
            self.instagram_api.post_photo(
                image_path,
                caption=content
            )
            print(f"Posted to Instagram: {content}")
            return True
        except Exception as e:
            print(f"Error posting to Instagram: {str(e)}")
            return None

    def post_to_linkedin(self, content, image_path=None):
        """Post to LinkedIn"""
        try:
            if image_path:
                # Upload image and get asset ID
                asset = self.linkedin_api.upload_image(image_path)
                
                # Post with image
                response = self.linkedin_api.submit_share(
                    comment=content,
                    image_ids=[asset['id']]
                )
            else:
                # Text-only post
                response = self.linkedin_api.submit_share(
                    comment=content
                )
            
            print(f"Posted to LinkedIn: {content}")
            return response
        except Exception as e:
            print(f"Error posting to LinkedIn: {str(e)}")
            return None

    def create_and_post_content(self, text_prompt, image_prompt=None, platforms=None):
        """Generate and post content to specified platforms"""
        if platforms is None:
            platforms = ['x', 'instagram', 'linkedin']

        # Generate text content
        content = self.generate_text_content(text_prompt)
        if not content:
            return False

        # Generate image if prompt provided
        image_path = None
        if image_prompt:
            image_path = self.generate_image(image_prompt)

        # Post to each platform
        results = {}
        if 'x' in platforms:
            results['x'] = self.post_to_x(content, image_path)
        
        if 'instagram' in platforms and image_path:  # Instagram requires an image
            results['instagram'] = self.post_to_instagram(content, image_path)
            
        if 'linkedin' in platforms:
            results['linkedin'] = self.post_to_linkedin(content, image_path)

        return results

    def schedule_multi_platform_posts(self, prompts, times, platforms=None):
        """Schedule posts across multiple platforms"""
        if platforms is None:
            platforms = ['x', 'instagram', 'linkedin']

        for prompt, post_time in zip(prompts, times):
            schedule.every().day.at(post_time).do(
                self.create_and_post_content,
                text_prompt=prompt['text'],
                image_prompt=prompt.get('image'),
                platforms=platforms
            )

# Example usage
if __name__ == "__main__":
    manager = SocialMediaManager("config.json")
    
    # Example of a single multi-platform post
    text_prompt = "Write an engaging post about AI technology trends"
    image_prompt = "Create a futuristic visualization of AI networks"
    
    manager.create_and_post_content(
        text_prompt=text_prompt,
        image_prompt=image_prompt,
        platforms=['x', 'instagram', 'linkedin']
    )
    
    # Example of scheduled posts
    prompts = [
        {
            'text': "Write about machine learning best practices",
            'image': "Create an image of neural networks in action"
        },
        {
            'text': "Share tips about data science workflows",
            'image': "Visualize data science pipeline"
        }
    ]
    
    times = ["10:00", "15:00"]
    manager.schedule_multi_platform_posts(prompts, times)
    
    # Run the scheduler
    while True:
        schedule.run_pending()
        time.sleep(60)

Consider asking Claude to use instapy instead of instagram_private_api which seems like it doesnt exist by the name. I asked Claude that but ran out of tokens partially through.