-
Notifications
You must be signed in to change notification settings - Fork 3.7k
/
sponsored_products.py
207 lines (153 loc) · 7.21 KB
/
sponsored_products.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
#
# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
#
from abc import ABC
from http import HTTPStatus
from typing import Any, Iterable, List, Mapping, MutableMapping, Optional
import requests as requests
from airbyte_protocol.models import SyncMode
from source_amazon_ads.schemas import (
Keywords,
NegativeKeywords,
ProductAd,
ProductAdGroupBidRecommendations,
ProductAdGroups,
ProductAdGroupSuggestedKeywords,
ProductCampaign,
ProductTargeting,
)
from source_amazon_ads.streams.common import AmazonAdsStream, SubProfilesStream
class SponsoredProductCampaigns(SubProfilesStream):
"""
This stream corresponds to Amazon Advertising API - Sponsored Products Campaigns
https://advertising.amazon.com/API/docs/en-us/sponsored-display/3-0/openapi#/Campaigns
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.state_filter = kwargs.get("config", {}).get("state_filter")
primary_key = "campaignId"
state_filter = None
model = ProductCampaign
def path(self, **kvargs) -> str:
return "v2/sp/campaigns"
def request_params(self, *args, **kwargs):
params = super().request_params(*args, **kwargs)
if self.state_filter:
params["stateFilter"] = ",".join(self.state_filter)
return params
class SponsoredProductAdGroups(SubProfilesStream):
"""
This stream corresponds to Amazon Advertising API - Sponsored Products Ad groups
https://advertising.amazon.com/API/docs/en-us/sponsored-products/2-0/openapi#/Ad%20groups
"""
primary_key = "adGroupId"
model = ProductAdGroups
def path(self, **kvargs) -> str:
return "v2/sp/adGroups"
class SponsoredProductAdGroupsWithProfileId(SponsoredProductAdGroups):
"""Add profileId attr for each records in SponsoredProductAdGroups stream"""
def parse_response(self, *args, **kwargs) -> Iterable[Mapping]:
for record in super().parse_response(*args, **kwargs):
record["profileId"] = self._current_profile_id
yield record
class SponsoredProductAdGroupWithSlicesABC(AmazonAdsStream, ABC):
"""ABC Class for extraction of additional information for each known sp ad group"""
primary_key = "adGroupId"
def __init__(self, *args, **kwargs):
self.__args = args
self.__kwargs = kwargs
super().__init__(*args, **kwargs)
def request_headers(self, *args, **kvargs) -> MutableMapping[str, Any]:
headers = super().request_headers(*args, **kvargs)
headers["Amazon-Advertising-API-Scope"] = str(kvargs["stream_slice"]["profileId"])
return headers
def stream_slices(
self, *, sync_mode: SyncMode, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
) -> Iterable[Optional[Mapping[str, Any]]]:
yield from SponsoredProductAdGroupsWithProfileId(*self.__args, **self.__kwargs).read_records(
sync_mode=sync_mode, cursor_field=cursor_field, stream_slice=None, stream_state=stream_state
)
def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapping]:
resp = response.json()
if response.status_code == HTTPStatus.OK:
yield resp
if response.status_code == HTTPStatus.BAD_REQUEST:
# 400 error message for bids recommendation:
# Bid recommendation for AD group in Manual Targeted Campaign is not supported.
# 400 error message for keywords recommendation:
# Getting keyword recommendations for AD Group in Auto Targeted Campaign is not supported
self.logger.warning(
f"Skip current AdGroup because it does not support request {response.request.url} for "
f"{response.request.headers['Amazon-Advertising-API-Scope']} profile: {response.text}"
)
else:
response.raise_for_status()
class SponsoredProductAdGroupBidRecommendations(SponsoredProductAdGroupWithSlicesABC):
"""Docs:
Latest API:
https://advertising.amazon.com/API/docs/en-us/sponsored-display/3-0/openapi#/Bid%20Recommendations/getTargetBidRecommendations
POST /sd/targets/bid/recommendations
Note: does not work, always get "403 Forbidden"
V2 API:
https://advertising.amazon.com/API/docs/en-us/sponsored-products/2-0/openapi#/Bid%20recommendations/getAdGroupBidRecommendations
GET /v2/sp/adGroups/{adGroupId}/bidRecommendations
"""
model = ProductAdGroupBidRecommendations
def path(self, stream_slice: Mapping[str, Any] = None, **kwargs) -> str:
return f"v2/sp/adGroups/{stream_slice['adGroupId']}/bidRecommendations"
class SponsoredProductAdGroupSuggestedKeywords(SponsoredProductAdGroupWithSlicesABC):
"""Docs:
Latest API:
https://advertising.amazon.com/API/docs/en-us/sponsored-products/3-0/openapi/prod#/Keyword%20Targets/getRankedKeywordRecommendation
POST /sp/targets/keywords/recommendations
Note: does not work, always get "403 Forbidden"
V2 API:
https://advertising.amazon.com/API/docs/en-us/sponsored-products/2-0/openapi#/Suggested%20keywords
GET /v2/sp/adGroups/{{adGroupId}}>/suggested/keywords
"""
model = ProductAdGroupSuggestedKeywords
def path(self, stream_slice: Mapping[str, Any] = None, **kwargs) -> str:
return f"v2/sp/adGroups/{stream_slice['adGroupId']}/suggested/keywords"
class SponsoredProductKeywords(SubProfilesStream):
"""
This stream corresponds to Amazon Advertising API - Sponsored Products Keywords
https://advertising.amazon.com/API/docs/en-us/sponsored-products/2-0/openapi#/Keywords
"""
primary_key = "keywordId"
model = Keywords
def path(self, **kvargs) -> str:
return "v2/sp/keywords"
class SponsoredProductNegativeKeywords(SubProfilesStream):
"""
This stream corresponds to Amazon Advertising API - Sponsored Products Negative Keywords
https://advertising.amazon.com/API/docs/en-us/sponsored-products/2-0/openapi#/Negative%20keywords
"""
primary_key = "keywordId"
model = NegativeKeywords
def path(self, **kvargs) -> str:
return "v2/sp/negativeKeywords"
class SponsoredProductCampaignNegativeKeywords(SponsoredProductNegativeKeywords):
"""
This stream corresponds to Amazon Advertising API - Sponsored Products Negative Keywords
https://advertising.amazon.com/API/docs/en-us/sponsored-products/2-0/openapi#/Negative%20keywords
"""
def path(self, **kvargs) -> str:
return "v2/sp/campaignNegativeKeywords"
class SponsoredProductAds(SubProfilesStream):
"""
This stream corresponds to Amazon Advertising API - Sponsored Products Ads
https://advertising.amazon.com/API/docs/en-us/sponsored-products/2-0/openapi#/Product%20ads
"""
primary_key = "adId"
model = ProductAd
def path(self, **kvargs) -> str:
return "v2/sp/productAds"
class SponsoredProductTargetings(SubProfilesStream):
"""
This stream corresponds to Amazon Advertising API - Sponsored Products Targetings
https://advertising.amazon.com/API/docs/en-us/sponsored-products/2-0/openapi#/Product%20targeting
"""
primary_key = "targetId"
model = ProductTargeting
def path(self, **kvargs) -> str:
return "v2/sp/targets"