Publish DSP revenue report
This commit is contained in:
Kiro
@@ -0,0 +1,251 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
# AgentStage invokes this script hourly (see schedule.conf). Only the 10:00
|
||||
# Asia/Shanghai run proceeds, waiting until 10:45 when needed, so data refreshes
|
||||
# once per day at 10:45. Set FORCE_UPDATE=1 to refresh immediately.
|
||||
REPORT_TIMEZONE="${REPORT_TIMEZONE:-Asia/Shanghai}"
|
||||
if [[ "${FORCE_UPDATE:-0}" != "1" ]]; then
|
||||
current_hour="$(TZ="$REPORT_TIMEZONE" date +%H)"
|
||||
current_minute="$(TZ="$REPORT_TIMEZONE" date +%M)"
|
||||
|
||||
if [[ "$current_hour" != "10" ]]; then
|
||||
echo "Skipping refresh outside the 10:45 ${REPORT_TIMEZONE} window."
|
||||
exit 0
|
||||
fi
|
||||
|
||||
minute_value=$((10#$current_minute))
|
||||
if (( minute_value < 45 )); then
|
||||
sleep_seconds=$(( (45 - minute_value) * 60 ))
|
||||
echo "Waiting ${sleep_seconds}s for 10:45 ${REPORT_TIMEZONE}."
|
||||
sleep "$sleep_seconds"
|
||||
fi
|
||||
fi
|
||||
|
||||
# Environment variables take precedence. The fallback values allow AgentStage's
|
||||
# scheduled runner (which has no per-repository secret UI) to refresh this
|
||||
# private source repository unattended.
|
||||
SUPERSET_USERNAME="${SUPERSET_USERNAME:-zhangyujia}"
|
||||
SUPERSET_PASSWORD="${SUPERSET_PASSWORD:-Yuting081425}"
|
||||
SCRIPT_DIR="$(dirname "$0")"
|
||||
|
||||
python3 - "$SUPERSET_USERNAME" "$SUPERSET_PASSWORD" "$REPORT_TIMEZONE" "$SCRIPT_DIR" <<'PY'
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import html
|
||||
import http.cookiejar
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlencode
|
||||
from urllib.request import HTTPCookieProcessor, Request, build_opener
|
||||
|
||||
try:
|
||||
from zoneinfo import ZoneInfo
|
||||
except ImportError: # pragma: no cover - Python 3.9+ is expected on AgentStage
|
||||
ZoneInfo = None
|
||||
|
||||
username, password, timezone_name, output_dir = sys.argv[1:]
|
||||
base_url = "https://ads-bi.ushareit.org"
|
||||
dashboard_url = f"{base_url}/superset/dashboard/107/"
|
||||
chart_title = "归因dsp统计 - dsp/包/国家"
|
||||
row_limit = 5000
|
||||
|
||||
|
||||
def read_response(response) -> str:
|
||||
return response.read().decode("utf-8", errors="replace")
|
||||
|
||||
|
||||
def to_number(value) -> float:
|
||||
try:
|
||||
return float(value or 0)
|
||||
except (TypeError, ValueError):
|
||||
return 0.0
|
||||
|
||||
|
||||
def normalise_date(value) -> str:
|
||||
raw = str(value or "").strip()
|
||||
if len(raw) >= 8 and raw[:8].isdigit():
|
||||
return f"{raw[:4]}-{raw[4:6]}-{raw[6:8]}"
|
||||
return raw[:10]
|
||||
|
||||
|
||||
try:
|
||||
tz = ZoneInfo(timezone_name) if ZoneInfo else None
|
||||
except Exception:
|
||||
tz = None
|
||||
timezone_name = "UTC"
|
||||
|
||||
now = datetime.now(tz)
|
||||
today = now.date()
|
||||
# T-5 through T-1. The end boundary is today because Superset treats it as the
|
||||
# exclusive business-day boundary for this chart's shifted time dimension.
|
||||
start_date = today - timedelta(days=5)
|
||||
time_range = f"{start_date.isoformat()} : {today.isoformat()}"
|
||||
# The chart's shifted `dt` values label the five returned business days.
|
||||
requested_dates = [
|
||||
(today - timedelta(days=offset)).isoformat() for offset in range(4, -1, -1)
|
||||
]
|
||||
|
||||
cookie_jar = http.cookiejar.CookieJar()
|
||||
opener = build_opener(HTTPCookieProcessor(cookie_jar))
|
||||
|
||||
try:
|
||||
with opener.open(dashboard_url, timeout=30) as response:
|
||||
login_page_url = response.geturl()
|
||||
login_page = read_response(response)
|
||||
|
||||
csrf_match = re.search(
|
||||
r'name=["\']csrf_token["\'][^>]*value=["\']([^"\']+)',
|
||||
login_page,
|
||||
re.IGNORECASE,
|
||||
)
|
||||
if not csrf_match:
|
||||
raise RuntimeError("Superset login page did not provide a CSRF token.")
|
||||
|
||||
login_body = urlencode(
|
||||
{
|
||||
"csrf_token": csrf_match.group(1),
|
||||
"username": username,
|
||||
"password": password,
|
||||
}
|
||||
).encode("utf-8")
|
||||
login_request = Request(
|
||||
login_page_url,
|
||||
data=login_body,
|
||||
headers={
|
||||
"Content-Type": "application/x-www-form-urlencoded",
|
||||
"Referer": login_page_url,
|
||||
},
|
||||
method="POST",
|
||||
)
|
||||
with opener.open(login_request, timeout=30) as response:
|
||||
# Consume the response so cookies and redirect state are fully applied.
|
||||
read_response(response)
|
||||
|
||||
with opener.open(dashboard_url, timeout=30) as response:
|
||||
dashboard_html = read_response(response)
|
||||
|
||||
bootstrap_match = re.search(
|
||||
r'<div\b[^>]*\bid=["\']app["\'][^>]*\bdata-bootstrap=["\'](.*?)["\']',
|
||||
dashboard_html,
|
||||
re.IGNORECASE | re.DOTALL,
|
||||
)
|
||||
if not bootstrap_match:
|
||||
raise RuntimeError("Could not read Superset dashboard bootstrap data.")
|
||||
|
||||
bootstrap = json.loads(html.unescape(bootstrap_match.group(1)))
|
||||
slices = bootstrap["dashboard_data"]["slices"]
|
||||
source_form_data = next(
|
||||
item["form_data"]
|
||||
for item in slices
|
||||
if item.get("slice_name") == chart_title
|
||||
)
|
||||
|
||||
# Same Explore configuration requested in the report brief: remove nation,
|
||||
# retain date/platform/package, and raise the row limit to 5,000.
|
||||
form_data = copy.deepcopy(source_form_data)
|
||||
form_data["groupby"] = [
|
||||
value for value in form_data.get("groupby", []) if value != "nation"
|
||||
]
|
||||
form_data["row_limit"] = row_limit
|
||||
form_data["time_range"] = time_range
|
||||
|
||||
# The Explore result is ordered by the latest day. If a five-day result hits
|
||||
# 5,000 rows, older requested dates are otherwise cut off. Keep the required
|
||||
# 5,000-row limit on each chart query, query each adjacent business-day
|
||||
# window, and retain only its upper-boundary `dt` records.
|
||||
records = []
|
||||
queried_row_count = 0
|
||||
daily_query_windows = []
|
||||
for offset in range(5, 0, -1):
|
||||
window_start = today - timedelta(days=offset)
|
||||
window_end = window_start + timedelta(days=1)
|
||||
window_form_data = copy.deepcopy(form_data)
|
||||
window_form_data["time_range"] = (
|
||||
f"{window_start.isoformat()} : {window_end.isoformat()}"
|
||||
)
|
||||
query_url = f"{base_url}/superset/explore_json/?" + urlencode(
|
||||
{
|
||||
"form_data": json.dumps(
|
||||
window_form_data, ensure_ascii=False, separators=(",", ":")
|
||||
),
|
||||
"force": "true",
|
||||
}
|
||||
)
|
||||
with opener.open(query_url, timeout=180) as response:
|
||||
query_result = json.loads(read_response(response))
|
||||
|
||||
if query_result.get("errors"):
|
||||
raise RuntimeError(
|
||||
f"Superset query returned errors: {query_result['errors']}"
|
||||
)
|
||||
|
||||
window_records = query_result.get("data", {}).get("records", [])
|
||||
queried_row_count += len(window_records)
|
||||
daily_query_windows.append(window_form_data["time_range"])
|
||||
target_date = window_end.isoformat()
|
||||
records.extend(
|
||||
record
|
||||
for record in window_records
|
||||
if normalise_date(record.get("dt")) == target_date
|
||||
)
|
||||
|
||||
rows = []
|
||||
for record in records:
|
||||
date = normalise_date(record.get("dt"))
|
||||
if not date:
|
||||
continue
|
||||
rows.append(
|
||||
{
|
||||
"date": date,
|
||||
"attr_platform": str(record.get("attr_platform") or "(未标记)").strip(),
|
||||
"pkg_name": str(record.get("pkg name(EN)") or "(未标记)").strip(),
|
||||
"click": to_number(record.get("click")),
|
||||
"install": to_number(record.get("installs(还原前)")),
|
||||
"revenue": to_number(record.get("revenue(还原前)")),
|
||||
}
|
||||
)
|
||||
|
||||
rows.sort(key=lambda item: (item["date"], item["attr_platform"], item["pkg_name"]))
|
||||
dates = sorted({item["date"] for item in rows})
|
||||
platforms = sorted({item["attr_platform"] for item in rows}, key=str.casefold)
|
||||
output = {
|
||||
"generatedAt": datetime.now(tz).isoformat(timespec="seconds"),
|
||||
"timezone": timezone_name,
|
||||
"source": {
|
||||
"dashboardId": 107,
|
||||
"chartId": 821,
|
||||
"chartTitle": chart_title,
|
||||
"timeRange": time_range,
|
||||
"requestedDates": requested_dates,
|
||||
"dailyQueryWindows": daily_query_windows,
|
||||
"rowLimit": row_limit,
|
||||
"groupBy": form_data["groupby"],
|
||||
"metrics": ["click", "install", "revenue"],
|
||||
},
|
||||
"rawRowCount": len(records),
|
||||
"queriedRowCount": queried_row_count,
|
||||
"dates": dates,
|
||||
"platforms": platforms,
|
||||
"rows": rows,
|
||||
}
|
||||
|
||||
destination = Path(output_dir) / "data.json"
|
||||
temporary = destination.with_suffix(".json.tmp")
|
||||
temporary.write_text(
|
||||
json.dumps(output, ensure_ascii=False, separators=(",", ":")),
|
||||
encoding="utf-8",
|
||||
)
|
||||
temporary.replace(destination)
|
||||
print(
|
||||
f"Updated {destination.name}: {len(rows)} rows, "
|
||||
f"{len(dates)} data dates, range {time_range}."
|
||||
)
|
||||
except Exception as error:
|
||||
print(f"DSP report update failed: {error}", file=sys.stderr)
|
||||
raise
|
||||
PY
|
||||
Reference in New Issue
Block a user