dsp-package-revenue/update.sh

#!/usr/bin/env bash
set -euo pipefail

# AgentStage invokes this script hourly (see schedule.conf). Only the 10:00
# Asia/Shanghai check proceeds, waiting until 10:30 when needed, so data
# refreshes once per day at 10:30. 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:30 ${REPORT_TIMEZONE} window."
    exit 0
  fi

  minute_value=$((10#$current_minute))
  if (( minute_value < 30 )); then
    sleep_seconds=$(( (30 - minute_value) * 60 ))
    echo "Waiting ${sleep_seconds}s for 10:30 ${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:-Yujia081425}"
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) -> int:
    """Return a nearest whole number for report metrics."""
    try:
        number = float(value or 0)
    except (TypeError, ValueError):
        return 0
    # Metrics are non-negative in the source; this also handles any negative
    # values with conventional half-away-from-zero rounding.
    return int(number + 0.5) if number >= 0 else int(number - 0.5)


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)
# Use the last completed calendar day as the anchor: T-1 is always yesterday.
# All query windows below therefore exclude today's incomplete data.
today = now.date() - timedelta(days=1)
# T-5 through T-1, where the chart's shifted `dt` values are the five labels
# from five days before today through yesterday.
start_date = today - timedelta(days=5)
time_range = f"{start_date.isoformat()} : {today.isoformat()}"
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