#!/usr/bin/env python3 """Extract AppsFlyer click-gap reports and publish validated static JSON.""" from __future__ import annotations import argparse import datetime as dt import json import logging import math import os import re import shutil import sqlite3 import subprocess import sys import tempfile import time from pathlib import Path from typing import Any DB_PATH = "/Users/ads-center/Documents/database/af_click_gap.db" PID = "shareit" CLICK_NUM = 300 REPORT_COLUMNS = ["dt", "attr_platform", "pkg_name", "shareit_clicks", "af_clicks", "installs_ua", "clicks_diff", "clicks_diff_rate"] SPECIAL_PKGS = { "za.co.freeurl.easybetapp.fwdb": "za.co.freeurl.easybetapp.fwdb-Shareit", "org.melbet.client": "org.melbet.client-Custom", "org.xbet.client1": "org.xbet.client1-Custom", "org.betwinner_ng.client": "org.betwinner_ng.client-Custom", "org.betwinner_zm.client": "org.betwinner_zm.client-Custom", "org.betwinner_cm.client": "org.betwinner_cm.client-Custom", "org.betwinner_cd.client": "org.betwinner_cd.client-Custom", } EXPECTED_TYPES = {"pid": "TEXT", "dt": "TEXT", "attr_platform": "TEXT", "pkg_name": "TEXT", "shareit_clicks": "REAL", "af_clicks": "REAL", "installs_ua": "REAL", "clicks_diff": "REAL", "clicks_diff_rate": "REAL", "updated_at": "TEXT"} LOG_PATH = Path(os.path.expanduser("~/Library/Logs/af-click-gap/pipeline.log")) def logger_setup() -> logging.Logger: LOG_PATH.parent.mkdir(parents=True, exist_ok=True) log = logging.getLogger("af_click_gap") if not log.handlers: log.setLevel(logging.INFO) handler = logging.FileHandler(LOG_PATH, encoding="utf-8") handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s")) log.addHandler(handler) log.addHandler(logging.StreamHandler()) return log LOG = logger_setup() def date8(value: str) -> str: try: parsed = dt.datetime.strptime(value, "%Y%m%d") except ValueError as exc: raise ValueError(f"非法日期: {value}") from exc return parsed.strftime("%Y%m%d") def date8_from_iso(value: str) -> str: try: return dt.datetime.strptime(value, "%Y-%m-%d").strftime("%Y%m%d") except ValueError as exc: raise ValueError(f"日期必须为 YYYY-MM-DD: {value}") from exc def window(start: str | None, end: str | None) -> tuple[str, str]: today = dt.date.today() if start is None and end is None: first, last = today - dt.timedelta(days=6), today else: if not start or not end: raise ValueError("--start-date 与 --end-date 必须同时提供") first = dt.datetime.strptime(start, "%Y-%m-%d").date() last = dt.datetime.strptime(end, "%Y-%m-%d").date() if first > last or (last - first).days > 6: raise ValueError("日期区间必须为包含首尾且不超过 7 天") return first.strftime("%Y%m%d"), last.strftime("%Y%m%d") def finite_number(value: Any, field: str, allow_null: bool = False) -> int | float | None: if value is None or (isinstance(value, float) and math.isnan(value)): if allow_null: return None raise ValueError(f"{field} 不能为空") try: number = float(value) except (TypeError, ValueError) as exc: raise ValueError(f"{field} 必须为数值") from exc if not math.isfinite(number): raise ValueError(f"{field} 必须为有限数值") return int(number) if number.is_integer() else number def validate_existing_schema(path: str) -> None: db = Path(path) if not db.is_file(): raise RuntimeError(f"SQLite 文件不存在: {path}") try: with sqlite3.connect(f"file:{db}?mode=ro", uri=True) as conn: columns = conn.execute("PRAGMA table_info(click_gap)").fetchall() except sqlite3.Error as exc: raise RuntimeError(f"SQLite 不可读: {exc}") from exc if not columns: raise RuntimeError("click_gap 表不存在") by_name = {row[1]: row for row in columns} missing = set(EXPECTED_TYPES) - set(by_name) if missing: raise RuntimeError(f"click_gap 缺少列: {sorted(missing)}") for name, expected in EXPECTED_TYPES.items(): if str(by_name[name][2]).upper() != expected: raise RuntimeError(f"click_gap 列类型不兼容: {name}") if [(row[1], row[5]) for row in columns if row[5]] != [("pid", 1), ("dt", 2), ("pkg_name", 3)]: raise RuntimeError("click_gap 主键必须为 (pid, dt, pkg_name)") nullable_required = {"pid", "dt", "pkg_name", "shareit_clicks", "af_clicks", "installs_ua", "clicks_diff", "updated_at"} if any(by_name[name][3] != 1 for name in nullable_required): raise RuntimeError("click_gap 必填列不得允许 NULL") def read_existing_rows(path: str, start: str, end: str, pid: str = PID) -> list[dict[str, Any]]: validate_existing_schema(path) with sqlite3.connect(path) as conn: conn.row_factory = sqlite3.Row return [dict(row) for row in conn.execute("SELECT * FROM click_gap WHERE pid=? AND dt BETWEEN ? AND ?", (pid, start, end))] def upsert_report(conn: sqlite3.Connection, rows: list[dict[str, Any]], start: str, end: str) -> int: now = dt.datetime.now().astimezone().isoformat(timespec="seconds") conn.execute("DELETE FROM click_gap WHERE pid=? AND dt BETWEEN ? AND ?", (PID, start, end)) values = [] for row in rows: values.append((PID, row["dt"], row["attr_platform"], row["pkg_name"], row["shareit_clicks"], row["af_clicks"], row["installs_ua"], row["clicks_diff"], row["clicks_diff_rate"], now)) conn.executemany("INSERT INTO click_gap (pid,dt,attr_platform,pkg_name,shareit_clicks,af_clicks,installs_ua,clicks_diff,clicks_diff_rate,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?)", values) return len(values) def superset_rows(start: str, end: str) -> list[dict[str, Any]]: from superset_login import get_cookies import requests from requests_toolbelt import MultipartEncoder session, csrf = get_cookies() nextday = (dt.datetime.strptime(end, "%Y%m%d").date() + dt.timedelta(days=1)).strftime("%Y-%m-%d") start_iso = dt.datetime.strptime(start, "%Y%m%d").strftime("%Y-%m-%d") query = {"datasource": "114__table", "viz_type": "table", "slice_id": 1344, "url_params": {}, "time_range_endpoints": ["inclusive", "exclusive"], "granularity_sqla": "dt_as_time", "time_grain_sqla": "P1D", "time_range": f"{start_iso}T00:00:00 : {nextday}T00:00:00", "query_mode": "aggregate", "groupby": ["dt", "attr_platform", "pkg_name"], "metrics": ["clicks", "pb_install"], "all_columns": [], "array_join": [], "percent_metrics": [], "timeseries_limit_metric": ["dt_sort"], "order_by_cols": [], "row_limit": None, "order_desc": True, "adhoc_filters": [{"clause": "WHERE", "expressionType": "SIMPLE", "comparator": ["appsflyer"], "operator": "IN", "subject": "attr_platform"}, {"clause": "WHERE", "expressionType": "SIMPLE", "comparator": ["0"], "operator": "IN", "subject": "is_internal"}], "table_timestamp_format": "smart_date", "color_pn": True, "show_cell_bars": True} encoder = MultipartEncoder(fields={"form_data": json.dumps(query)}) response = session.post('https://ads-bi.ushareit.org/superset/explore_json/?form_data={"slice_id":1344}&force=true', headers={"x-csrftoken": csrf, "content-type": encoder.content_type}, data=encoder, timeout=(10, 60)) if response.status_code != 200: raise RuntimeError(f"Superset 请求失败: HTTP {response.status_code}") payload = response.json() data = payload.get("data") if isinstance(payload, dict) else None if not isinstance(data, dict) or not isinstance(data.get("records"), list) or not isinstance(data.get("columns"), list): raise RuntimeError("Superset 响应格式错误") return [dict(zip(data["columns"], row)) if isinstance(row, (list, tuple)) else row for row in data["records"]] def pkg_list(rows: list[dict[str, Any]]) -> list[str]: totals: dict[str, float] = {} for row in rows: pkg = str(row.get("pkg_name") or "").strip() if pkg: totals[pkg] = totals.get(pkg, 0) + float(row.get("clicks") or 0) return sorted(pkg for pkg, clicks in totals.items() if clicks > CLICK_NUM) def appsflyer_rows(packages: list[str], start: str, end: str, browser: Any) -> list[dict[str, Any]]: import requests cookie = browser.get_cookie_str() if not cookie: raise RuntimeError("AppsFlyer cookie 为空") payload = {"dates": {"start": dt.datetime.strptime(start, "%Y%m%d").strftime("%Y-%m-%d"), "end": dt.datetime.strptime(end, "%Y%m%d").strftime("%Y-%m-%d")}, "filters": {"app-id": []}, "view-type": "unified", "localization": {"timezone": "UTC", "currency": "USD"}, "groupings": [{"dimension": "date", "limit": 200}], "summations": ["totals", "others"], "metrics": [{"metric-id": "clicks"}, {"metric-id": "installs-ua"}], "format": "json", "granularity": "days", "implicit-sorting": True} result: list[dict[str, Any]] = [] url = "https://hq1.appsflyer.com/platform/dashboard?widget=platform-table__performance:3" for package in packages: app_id = SPECIAL_PKGS.get(package, package) payload["filters"]["app-id"] = [app_id] last_error = None for attempt in range(4): try: response = requests.post(url, data=json.dumps(payload), headers={"content-type": "application/json;charset=UTF-8", "cookie": cookie}, timeout=(10, 60)) if response.status_code == 429 or response.status_code >= 500: raise RuntimeError(f"HTTP {response.status_code}") if response.status_code != 200: raise RuntimeError(f"HTTP {response.status_code}") data = response.json().get("data") if not isinstance(data, list) or any(not isinstance(item, dict) for item in data): raise RuntimeError("响应格式错误") for item in data: item["app_id"] = package result.extend(data) break except Exception as exc: # do not leak request contents last_error = exc if attempt < 3: time.sleep(2 ** (attempt + 1)) else: raise RuntimeError(f"AppsFlyer 查询失败 pkg={package}: {last_error}") return result def build_report(shareit: list[dict[str, Any]], af: list[dict[str, Any]], packages: list[str]) -> list[dict[str, Any]]: shared: dict[tuple[str, str, str], float] = {} for row in shareit: if str(row.get("attr_platform") or "") != "appsflyer": continue pkg = str(row.get("pkg_name") or "").strip() if not pkg or pkg not in packages: continue raw = str(row.get("dt") or "")[:10].replace("-", "") date = date8(raw) shared[(date, "appsflyer", pkg)] = shared.get((date, "appsflyer", pkg), 0) + float(row.get("clicks") or 0) af_totals: dict[tuple[str, str], list[float]] = {} for row in af: raw = str(row.get("date") or "")[:10].replace("-", "") date = date8(raw) pkg = str(row.get("app_id") or "").strip() key = (date, pkg) values = af_totals.setdefault(key, [0.0, 0.0]) values[0] += float(row.get("filtersGranularityMetricIdClicksPeriod", row.get("clicks", 0)) or 0) values[1] += float(row.get("attributionSourceAppsflyerFiltersGranularityMetricIdInstallsUaPeriod", row.get("installs_ua", 0)) or 0) result = [] for (date, platform, pkg), share_clicks in shared.items(): af_clicks, installs = af_totals.get((date, pkg), [0.0, 0.0]) diff = share_clicks - af_clicks result.append({"dt": date, "attr_platform": platform, "pkg_name": pkg, "shareit_clicks": finite_number(share_clicks, "shareit_clicks"), "af_clicks": finite_number(af_clicks, "af_clicks"), "installs_ua": finite_number(installs, "installs_ua"), "clicks_diff": finite_number(diff, "clicks_diff"), "clicks_diff_rate": None if share_clicks == 0 else finite_number(round(diff / share_clicks, 3), "clicks_diff_rate", True)}) result.sort(key=lambda row: (row["dt"], row["shareit_clicks"]), reverse=True) if len({(r["dt"], r["pkg_name"]) for r in result}) != len(result): raise RuntimeError("报告存在重复业务键") return result def validate_rows(rows: list[dict[str, Any]], start: str, end: str) -> None: for row in rows: if list(row) != REPORT_COLUMNS or date8(row["dt"]) < start or date8(row["dt"]) > end or not str(row["pkg_name"]).strip(): raise ValueError("报告字段、日期或包名不合法") for field in REPORT_COLUMNS[3:]: finite_number(row[field], field, field == "clicks_diff_rate") def make_files(rows: list[dict[str, Any]], dates: list[str], out: Path) -> None: out.mkdir(parents=True, exist_ok=True) for date in dates: content = [row for row in rows if row["dt"] == date] target = out / f"{date}.json" target.write_text(json.dumps(content, ensure_ascii=False, allow_nan=False, separators=(",", ":")) + "\n", encoding="utf-8") index = {"schema_version": 1, "generated_at": dt.datetime.now().astimezone().isoformat(timespec="seconds"), "files": [{"date": date, "path": f"data/{date}.json"} for date in sorted(dates, reverse=True)]} (out / "index.json").write_text(json.dumps(index, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def sensitive_check(paths: list[Path]) -> None: pattern = re.compile(r"cookie|authorization|password|api[_-]?key|token", re.I) for path in paths: if path.name.startswith(".env") or path.suffix in {".db", ".log"}: raise RuntimeError(f"禁止发布敏感文件: {path.name}") if path.suffix in {".json", ".html", ".md"} and pattern.search(path.read_text(encoding="utf-8")): raise RuntimeError(f"文件疑似包含敏感字段: {path.name}") def publish_files(repo: Path, generated: Path, dates: list[str]) -> None: data = repo / "data" data.mkdir(exist_ok=True) for date in dates: source = generated / f"{date}.json" tmp = data / f".{date}.tmp" shutil.copyfile(source, tmp) os.replace(tmp, data / f"{date}.json") os.replace(generated / "index.json", data / "index.json") def git_run(repo: Path, args: list[str]) -> subprocess.CompletedProcess[str]: return subprocess.run(["git", *args], cwd=repo, text=True, capture_output=True, timeout=120, check=False) def git_publish(repo: Path, start: str, end: str, no_push: bool) -> str: if not (repo / ".git").exists(): raise RuntimeError("仓库尚未初始化；请先执行 git init 并配置 remote") status = git_run(repo, ["status", "--porcelain"]).stdout.strip().splitlines() allowed = {"README.md", ".gitignore", "index.html", "scripts/af_click_gap_pipeline.py", "launchd/com.roasbloom.af-click-gap.plist.template"} allowed.update({f"data/{date}.json" for date in (repo / "data").glob("*.json")}) unexpected = [line for line in status if line[3:] not in allowed] if unexpected: raise RuntimeError("工作区已有未预期改动: " + ", ".join(unexpected)) paths = ["README.md", ".gitignore", "index.html", "scripts/af_click_gap_pipeline.py", "launchd/com.roasbloom.af-click-gap.plist.template", "data"] add = git_run(repo, ["add", *paths]) if add.returncode: raise RuntimeError("git add 失败: " + add.stderr.strip()) staged = git_run(repo, ["diff", "--cached", "--name-only"]) if staged.returncode or not staged.stdout.strip(): raise RuntimeError("没有可提交的文件") commit = git_run(repo, ["commit", "-m", f"feat: publish click gap {start}-{end}"]) if commit.returncode: raise RuntimeError("git commit 失败: " + commit.stderr.strip()) commit_hash = git_run(repo, ["rev-parse", "HEAD"]).stdout.strip() if no_push: return commit_hash + " (未 push)" remote = git_run(repo, ["remote", "-v"]) if remote.returncode or not remote.stdout.strip(): raise RuntimeError("未配置 Git remote；本地 commit 已保留") push = git_run(repo, ["push"]) if push.returncode: raise RuntimeError("git push 失败（本地 commit 已保留）: " + push.stderr.strip()) return commit_hash + " (已 push)" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Extract and publish AppsFlyer click-gap JSON") parser.add_argument("--start-date", help="补数开始日期 YYYY-MM-DD") parser.add_argument("--end-date", help="补数结束日期 YYYY-MM-DD") parser.add_argument("--dry-run", action="store_true", help="只提取并校验，不写正式文件、数据库或 Git") group = parser.add_mutually_exclusive_group() group.add_argument("--no-push", action="store_true", help="写文件和数据库但不 push") group.add_argument("--push", action="store_true", help="写文件、数据库并 commit/push（默认）") parser.add_argument("--db-path", default=DB_PATH, help=argparse.SUPPRESS) return parser.parse_args() def main() -> int: args = parse_args() try: if args.dry_run and (args.no_push or args.push): raise ValueError("--dry-run 不可与 --push/--no-push 同时使用") start, end = window(args.start_date, args.end_date) validate_existing_schema(args.db_path) shareit = superset_rows(start, end) packages = pkg_list(shareit) af: list[dict[str, Any]] = [] browser = None if packages: from af_browser import AFBrowser browser = AFBrowser() try: af = appsflyer_rows(packages, start, end, browser) finally: try: browser.close() except Exception: LOG.warning("关闭浏览器失败") rows = build_report(shareit, af, packages) validate_rows(rows, start, end) dates = [(dt.datetime.strptime(start, "%Y%m%d").date() + dt.timedelta(days=i)).strftime("%Y%m%d") for i in range((dt.datetime.strptime(end, "%Y%m%d").date() - dt.datetime.strptime(start, "%Y%m%d").date()).days + 1)] repo = Path(__file__).resolve().parents[1] with tempfile.TemporaryDirectory(prefix="af-click-gap-") as temp: generated = Path(temp) / "data" make_files(rows, dates, generated) sensitive_check(list(generated.glob("*"))) if args.dry_run: LOG.info("dry-run 成功: %s-%s, %d 行", start, end, len(rows)) return 0 publish_files(repo, generated, dates) with sqlite3.connect(args.db_path) as conn: conn.execute("BEGIN") upsert_report(conn, rows, start, end) conn.commit() if args.no_push: LOG.info("已发布本地文件和 SQLite，未执行 Git") else: result = git_publish(repo, start, end, False) LOG.info("Git 结果: %s", result) return 0 except ValueError as exc: LOG.error("输入失败: %s", exc) return 2 except Exception as exc: LOG.error("任务失败: %s", exc) return 1 if __name__ == "__main__": raise SystemExit(main())