Python × Flaskで自作ブログとInstagram・Threadsを連携し、同時投稿を自動化する

はじめに

私は2025年9月からPythonの学習を始めました。

学習を始めた当初、知人が経営する施設で「ホームページを作りたい」という話をしているのを耳にしたことが、本プロジェクトのきっかけです。

学習の過程で、フロントエンドとFlaskを用いたWebアプリケーション開発を学び、そのアウトプットとして、ブログ機能や施設概要の編集ができるWebアプリを作成しました。

このアプリをポートフォリオとして営業したところ、実際に制作の依頼をいただくことができました。本記事では、そのポートフォリオの一部機能として実装した、ブログ投稿をInstagram・Threadsへ自動連携する仕組みについてまとめます。

ホーム画面イメージ

なぜ同時投稿を自動化したのか

Instagramの投稿

知人が経営する施設では、日常的な情報発信にInstagramを利用していました。
ブログとInstagramの両方に同じ内容を投稿する作業は手間がかかるため、ブログ投稿を起点としてInstagramへも同時に投稿できれば、運用負荷を減らせると考え、自動化を行いました。

Threadsの投稿

Threadsについては、現在入校しているスクールにて、需要がありそうだという話を聞いたことがきっかけです。
ポートフォリオ作成の一環として、Instagramに加えてThreadsへの投稿も自動化することで、複数SNS連携の実装経験を積み、技術力向上につなげたいと考えました

ブログ機能の内容

1. ブログの投稿フォーム画面

ブログ投稿時に、以下の情報を入力できるフォームを用意しました。

• タイトル
• 本文
• 画像

また、投稿時に

• Instagramへ投稿
• Threadsへ投稿

のチェックボックスを選択することで、ブログ投稿と同時に各SNSへ自動投稿される仕組みになっています。

2. ブログ画面

投稿した記事は、ブログ一覧および詳細画面として表示されます。

3. Instagramへの自動投稿

ブログ投稿時にInstagramへの投稿を選択した場合、本文と画像がInstagramにも自動で投稿されます。

4. Threadsへの自動投稿

同様に、Threadsへの投稿を選択した場合は、ブログの内容がThreadsにも自動で投稿されます。

開発で行ったこと

APIを使用するための事前準備

1.アカウント登録

• Facebook
• Instagramビジネスアカウント（プロアカウント）変更する
• Threads

2.Facebookで公開ページの作成

Facebookのアイコンをクリックしてページ作成を選択

公開ページを選択

必要な情報を入力してページを作成

3.Instagram（プロアカウント）をFacebookの公開ページにリンクする

設定とプライバシーを選択

リンクするアカウントを選択

Instagramを選択してアカウントとリンクする

4. Instagram投稿のトークンを取得する

4-1.Meta For Developersでアプリの作成

Facebookのアカウントでログインする
https://developers.facebook.com/

アプリの作成をクリック

アプリ名を入力

Instagaramでメッセージとコンテンツを管理を選択

現時点ではビジネスポートフォリオをリンクしないを選択

次へを選択

ダッシュボードに移動を選択

4-2. アクセス許可の設定

ダッシュボードでユースケースをカスタマイズを選択

• FacebookログインによるAPI設定のタブを開く
• Instagramでコンテンツを処理のアクセス許可を選択

上のタブのツールを選択して
グラフAPIエクスプローラを選択

4-3. トークンの取得

トークンの取得は3段階あります。
1 短期トークン（1時間）
2 長期トークン（60日）
3 無期限トークン

エクスプローラで　
右下のアクセス許可をクリック

下記アクセス許可を選択して
Generate Access Tokenをクリック

• business_management
• pages_read_engagement
• pages_show_list
• instagram_basic
• instagram_content_publish

Instagramアカウントが紐づいているfacebookの公開ページを選択

Instagramアカウントの選択

認証の操作が終われば短期トークンが生成される

4-4. 長期トークンの取得

短期トークンが生成されている
トークン入力欄の横の「i」をクリック

アクセストークンツールで開くを選択

アクセストークン情報が表示される
有効期限が1時間以内となっている

左下のアクセストークンの延長を選択
下に長期トークンが表示される

デバッグをクリック

有効期限が2か月となっているので
長期トークンの取得が完了

4-5. 無期限トークンの取得

先ほど取得した長期トークンをコピー

エクスプローラに戻り
長期トークンを貼り付け

URLの末尾に「me/accounts」と入力
送信ボタンをクリック

dataの中のaccess_tokenが無期限のトークンとなります

以上で無期限トークンの取得が完了

無制限トークンを貼り付けて再度「i」のアイコンを押すと有効期限が受け取らないとなっていたらOK
下の方のinstagram_basicの横に記載してある数字がinstagramのIDです。
投稿に必要な番号なのでメモ

5.Threadsのトークンを取得

Threadsのトークンは

• 短期トークン（1時間）
• 長期トークン（60日）
  があります

5-1.アプリの作成

ThreadsAPIにアクセスを選択

現時点ではビジネスポートフォリオをリンクしないを選択

ダッシュボードに移動クリック

5-2.トークンの取得準備

「Threads APIへのアクセス」ユースケースをカスタマイズをクリック

設定のタブを開いてThreadsのapp secretを表示してメモ（長期トークンの発行に使用）
Threadsテスターを追加または削除を選択

メンバーを追加を選択

Threadsテスターを選択して、Threadsのユーザーネームを入力

Threadsのアカウントが追加される
ステータスが承認待ちとなっているので
Threadsを開いて承認する必要があります。

Threadsの設定→アカウント→ウェブサイトのアクセス許可を選択

同意するをクリック

アプリの役割の画面に戻ってリロードするとステータスの承認待ちが消える

5-3.短期トークンの取得

ダッシュボードから先ほどの画面に戻り右下の
アクセストークンを生成をクリックを入力するとトークンが生成されます。

エラーが出た場合

このようなエラーが出ることがあります。
正直、今回はこのエラーに悩まされました。
色々調べましたがこのエラーの原因はわかりませんでした。

私が生成した方法

手探りで色々試してみて何とか生成することができたやり方が以下となります。
このやり方が正しいかはわかりません。

上のツールタブからグラフAPIエクスプローラを開く

URLをthreads.netを選択
metaアプリは現在作成しているアプリ名を選択
Generate Threads Access Tokenをクリック

アクセストークンが生成されます。
その後右上の送信をクリックすると
user_idが返ってくるので、こちらのユーザーIDもメモ

5-3.長期トークンの取得

グラフAPIエクスプローラで
access_token?grant_type=th_exchange_token&client_secret=Threadsのapp secret
を入力して送信をクリック

access_tokenに長期トークンがあります。
以上でトークンの取得は終了

環境構築

1.使用する言語

言語	バージョン
Python	3.11.9

pyenvとpoetryを使用して環境構築を行いました。
（使用OS：Windows）

2. 使用するライブラリ

使用目的	ライブラリ	バージョン
Webアプリケーションフレームワーク	Flask	>=3.1.2,<4.0.0
ローカルでの環境変数の管理	dotenv	>=0.9.9,<0.10.0
外部API通信	requests	>=2.32.5,<3.0.0
DBの操作	Flask-SQLAlchemy	==3.1.1
DBマイグレーション	Flask-Migrate	==4.0.7
PostgreSQL接続	psycopg[binary]	==3.2.3
タイムゾーン管理	pytz	>=2025.2,<2026.0
認証・ログイン管理	Flask-Login	==0.6.3
画像処理	Pillow	>=12.0.0,<13.0.0
HEIF画像対応	pillow-heif	>=1.1.1,<2.0.0
WSGIサーバー	gunicorn	>=23.0.0,<24.0.0
S3互換ストレージ（Cloudflare R2）操作	boto3	>=1.42.25,<2.0.0

フォルダ構成

InstagaramやThreadsへの投稿や他ロジックの部分はserviceフォルダに入れています。

app/
├─ blueprints/
│  └─ 各Blueprintのルーティング処理
├─ models/
│  └─ データベースモデル定義
├─ services/
│  └─ Instagram / Threads 投稿処理などのロジック部分
├─ static/
│  └─ CSS / JavaScript / 画像ファイル
├─ templates/
│  └─ HTMLテンプレート
├─ __init__.py
│  └─ create_app() の定義、Blueprintの登録
├─ config.py
│  └─ 環境別設定（Local / Production）
└─ extensions.py
   └─ db / migrate / login_manager の初期化

コードの詳細

本記事では、プロジェクト全体のコードではなく、自動投稿処理に関わる主要な部分のコードを中心に掲載します。

1.app/init.py

app/__init__.py

import logging
import os

from dotenv import load_dotenv
from flask import Flask

from app.config import LocalConfig, ProductionConfig
from app.extensions import db, migrate, login_manager
from .blueprints.main import bp as main_bp
from .blueprints.admin import bp as admin_bp


def create_app():
    env = os.getenv('ENV', 'local')
    if env == 'local':
        load_dotenv('.env')

    app = Flask(__name__)

    if env == 'production':
        app.config.from_object(ProductionConfig)
        app.logger.setLevel(logging.INFO)
        missing = [k for k in ('SECRET_KEY', 'SQLALCHEMY_DATABASE_URI') if not app.config.get(k)]
        
        if missing:
            raise RuntimeError(f"Missing required config: {missing}")
    else:
        app.config.from_object(LocalConfig)
        app.logger.setLevel(logging.DEBUG)

    db.init_app(app)
    migrate.init_app(app, db)
    login_manager.init_app(app)

    from app import models

    app.register_blueprint(main_bp)
    app.register_blueprint(admin_bp, url_prefix='/admin')
    return app

アプリの初期化とBlueprintの登録
ローカル環境と本番環境で環境に応じて設定を切り替えています。

2.app/config.py

app/config.py

import os


class BaseConfig:
    # Instagram Graph API
    INSTAGRAM_MAIN_URL = "https://graph.facebook.com"
    INSTAGRAM_VERSION = os.getenv("INSTAGRAM_VERSION", "v23.0")
    INSTAGRAM_ACCESS_TOKEN = os.getenv("INSTAGRAM_ACCESS_TOKEN")
    INSTAGRAM_PAGE_ID = os.getenv("INSTAGRAM_PAGE_ID")
    INSTAGRAM_USER_ID = os.getenv("INSTAGRAM_USER_ID")
    # threds API
    THREADS_MAIN_URL = "https://graph.threads.net"
    THREADS_VERSION = os.getenv("THREADS_VERSION", "v1.0")
    THREADS_ACCESS_TOKEN = os.getenv("THREADS_ACCESS_TOKEN")
    THREADS_USER_ID = os.getenv("THREADS_USER_ID")
    # Cloudflare R2
    R2_ENDPOINT_URL = os.getenv('R2_ENDPOINT_URL')
    R2_ACCESS_KEY_ID = os.getenv('R2_ACCESS_KEY_ID')
    R2_SECRET_ACCESS_KEY = os.getenv('R2_SECRET_ACCESS_KEY')
    # 画像のベースURL
    R2_STRAGE_BASE_URL = os.getenv('R2_STRAGE_BASE_URL')
    BUCKET_NAME = os.getenv('BUCKET_NAME')

class LocalConfig(BaseConfig):
    DEBUG = True
    SECRET_KEY = os.getenv('SECRET_KEY', 'dev-secret')    
    SQLALCHEMY_DATABASE_URI = os.getenv('SQLALCHEMY_DATABASE_URI')

class ProductionConfig(BaseConfig):
    DEBUG = False

    SECRET_KEY = os.getenv('SECRET_KEY')
    db_url = os.getenv('DATABASE_URL')
    SQLALCHEMY_DATABASE_URI = (
        db_url.replace('postgres://', 'postgresql+psycopg://')
        if db_url else None
    )

トークンやデータベース接続情報などの機密情報は環境変数から取得し、BaseConfig を継承した設定クラスで開発環境・本番環境を切り替える構成としています。

3. app/blueprints/admin/routes.py

app/blueprints/admin/routes.py

from flask import Blueprint, render_template, request, redirect, current_app, url_for, flash
from flask_login import login_user, logout_user, login_required, current_user
from werkzeug.security import check_password_hash

from app.extensions import db
from app.models.user_db import User
from app.services import blog_services
from app.services import info_services

bp = Blueprint('admin', __name__)

CHECK_IMG_TYPES = {'jpg', 'png', 'gif', 'tiff', 'webp', 'heic', 'jpeg'}

@bp.route('/blog_create',  methods=['GET', 'POST'])
@login_required
def blog_create():
    if request.method == 'POST':
        title = request.form.get('title')
        body = request.form.get('body')
        is_insta = request.form.get('post_to_instagram')
        is_threads = request.form.get('post_to_threads')
        # 画像情報の取得
        file = request.files['img']
        filename = file.filename
        
        if file and file.filename:
            if not filename.rsplit('.')[-1].lower() in CHECK_IMG_TYPES:
                current_app.logger.warning("file failed reason=not imagefile file=%s", filename)
                flash('画像ファイルを選択してください', 'danger')
                return redirect(url_for('admin.blog_create'))
            try:
                uuid_filename = blog_services.save_image_to_cloud_storage(file) 
            except Exception:
                current_app.logger.exception('img upload to storage error')
                uuid_filename = 'default.jpg'
                flash('画像のアップロードが失敗したのでデフォルト画像で投稿しました', 'warning')
        else:
            uuid_filename = 'default.jpg'

        blog_services.create_blog(title, body, uuid_filename)
        current_app.logger.info("blog post success title=%s" , title)

        if is_insta == '1' or is_threads == '1':
            caption = blog_services.create_caption(title=title, body=body)
            base_url = current_app.config['R2_STRAGE_BASE_URL']
            file_url = f'{base_url}/{uuid_filename}'
            if is_insta == '1':
                try:
                    blog_services.instagram_post(file_url, caption)
                except Exception:
                    flash('Instagramの投稿が失敗しました。', 'danger')
            
            if is_threads == '1':
                try:
                    blog_services.threads_post(file_url, caption)
                except Exception:
                    flash('Threadsの投稿が失敗しました。', 'danger')
        
        flash('投稿が完了しました', 'success')
        return redirect(url_for('admin.admin_blog'))
    
    elif request.method == 'GET':
        return render_template('admin/blog_create.html') 

ブログ投稿・Instagram投稿・Threads投稿を行うルーティング部分です。
実際の処理は blog_services.py に切り出しており、ルーティングでは主に入力の受け取りと処理の分岐を担当しています。

ユーザーが画像ファイルを指定しなかった場合は、デフォルトの画像を使用する仕様としています。

処理の流れ

1. フォームから title、body、file とチェックボックス（Instagram / Threads）を取得
2. 画像ファイルの拡張子をチェック（画像以外は弾く）
3. Cloudflare R2へ画像をアップロード（失敗時はデフォルト画像を使用）
4. ブログ記事をDBに保存
5. チェックが入っている場合はキャプションを生成し、Instagram / Threadsへ投稿
6. 処理結果を flash で画面に表示

4. app/servises

4-1. servicesフォルダ構成

services/
|
├─ __init__.py
│  
├─ blog_services.py
│  └─ ブログ作成・キャプション生成などの共通処理
├─ cloudflare_r2.py
│  └─ Cloudflare R2（S3互換ストレージ）への画像アップロード処理
├─ info_services.py
│  └─ 施設情報などの補助的なビジネスロジック
├─ instagram_posts.py
│  └─ Instagram投稿処理
├─ threads_posts.py
│  └─ Threads投稿処理
└─ utils.py
   └─ 共通ユーティリティ関数（画像の変換処理）

4-2.blog_services.py

blog_services.py

from flask import render_template

from app.extensions import db
from app.services.cloudflare_r2 import StorageManager
from app.services.instagram_posts import InstagramPublisher
from app.services.threads_posts import ThreadsPublisher
from app.models.blog_db import Blog


def create_blog(title, body, filename):
    # データベースに各テキストファイルを保存
    blog_post = Blog(title=title,body=body, img_name=filename)
    
    db.session.add(blog_post)
    db.session.commit()

def save_image_to_cloud_storage(img_file):
    storage_manager = StorageManager()
    uuid_filename = storage_manager.upload(img_file)
    return uuid_filename

def create_caption(**kwargs):
    return render_template('admin/caption.txt', **kwargs).strip()

def instagram_post(file_path, caption):
    insta = InstagramPublisher()
    insta.publish_media(file_path,caption=caption)

def threads_post(file_path, text):
    threads = ThreadsPublisher()
    threads.publish_media(file_path,text=text)

ブログに関する処理をまとめたファイルです。
本記事では、ブログ投稿の処理に関するコードを抜粋して掲載しています。

各関数の役割

• create_blog()：ブログ記事をDBへ保存
• save_image_to_cloud_storage()：画像をCloudflare R2へアップロード
• create_caption()：テンプレートからSNS投稿用キャプションを生成
• instagram_post()：Instagram投稿処理を呼び出し
• threads_post()：Threads投稿処理を呼び出し

create_caption() では、templates フォルダに用意したテキストテンプレートを使用して、SNS投稿用のキャプションを生成しています。

caption.txt

○○園

【{{title}}】

{{body}}

4-3.cloudflare_r2.py

cloudflare_r2.py

import logging
import uuid

import boto3
from flask import current_app

from app.services.utils import save_image_bytes_as_jpg


logger = logging.getLogger(__name__)

class StorageManager:

    def __init__(self):
        config = current_app.config
        self.s3 = boto3.client(
            service_name="s3",
            endpoint_url = config['R2_ENDPOINT_URL'],
            aws_access_key_id = config['R2_ACCESS_KEY_ID'],
            aws_secret_access_key = config['R2_SECRET_ACCESS_KEY'],
            region_name="auto"
        )
        self.bucket = config['BUCKET_NAME']

    def upload(self, img_file):
        buf_img = save_image_bytes_as_jpg(img_file)
        if buf_img is None:
            logger.error('Invalid image file buf_img=NONE')
            raise ValueError("Invalid image file")
        
        img_name = f"{uuid.uuid4().hex}.jpg"
        try:
            self.s3.upload_fileobj(
                buf_img, 
                self.bucket, 
                img_name
                )
        except Exception:
            logger.exception('file upload failed filename=%s', img_name)
            raise
        logger.info('image upload success img_name=%s', img_name)
        return img_name
    
    def delete(self, img_name):
        try:
            self.s3.delete_object(Bucket=self.bucket, Key=img_name)
        except Exception:
            logger.exception('file delete failed filename=%s', img_name)
            raise
        
        logger.info('storage file delete success filename=%s', img_name)

Cloudflare R2（S3互換ストレージ）を操作するためのクラスです。

upload() メソッドでは、受け取った画像ファイルを一度バイナリデータとして処理し、JPEG形式に変換したうえでストレージへアップロードします。

Instagram および Threads では、画像投稿時に使用できる形式が JPEG / PNG に限定されているため、utils.py に定義した save_image_bytes_as_jpg() を利用して、必ず JPEG 形式に変換しています。

アップロード時には UUID を用いたファイル名を生成し、重複を防いでいます。
upload() の戻り値としては、保存した画像のファイル名を返す仕様としています。

4-4. utils.py

utils.py

import io
import os

import pillow_heif
from PIL import Image, ImageOps


pillow_heif.register_heif_opener()

def save_image_bytes_as_jpg(file, max_width: int=1080):

    try:
        img = Image.open(file)
        img.verify()
    except Exception:
        return None
    
    file.seek(0)
    img = Image.open(file)
    img = ImageOps.exif_transpose(img)
    img = img.convert('RGB')

    if img.width > max_width:
        ratio = max_width / img.width
        img = img.resize((max_width, int(img.height * ratio)), Image.LANCZOS)

    buf = io.BytesIO()
    img.save(buf,'JPEG', quality=85, optimize=True)
    buf.seek(0)
    return buf

画像ファイルとして有効かを確認したうえで、JPEG形式へ変換し、バイナリデータとして返却しています。
※Instagramでは投稿画像の推奨最大横幅が 1080px とされているため、max_width を 1080 に設定しています。

4-5. instagram_posts.py

instagram_posts.py

import logging
import time

from flask import current_app
import requests


logger = logging.getLogger(__name__)

class InstagramPublisher:
    def __init__(self):
        config = current_app.config
        self.main_url = config["INSTAGRAM_MAIN_URL"]
        self.access_token = config["INSTAGRAM_ACCESS_TOKEN"]
        self.instagram_id = config["INSTAGRAM_USER_ID"]
        self.version = config["INSTAGRAM_VERSION"]

    def _create_container(self, img_url, caption=''):
        """
        インスタに投稿するためのコンテナを作成
        """
        params = {
        'image_url': img_url,
        'caption': caption,
        'access_token': self.access_token
    }
        res = requests.post(f'{self.main_url}/{self.version}/{self.instagram_id}/media', data=params)

        if res.status_code != 200:
            logger.error("Failed to create container: status=%s text=%s", res.status_code, res.text)
            raise RuntimeError("Failed to create container")
        
        logger.info("create container response status=%s", res.status_code)
        container = res.json()
        return container

    def get_container_id(self,  img_url, caption=''):
        """
        画像URLとキャプションからInstagram投稿用の
        メディアコンテナIDを取得する。
        """
        container = self._create_container(img_url, caption)
        return container['id']
    
    def _wait_until_ready(self, container_id, timeout=120):
        """
        メディアコンテナが公開可能（FINISHED）になるまで待機する。
        """
        
        url = f'{self.main_url}/{self.version}/{container_id}'

        logger.info("投稿準備開始 container_id=%s timeout=%s", container_id, timeout)
        for i in range(timeout):
            res = requests.get(
                url,
                params={
                    "fields": "status_code",
                    "access_token": self.access_token
                },
                timeout=10
            )

            result = res.json()
            status = result.get('status_code')

            if status == "FINISHED":
                logger.info("container投稿準備完了 status=%s",status)
                return
            if status == "ERROR":
                logger.error("containerの投稿準備ができません result=%s", result)
                raise RuntimeError("Media processing failed")

            time.sleep(1)
        raise TimeoutError("Media processing timeout")
 
    def publish_media(self, img_url, caption=''):
        """
        作成済みのメディアコンテナをInstagramに公開（投稿確定）する。
        """
        container_id = self.get_container_id(img_url, caption)
        
        self._wait_until_ready(container_id)

        data = {
                'creation_id': container_id,
                'access_token': self.access_token
        }
        res = requests.post(f'{self.main_url}/{self.version}/{self.instagram_id}/media_publish',data=data)
        if res.status_code != 200:
            logger.error("投稿失敗 status_code=%s response=%s", res.status_code, res.text)
            raise RuntimeError("Failed to publish media")
        
        logger.info("投稿完了 status=%s publish_id=%s", res.status_code, res.text)

Instagramへの画像投稿を行うクラスです。

Instagram Graph API では、投稿処理が2段階になっており、
まずメディアコンテナを作成してコンテナIDを取得し、その後コンテナIDを指定して投稿を確定させる必要があります。

処理の流れ

1. 画像URLとキャプションを指定してメディアコンテナを作成
2. 返却されたコンテナIDを取得
3. コンテナのステータスが FINISHED になるまで待機
4. コンテナIDを指定して投稿を確定（公開）

Instagramのメディアコンテナ作成は非同期処理となっているため、
コンテナ作成直後に投稿を確定しようとするとエラーになります。

そのため _wait_until_ready() で一定時間ポーリングを行い、
ステータスが FINISHED になったことを確認してから投稿処理を行うようにしています。

4-6. threads_posts.py

threads_posts.py

import logging
import time

from flask import current_app
import requests


logger = logging.getLogger(__name__)

class ThreadsPublisher:

    def __init__(self):
        config = current_app.config
        self.main_url = config["THREADS_MAIN_URL"]
        self.access_token = config["THREADS_ACCESS_TOKEN"]
        self.user_id = config["THREADS_USER_ID"]
        self.version = config["THREADS_VERSION"]

    def _create_container(self, img_url, text=''):
        params = {
            'media_type':'IMAGE',
            'image_url': img_url,
            'text': text,
            'access_token': self.access_token
    }

        res = requests.post(f'{self.main_url}/{self.user_id}/threads', data=params)
        if res.status_code != 200:
            logger.error("Failed to create container: status=%s text=%s", res.status_code, res.text)
            raise RuntimeError(f"Failed to create container")
        
        logger.info("create container response status=%s", res.status_code)
        container = res.json()
        return container
    
    def get_container_id(self,  img_url, text=''):

        container = self._create_container(img_url, text)
        return container['id']
    
    def _wait_until_ready(self, container_id, timeout=120):
        """
        メディアコンテナが公開可能（FINISHED）になるまで待機する。
        """
        url = f'{self.main_url}/{self.version}/{container_id}'

        logger.info("投稿準備開始 container_id=%s timeout=%s", container_id, timeout)
        for i in range(timeout):
            res = requests.get(
                url,
                params={
                    "fields": "status",
                    "access_token": self.access_token
                },
                timeout=10
            )

            result = res.json()
            status = result.get('status')

            if status == "FINISHED":
                logger.info("container投稿準備完了 status=%s",status)
                return
            if status == "ERROR":
                logger.error("containerの投稿準備ができません result=%s", result)
                raise RuntimeError("Media processing failed")

            time.sleep(1)

        raise TimeoutError("Media processing timeout")
    
    def publish_media(self, img_url, text=''):
        """
        作成済みのメディアコンテナをInstagramに公開（投稿確定）する。
        """
        container_id = self.get_container_id(img_url, text)
        
        self._wait_until_ready(container_id)

        data = {
                'creation_id': container_id,
                'access_token': self.access_token
        }
        res = requests.post(f'{self.main_url}/{self.version}/{self.user_id}/threads_publish',data=data)
        if res.status_code != 200:
            logger.error("投稿失敗 status_code=%s response=%s", res.status_code, res.text)
            raise RuntimeError("Failed to publish media")
        
        logger.info("投稿完了 status=%s publish_id=%s", res.status_code, res.text)

Threadsへの画像投稿を行うクラスです。

Threads API でも Instagram と同様に「コンテナ作成 → 投稿確定」の2段階になっているため、
まず投稿用コンテナを作成して id を取得し、コンテナが FINISHED になるまで待機したうえで投稿を確定させます。

非同期処理のため、コンテナ作成直後に投稿確定を行うと失敗することがあるため、
_wait_until_ready() でステータスをポーリングしてから threads_publish を呼び出すようにしています。

Instagram側との主な違い

• コンテナ作成のエンドポイント：/{user_id}/threads
• 作成時に media_type=IMAGE を指定
• ステータス確認のフィールド：fields=status（Instagramは status_code）
• 投稿確定のエンドポイント：/{user_id}/threads_publish

開発の振り返り

1. 大変だったこと

トークンの取得と設定

公式ドキュメントや技術記事を参考にしながら進めましたが、仕様変更が多く、自分が知りたい情報が公式ドキュメントのどこに記載されているのかを探すのに苦労しました。

特にトークンの取得や権限設定の部分で詰まることが多く、その際は公式ドキュメントをAIに読み解いてもらいながら、一つずつ確認して進めていきました。

Herokuデプロイ後に発覚したエラーによる仕様変更

当初は Cloudflare R2 を使用せず、ブログ投稿時にアップロードした画像を static/uploads フォルダに保存し、そこから表示する構成にしていました。

初回デプロイ時は問題なく動作していましたが、ファイルを修正して再デプロイした際に、アプリ側でアップロードした画像が表示されなくなる問題が発生しました。

uploads フォルダを .gitignore に追加して管理対象外にするなどの対応も試しましたが、この問題は解消されませんでした。

そのため、画像の保存方法を見直し、Cloudflare R2（S3互換ストレージ）を利用する構成へ仕様を変更しました。

2. AIの活用に関して

公式ドキュメントの解読

公式ドキュメントから必要な情報にたどり着けない場合は、該当する内容のURLを教えてもらい、また内容の理解が難しい場合は、該当しそうな部分を抜粋してChatGPTに貼り付け、要点を整理しながら読み進めました。

フロントエンド部分

フロントエンド部分は、HTML と Bootstrap を使用し、Grid や Card などの基本的な構造のみを自分で実装しました。
細かな配色やデザインについては、部分的にコードをChatGPTへ貼り付けながら調整を行いました。

次回は、Figmaに関しても少し学び、FigmaのAI機能も活用していきたいと考えています。

プロジェクトのフォルダ構成について

今回のプロジェクトでは、ロジック部分や設定部分を1つのPythonファイルにまとめると規模が大きくなってしまうと感じたこと、また今後の機能追加や拡張を見据えて開発を進めていたことから、Pythonファイルを役割ごとに分割する構成を意識しました。

Flaskの基礎講座では、1つのPythonファイルにすべての処理を記述する方法のみを学んでいたため、実際の開発では技術記事を参考にしながらフォルダ構成を検討しました。
その際、フォルダ構成やファイル名の付け方については、技術記事の内容をもとにChatGPTからヒントをもらいながら進めました。

3. 開発を通して学んだこと

今回の開発を通して、Flaskを用いたWebアプリケーションの作成やフォルダ構成の考え方、Instagram・Threadsとの連携、Cloudflare R2の使い方などを学びました。

特にトークンの設定部分については、本記事で細かくまとめたことで、自分の中でも理解が整理され、より深く理解できたと感じています。
また、Flaskでファイルを分割していく中で必要となる Blueprint や current_app などの仕組みについても学ぶことができ、とても良い勉強になりました。

今後の展望

現在、本ポートフォリオをベースに実際の制作依頼を受けているため、追記ページや機能追加について打ち合わせを行いながら、開発・調整を進めていく予定です。

また、Threadsのアクセストークンは無期限トークンを取得できず、60日で有効期限が切れてしまいます。
有効期限内であれば、現在取得しているトークンを使用して新しいトークンを生成できるようなので、定期実行によってトークンをリフレッシュする処理の実装にも取り組みたいと考えています。