How to Automate Instagram Analytics using the API

1#!/usr/bin/env python3
2"""Instagram Weekly Analytics Report - run as weekly cron job."""
3import os
4import time
5import json
6import logging
7import requests
8from datetime import datetime, timedelta
9
10logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
11log = logging.getLogger(__name__)
12
13IG_USER_ID = os.environ['IG_USER_ID']
14ACCESS_TOKEN = os.environ['IG_ACCESS_TOKEN']
15SLACK_WEBHOOK = os.environ.get('SLACK_WEBHOOK_URL')
16BASE = 'https://graph.instagram.com/v25.0'
17
18def ig_get(path, params):
19    params['access_token'] = ACCESS_TOKEN
20    r = requests.get(f'{BASE}{path}', params=params, timeout=30)
21    r.raise_for_status()
22    return r.json()
23
24def get_account_insights():
25    since = (datetime.now() - timedelta(days=7)).strftime('%Y-%m-%d')
26    until = datetime.now().strftime('%Y-%m-%d')
27    data = ig_get(f'/{IG_USER_ID}/insights', {
28        'metric': 'impressions,reach,profile_views',
29        'period': 'day',
30        'since': since,
31        'until': until,
32    })
33    return {item['name']: sum(v['value'] for v in item['values']) for item in data['data']}
34
35def get_top_posts(limit=20):
36    data = ig_get(f'/{IG_USER_ID}/media', {
37        'fields': 'id,caption,media_type,timestamp,like_count,comments_count',
38        'limit': limit,
39    })
40    posts = data['data']
41    posts.sort(key=lambda p: p.get('like_count', 0) + p.get('comments_count', 0), reverse=True)
42    return posts[:10]
43
44def get_post_insights(media_id, media_type):
45    metrics = 'impressions,reach,likes,comments,shares,saved'
46    if media_type == 'VIDEO':
47        metrics += ',plays'
48    try:
49        data = ig_get(f'/{media_id}/insights', {'metric': metrics, 'period': 'lifetime'})
50        return {item['name']: item['values'][0]['value'] for item in data['data']}
51    except Exception as e:
52        log.warning(f'Failed to get insights for {media_id}: {e}')
53        return {}
54
55def generate_report():
56    log.info('Fetching account insights...')
57    account = get_account_insights()
58    log.info(f'Account: {account}')
59    log.info('Fetching recent posts...')
60    posts = get_top_posts()
61    enriched = []
62    for post in posts:
63        insights = get_post_insights(post['id'], post.get('media_type', 'IMAGE'))
64        enriched.append({**post, **insights})
65        time.sleep(0.3)
66    enriched.sort(key=lambda p: p.get('reach', 0), reverse=True)
67    total_eng = sum(p.get('likes',0)+p.get('comments',0)+p.get('shares',0)+p.get('saved',0) for p in enriched)
68    return {'account': account, 'top_posts': enriched[:5], 'total_engagement': total_eng}
69
70def send_slack(report):
71    if not SLACK_WEBHOOK:
72        print(json.dumps({k: v for k, v in report.items() if k != 'top_posts'}, indent=2))
73        return
74    acc = report['account']
75    top = report['top_posts'][0] if report['top_posts'] else {}
76    text = (
77        f'*Instagram Weekly Report*\n'
78        f'Impressions: {acc.get("impressions",0):,}  Reach: {acc.get("reach",0):,}  '
79        f'Profile Views: {acc.get("profile_views",0):,}\n'
80        f'Total Engagement: {report["total_engagement"]:,}\n'
81        f'Top Post: {top.get("reach",0):,} reach | {top.get("saved",0)} saves'
82    )
83    requests.post(SLACK_WEBHOOK, json={'text': text}, timeout=10).raise_for_status()
84    log.info('Report sent to Slack')
85
86if __name__ == '__main__':
87    report = generate_report()
88    send_slack(report)
89    print('Done')