""" Instagram Graph API Client Instagram Graph API를 사용한 콘텐츠 게시 및 조회를 위한 비동기 클라이언트입니다. 멀티테넌트 지원 - 각 사용자가 자신의 access_token으로 인스턴스를 생성합니다. Example: ```python async with InstagramClient(access_token="YOUR_TOKEN") as client: media = await client.publish_image( image_url="https://example.com/image.jpg", caption="Hello Instagram!" ) print(f"게시 완료: {media.permalink}") ``` """ import asyncio import logging import time from typing import Any, Optional import httpx from .exceptions import ( ContainerStatusError, ContainerTimeoutError, InstagramAPIError, RateLimitError, create_exception_from_error, ) from .models import ErrorResponse, Media, MediaContainer, MediaList logger = logging.getLogger(__name__) class InstagramClient: """ Instagram Graph API 비동기 클라이언트 멀티테넌트 지원 - 각 사용자가 자신의 access_token으로 인스턴스를 생성합니다. 비동기 컨텍스트 매니저로 사용해야 합니다. Example: ```python async with InstagramClient(access_token="USER_TOKEN") as client: media = await client.publish_image( image_url="https://example.com/image.jpg", caption="My photo!" ) print(f"게시됨: {media.permalink}") ``` """ DEFAULT_BASE_URL = "https://graph.instagram.com/v21.0" def __init__( self, access_token: str, *, base_url: Optional[str] = None, timeout: float = 30.0, max_retries: int = 3, container_timeout: float = 300.0, container_poll_interval: float = 5.0, ): """ 클라이언트 초기화 Args: access_token: Instagram 액세스 토큰 (필수) base_url: API 기본 URL (기본값: https://graph.instagram.com/v21.0) timeout: HTTP 요청 타임아웃 (초) max_retries: 최대 재시도 횟수 container_timeout: 컨테이너 처리 대기 타임아웃 (초) container_poll_interval: 컨테이너 상태 확인 간격 (초) """ if not access_token: raise ValueError("access_token은 필수입니다.") self.access_token = access_token self.base_url = base_url or self.DEFAULT_BASE_URL self.timeout = timeout self.max_retries = max_retries self.container_timeout = container_timeout self.container_poll_interval = container_poll_interval self._client: Optional[httpx.AsyncClient] = None self._account_id: Optional[str] = None self._account_id_lock: asyncio.Lock = asyncio.Lock() async def __aenter__(self) -> "InstagramClient": """비동기 컨텍스트 매니저 진입""" self._client = httpx.AsyncClient( timeout=httpx.Timeout(self.timeout), follow_redirects=True, ) logger.debug("[InstagramClient] HTTP 클라이언트 초기화 완료") return self async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: """비동기 컨텍스트 매니저 종료""" if self._client: await self._client.aclose() self._client = None logger.debug("[InstagramClient] HTTP 클라이언트 종료") def _get_client(self) -> httpx.AsyncClient: """HTTP 클라이언트 반환""" if self._client is None: raise RuntimeError( "InstagramClient는 비동기 컨텍스트 매니저로 사용해야 합니다. " "예: async with InstagramClient(access_token=...) as client:" ) return self._client def _build_url(self, endpoint: str) -> str: """API URL 생성""" return f"{self.base_url}/{endpoint}" async def _request( self, method: str, endpoint: str, params: Optional[dict[str, Any]] = None, data: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: """ 공통 HTTP 요청 처리 - Rate Limit 시 지수 백오프 재시도 - 에러 응답 시 InstagramAPIError 발생 """ client = self._get_client() url = self._build_url(endpoint) params = params or {} params["access_token"] = self.access_token retry_base_delay = 1.0 last_exception: Optional[Exception] = None for attempt in range(self.max_retries + 1): try: logger.debug( f"[API] {method} {endpoint} (attempt {attempt + 1}/{self.max_retries + 1})" ) response = await client.request( method=method, url=url, params=params, data=data, ) # Rate Limit 체크 (429) if response.status_code == 429: retry_after = int(response.headers.get("Retry-After", 60)) if attempt < self.max_retries: wait_time = max(retry_base_delay * (2**attempt), retry_after) logger.warning(f"Rate limit 초과. {wait_time}초 후 재시도...") await asyncio.sleep(wait_time) continue raise RateLimitError( message="Rate limit 초과 (최대 재시도 횟수 도달)", retry_after=retry_after, ) # 서버 에러 재시도 (5xx) if response.status_code >= 500: if attempt < self.max_retries: wait_time = retry_base_delay * (2**attempt) logger.warning(f"서버 에러 {response.status_code}. {wait_time}초 후 재시도...") await asyncio.sleep(wait_time) continue response.raise_for_status() # JSON 파싱 response_data = response.json() # API 에러 체크 (Instagram API는 200 응답에도 error 포함 가능) if "error" in response_data: error_response = ErrorResponse.model_validate(response_data) err = error_response.error logger.error(f"[API Error] code={err.code}, message={err.message}") raise create_exception_from_error( message=err.message, code=err.code, subcode=err.error_subcode, fbtrace_id=err.fbtrace_id, ) return response_data except InstagramAPIError: raise except httpx.HTTPError as e: last_exception = e if attempt < self.max_retries: wait_time = retry_base_delay * (2**attempt) logger.warning(f"HTTP 에러: {e}. {wait_time}초 후 재시도...") await asyncio.sleep(wait_time) continue raise # 이 지점에 도달하면 안 되지만, 타입 체커를 위해 명시적 raise raise last_exception or InstagramAPIError("최대 재시도 횟수 초과") async def _wait_for_container( self, container_id: str, timeout: Optional[float] = None, ) -> MediaContainer: """컨테이너 상태가 FINISHED가 될 때까지 대기""" timeout = timeout or self.container_timeout start_time = time.monotonic() logger.debug(f"[Container] 대기 시작: {container_id}, timeout={timeout}s") while True: elapsed = time.monotonic() - start_time if elapsed >= timeout: raise ContainerTimeoutError( f"컨테이너 처리 타임아웃 ({timeout}초 초과): {container_id}" ) response = await self._request( method="GET", endpoint=container_id, params={"fields": "status_code,status"}, ) container = MediaContainer.model_validate(response) logger.debug(f"[Container] status={container.status_code}, elapsed={elapsed:.1f}s") if container.is_finished: logger.info(f"[Container] 완료: {container_id}") return container if container.is_error: raise ContainerStatusError(f"컨테이너 처리 실패: {container.status}") await asyncio.sleep(self.container_poll_interval) async def _get_account_id(self) -> str: """계정 ID 조회 (캐시됨, 동시성 안전)""" if self._account_id: return self._account_id async with self._account_id_lock: # Double-check after acquiring lock if self._account_id: return self._account_id response = await self._request( method="GET", endpoint="me", params={"fields": "id"}, ) account_id: str = response["id"] self._account_id = account_id logger.debug(f"[Account] ID 조회 완료: {account_id}") return account_id async def get_media_list( self, limit: int = 25, after: Optional[str] = None, ) -> MediaList: """ 미디어 목록 조회 Args: limit: 조회할 미디어 수 (최대 100) after: 페이지네이션 커서 Returns: MediaList: 미디어 목록 Raises: httpx.HTTPStatusError: API 에러 발생 시 """ logger.info(f"[get_media_list] limit={limit}") account_id = await self._get_account_id() params: dict[str, Any] = { "fields": "id,media_type,media_url,thumbnail_url,caption,timestamp,permalink,like_count,comments_count", "limit": min(limit, 100), } if after: params["after"] = after response = await self._request( method="GET", endpoint=f"{account_id}/media", params=params, ) result = MediaList.model_validate(response) logger.info(f"[get_media_list] 완료: {len(result.data)}개") return result async def get_media(self, media_id: str) -> Media: """ 미디어 상세 조회 Args: media_id: 미디어 ID Returns: Media: 미디어 상세 정보 Raises: httpx.HTTPStatusError: API 에러 발생 시 """ logger.info(f"[get_media] media_id={media_id}") response = await self._request( method="GET", endpoint=media_id, params={ "fields": "id,media_type,media_url,thumbnail_url,caption,timestamp,permalink,like_count,comments_count,children{id,media_type,media_url}", }, ) result = Media.model_validate(response) logger.info(f"[get_media] 완료: type={result.media_type}, likes={result.like_count}") return result async def publish_image( self, image_url: str, caption: Optional[str] = None, ) -> Media: """ 이미지 게시 Args: image_url: 공개 접근 가능한 이미지 URL (JPEG 권장) caption: 게시물 캡션 Returns: Media: 게시된 미디어 정보 Raises: httpx.HTTPStatusError: API 에러 발생 시 TimeoutError: 컨테이너 처리 타임아웃 """ logger.info(f"[publish_image] 시작: {image_url[:50]}...") account_id = await self._get_account_id() # Step 1: Container 생성 container_params: dict[str, Any] = {"image_url": image_url} if caption: container_params["caption"] = caption container_response = await self._request( method="POST", endpoint=f"{account_id}/media", params=container_params, ) container_id = container_response["id"] logger.debug(f"[publish_image] Container 생성: {container_id}") # Step 2: Container 상태 대기 await self._wait_for_container(container_id) # Step 3: 게시 publish_response = await self._request( method="POST", endpoint=f"{account_id}/media_publish", params={"creation_id": container_id}, ) media_id = publish_response["id"] result = await self.get_media(media_id) logger.info(f"[publish_image] 완료: {result.permalink}") return result async def publish_video( self, video_url: str, caption: Optional[str] = None, share_to_feed: bool = True, ) -> Media: """ 비디오/릴스 게시 Args: video_url: 공개 접근 가능한 비디오 URL (MP4 권장) caption: 게시물 캡션 share_to_feed: 피드에 공유 여부 Returns: Media: 게시된 미디어 정보 Raises: httpx.HTTPStatusError: API 에러 발생 시 TimeoutError: 컨테이너 처리 타임아웃 """ logger.info(f"[publish_video] 시작: {video_url[:50]}...") account_id = await self._get_account_id() # Step 1: Container 생성 container_params: dict[str, Any] = { "media_type": "REELS", "video_url": video_url, "share_to_feed": str(share_to_feed).lower(), } if caption: container_params["caption"] = caption container_response = await self._request( method="POST", endpoint=f"{account_id}/media", params=container_params, ) container_id = container_response["id"] logger.debug(f"[publish_video] Container 생성: {container_id}") # Step 2: Container 상태 대기 (비디오는 더 오래 걸림) await self._wait_for_container(container_id, timeout=self.container_timeout * 2) # Step 3: 게시 publish_response = await self._request( method="POST", endpoint=f"{account_id}/media_publish", params={"creation_id": container_id}, ) media_id = publish_response["id"] result = await self.get_media(media_id) logger.info(f"[publish_video] 완료: {result.permalink}") return result async def publish_carousel( self, media_urls: list[str], caption: Optional[str] = None, ) -> Media: """ 캐러셀(멀티 이미지) 게시 Args: media_urls: 이미지 URL 목록 (2-10개) caption: 게시물 캡션 Returns: Media: 게시된 미디어 정보 Raises: ValueError: 이미지 수가 2-10개가 아닌 경우 httpx.HTTPStatusError: API 에러 발생 시 TimeoutError: 컨테이너 처리 타임아웃 """ if len(media_urls) < 2 or len(media_urls) > 10: raise ValueError("캐러셀은 2-10개의 이미지가 필요합니다.") logger.info(f"[publish_carousel] 시작: {len(media_urls)}개 이미지") account_id = await self._get_account_id() # Step 1: 각 이미지의 Container 병렬 생성 async def create_item_container(url: str, index: int) -> str: response = await self._request( method="POST", endpoint=f"{account_id}/media", params={"image_url": url, "is_carousel_item": "true"}, ) logger.debug(f"[publish_carousel] 이미지 {index + 1} Container 생성 완료") return response["id"] children_ids = await asyncio.gather( *[create_item_container(url, i) for i, url in enumerate(media_urls)] ) logger.debug(f"[publish_carousel] 모든 Container 생성 완료: {len(children_ids)}개") # Step 2: 캐러셀 Container 생성 carousel_params: dict[str, Any] = { "media_type": "CAROUSEL", "children": ",".join(children_ids), } if caption: carousel_params["caption"] = caption carousel_response = await self._request( method="POST", endpoint=f"{account_id}/media", params=carousel_params, ) carousel_id = carousel_response["id"] # Step 3: Container 상태 대기 await self._wait_for_container(carousel_id) # Step 4: 게시 publish_response = await self._request( method="POST", endpoint=f"{account_id}/media_publish", params={"creation_id": carousel_id}, ) media_id = publish_response["id"] result = await self.get_media(media_id) logger.info(f"[publish_carousel] 완료: {result.permalink}") return result