Add KPI event and CAC data pipeline

refresh_events.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+import csv
+import json
+import os
+import re
+import sys
+import tempfile
+import urllib.request
+import zipfile
+from pathlib import Path
+
+from tableauhyperapi import Connection, HyperProcess, Telemetry
+
+
+SERVER = os.environ.get("TABLEAU_SERVER", "https://mars.ushareit.org").rstrip("/")
+API_VERSION = os.environ.get("TABLEAU_API_VERSION", "3.15")
+SITE_CONTENT_URL = os.environ.get("TABLEAU_SITE_CONTENT_URL", "")
+WORKBOOK_ID = os.environ.get("TABLEAU_EVENT_WORKBOOK_ID", "ad567777-aa52-4c2f-89c3-5acfed4fdbba")
+
+
+def request_json(url, *, body=None, headers=None):
+    request = urllib.request.Request(url, data=body, headers=headers or {})
+    with urllib.request.urlopen(request, timeout=120) as response:
+        return json.load(response)
+
+
+def main():
+    if len(sys.argv) != 2:
+        raise SystemExit("usage: refresh_events.py OUTPUT.csv")
+    output_path = sys.argv[1]
+    pat_name = os.environ.get("TABLEAU_PAT_NAME", "")
+    pat_secret = os.environ.get("TABLEAU_PAT_SECRET", "")
+    if not pat_name or not pat_secret:
+        raise SystemExit("TABLEAU_PAT_NAME and TABLEAU_PAT_SECRET are required")
+
+    signin_body = json.dumps({"credentials": {
+        "personalAccessTokenName": pat_name,
+        "personalAccessTokenSecret": pat_secret,
+        "site": {"contentUrl": SITE_CONTENT_URL},
+    }}).encode()
+    signin = request_json(
+        f"{SERVER}/api/{API_VERSION}/auth/signin",
+        body=signin_body,
+        headers={"Content-Type": "application/json", "Accept": "application/json"},
+    )["credentials"]
+    auth_token = signin["token"]
+    site_id = signin["site"]["id"]
+
+    try:
+        with tempfile.TemporaryDirectory(prefix="manta-events-") as temp_dir:
+            archive_path = Path(temp_dir) / "workbook.twbx"
+            download = urllib.request.Request(
+                f"{SERVER}/api/{API_VERSION}/sites/{site_id}/workbooks/{WORKBOOK_ID}/content",
+                headers={"X-Tableau-Auth": auth_token},
+            )
+            with urllib.request.urlopen(download, timeout=180) as response:
+                archive_path.write_bytes(response.read())
+            with zipfile.ZipFile(archive_path) as archive:
+                members = [name for name in archive.namelist() if name.endswith(".hyper")]
+                if len(members) != 1:
+                    raise SystemExit(f"expected one Hyper extract, found {len(members)}")
+                archive.extract(members[0], temp_dir)
+            hyper_path = Path(temp_dir) / members[0]
+            query = '''
+                SELECT "dt", "pkg_name", "adping_id", "offer_name", "event_name",
+                       SUM("event_num")
+                FROM "Extract"."Extract"
+                GROUP BY "dt", "pkg_name", "adping_id", "offer_name", "event_name"
+                ORDER BY "dt", "pkg_name", "offer_name", "event_name"
+            '''
+            with HyperProcess(Telemetry.DO_NOT_SEND_USAGE_DATA_TO_TABLEAU) as process:
+                with Connection(process.endpoint, str(hyper_path)) as connection:
+                    rows = connection.execute_list_query(query)
+            with open(output_path, "w", encoding="utf-8", newline="") as handle:
+                writer = csv.writer(handle)
+                writer.writerow(("dt", "pkg_name", "adping_id", "offer_name", "event_name", "event_num"))
+                writer.writerows(rows)
+            dates = sorted({str(row[0]) for row in rows})
+            print(json.dumps({"rows": len(rows), "dates": dates}))
+    finally:
+        try:
+            signout = urllib.request.Request(
+                f"{SERVER}/api/{API_VERSION}/auth/signout", data=b"",
+                headers={"X-Tableau-Auth": auth_token},
+            )
+            urllib.request.urlopen(signout, timeout=30).close()
+        except Exception:
+            pass
+
+
+if __name__ == "__main__":
+    main()