Source Amazon Seller Partner: Add GET_BRAND_ANALYTICS_SEARCH_TERMS_REPORT report (#8179)

* Source Amazon Seller Partner: Add GET_BRAND_ANALYTICS_SEARCH_TERMS_REPORT report

* loads

* fixes and refactors

* docs

* fixes

* Field definitions

* versions

* versions

* refactor report options to scope by REPORT name

* fix _report_data

* 1

* make _augmented_data a static method

* parse_document is static too

Author: lizdeika

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/e55879a8-0ef8-4557-abcf-ab34c53ec460.json

@@ -2,7 +2,7 @@
   "sourceDefinitionId": "e55879a8-0ef8-4557-abcf-ab34c53ec460",
   "name": "Amazon Seller Partner",
   "dockerRepository": "airbyte/source-amazon-seller-partner",
-  "dockerImageTag": "0.2.5",
+  "dockerImageTag": "0.2.6",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/amazon-seller-partner",
   "icon": "amazonsellerpartner.svg"
 }

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -22,7 +22,7 @@
 - name: Amazon Seller Partner
   sourceDefinitionId: e55879a8-0ef8-4557-abcf-ab34c53ec460
   dockerRepository: airbyte/source-amazon-seller-partner
-  dockerImageTag: 0.2.5
+  dockerImageTag: 0.2.6
   sourceType: api
   documentationUrl: https://docs.airbyte.io/integrations/sources/amazon-seller-partner
   icon: amazonsellerpartner.svg

airbyte-config/init/src/main/resources/seed/source_specs.yaml

@@ -153,7 +153,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-amazon-seller-partner:0.2.5"
+- dockerImage: "airbyte/source-amazon-seller-partner:0.2.6"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/amazon-seller-partner"
     changelogUrl: "https://docs.airbyte.io/integrations/sources/amazon-seller-partner"

airbyte-integrations/connectors/source-amazon-seller-partner/Dockerfile

@@ -12,5 +12,5 @@ RUN pip install .
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.2.5
+LABEL io.airbyte.version=0.2.6
 LABEL io.airbyte.name=airbyte/source-amazon-seller-partner

airbyte-integrations/connectors/source-amazon-seller-partner/integration_tests/configured_catalog.json

@@ -104,6 +104,15 @@
       "sync_mode": "incremental",
       "destination_sync_mode": "append",
       "cursor_field": ["Date"]
+    },
+    {
+      "stream": {
+        "name": "GET_BRAND_ANALYTICS_SEARCH_TERMS_REPORT",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"]
+      },
+      "sync_mode": "full_refresh",
+      "destination_sync_mode": "overwrite"
     }
   ]
 }

airbyte-integrations/connectors/source-amazon-seller-partner/integration_tests/sample_state.json

@@ -28,5 +28,8 @@
   },
   "GET_SELLER_FEEDBACK_DATA": {
     "createdTime": "2021-07-01T00:00:00Z"
+  },
+  "GET_BRAND_ANALYTICS_SEARCH_TERMS_REPORT": {
+    "createdTime": "2021-07-01T00:00:00Z"
   }
 }

airbyte-integrations/connectors/source-amazon-seller-partner/source_amazon_seller_partner/schemas/GET_BRAND_ANALYTICS_SEARCH_TERMS_REPORT.json

@@ -0,0 +1,29 @@
+{
+    "title": "Brand Analytics Search Terms Reports",
+    "description": "Brand Analytics Search Terms Reports",
+    "type": "object",
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "properties": {
+      "departmentName": {
+        "type": ["null", "string"]
+      },
+      "searchTerm": {
+        "type": ["null", "string"]
+      },
+      "searchFrequencyRank": {
+        "type": ["null", "number"]
+      },
+      "clickedAsin": {
+        "type": ["null", "string"]
+      },
+      "clickShareRank": {
+        "type": ["null", "number"]
+      },
+      "clickShare": {
+        "type": ["null", "number"]
+      },
+      "conversionShare": {
+        "type": ["null", "number"]
+      }
+    }
+  }

airbyte-integrations/connectors/source-amazon-seller-partner/source_amazon_seller_partner/source.py

@@ -26,6 +26,7 @@
     SellerFeedbackReports,
     VendorDirectFulfillmentShipping,
     VendorInventoryHealthReports,
+    BrandAnalyticsSearchTermsReports,
 )
 
 
@@ -44,6 +45,11 @@ class Config:
         description="Will be used for stream slicing for initial full_refresh sync when no updated state is present for reports that support sliced incremental sync.",
         examples=["30", "365"],
     )
+    report_options: str = Field(
+        None,
+        description="Additional information passed to reports. This varies by report type. Must be a valid json string.",
+        examples=['{"GET_BRAND_ANALYTICS_SEARCH_TERMS_REPORT": {"reportPeriod": "WEEK"}}', '{"GET_SOME_REPORT": {"custom": "true"}}'],
+    )
     refresh_token: str = Field(
         description="The Refresh Token obtained via OAuth flow authorization.",
         title="Refresh Token",
@@ -98,6 +104,7 @@ def _get_stream_kwargs(self, config: ConnectorConfig) -> Mapping[str, Any]:
             "replication_start_date": config.replication_start_date,
             "marketplace_ids": [marketplace_id],
             "period_in_days": config.period_in_days,
+            "report_options": config.report_options,
         }
         return stream_kwargs
 
@@ -136,6 +143,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
             VendorInventoryHealthReports(**stream_kwargs),
             Orders(**stream_kwargs),
             SellerFeedbackReports(**stream_kwargs),
+            BrandAnalyticsSearchTermsReports(**stream_kwargs),
         ]
 
     def spec(self, *args, **kwargs) -> ConnectorSpecification:

airbyte-integrations/connectors/source-amazon-seller-partner/source_amazon_seller_partner/spec.json

@@ -19,6 +19,12 @@
         "type": "integer",
         "default": 30
       },
+      "report_options": {
+        "title": "Report Options",
+        "description": "Additional information passed to reports. This varies by report type. Must be a valid json string.",
+        "examples":["{\"GET_BRAND_ANALYTICS_SEARCH_TERMS_REPORT\": {\"reportPeriod\": \"WEEK\"}}", "{\"GET_SOME_REPORT\": {\"custom\": \"true\"}}"],
+        "type": "string"
+      },
       "refresh_token": {
         "title": "Refresh Token",
         "description": "The Refresh Token obtained via OAuth flow authorization.",

airbyte-integrations/connectors/source-amazon-seller-partner/source_amazon_seller_partner/streams.py

@@ -35,7 +35,6 @@
 
 DATE_TIME_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
 
-
 class AmazonSPStream(HttpStream, ABC):
     data_field = "payload"
 
@@ -46,6 +45,7 @@ def __init__(
         replication_start_date: str,
         marketplace_ids: List[str],
         period_in_days: Optional[int],
+        report_options: Optional[str],
         *args,
         **kwargs,
     ):
@@ -151,6 +151,7 @@ def __init__(
         replication_start_date: str,
         marketplace_ids: List[str],
         period_in_days: Optional[int],
+        report_options: Optional[str],
         authenticator: HttpAuthenticator = NoAuth(),
     ):
         self._authenticator = authenticator
@@ -160,6 +161,7 @@ def __init__(
         self._replication_start_date = replication_start_date
         self.marketplace_ids = marketplace_ids
         self.period_in_days = period_in_days
+        self._report_options = report_options
 
     @property
     def url_base(self) -> str:
@@ -282,9 +284,16 @@ def parse_response(self, response: requests.Response) -> Iterable[Mapping]:
             payload,
         )
 
-        document_records = csv.DictReader(StringIO(document), delimiter="\t")
+        document_records = self.parse_document(document)
         yield from document_records
 
+    @staticmethod
+    def parse_document(document):
+        return csv.DictReader(StringIO(document), delimiter="\t")
+
+    def report_options(self) -> Mapping[str, Any]:
+        return json_lib.loads(self._report_options).get(self.name)
+
     def read_records(
         self,
         sync_mode: SyncMode,
@@ -377,6 +386,69 @@ class FbaShipmentsReports(ReportsAmazonSPStream):
 class VendorInventoryHealthReports(ReportsAmazonSPStream):
     name = "GET_VENDOR_INVENTORY_HEALTH_AND_PLANNING_REPORT"
 
+class BrandAnalyticsSearchTermsReports(ReportsAmazonSPStream):
+    """
+    Field definitions: https://sellercentral.amazon.co.uk/help/hub/reference/G5NXWNY8HUD3VDCW
+    """
+
+    name = "GET_BRAND_ANALYTICS_SEARCH_TERMS_REPORT"
+
+    @staticmethod
+    def parse_document(document):
+        parsed = json_lib.loads(document)
+        return parsed.get("dataByDepartmentAndSearchTerm", {})
+
+    def _report_data(
+        self,
+        sync_mode: SyncMode,
+        cursor_field: List[str] = None,
+        stream_slice: Mapping[str, Any] = None,
+        stream_state: Mapping[str, Any] = None,
+    ) -> Mapping[str, Any]:
+         data = super()._report_data(sync_mode, cursor_field, stream_slice, stream_state)
+         options = self.report_options()
+         if options is not None:
+             data.update(self._augmented_data(options))
+
+         return data
+
+    @staticmethod
+    def _augmented_data(report_options) -> Mapping[str, Any]:
+        if report_options.get("reportPeriod") is None:
+            return {}
+        else:
+            now = pendulum.now("utc")
+            if report_options["reportPeriod"] == "DAY":
+                now = now.subtract(days=1)
+                data_start_time = now.start_of("day")
+                data_end_time = now.end_of("day")
+            elif report_options["reportPeriod"] == "WEEK":
+                now = now.subtract(weeks=1)
+
+                # According to report api docs
+                # dataStartTime must be a Sunday and dataEndTime must be the following Saturday
+                pendulum.week_starts_at(pendulum.SUNDAY)
+                pendulum.week_ends_at(pendulum.SATURDAY)
+
+                data_start_time = now.start_of("week")
+                data_end_time = now.end_of("week")
+
+                # Reset week start and end
+                pendulum.week_starts_at(pendulum.MONDAY)
+                pendulum.week_ends_at(pendulum.SUNDAY)
+            elif report_options["reportPeriod"] == "MONTH":
+                now = now.subtract(months=1)
+                data_start_time = now.start_of("month")
+                data_end_time = now.end_of("month")
+            else:
+                raise Exception([{"message": "This reportPeriod is not implemented."}])
+
+            return {
+                "dataStartTime": data_start_time.strftime(DATE_TIME_FORMAT),
+                "dataEndTime": data_end_time.strftime(DATE_TIME_FORMAT),
+                "reportOptions": report_options,
+            }
+
 
 class IncrementalReportsAmazonSPStream(ReportsAmazonSPStream):
     @property

airbyte-integrations/connectors/source-amazon-seller-partner/unit_tests/test_repots_streams_rate_limits.py

@@ -28,6 +28,7 @@ def reports_stream():
         marketplace_ids=["id"],
         authenticator=NoAuth(),
         period_in_days=0,
+        report_options=None,
     )
     return stream
 

docs/integrations/sources/amazon-seller-partner.md

@@ -24,6 +24,7 @@ This source is capable of syncing the following streams:
 * [Orders](https://github.com/amzn/selling-partner-api-docs/blob/main/references/orders-api/ordersV0.md) \(incremental\)
 * [VendorDirectFulfillmentShipping](https://github.com/amzn/selling-partner-api-docs/blob/main/references/vendor-direct-fulfillment-shipping-api/vendorDirectFulfillmentShippingV1.md)
 * [Seller Feedback Report](https://github.com/amzn/selling-partner-api-docs/blob/main/references/reports-api/reporttype-values.md#performance-reports)
+* [Brand Analytics Search Terms Report](https://github.com/amzn/selling-partner-api-docs/blob/main/references/reports-api/reporttype-values.md#brand-analytics-reports)
 
 ## Getting started
 
@@ -63,6 +64,7 @@ Information about rate limits you may find [here](https://github.com/amzn/sellin
 
 | Version | Date | Pull Request | Subject |
 | :--- | :--- | :--- | :--- |
+| `0.2.6` | 2021-12-10 | [\#8179](https://github.com/airbytehq/airbyte/pull/8179) | Add GET_BRAND_ANALYTICS_SEARCH_TERMS_REPORT report |
 | `0.2.5` | 2021-12-06 | [\#8425](https://github.com/airbytehq/airbyte/pull/8425) | Update title, description fields in spec |
 | `0.2.4` | 2021-11-08 | [\#8021](https://github.com/airbytehq/airbyte/pull/8021) | Added GET_SELLER_FEEDBACK_DATA report with incremental sync capability |
 | `0.2.3` | 2021-11-08 | [\#7828](https://github.com/airbytehq/airbyte/pull/7828) | Remove datetime format from all streams |