AI生成画像をInstagramに自動投稿するアプリを作ってみた

はじめに

OpenAIのDALL·EとStable Diffusionの画像生成エンジンを利用しInstagramに生成画像を自動投稿するアプリを作成しました。
アプリで実際に投稿しているInstagramアカウントはこちらです。

インフラはGCPを利用しています。
Cloud RunにPython Flaskアプリのコンテナをデプロイして動作させます。
特にクリティカルな処理ではないのでサーバーレス環境で運用しています。

処理フロー

Cloud Schedulerによる完全自動運転を実現しています。
画像生成に使うプロンプトはOpenAIのチャットAPIで作成させています。
それぞれの画像生成エンジンにて生成された画像ファイルをInstagramのAPIを利用して投稿します。

処理解説

インストールの必要なPythonライブラリはこちらを参照してください。

RUN pip install requests
RUN pip install openai
RUN pip install google-cloud-storage
RUN pip install stability-sdk

今回は各REST APIをラッピングしたPython SDKを使って実装してます。

定数群

main.py

PAGE_ACCESS_TOKEN = os.environ.get('INSTA_PAGE_ACCESS_TOKEN', '')
VERIFY_TOKEN = os.environ.get('INSTA_PAGE_VERIFY_TOKEN', '')
BUSINESS_ACCOUNT_ID = os.environ.get('INSTA_BUSINESS_ACCOUNT_ID', '')
STABILITY_KEY = os.environ.get('STABILITY_KEY', '')
openai.api_key = os.environ.get('OPENAI_TOKEN', '')

処理に必要なAPIトークン類はあらかじめ定数保持をします。
環境変数に仕込まれるトークン類はGCPのSecret Managerに保管しています。
Cloud Runのコンテナ設定でどのシークレット値を環境変数に入れるのかマッピングを定義することができます。Instagram・OpenAI・Stable Diffusion（こちらはDream StudioでAPIキーを取得してください）のトークンがそれぞれ必要になります。

プロンプト生成

main.py

topic = [
   "city",
   "world heritage",
   "sightseeing place",
   "airport",
   "train station",
   "sea port",
   "bridge",
   "zoo",
   "aquarium",
   "theme park"
]

place = [
    "North America",
    "South America",
    "Asia",
    "Europe",
    "Africa",
    "Oceania"
]
    # pick topic and place randomly
    picked_topic = random.choice(topic)
    picked_place = random.choice(place)

    # make openai parameter
    input = []
    text = f'pick one {picked_topic} in {picked_place} countries.'
    # text = 'pick one place all over the world'
    new_message = {"role":"user", "content":text}
    input.append(new_message)

    # send message to openai api
    result = openai.ChatCompletion.create(model=AI_ENGINE, messages=input)    
    ai_response = result.choices[0].message.content
    print(ai_response)

トピックはChatGPTにてランダムに作成してもらいます。

画像生成

AIエンジンで生成した生成した画像ファイルをいったんコンテナ内に一時保管します。

以下はStable Diffusionの場合

main.py

    # generate image by stability
    stability_api = client.StabilityInference(key=STABILITY_KEY, verbose=True)
    answers = stability_api.generate(prompt=ai_response)

    # save image as file
    image_path = f"/tmp/image_{BUSINESS_ACCOUNT_ID}.png"
    for resp in answers:
        for artifact in resp.artifacts:
            if artifact.finish_reason == generation.FILTER:
                print("NSFW")
            if artifact.type == generation.ARTIFACT_IMAGE:
                img = Image.open(io.BytesIO(artifact.binary))
                img.save(image_path)

以下はOpenAIのDALL·Eの場合

main.py

    # generate image by openai
    response = openai.Image.create(
        prompt=ai_response,
        n=1,
        size="512x512",
        response_format="b64_json",
    )

    # save image as file
    image_path = f"/tmp/image_{BUSINESS_ACCOUNT_ID}.png"
    for data, n in zip(response["data"], range(1)):
        img_data = base64.b64decode(data["b64_json"])
        with open(image_path, "wb") as f:
            f.write(img_data)

ai_responseへChatGPTに作成してもらったプロンプトを格納しAI画像エンジンに渡して絵を描いてもらいます。

画像保管

生成された画像をGCP Cloud Storageにバックアップします。
ファイルを直接ダウンロードできるURLを発行します。
これをInstagram Graph APIのパラメータに設定します。

main.py

    current_time = int(time.time())
    current_time_string = str(current_time)

    # Uploads a file to the Google Cloud Storage bucket
    image_url = upload_to_bucket(current_time_string, image_path, "ai-bot-app-insta")
    print(image_url)

#  Uploads a file to the Google Cloud Storage bucket
def upload_to_bucket(blob_name, file_path, bucket_name):
    # Create a Cloud Storage client
    storage_client = storage.Client()

    # Get the bucket that the file will be uploaded to
    bucket = storage_client.bucket(bucket_name)

    # Create a new blob and upload the file's content
    blob = bucket.blob(blob_name)
    blob.upload_from_filename(file_path)

    # Make the blob publicly viewable
    blob.make_public()

    # Return the public URL of the uploaded file
    return blob.public_url

ファイルを格納するバケットの作成が事前に必要となります。
作成したバケット名をbucket_nameに設定します。

Instagramへ送信

これはStable Diffusionで生成した画像をファイルを送信する例です。
captionにて投稿用の文言をあらかじめ作成しておきます。
image_urlに格納されたCloud StrageのURLを渡してあげます。
アップロードに成功するとmedia_idが取得できますので今度はそれを次のエンドポイントの引数として渡してあげることで実際に投稿されます。

main.py

    caption = f"This is an image of {ai_response} created by image generation Stable Diffusion API #stablediffusion #texttoimage #api"

    # Upload the image to Facebook
    url = f"https://graph.facebook.com/{BUSINESS_ACCOUNT_ID}/media"
    params = {'access_token': PAGE_ACCESS_TOKEN, 'image_url':image_url, 'caption':caption}
    response = requests.post(url, params=params)
    if response.status_code != 200:
        raise Exception(f"Failed to upload image: {response.text}")
    media_id = response.json()['id']

    # Publish the photo to Instagram
    url = f"https://graph.facebook.com/{BUSINESS_ACCOUNT_ID}/media_publish"
    params = {'access_token': PAGE_ACCESS_TOKEN, 'creation_id': media_id}
    response = requests.post(url, params=params)
    if response.status_code != 200:
        raise Exception(f"Failed to publish photo: {response.text}")

    print('Image uploaded and published successfully!')

資料置き場

ソースコード
OpenAI API
Stability.ai REST API
InstagramグラフAPI