add ingesting parent segment configuration function to monitoring wor…

…kflow

scenarios/monitoring/cdp_monitoring/common/settings.yaml

@@ -13,6 +13,8 @@ td:
     journey_activation_history: journey_activation_history
     activations: activations
     activations_history: activations_history
+    parent_segments_configuration: parent_segments_configuration
+    parent_segments_configuration_history: parent_segments_configuration_history
   api_endpoint: api.treasuredata.com
   cdp_api_endpoint: api-cdp.treasuredata.com
 

scenarios/monitoring/cdp_monitoring/incremental_ingest.dig

@@ -23,6 +23,24 @@ schedule:
         _env:
             TD_API_KEY: ${secret:td.apikey}
 
++incremental_ingest_ps_configuration:
+    +append_ps_configuration_history:
+        td>: 
+        query: select * from ${td.tables.parent_segments_configuration}
+        insert_into: ${td.tables.parent_segments_configuration_history}
+        database: ${td.database}
+    +ingest_ps_configuration:
+        py>: scripts.ingest_ps_configuration.run
+        session_unixtime: ${session_unixtime}
+        dest_db: ${td.database}
+        dest_table: ${td.tables.parent_segments_configuration}
+        api_endpoint: ${td.api_endpoint}
+        cdp_api_endpoint: ${td.cdp_api_endpoint}
+        docker:
+            image: "digdag/digdag-python:3.9"
+        _env:
+            TD_API_KEY: ${secret:td.apikey}
+
 +incremental_ingest_entities:
     +append_entities_history:
         td>: 

scenarios/monitoring/cdp_monitoring/initial_ingest.dig

@@ -17,7 +17,19 @@ _export:
     api_endpoint: ${td.api_endpoint}
     cdp_api_endpoint: ${td.cdp_api_endpoint}
     docker:
-        image: "digdag/digdag-python:3.9"
+        image: "digdag/digdag-python:3.10.1"
+    _env:
+        TD_API_KEY: ${secret:td.apikey}
+
++initial_ingest_ps_configuration:
+  py>: scripts.ingest_ps_configuration.run
+  session_unixtime: ${session_unixtime}
+  dest_db: ${td.database}
+  dest_table: ${td.tables.parent_segments_configuration}
+    api_endpoint: ${td.api_endpoint}
+    cdp_api_endpoint: ${td.cdp_api_endpoint}
+    docker:
+        image: "digdag/digdag-python:3.10.1"
     _env:
         TD_API_KEY: ${secret:td.apikey}
 

scenarios/monitoring/cdp_monitoring/scripts/ingest_ps_configuration.py

@@ -0,0 +1,37 @@
+import requests
+import pandas as pd
+import pytd
+import os
+import json
+
+def convert_to_json(s):
+  return json.dumps(s)
+
+def get_all_parent_segment_configuration(url, headers):
+    print(url)
+    res = requests.get(url=url, headers=headers)
+    if res.status_code != requests.codes.ok:
+        res.raise_for_status()
+
+    data = res.json()
+    for d in data:
+        for k in d.keys():
+            if type(d[k]) is dict:
+                d[k] = json.dumps(d[k])
+
+    return data
+
+def run(session_unixtime, dest_db, dest_table, api_endpoint='api.treasuredata.com', cdp_api_endpoint='api-cdp.treasuredata.com'):
+    url = 'https://%s/audiences' % cdp_api_endpoint
+    headers = {'Authorization': 'TD1 %s' % os.environ['TD_API_KEY']}
+    l = get_all_parent_segment_configuration(url, headers)
+    if len(l) == 0:
+      print('no import record')
+      return
+    df = pd.DataFrame(l)
+    df['time'] = int(session_unixtime)
+    df['attributes'] = df['attributes'].apply(convert_to_json)
+    df['behaviors'] = df['behaviors'].apply(convert_to_json)
+
+    client = pytd.Client(apikey=os.environ['TD_API_KEY'], endpoint='https://%s' % api_endpoint, database=dest_db)
+    client.load_table_from_dataframe(df, dest_table, if_exists='overwrite', fmt='msgpack')