How to Automate Amazon Review Monitoring using the API

1import requests
2import os
3import time
4import gzip
5import json
6import csv
7import io
8import logging
9from datetime import datetime, timedelta
10
11logging.basicConfig(level=logging.INFO)
12log = logging.getLogger(__name__)
13
14BASE_URL = 'https://sellingpartnerapi-na.amazon.com'
15SLACK_WEBHOOK = os.environ.get('SLACK_WEBHOOK_URL')
16
17class SPAPIClient:
18    def __init__(self):
19        self._token = None
20        self._expires_at = 0
21
22    def token(self):
23        if time.time() >= self._expires_at - 300:
24            resp = requests.post('https://api.amazon.com/auth/o2/token', data={
25                'grant_type': 'refresh_token',
26                'refresh_token': os.environ['AMAZON_LWA_REFRESH_TOKEN'],
27                'client_id': os.environ['AMAZON_LWA_CLIENT_ID'],
28                'client_secret': os.environ['AMAZON_LWA_CLIENT_SECRET']
29            })
30            resp.raise_for_status()
31            d = resp.json()
32            self._token = d['access_token']
33            self._expires_at = time.time() + d['expires_in']
34        return self._token
35
36    def headers(self):
37        return {'x-amz-access-token': self.token(), 'Content-Type': 'application/json'}
38
39client = SPAPIClient()
40
41def request_report(report_type, start, end, marketplace='ATVPDKIKX0DER'):
42    resp = requests.post(f'{BASE_URL}/reports/2021-06-30/reports',
43        headers=client.headers(),
44        json={
45            'reportType': report_type,
46            'dataStartTime': f'{start}T00:00:00Z',
47            'dataEndTime': f'{end}T23:59:59Z',
48            'reportOptions': {'reportPeriod': 'MONTH'},
49            'marketplaceIds': [marketplace]
50        }
51    )
52    resp.raise_for_status()
53    return resp.json()['reportId']
54
55def await_report(report_id, timeout=1800):
56    deadline = time.time() + timeout
57    while time.time() < deadline:
58        resp = requests.get(f'{BASE_URL}/reports/2021-06-30/reports/{report_id}',
59            headers=client.headers())
60        data = resp.json()
61        log.info(f'Status: {data["processingStatus"]}')
62        if data['processingStatus'] == 'DONE':
63            return data['reportDocumentId']
64        if data['processingStatus'] in ('FATAL', 'CANCELLED'):
65            raise Exception(f'Report failed: {data["processingStatus"]}')
66        time.sleep(30)
67    raise TimeoutError('Report timeout')
68
69def download(doc_id):
70    resp = requests.get(f'{BASE_URL}/reports/2021-06-30/documents/{doc_id}',
71        headers=client.headers())
72    doc = resp.json()
73    file_resp = requests.get(doc['url'])
74    if doc.get('compressionAlgorithm') == 'GZIP':
75        return gzip.decompress(file_resp.content).decode('utf-8')
76    return file_resp.text
77
78def send_slack(msg):
79    if SLACK_WEBHOOK:
80        requests.post(SLACK_WEBHOOK, json={'text': msg})
81
82def main():
83    # Previous month
84    today = datetime.utcnow()
85    first = today.replace(day=1) - timedelta(days=1)
86    start = first.replace(day=1).strftime('%Y-%m-%d')
87    end = first.strftime('%Y-%m-%d')
88
89    log.info(f'Requesting Brand Analytics report for {start} to {end}')
90    report_id = request_report('GET_BRAND_ANALYTICS_SEARCH_TERMS_REPORT', start, end)
91    doc_id = await_report(report_id)
92    content = download(doc_id)
93
94    # Parse tab-delimited report
95    reader = csv.DictReader(io.StringIO(content), delimiter='\t')
96    rows = list(reader)
97    log.info(f'Report has {len(rows)} rows')
98
99    msg = f'*Amazon Brand Analytics — {start[:7]}*\n'
100    msg += f'Search terms analyzed: {len(rows)}\n'
101    if rows:
102        top = rows[:3]
103        msg += 'Top search terms:\n'
104        for r in top:
105            term = list(r.values())[0] if r else 'N/A'
106            msg += f'  • {term}\n'
107
108    send_slack(msg)
109    log.info('Done')
110
111if __name__ == '__main__':
112    main()