326 lines
11 KiB
Python
326 lines
11 KiB
Python
"""
|
|
Instagram Graph API POC 테스트
|
|
|
|
이 파일은 InstagramClient의 각 기능을 테스트합니다.
|
|
|
|
실행 방법:
|
|
```bash
|
|
# 환경변수 설정
|
|
export INSTAGRAM_ACCESS_TOKEN="your_access_token"
|
|
|
|
# 실행
|
|
python -m poc.instagram.main
|
|
```
|
|
|
|
주의사항:
|
|
- 게시 테스트는 실제로 Instagram에 게시됩니다.
|
|
- 테스트 전 토큰이 올바른지 확인하세요.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import sys
|
|
|
|
# 로깅 설정
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(name)s - %(message)s",
|
|
handlers=[logging.StreamHandler(sys.stdout)],
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_access_token() -> str:
|
|
"""환경변수에서 액세스 토큰 가져오기"""
|
|
token = "EAAmAhD98ZBY8BQg4PjcQrQFnHPoLLgMdbAsPz80oIVVQxAGjlAHgO1lyjzGsBi5ugIHPanmozFVyZAN4OZACESqeASAgn4rdxnyGYiWiGTME0uAm9dUmtYRpNJtlyslCkn9ee1YQVlZBgyS5PpVfXP1tV7cPJh2EHUZBwvsXnAZAYVDfdAKVZAy3kZB62VTugBt7"
|
|
if not token:
|
|
print("=" * 60)
|
|
print("오류: INSTAGRAM_ACCESS_TOKEN 환경변수가 설정되지 않았습니다.")
|
|
print()
|
|
print("설정 방법:")
|
|
print(" Windows PowerShell:")
|
|
print(' $env:INSTAGRAM_ACCESS_TOKEN = "your_token_here"')
|
|
print()
|
|
print(" Windows CMD:")
|
|
print(" set INSTAGRAM_ACCESS_TOKEN=your_token_here")
|
|
print()
|
|
print(" Linux/macOS:")
|
|
print(' export INSTAGRAM_ACCESS_TOKEN="your_token_here"')
|
|
print("=" * 60)
|
|
sys.exit(1)
|
|
return token
|
|
|
|
|
|
async def test_get_media_list():
|
|
"""미디어 목록 조회 테스트"""
|
|
from poc.instagram.client import InstagramClient
|
|
|
|
print("\n" + "=" * 60)
|
|
print("1. 미디어 목록 조회 테스트")
|
|
print("=" * 60)
|
|
|
|
access_token = get_access_token()
|
|
|
|
try:
|
|
async with InstagramClient(access_token=access_token) as client:
|
|
media_list = await client.get_media_list(limit=5)
|
|
|
|
print(f"\n최근 게시물 ({len(media_list.data)}개)")
|
|
print("-" * 50)
|
|
|
|
for i, media in enumerate(media_list.data, 1):
|
|
caption_preview = (
|
|
media.caption[:40] + "..."
|
|
if media.caption and len(media.caption) > 40
|
|
else media.caption or "(캡션 없음)"
|
|
)
|
|
print(f"\n{i}. [{media.media_type}] {caption_preview}")
|
|
print(f" ID: {media.id}")
|
|
print(f" 좋아요: {media.like_count:,}")
|
|
print(f" 댓글: {media.comments_count:,}")
|
|
print(f" 게시일: {media.timestamp}")
|
|
print(f" 링크: {media.permalink}")
|
|
|
|
if media_list.next_cursor:
|
|
print(f"\n다음 페이지 있음 (cursor: {media_list.next_cursor[:20]}...)")
|
|
|
|
print("\n[성공] 미디어 목록 조회 완료")
|
|
|
|
except Exception as e:
|
|
print(f"\n[실패] 에러: {e}")
|
|
raise
|
|
|
|
|
|
async def test_get_media_detail():
|
|
"""미디어 상세 조회 테스트"""
|
|
from poc.instagram.client import InstagramClient
|
|
|
|
print("\n" + "=" * 60)
|
|
print("2. 미디어 상세 조회 테스트")
|
|
print("=" * 60)
|
|
|
|
access_token = get_access_token()
|
|
|
|
try:
|
|
async with InstagramClient(access_token=access_token) as client:
|
|
# 먼저 목록에서 첫 번째 미디어 ID 가져오기
|
|
media_list = await client.get_media_list(limit=1)
|
|
if not media_list.data:
|
|
print("\n게시물이 없습니다.")
|
|
return
|
|
|
|
media_id = media_list.data[0].id
|
|
print(f"\n조회할 미디어 ID: {media_id}")
|
|
|
|
# 상세 조회
|
|
media = await client.get_media(media_id)
|
|
|
|
print("\n미디어 상세 정보")
|
|
print("-" * 50)
|
|
print(f"ID: {media.id}")
|
|
print(f"타입: {media.media_type}")
|
|
print(f"URL: {media.media_url}")
|
|
print(f"게시일: {media.timestamp}")
|
|
print(f"좋아요: {media.like_count:,}")
|
|
print(f"댓글: {media.comments_count:,}")
|
|
print(f"퍼머링크: {media.permalink}")
|
|
|
|
if media.caption:
|
|
print("\n캡션:")
|
|
print(f" {media.caption}")
|
|
|
|
if media.children:
|
|
print(f"\n캐러셀 하위 미디어 ({len(media.children)}개)")
|
|
for j, child in enumerate(media.children, 1):
|
|
print(f" {j}. [{child.media_type}] {child.media_url}")
|
|
|
|
print("\n[성공] 미디어 상세 조회 완료")
|
|
|
|
except Exception as e:
|
|
print(f"\n[실패] 에러: {e}")
|
|
raise
|
|
|
|
|
|
async def test_publish_image():
|
|
"""이미지 게시 테스트 (주석 처리됨 - 실제 게시됨)"""
|
|
|
|
print("\n" + "=" * 60)
|
|
print("3. 이미지 게시 테스트")
|
|
print("=" * 60)
|
|
|
|
# 테스트 설정 (공개 접근 가능한 이미지 URL 필요)
|
|
TEST_IMAGE_URL = "https://example.com/test-image.jpg"
|
|
TEST_CAPTION = "Test post from Instagram POC #test"
|
|
|
|
print("\n이 테스트는 실제로 게시물을 작성합니다!")
|
|
print(f" 이미지 URL: {TEST_IMAGE_URL}")
|
|
print(f" 캡션: {TEST_CAPTION}")
|
|
print("\n테스트하려면 아래 코드의 주석을 해제하세요.")
|
|
print("[건너뜀] 이미지 게시 테스트 (주석 처리됨)")
|
|
|
|
# ==========================================================================
|
|
# 실제 테스트 - 주석 해제 시 실행됨
|
|
# ==========================================================================
|
|
# from poc.instagram.exceptions import InstagramAPIError
|
|
# access_token = get_access_token()
|
|
#
|
|
# try:
|
|
# async with InstagramClient(access_token=access_token) as client:
|
|
# media = await client.publish_image(
|
|
# image_url=TEST_IMAGE_URL,
|
|
# caption=TEST_CAPTION,
|
|
# )
|
|
# print(f"\n[성공] 게시 완료!")
|
|
# print(f" 미디어 ID: {media.id}")
|
|
# print(f" 링크: {media.permalink}")
|
|
#
|
|
# except InstagramAPIError as e:
|
|
# print(f"\n[실패] 게시 실패: {e}")
|
|
# except Exception as e:
|
|
# print(f"\n[실패] 에러: {e}")
|
|
|
|
|
|
async def test_publish_video():
|
|
"""비디오/릴스 게시 테스트 (주석 처리됨 - 실제 게시됨)"""
|
|
|
|
print("\n" + "=" * 60)
|
|
print("4. 비디오/릴스 게시 테스트")
|
|
print("=" * 60)
|
|
|
|
TEST_VIDEO_URL = "https://f002.backblazeb2.com/file/creatomate-c8xg3hsxdu/9b1a680b-3481-4b22-94d4-a5cfd3e19f95.mp4"
|
|
TEST_CAPTION = "Test video from Instagram POC #test"
|
|
|
|
print("\n이 테스트는 실제로 게시물을 작성합니다!")
|
|
print(f" 비디오 URL: {TEST_VIDEO_URL}")
|
|
print(f" 캡션: {TEST_CAPTION}")
|
|
print("\n테스트하려면 아래 코드의 주석을 해제하세요.")
|
|
print("[건너뜀] 비디오 게시 테스트 (주석 처리됨)")
|
|
|
|
# ==========================================================================
|
|
# 실제 테스트 - 주석 해제 시 실행됨
|
|
# ==========================================================================
|
|
# from poc.instagram.exceptions import InstagramAPIError
|
|
# access_token = get_access_token()
|
|
#
|
|
# try:
|
|
# async with InstagramClient(access_token=access_token) as client:
|
|
# media = await client.publish_video(
|
|
# video_url=TEST_VIDEO_URL,
|
|
# caption=TEST_CAPTION,
|
|
# share_to_feed=True,
|
|
# )
|
|
# print(f"\n[성공] 게시 완료!")
|
|
# print(f" 미디어 ID: {media.id}")
|
|
# print(f" 링크: {media.permalink}")
|
|
#
|
|
# except InstagramAPIError as e:
|
|
# print(f"\n[실패] 게시 실패: {e}")
|
|
# except Exception as e:
|
|
# print(f"\n[실패] 에러: {e}")
|
|
|
|
|
|
async def test_publish_carousel():
|
|
"""캐러셀 게시 테스트 (주석 처리됨 - 실제 게시됨)"""
|
|
|
|
print("\n" + "=" * 60)
|
|
print("5. 캐러셀(멀티 이미지) 게시 테스트")
|
|
print("=" * 60)
|
|
|
|
TEST_IMAGE_URLS = [
|
|
"https://example.com/image1.jpg",
|
|
"https://example.com/image2.jpg",
|
|
"https://example.com/image3.jpg",
|
|
]
|
|
TEST_CAPTION = "Test carousel from Instagram POC #test"
|
|
|
|
print("\n이 테스트는 실제로 게시물을 작성합니다!")
|
|
print(f" 이미지 수: {len(TEST_IMAGE_URLS)}개")
|
|
print(f" 캡션: {TEST_CAPTION}")
|
|
print("\n테스트하려면 아래 코드의 주석을 해제하세요.")
|
|
print("[건너뜀] 캐러셀 게시 테스트 (주석 처리됨)")
|
|
|
|
# ==========================================================================
|
|
# 실제 테스트 - 주석 해제 시 실행됨
|
|
# ==========================================================================
|
|
# from poc.instagram.exceptions import InstagramAPIError
|
|
# access_token = get_access_token()
|
|
#
|
|
# try:
|
|
# async with InstagramClient(access_token=access_token) as client:
|
|
# media = await client.publish_carousel(
|
|
# media_urls=TEST_IMAGE_URLS,
|
|
# caption=TEST_CAPTION,
|
|
# )
|
|
# print(f"\n[성공] 게시 완료!")
|
|
# print(f" 미디어 ID: {media.id}")
|
|
# print(f" 링크: {media.permalink}")
|
|
#
|
|
# except InstagramAPIError as e:
|
|
# print(f"\n[실패] 게시 실패: {e}")
|
|
# except Exception as e:
|
|
# print(f"\n[실패] 에러: {e}")
|
|
|
|
|
|
async def test_error_handling():
|
|
"""에러 처리 테스트"""
|
|
from poc.instagram.client import InstagramClient
|
|
from poc.instagram.exceptions import (
|
|
AuthenticationError,
|
|
InstagramAPIError,
|
|
RateLimitError,
|
|
)
|
|
|
|
print("\n" + "=" * 60)
|
|
print("6. 에러 처리 테스트")
|
|
print("=" * 60)
|
|
|
|
# 잘못된 토큰으로 테스트
|
|
print("\n잘못된 토큰으로 요청 테스트:")
|
|
|
|
try:
|
|
async with InstagramClient(access_token="INVALID_TOKEN") as client:
|
|
await client.get_media_list(limit=1)
|
|
print("[실패] 예외가 발생하지 않음")
|
|
|
|
except AuthenticationError as e:
|
|
print(f"[성공] AuthenticationError 발생: {e}")
|
|
|
|
except RateLimitError as e:
|
|
print(f"[성공] RateLimitError 발생: {e}")
|
|
if e.retry_after:
|
|
print(f" 재시도 대기 시간: {e.retry_after}초")
|
|
|
|
except InstagramAPIError as e:
|
|
print(f"[성공] InstagramAPIError 발생: {e}")
|
|
print(f" 코드: {e.code}, 서브코드: {e.subcode}")
|
|
|
|
except Exception as e:
|
|
print(f"[성공] 예외 발생: {type(e).__name__}: {e}")
|
|
|
|
|
|
async def main():
|
|
"""모든 테스트 실행"""
|
|
print("\n" + "=" * 60)
|
|
print("Instagram Graph API POC 테스트")
|
|
print("=" * 60)
|
|
|
|
# 조회 테스트 (안전)
|
|
await test_get_media_list()
|
|
await test_get_media_detail()
|
|
|
|
# 게시 테스트 (기본 비활성화)
|
|
await test_publish_image()
|
|
await test_publish_video()
|
|
await test_publish_carousel()
|
|
|
|
# 에러 처리 테스트
|
|
await test_error_handling()
|
|
|
|
print("\n" + "=" * 60)
|
|
print("모든 테스트 완료")
|
|
print("=" * 60)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|