Instagram へ自動投稿する機能を開発する

はじめに

この記事は自分の作業メモです。
元々、TwitterとFacebookへの自動投稿は機能として持ってましたが、Instagram への自動投稿も欲しいということで作成しました。

まずは 生成AIさんにやり方を質問します。

インスタグラムのAPIを利用して、自動投稿を行いたいです。できるだけ詳しい手順を教えてください。

結果を見ていきながら、実装していきます。

開発者登録と権限設定

Meta for Developersアカウント作成
開発者ポータルにログイン後、「アプリの作成」から新規アプリを登録
これはFacebookk投稿で元々アプリを作成してました。

「アプリに製品を追加」の「Instagram」を追加
追加すると、右のメニューに「Instagram」が追加されます。

必要な情報を登録し、アクセストークンを生成する
「InstagramビジネスログインによるAPI設定」で「Instagramアプリ名」などを登録すると「InstagramアプリID」や「Instagram app secret」を取得できます。

また、 「アクセストークンを生成する」でFacebookアカウントを紐付けると「長期アクセストークン」を取得できます。APIを利用して投稿する場合はこのアクセストークンを利用します。（私はここで生成されるトークンは短期アクセストークンと思ってましたが、ここで表示されるトークンは「長期アクセストークン」でした）

なお、「 Instagramビジネスログインを設定する」を利用すると、OAuth認証を利用して、「短期アクセストークン」を生成できます。こちらの場合は「短期アクセストークン」から「長期アクセストークン」を生成することを忘れないようにしましょう。

トークンの管理

生成したトークンはデータベースに保存するようにします。これは、長期アクセストークンは永続トークンではなく、定期的に更新しなければならないためです。

以下のマイグレーションファイルを作成しました。

class CreateInstagramTokens < ActiveRecord::Migration
  def change
    create_table :instagram_tokens do |t|
      t.string :access_token, null: false
      t.string :refresh_token
      t.integer :expires_in
      t.string :scope
      t.datetime :expires_at
      t.json :raw_response
      t.boolean :active, default: true
      t.string :instagram_user_id, :string

      t.timestamps
    end

    add_index :instagram_tokens, :active
  end
end

APIを呼び出すクライアントを作成

実際にAPI呼び出しには以下のクライアントを介するようにしました。（なお、コードは生成AIで作成しています）
なお、client_id は InstagramアプリID であり、client_secret は Instagram app secret です。それぞれ、環境変数として登録します。

require 'faraday'
require 'json'

module Instagram
  class InstagramError < StandardError; end
  class TokenError < InstagramError; end
  class ApiError < InstagramError; end
  class ParseError < InstagramError; end

  class Client
    BASE_URL = 'https://graph.instagram.com'.freeze
    AUTH_URL = 'https://www.instagram.com'.freeze
    API_URL = 'https://api.instagram.com'.freeze
    API_VERSION = 'v18.0'.freeze

    attr_reader :access_token, :client_id, :client_secret, :user_id

    def initialize(access_token = nil, user_id = nil)
      @client_id = ENV['instagram__app_id']
      @client_secret = ENV['instagram__app_secret']
      
      # アクセストークンの取得優先順位:
      # 1. 引数で渡されたトークン
      # 2. データベースの最新のアクティブなトークン
      @access_token = access_token
      @user_id = user_id
      @conn = create_connection()
    end

    # ユーザー情報を取得
    def me
      response = @conn.get('/v22.0/me')
      handle_response(response)
    end

    # メディアを投稿（Instagram Graph APIでは直接投稿できないため、Facebookページを経由）
    def create_media(image_url, caption)
      
      # コンテナを作成（投稿準備）
      response = @conn.post do |req|
        req.url "/#{user_id}/media"
        req.params = {
          access_token: access_token,
          image_url: image_url,
          caption: caption
        }
      end
      
      result = handle_response(response)
      return nil unless result && result['id']
      
      creation_id = result['id']
      publish_media(creation_id)
    end

    # 複数画像の投稿
    def create_carousel(image_urls, caption)
      # Instagramの制限：カルーセル投稿は最大10枚まで
      if image_urls.size > 10
        Rails.logger.warn("Instagram carousel post limited to 10 images, truncating #{image_urls.size} to 10")
        image_urls = image_urls.take(10)
      end
      
      # 各画像のコンテナIDを取得
      Rails.logger.info("Creating carousel with #{image_urls.size} images")
      media_ids = []
      
      image_urls.each_with_index do |url, index|
        begin
          Rails.logger.info("Creating carousel item #{index + 1}/#{image_urls.size}: #{url}")
          response = @conn.post do |req|
            req.url "/#{user_id}/media"
            req.params = {
              access_token: access_token,
              image_url: url,
              is_carousel_item: true
            }
          end
          
          result = handle_response(response)
          if result && result['id']
            media_ids << result['id']
            Rails.logger.info("Successfully created carousel item #{index + 1}, container ID: #{result['id']}")
          else
            Rails.logger.error("Failed to create carousel item #{index + 1}")
          end
        rescue StandardError => e
          Rails.logger.error("Error creating carousel item #{index + 1}: #{e.message}")
        end
      end
      
      if media_ids.empty?
        Rails.logger.error("No valid media containers created, aborting carousel post")
        return nil
      end
      
      # カルーセル投稿を作成
      Rails.logger.info("Creating carousel container with #{media_ids.size} items")
      begin
        response = @conn.post do |req|
          req.url "/#{user_id}/media"
          req.params = {
            access_token: access_token,
            caption: caption,
            media_type: 'CAROUSEL',
            children: media_ids.join(',')
          }
        end
        
        result = handle_response(response)
        unless result && result['id']
          Rails.logger.error("Failed to create carousel container")
          return nil
        end
        
        creation_id = result['id']
        Rails.logger.info("Successfully created carousel container, ID: #{creation_id}")
        
        # カルーセルを公開
        Rails.logger.info("Publishing carousel")
        publish_result = publish_media(creation_id)
        
        if publish_result && publish_result['id']
          Rails.logger.info("Successfully published carousel, ID: #{publish_result['id']}")
        else
          Rails.logger.error("Failed to publish carousel")
        end
        
        publish_result
      rescue StandardError => e
        Rails.logger.error("Error creating or publishing carousel: #{e.message}")
        nil
      end
    end

    # 投稿を公開
    def publish_media(creation_id)  
      response = @conn.post do |req|
        req.url "/#{user_id}/media_publish"
        req.params = {
          access_token: access_token,
          creation_id: creation_id
        }
      end
      
      handle_response(response)
    end

    # 投稿一覧を取得
    def media(limit = 25)    
      response = @conn.get do |req|
        req.url "/#{user_id}/media"
        req.params = {
          access_token: access_token,
          fields: 'id,caption,media_type,media_url,permalink,thumbnail_url,timestamp,username',
          limit: limit
        }
      end
      
      handle_response(response)
    end

    # 特定の投稿の詳細を取得
    def media_details(media_id)
      response = @conn.get do |req|
        req.url "/#{media_id}"
        req.params = {
          access_token: access_token,
          fields: 'id,caption,media_type,media_url,permalink,thumbnail_url,timestamp,username'
        }
      end
      
      handle_response(response)
    end

    # アクセストークンを更新（長期トークンの取得）
    def refresh_token
      response = @conn.get do |req|
        req.url '/refresh_access_token'
        req.params = {
          grant_type: 'ig_refresh_token',
          access_token: access_token
        }
      end
      
      result = handle_response(response)
      if result && result['access_token']
        @access_token = result['access_token']
        @conn = create_connection
      end
      
      result
    end
    
    # 認証コードからアクセストークンを取得
    def exchange_code_for_token(code, redirect_uri)
      if client_id.blank? || client_secret.blank?
        raise ArgumentError, "Client ID and Client Secret are required for code exchange"
      end
      
      api_conn = Faraday.new(url: API_URL)
      response = api_conn.post("/oauth/access_token") do |req|
        req.body = {
          client_id: client_id,
          client_secret: client_secret,
          redirect_uri: redirect_uri,
          code: code,
          grant_type: 'authorization_code'
        }
      end
      
      result = handle_api_response(response)
      if result && result['access_token']
        @access_token = result['access_token']
        @conn = create_connection
      end
      
      result
    end
    
    # 短期アクセストークンから長期アクセストークンを取得
    def exchange_token(short_lived_token = nil)
      token_to_exchange = short_lived_token || access_token
      
      if client_id.blank? || client_secret.blank?
        raise ArgumentError, "Client ID and Client Secret are required for token exchange"
      end

      response = @conn.get do |req|
        req.url "/access_token"
        req.params = {
          grant_type: 'ig_exchange_token',
          client_secret: client_secret,
          access_token: token_to_exchange
        }
      end
      
      result = handle_response(response)
      if result && result['access_token']
        @access_token = result['access_token']
        @conn = create_connection
      end
      
      result
    end
    
    # 認証URLを生成
    def self.generate_auth_url(redirect_uri)
      client_id = ENV['instagram__app_id']
      
      if client_id.blank?
        raise ArgumentError, "Client ID is required for generating auth URL"
      end
      
      # Instagram Business Login用のURLを生成
      AUTH_URL + 
      "/oauth/authorize?" +
      "enable_fb_login=0" +
      "&force_authentication=1" +
      "&client_id=#{client_id}" +
      "&redirect_uri=#{CGI.escape(redirect_uri)}" +
      "&scope=#{CGI.escape('instagram_business_basic,instagram_business_manage_messages,instagram_business_manage_comments,instagram_business_content_publish,instagram_business_manage_insights')}" +
      "&response_type=code" +
      "&state=#{SecureRandom.hex(10)}"
    end
    
    # ユーザーのInstagramアカウント情報を取得
    def get_user_account_info
      response = @conn.get do |req|
        req.url '/me'
        req.params = {
          fields: 'id,username',
          access_token: access_token
        }
      end
      
      handle_response(response)
    end
    
    # 認証コードからInstagram Business Accountまでの全プロセスを実行
    def process_auth_code(code, redirect_uri)
      # 認証コードをアクセストークンに交換
      token_result = exchange_code_for_token(code, redirect_uri)
      return nil unless token_result && token_result['access_token'].present?
      
      short_lived_token = token_result['access_token']
      user_id = token_result['user_id'] # Instagram APIはユーザーIDも返す
      
      # 短期トークンを長期トークンに交換
      long_lived_result = exchange_token(short_lived_token)
      return nil unless long_lived_result && long_lived_result['access_token'].present?
      
      # ユーザーIDが取得できなかった場合はAPIから取得
      if user_id.nil?
        begin
          # アクセストークンを使ってユーザー情報を取得
          @access_token = long_lived_result['access_token']
          @conn = create_connection
          
          user_info = get_user_account_info
          user_id = user_info['id'] if user_info && user_info['id'].present?
        rescue => e
          Rails.logger.error("Failed to fetch Instagram user ID: #{e.message}")
        end
      end
      
      # 結果を返す
      {
        'access_token' => long_lived_result['access_token'],
        'refresh_token' => long_lived_result['refresh_token'],
        'expires_in' => long_lived_result['expires_in'],
        'scope' => long_lived_result['scope'],
        'instagram_user_id' => user_id,
        'raw_response' => long_lived_result
      }
    end

    private

    def create_connection
      Faraday.new(url: BASE_URL) do |faraday|
        faraday.request :url_encoded
        faraday.response :logger if Rails.env.development?
        faraday.adapter Faraday.default_adapter
      end
    end

    def handle_response(response)      
      begin
        res = JSON.parse(response.body)
        unless response.status.between?(200, 299)
          error_message = res.dig("error", "message") || response.body
          Rails.logger.error("Instagram API request failed: #{response.status} - #{error_message}")
          
          # セッションキー無効エラーの場合はTokenErrorを発生
          if error_message.include?("Session key invalid")
            raise TokenError, "Instagram token is invalid or expired: #{error_message}"
          else
            raise ApiError, "Instagram API request failed: #{response.status} - #{error_message}"
          end
        end
        res
      rescue JSON::ParserError => e
        Rails.logger.error("Instagram API response parsing error: #{e.message}")
        raise ParseError, "Instagram API response parsing error: #{e.message}"
      end
    end
    
    def handle_api_response(response)
      begin
        res = JSON.parse(response.body)
        unless response.status.between?(200, 299)
          error_message = res.dig("error", "message") || res.dig("error_message") || response.body
          Rails.logger.error("Instagram API request failed: #{response.status} - #{error_message}")
          raise ApiError, "Instagram API request failed: #{response.status} - #{error_message}"
        end
        res
      rescue JSON::ParserError => e
        Rails.logger.error("Instagram API response parsing error: #{e.message}")
        raise ParseError, "Instagram API response parsing error: #{e.message}"
      end
    end
  end
end

トークン更新タスク

トークンを更新するタスクも作成しました。

namespace :instagram do
  desc 'Refresh Instagram access token to prevent expiration'
  task refresh_token: :environment do
    token = InstagramToken.current
    
    if token.blank?
      puts "No active Instagram token found."
      next
    end
    
    if token.expires_at.present? && token.expires_at > 1.week.from_now
      puts "Instagram token is still valid for more than a week (expires at #{token.expires_at}). Skipping refresh."
      next
    end
    
    begin
      puts "Refreshing Instagram token (ID: #{token.id}, created at: #{token.created_at})..."
      client = Instagram::Client.new(token.access_token)
      result = client.refresh_token
      
      if result && result['access_token'].present?
        token.update_from_response(result)
        puts "Instagram token successfully refreshed. New expiration: #{token.expires_at}"
      else
        puts "Failed to refresh Instagram token. No valid response received."
      end
    rescue StandardError => e
      puts "Error refreshing Instagram token: #{e.message}"
    end
  end

テスト

トークンを登録し、実際に投稿できるか試してみます。
以下のような画面を用意し、指定した画像を投稿してみました。

結果

ちゃんと投稿できました。

まとめ

この記事では Instagram への投稿をAPIで実施しました。
Facebookとは異なり、永続トークンが存在しないため、めんどくさいなと思ってましたが、生成AIを利用することでかなりの部分を機械的に作成できました。
もちろん、生成AIだけでは URLを間違っていたりして動作しませんので、そのあたりはドキュメントの内容を提示して修正を行いました。
次は TikTok の API 投稿もやってみようと思います。