o2o-castad-backend/poc/instagram/main.py

113 lines
4.3 KiB
Python

"""
Instagram Graph API POC - 비디오 업로드 테스트
실행 방법:
python -m poc.instagram.main
"""
import asyncio
import logging
import sys
from poc.instagram import InstagramClient, ErrorState, parse_instagram_error
# 로깅 설정
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s - %(message)s",
handlers=[logging.StreamHandler(sys.stdout)],
)
logger = logging.getLogger(__name__)
# 설정
ACCESS_TOKEN = ""
VIDEO_URL2 = "https://f002.backblazeb2.com/file/creatomate-c8xg3hsxdu/9b1a680b-3481-4b22-94d4-a5cfd3e19f95.mp4"
VIDEO_URL3 = "https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/019c1c1c-e311-756d-8635-bfe62898f73e/019c1c1d-1a3e-78c9-819a-a9de16f487c7/video/스테이머뭄.mp4"
VIDEO_URL1 = "https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/019c1d13-db76-7bfa-849f-02803d9e39fb/019c1d21-b686-7dee-b04e-97c8ffe99c28/video/스테이 머뭄.mp4"
VIDEO_URL = "https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/019c1d13-db76-7bfa-849f-02803d9e39fb/019c1d21-b686-7dee-b04e-97c8ffe99c28/video/28aa6541ddd74c348c5aae730a232454.mp4"
VIDEO_CAPTION = "Test video from Instagram POC #test"
async def main():
"""비디오 업로드 POC 실행"""
print("\n" + "=" * 60)
print("Instagram Graph API - 비디오 업로드 POC")
print("=" * 60)
async with InstagramClient(access_token=ACCESS_TOKEN) as client:
# Step 1: 접속 테스트
print("\n[Step 1] 접속 테스트")
print("-" * 40)
try:
account_id = await client.get_account_id()
print("[성공] 접속 확인 완료")
print(f" Account ID: {account_id}")
except Exception as e:
error_state, message, extra_info = parse_instagram_error(e)
if error_state == ErrorState.AUTH_ERROR:
print(f"[실패] 인증 실패: {message}")
else:
print(f"[실패] 접속 실패: {message}")
return
# Step 2: 비디오 업로드
print("\n[Step 2] 비디오 업로드")
print("-" * 40)
print(f" 비디오 URL: {VIDEO_URL}")
print(f" 캡션: {VIDEO_CAPTION}")
print("\n업로드 중... (비디오 처리에 시간이 걸릴 수 있습니다)")
try:
media = await client.publish_video(
video_url=VIDEO_URL,
caption=VIDEO_CAPTION,
share_to_feed=True,
)
print("\n[성공] 비디오 업로드 완료!")
print(f" 미디어 ID: {media.id}")
print(f" 링크: {media.permalink}")
except Exception as e:
error_state, message, extra_info = parse_instagram_error(e)
if error_state == ErrorState.RATE_LIMIT:
retry_after = extra_info.get("retry_after", 60)
print(f"\n[실패] Rate Limit: {message} (재시도: {retry_after}초)")
elif error_state == ErrorState.CONTAINER_TIMEOUT:
print(f"\n[실패] 타임아웃: {message}")
elif error_state == ErrorState.CONTAINER_ERROR:
status = extra_info.get("status", "UNKNOWN")
print(f"\n[실패] 컨테이너 에러: {message} (상태: {status})")
else:
print(f"\n[실패] 업로드 실패: {message}")
return
# Step 3: 업로드 확인
print("\n[Step 3] 업로드 확인")
print("-" * 40)
try:
verified_media = await client.get_media(media.id)
print("[성공] 업로드 확인 완료!")
print(f" 미디어 ID: {verified_media.id}")
print(f" 타입: {verified_media.media_type}")
print(f" URL: {verified_media.media_url}")
print(f" 퍼머링크: {verified_media.permalink}")
print(f" 게시일: {verified_media.timestamp}")
if verified_media.caption:
print(f" 캡션: {verified_media.caption}")
except Exception as e:
error_state, message, _ = parse_instagram_error(e)
print(f"[실패] 확인 실패: {message}")
return
print("\n" + "=" * 60)
print("모든 단계 완료!")
print("=" * 60)
if __name__ == "__main__":
asyncio.run(main())