Instagram Platform API Integration Guide

# Instagram Platform API Configuration
INSTAGRAM_APP_ID=your_instagram_app_id
INSTAGRAM_APP_SECRET=your_instagram_app_secret
INSTAGRAM_GRAPH_API_VERSION=v24.0
INSTAGRAM_REDIRECT_URI=http://localhost:8000/api/instagram/oauth-callback

# Frontend URL for OAuth redirects
FRONTEND_URL=http://localhost:3000

from sqlalchemy import Column, String, Boolean, DateTime, Integer
from sqlalchemy.dialects.postgresql import UUID
import uuid
from datetime import datetime

class InstagramAccount(Base):
    __tablename__ = "instagram_accounts"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    user_id = Column(String, nullable=False)
    product_id = Column(String, nullable=False)
    instagram_business_account_id = Column(String, nullable=True)
    instagram_user_id = Column(String, nullable=False)
    access_token = Column(String, nullable=False)
    username = Column(String, nullable=False)
    name = Column(String, nullable=True)
    profile_picture_url = Column(String, nullable=True)
    followers_count = Column(Integer, nullable=True)
    is_active = Column(Boolean, default=True)
    auto_post = Column(Boolean, default=False)
    token_expires_at = Column(DateTime, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)

from sqlalchemy import Column, String, DateTime, Integer, Text
from sqlalchemy.dialects.postgresql import UUID
import uuid
from datetime import datetime
from enum import Enum

class InstagramPostStatus(str, Enum):
    PENDING = "pending"
    POSTED = "posted"
    FAILED = "failed"
    SCHEDULED = "scheduled"

class InstagramPost(Base):
    __tablename__ = "instagram_posts"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    instagram_account_id = Column(UUID(as_uuid=True), ForeignKey("instagram_accounts.id"))
    post_id = Column(UUID(as_uuid=True))
    caption = Column(Text, nullable=False)
    image_url = Column(String, nullable=False)
    hashtags = Column(String, nullable=True)
    status = Column(String, default=InstagramPostStatus.PENDING)
    instagram_media_id = Column(String, nullable=True)
    instagram_permalink = Column(String, nullable=True)
    scheduled_for = Column(DateTime, nullable=True)
    posted_at = Column(DateTime, nullable=True)
    error_message = Column(Text, nullable=True)
    retry_count = Column(Integer, default=0)
    last_retry_at = Column(DateTime, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)

import httpx
from typing import Optional, Dict, Any
from datetime import datetime, timedelta
from urllib.parse import urlencode

class InstagramOAuthService:
    def __init__(self):
        self.app_id = os.getenv('INSTAGRAM_APP_ID')
        self.app_secret = os.getenv('INSTAGRAM_APP_SECRET')
        self.redirect_uri = os.getenv('INSTAGRAM_REDIRECT_URI')

    def get_auth_url(self, state: str) -> str:
        """Generate the OAuth authorization URL for Instagram Platform API"""
        params = {
            'client_id': self.app_id,
            'redirect_uri': self.redirect_uri,
            'scope': 'instagram_business_basic,instagram_business_content_publish',
            'state': state,
            'response_type': 'code',
        }
        return f'https://api.instagram.com/oauth/authorize?{urlencode(params)}'

    async def exchange_code_for_token(self, code: str) -> Dict[str, Any]:
        """Exchange authorization code for access token"""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                'https://api.instagram.com/oauth/access_token',
                data={
                    'client_id': self.app_id,
                    'client_secret': self.app_secret,
                    'grant_type': 'authorization_code',
                    'redirect_uri': self.redirect_uri,
                    'code': code,
                },
            )
            response.raise_for_status()
            data = response.json()

            # Exchange short-lived token for long-lived token
            if 'access_token' in data:
                long_lived_token_data = await self._get_long_lived_token(data['access_token'])
                data.update(long_lived_token_data)

            return data

    async def _get_long_lived_token(self, short_token: str) -> Dict[str, Any]:
        """Exchange short-lived token for long-lived token (60 days)"""
        try:
            async with httpx.AsyncClient() as client:
                response = await client.get(
                    'https://graph.instagram.com/access_token',
                    params={
                        'grant_type': 'ig_exchange_token',
                        'client_secret': self.app_secret,
                        'access_token': short_token,
                    },
                )
                response.raise_for_status()
                data = response.json()

                # Add expiry timestamp
                if 'expires_in' in data:
                    data['expires_at'] = (datetime.utcnow() + timedelta(seconds=data['expires_in'])).isoformat()

                return data
        except httpx.HTTPError as e:
            print(f'Failed to get long-lived token: {str(e)}')
            return {'access_token': short_token}

    async def get_instagram_accounts(self, access_token: str) -> list[Dict[str, Any]]:
        """Get Instagram account info using Instagram Platform API"""
        async with httpx.AsyncClient() as client:
            response = await client.get(
                'https://graph.instagram.com/me',
                params={
                    'fields': 'id,username,name,profile_picture_url,followers_count,media_count,account_type',
                    'access_token': access_token,
                },
            )
            response.raise_for_status()
            ig_data = response.json()

            return [{
                'id': ig_data['id'],
                'username': ig_data.get('username', ''),
                'name': ig_data.get('name', ig_data.get('username', '')),
                'profile_picture_url': ig_data.get('profile_picture_url'),
                'followers_count': ig_data.get('followers_count'),
                'media_count': ig_data.get('media_count', 0),
                'account_type': ig_data.get('account_type', 'BUSINESS'),
                'access_token': access_token,
            }]

    async def refresh_token(self, access_token: str) -> Optional[Dict[str, Any]]:
        """Refresh a long-lived token before it expires"""
        try:
            async with httpx.AsyncClient() as client:
                response = await client.get(
                    'https://graph.instagram.com/refresh_access_token',
                    params={
                        'grant_type': 'ig_refresh_token',
                        'access_token': access_token,
                    },
                )
                response.raise_for_status()
                data = response.json()

                if 'expires_in' in data:
                    data['expires_at'] = (datetime.utcnow() + timedelta(seconds=data['expires_in'])).isoformat()

                return data
        except httpx.HTTPError:
            return None

import httpx
import asyncio
from typing import Optional, Dict, Any
from datetime import datetime

class InstagramPublisher:
    def __init__(self):
        self.graph_base_url = 'https://graph.instagram.com'

    async def publish_post(self, account: InstagramAccount, image_url: str, caption: str) -> Dict[str, Any]:
        """Publish a post to Instagram"""
        try:
            # Use instagram_business_account_id or fallback to instagram_user_id
            account_id = account.instagram_business_account_id or account.instagram_user_id

            if not account_id:
                raise Exception('No Instagram account ID found. Please reconnect your Instagram account.')

            # Create media container
            container_id = await self._create_media_container(
                account_id=account_id,
                access_token=account.access_token,
                image_url=image_url,
                caption=caption,
            )

            if not container_id:
                raise Exception('Failed to create media container')

            # Wait for Instagram to process the media
            print(f'Media container created: {container_id}. Waiting for Instagram to process...')
            await asyncio.sleep(10)

            # Publish the container
            media_id = await self._publish_container(
                account_id=account_id,
                access_token=account.access_token,
                container_id=container_id,
            )

            if not media_id:
                raise Exception('Failed to publish media')

            # Get the published post details
            post_details = await self._get_post_details(media_id=media_id, access_token=account.access_token)

            print(f'Successfully published Instagram post {media_id}')

            return {
                'success': True,
                'media_id': media_id,
                'permalink': post_details.get('permalink'),
                'post_details': post_details,
            }

        except Exception as e:
            print(f'Failed to publish Instagram post: {str(e)}')
            return {'success': False, 'error': str(e)}

    async def _create_media_container(self, account_id: str, access_token: str, image_url: str, caption: str) -> Optional[str]:
        """Create a media container for Instagram"""
        try:
            async with httpx.AsyncClient() as client:
                response = await client.post(
                    f'{self.graph_base_url}/{account_id}/media',
                    data={'image_url': image_url, 'caption': caption, 'access_token': access_token},
                )
                response.raise_for_status()
                data = response.json()
                return data.get('id')
        except httpx.HTTPError as e:
            print(f'Failed to create media container: {str(e)}')
            return None

    async def _publish_container(self, account_id: str, access_token: str, container_id: str) -> Optional[str]:
        """Publish a media container to Instagram"""
        try:
            async with httpx.AsyncClient() as client:
                response = await client.post(
                    f'{self.graph_base_url}/{account_id}/media_publish',
                    data={'creation_id': container_id, 'access_token': access_token},
                )
                response.raise_for_status()
                data = response.json()
                return data.get('id')
        except httpx.HTTPError as e:
            print(f'Failed to publish container: {str(e)}')
            return None

    async def _get_post_details(self, media_id: str, access_token: str) -> Dict[str, Any]:
        """Get details of a published Instagram post"""
        try:
            async with httpx.AsyncClient() as client:
                response = await client.get(
                    f'{self.graph_base_url}/{media_id}',
                    params={
                        'fields': 'id,media_type,media_url,permalink,timestamp,caption',
                        'access_token': access_token,
                    },
                )
                response.raise_for_status()
                return response.json()
        except httpx.HTTPError as e:
            print(f'Failed to get post details: {str(e)}')
            return {}

    def validate_caption(self, caption: str) -> Dict[str, Any]:
        """Validate Instagram caption against platform limits"""
        errors = []
        warnings = []

        # Check length (2200 character limit)
        if len(caption) > 2200:
            errors.append(f'Caption too long: {len(caption)} characters (max 2200)')

        # Check hashtag count (30 hashtag limit)
        hashtags = [word for word in caption.split() if word.startswith('#')]
        if len(hashtags) > 30:
            errors.append(f'Too many hashtags: {len(hashtags)} (max 30)')

        # Check mention count (unofficial limit around 30)
        mentions = [word for word in caption.split() if word.startswith('@')]
        if len(mentions) > 30:
            warnings.append(f'Many mentions: {len(mentions)} (may affect reach)')

        return {
            'valid': len(errors) == 0,
            'errors': errors,
            'warnings': warnings,
            'hashtag_count': len(hashtags),
            'mention_count': len(mentions),
            'character_count': len(caption),
        }

    def format_caption_with_hashtags(self, caption: str, hashtags: list[str] = None) -> str:
        """Format caption with hashtags properly"""
        if not hashtags:
            return caption

        # Ensure hashtags start with #
        formatted_hashtags = []
        for tag in hashtags:
            if not tag.startswith('#'):
                tag = f'#{tag}'
            formatted_hashtags.append(tag)

        # Add hashtags with proper spacing
        if caption and not caption.endswith('\n'):
            caption += '\n\n'

        caption += ' '.join(formatted_hashtags)
        return caption

from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import RedirectResponse

router = APIRouter(prefix="/api/instagram")

@router.post('/connect')
async def connect_instagram_account(request: dict):
    """Start OAuth flow"""
    user_id = request.get('user_id')
    product_id = request.get('product_id')

    state = f"{user_id}:{product_id}"
    auth_url = instagram_oauth.get_auth_url(state)

    return {'auth_url': auth_url, 'message': 'Redirect user to this URL'}

@router.get('/oauth-callback')
async def oauth_callback(code: str = Query(...), state: str = Query(...)):
    """Handle Instagram OAuth callback"""
    try:
        # Parse state to get user_id and product_id
        user_id, product_id = state.split(':')

        # Exchange code for token
        token_data = await instagram_oauth.exchange_code_for_token(code)
        if not token_data.get('access_token'):
            raise Exception('Failed to get access token')

        # Get Instagram accounts
        accounts = await instagram_oauth.get_instagram_accounts(token_data['access_token'])
        if not accounts:
            raise Exception('No Instagram accounts found')

        # Save new account to database
        for account_data in accounts:
            instagram_account = InstagramAccount(
                user_id=user_id,
                product_id=product_id,
                instagram_business_account_id=account_data['id'],
                instagram_user_id=account_data['id'],
                access_token=token_data['access_token'],
                username=account_data.get('username', ''),
                name=account_data.get('name'),
                profile_picture_url=account_data.get('profile_picture_url'),
                followers_count=account_data.get('followers_count'),
                token_expires_at=datetime.fromisoformat(token_data.get('expires_at')) if token_data.get('expires_at') else None,
            )

        # Redirect to frontend with success message
        frontend_url = os.getenv('FRONTEND_URL')
        return RedirectResponse(
            url=f'{frontend_url}/instagram/callback?success=true&message=Successfully connected Instagram account',
            status_code=302,
        )

    except Exception as e:
        frontend_url = os.getenv('FRONTEND_URL')
        return RedirectResponse(
            url=f'{frontend_url}/instagram/callback?error=true&message={str(e)}',
            status_code=302
        )

import { useState, useEffect } from 'react'

interface InstagramAccount {
  id: string
  username: string
  name?: string
  profile_picture_url?: string
  followers_count?: number
  is_active: boolean
  auto_post: boolean
}

export const useInstagram = (productId: string) => {
  const [accounts, setAccounts] = useState<InstagramAccount[]>([])
  const [loading, setLoading] = useState(false)

  const connectAccount = async () => {
    try {
      const response = await fetch('/api/instagram/connect', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ product_id: productId })
      })
      const data = await response.json()

      if (data.auth_url) {
        window.location.href = data.auth_url
      }
    } catch (error) {
      console.error('Failed to connect Instagram account:', error)
    }
  }

  const disconnectAccount = async (accountId: string) => {
    try {
      await fetch(`/api/instagram/accounts/${accountId}`, {
        method: 'DELETE'
      })
      await fetchAccounts()
    } catch (error) {
      console.error('Failed to disconnect account:', error)
    }
  }

  const publishPost = async (data: {
    post_id: string
    caption: string
    hashtags?: string[]
    image_url: string
  }) => {
    try {
      const response = await fetch('/api/instagram/publish', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify(data)
      })
      return await response.json()
    } catch (error) {
      console.error('Failed to publish post:', error)
      throw error
    }
  }

  const fetchAccounts = async () => {
    try {
      setLoading(true)
      const response = await fetch(`/api/instagram/accounts?product_id=${productId}`)
      const data = await response.json()
      setAccounts(data.accounts || [])
    } catch (error) {
      console.error('Failed to fetch accounts:', error)
    } finally {
      setLoading(false)
    }
  }

  useEffect(() => {
    if (productId) {
      fetchAccounts()
    }
  }, [productId])

  return {
    accounts,
    loading,
    connectAccount,
    disconnectAccount,
    publishPost,
    refetch: fetchAccounts
  }
}