Automated social media manager using chatGPT and generative AI
import logging
import openai
import os

from insta_gpt.utils import (
    download_image,
    load_topic,
    load_yaml_file,
    publish,
    update_topic
)

# directory path
DIR_PATH = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

# configurations
config = load_yaml_file(os.path.join(DIR_PATH, "config.yaml"))

# logging
logger = logging.getLogger(__name__)


def generate_post():
    topic = load_topic(os.path.join(DIR_PATH, config["topic"]["csv"]))
    try:
        logger.info(f"Generating content for topic {topic}")
        caption = generate_caption(topic)
        logger.info(f"Generated caption")
        image_url = generate_image(topic)
        logger.info(f"Generated image")
        publish(caption, image_url)
        update_topic(os.path.join(DIR_PATH, config["topic"]["csv"]))
    except Exception as e:
        logger.error(str(e))
        raise ValueError(str(e))


def generate_caption(topic):
    prompt = generate_caption_prompt(topic)
    response = openai.ChatCompletion.create(
        model=config["caption"]["model"],
        messages=[
            {"role": "system", "content": config["caption"]["system_msg"]},
            {"role": "user", "content": prompt}
        ],
        temperature=config["caption"]["temperature"],
    )

    return response.choices[0]["message"]["content"]


def generate_image(topic):
    prompt = generate_image_prompt(topic)
    response = openai.Image.create(
        prompt=prompt,
        n=1,
        size=config["scene"]["size"]
    )

    return response['data'][0]['url']


def generate_caption_prompt(topic):
    file_path = os.path.join(
        DIR_PATH,
        "insta_gpt",
        "prompt_design",
        "caption_template.txt"
    )
    with open(file_path) as f:
        template = f.read()

    return template + f" The topic of the post is {topic}."


def generate_image_prompt(topic):
    file_path = os.path.join(
        DIR_PATH,
        "insta_gpt",
        "prompt_design",
        "scene_template.txt"
    )
    with open(file_path) as f:
        template = f.read()

    full_template = "photograph, " + template

    return full_template + f" featuring {topic}, realistic, dslr, square"

insta_gpt/generator.py
philippmoehl/insta-gpt
from dotenv import load_dotenv
import io
import logging
import os
import pandas as pd
from PIL import Image
import requests
import yaml

from instagrapi import Client
import openai


DIR_PATH = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
TMP_PATH = os.path.join(DIR_PATH, "temp.jpg")

logger = logging.getLogger(__name__)


def download_image(url, save_path):
    response = requests.get(url)
    image = Image.open(io.BytesIO(response.content))

    # Convert the image to RGB format if it has an alpha channel
    if image.mode == 'RGBA':
        image = image.convert('RGB')

    # Save the image as JPEG
    image.save(save_path, 'JPEG')
    logger.info(f"Image downloaded and saved as {save_path}")


def setup_logging():
    logging.basicConfig(level=logging.INFO)

    # Create a FileHandler and set its output file
    file_handler = logging.FileHandler(
        os.path.join(DIR_PATH, "application.log"))
    file_handler.setLevel(logging.INFO)

    # Define a formatter for the file handler
    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
    file_handler.setFormatter(formatter)

    # Add the file handler to the root logger
    logging.getLogger('').addHandler(file_handler)


def setup():
    load_dotenv()
    setup_logging()
    openai.api_key = os.getenv("OPENAI_API_KEY")
    logger.info("Initialized")


def publish(caption, image_url):
    # TODO: use native instagram api in order to post from image url
    # save image in order to post image via instagrapi
    download_image(image_url, TMP_PATH)

    # instagram api
    logger.info(f"Publishing image...")
    cl = Client()
    cl.login(os.getenv("USER_ID"), os.getenv("PASSWORD"))

    media = cl.photo_upload(
        TMP_PATH,
        caption
    )

    # remove the temporarily saved jpeg
    os.remove(TMP_PATH)
    logger.info("Image published on Instagram")


def load_yaml_file(file_path):
    with open(file_path, 'r') as file:
        try:
            config = yaml.safe_load(file)
            return config
        except yaml.YAMLError as e:
            logger.error(f"Error loading YAML file {file_path}")
            raise yaml.YAMLError(f"Error loading YAML file {file_path}")


def load_topic(filename):
    df = pd.read_csv(filename, header=0)
    if not df["posted"].all():
        if len(df[~df["posted"]]) < 5:
            logger.warning(f"Only {len(df[~df['posted']])} topics from "
                           f"{filename} left for posting")
        return df.loc[~df["posted"], "topic"].iloc[0]
    else:
        logger.error(f"All topics from {filename} are posted")
        raise ValueError(f"All topics from {filename} are posted")


def update_topic(filename):
    df = pd.read_csv(filename, header=0)
    row_index = df.loc[~df["posted"], :].index[0]
    df.loc[row_index, "posted"] = True
    df.to_csv(filename, index=False)
    logger.info(f"Updated {filename} - Marked topic as posted")

insta_gpt/utils.py
philippmoehl/insta-gpt
from insta_gpt.generator import generate_post
from insta_gpt.utils import setup


if __name__ == '__main__':
    setup()
    generate_post()

main.py
philippmoehl/insta-gpt