🌁
instagramに複数の画像をAPI経由で投稿する
何はともあれ公式を見る
の続き
ドキュメントによるとカルーセル投稿という方式で複数画像が投稿できるらしい。
- アイテムコンテナの作成
- カルーセルコンテナの作成
- カルーセルコンテナの公開
この順序でやれば良い。またアップロードする画像はインターネット上に存在していなければならないので、パブリックアクセス可能なS3バケットを用意しておき、インスタへのアップが終わったら消す方式にする。
今回使うライブラリはこれら
requests==2.31.0
pillow==10.2.0
boto3==1.34.23
投稿できる画像の縛り
ドキュメントによると以下の画像しか投稿できないのに注意。
フォーマット: JPEG
ファイルサイズ: 最大8 MB。
アスペクト比: 4:5から1.91:1までの範囲内
最小幅: 320 (必要な場合、この最小幅まで拡大されます)
最大幅: 1440 (必要な場合、この最大幅まで縮小されます)
高さ: 幅とアスペクト比に応じて可変
カラースペース: sRGB。画像で他の色空間を使用している場合、sRGBに変換されます。
必要に応じてリサイズ
必要に応じ、このようなリサイズ関数を用意したほうがいいかもしれない。これはかなり無理矢理に作ったものなので参考にしかしないほうがいい
def resize_image(input_path, output_path, target_aspect_ratio_range):
# 元画像を開く
original_image = Image.open(input_path)
# 元画像のサイズを取得
original_width, original_height = original_image.size
# 元画像のアスペクト比を計算
original_aspect_ratio = original_width / original_height
# 目標のアスペクト比の範囲を指定
min_aspect_ratio, max_aspect_ratio = target_aspect_ratio_range
# 目標のアスペクト比に対する新しいサイズを計算
if original_aspect_ratio < min_aspect_ratio:
new_width = int(original_height * min_aspect_ratio)
new_height = original_height
elif original_aspect_ratio > max_aspect_ratio:
new_width = original_width
new_height = int(original_width / max_aspect_ratio)
else:
new_width = original_width
new_height = original_height
# 新しいサイズで画像をリサイズ
resized_image = original_image.resize(
(new_width, new_height), Image.LANCZOS)
# 新しい画像のアスペクト比を計算
new_aspect_ratio = new_width / new_height
# 余白を計算
left_padding = 0
top_padding = 0
right_padding = 0
bottom_padding = 0
if new_aspect_ratio < max_aspect_ratio:
# 右側に余白を追加
right_padding = int((max_aspect_ratio * new_height - new_width) / 2)
else:
# 下側に余白を追加
bottom_padding = int((new_width / max_aspect_ratio - new_height) / 2)
# 余白を追加して新しいサイズに調整
padded_image = Image.new('RGB', (new_width + left_padding + right_padding,
new_height + top_padding + bottom_padding), (255, 255, 255))
padded_image.paste(resized_image, (left_padding, top_padding))
# 出力先に保存
padded_image.save(output_path.replace("png", "jpeg"))
S3へのアップロード
def upload_images_to_s3(local_folder, s3_bucket_name):
s3 = boto3.client('s3')
s3_public_urls = []
for root, _, files in os.walk(local_folder):
for file in files:
local_file_path = os.path.join(root, file)
s3_object_key = f'{os.path.basename(root)}/{file}'
try:
# ファイルをS3にアップロードします
s3.upload_file(local_file_path, s3_bucket_name, s3_object_key)
# アップロードされたファイルの公開URLを生成します
s3_public_url = f'https://{s3_bucket_name}.s3.amazonaws.com/{s3_object_key}'
s3_public_urls.append(s3_public_url)
print(
f'Successfully uploaded {local_file_path} to {s3_public_url}')
except Exception as e:
print(f'Error uploading {local_file_path}: {e}')
return s3_public_urls
投稿の前準備
こんなJsonを投げる
{
'id': 1,
'media_url': https:/hogehoge,
'type': 'IMAGE'
},
{
'id': 2,
'media_url': https:/forbar,
'type': 'IMAGE'
}
APIを叩く用途のラッパー関数
たまにリクエストが失敗するのでリトライを入れたおいたほうがいい
def instagram_api(url, method, post_data):
# APIを叩く関数
max_retries = 5
retry_delay = 1 # seconds
for retry_count in range(max_retries):
try:
data = post_data
headers = {
'Authorization': 'Bearer ' + accessToken,
'Content-Type': 'application/json',
}
options = {
'headers': headers,
'data': json.dumps(data),
'timeout': 60, # Add timeout (in seconds) as needed
}
response = requests.request(method, url, **options)
if response.status_code == 200:
return response
else:
print(
f'Instagram APIのリクエストが失敗しました。ステータスコード: {response.status_code}')
except Exception as error:
print(f'Instagram APIのリクエスト中にエラーが発生しました: {error}')
if retry_count < max_retries - 1:
print(f'リトライ {retry_count + 1}/{max_retries} を待機中...')
time.sleep(retry_delay)
print(f'{max_retries}回のリトライで成功しなかったため、処理を中止します。')
return None
アイテムコンテナ作成
こんなjson({'id':XXXXXX})が返ってくる。
def make_contena_api(pre_post_data):
# ステップ①: 画像と動画を登録し、コンテナIDを画像と動画の分取得する
contena_ids = []
for data in pre_post_data:
if data['type'] == 'IMAGE':
post_data = {
'image_url': data['media_url'],
'media_type': '',
'is_carousel_item': True
}
else:
print("画像以外が登録されてようとされている")
print(data)
sys.exit(1)
url = f'https://graph.facebook.com/v17.0/{instaBusinessId}/media?'
response = instagram_api(url, 'POST', post_data)
try:
if response:
data = response.json()
contena_ids.append(data['id'])
print("正常にデータが登録された")
print(data)
else:
print('Instagram APIのリクエストでエラーが発生しました。')
print("エラーになったデータ")
pprint(post_data)
print("エラーレスポンス")
print(response.text)
sys.exit(1)
except Exception as error:
print('Instagram APIのレスポンスの解析中にエラーが発生しました:', error)
return None
return contena_ids
カルーセルコンテナ作成
contenaだお(^ω^)
def make_group_contena_api(caption, pre_post_data):
contena_ids = make_contena_api(pre_post_data)
time.sleep(20) # DB登録を待つため一旦ストップ
post_data = {
'media_type': 'CAROUSEL',
'caption': caption,
'children': contena_ids
}
# グループコンテナID取得
url = f'https://graph.facebook.com/v17.0/{instaBusinessId}/media?'
response = instagram_api(url, 'POST', post_data)
try:
if response:
data = response.json()
return data['id']
else:
print('Instagram APIのリクエストでエラーが発生しました。')
return None
except Exception as error:
print('Instagram APIのレスポンスの解析中にエラーが発生しました:', error)
return None
カルーセルコンテナの公開
さっきのを登録するお(^ω^)
def content_publish_api(description, pre_post_data):
group_contena_id = make_group_contena_api(description, pre_post_data)
time.sleep(20) # DB登録を待つため一旦ストップ
# グループコンテナIDを使って投稿
contena_group_id = group_contena_id
post_data = {
'media_type': 'CAROUSEL',
'creation_id': contena_group_id
}
url = f'https://graph.facebook.com/v17.0/{instaBusinessId}/media_publish?'
response = instagram_api(url, 'POST', post_data)
try:
if response:
data = response.json()
return data
else:
print('Instagram APIのリクエストでエラーが発生しました。')
return None
except Exception as error:
print('Instagram APIのレスポンスの解析中にエラーが発生しました:', error)
return None
感想
終わった画像はaws s3 rm s3://{bucket-name}d/img/ --recursiveとかで消せばいい、でも若干の課金は発生する。まあまあ癖あってめんどい。
Discussion