277 lines
12 KiB
Python
Executable File
277 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Read click-gap data from SQLite 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 tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
DB_PATH = "/Users/ads-center/Documents/database/af_click_gap.db"
|
|
PID = "shareit"
|
|
REPORT_COLUMNS = ["dt", "attr_platform", "pkg_name", "shareit_clicks", "af_clicks", "installs_ua", "clicks_diff", "clicks_diff_rate"]
|
|
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 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 必须同时提供")
|
|
try:
|
|
first = dt.date.fromisoformat(start)
|
|
last = dt.date.fromisoformat(end)
|
|
except ValueError as exc:
|
|
raise ValueError("日期必须为 YYYY-MM-DD") from exc
|
|
if first > last or (last - first).days > 6:
|
|
raise ValueError("日期区间必须为包含首尾且不超过 7 天")
|
|
return first.strftime("%Y%m%d"), last.strftime("%Y%m%d")
|
|
|
|
|
|
def date8(value: str) -> str:
|
|
try:
|
|
return dt.datetime.strptime(value, "%Y%m%d").strftime("%Y%m%d")
|
|
except ValueError as exc:
|
|
raise ValueError(f"非法日期: {value}") from exc
|
|
|
|
|
|
def finite_number(value: Any, field: str, allow_null: bool = False) -> int | float | None:
|
|
if value is None:
|
|
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}")
|
|
primary_key = [(row[1], row[5]) for row in columns if row[5]]
|
|
if primary_key != [("pid", 1), ("dt", 2), ("pkg_name", 3)]:
|
|
raise RuntimeError("click_gap 主键必须为 (pid, dt, pkg_name)")
|
|
|
|
|
|
def read_rows(path: str, start: str | None, end: str | None, pid: str = PID) -> list[dict[str, Any]]:
|
|
validate_existing_schema(path)
|
|
try:
|
|
with sqlite3.connect(f"file:{path}?mode=ro", uri=True) as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
query = (
|
|
"SELECT dt, attr_platform, pkg_name, shareit_clicks, af_clicks, installs_ua, clicks_diff, clicks_diff_rate "
|
|
"FROM click_gap WHERE pid=?"
|
|
)
|
|
parameters: list[str] = [pid]
|
|
if start is not None and end is not None:
|
|
query += " AND dt BETWEEN ? AND ?"
|
|
parameters.extend((start, end))
|
|
query += " ORDER BY dt DESC, shareit_clicks DESC, pkg_name"
|
|
result = conn.execute(query, parameters)
|
|
return [dict(row) for row in result]
|
|
except sqlite3.Error as exc:
|
|
raise RuntimeError(f"SQLite 查询失败: {exc}") from exc
|
|
|
|
|
|
def available_dates(rows: list[dict[str, Any]]) -> list[str]:
|
|
dates = set()
|
|
for row in rows:
|
|
dates.add(date8(str(row["dt"])))
|
|
return sorted(dates, reverse=True)
|
|
|
|
|
|
def validate_rows(rows: list[dict[str, Any]], start: str, end: str) -> list[dict[str, Any]]:
|
|
normalized = []
|
|
for row in rows:
|
|
if list(row) != REPORT_COLUMNS:
|
|
raise ValueError("报告字段不合法")
|
|
normalized_row = dict(row)
|
|
normalized_row["attr_platform"] = "" if row["attr_platform"] is None else row["attr_platform"]
|
|
normalized_row["pkg_name"] = "" if row["pkg_name"] is None else row["pkg_name"]
|
|
normalized_row["dt"] = date8(str(row["dt"]))
|
|
if not start <= normalized_row["dt"] <= end or not str(normalized_row["pkg_name"]).strip():
|
|
raise ValueError("报告日期、平台或包名不合法")
|
|
for field in REPORT_COLUMNS[3:]:
|
|
normalized_row[field] = finite_number(row[field], field, field == "clicks_diff_rate")
|
|
normalized.append(normalized_row)
|
|
keys = [(row["dt"], row["pkg_name"]) for row in normalized]
|
|
if len(set(keys)) != len(keys):
|
|
raise ValueError("报告存在重复业务键")
|
|
return normalized
|
|
|
|
|
|
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("仓库尚未初始化")
|
|
status_output = git_run(repo, ["status", "--porcelain", "--untracked-files=all"]).stdout
|
|
status = status_output.splitlines()
|
|
allowed = re.compile(r"data/(?:index\.json|[0-9]{8}\.json)$")
|
|
allowed_paths = []
|
|
unexpected = []
|
|
for line in status:
|
|
prefix = line[:2] if len(line) >= 2 else ""
|
|
path = line[3:] if len(line) >= 4 and line[2] == " " else ""
|
|
if prefix not in {" M", "M ", "??", "D ", " D"} or not allowed.fullmatch(path):
|
|
unexpected.append(line)
|
|
else:
|
|
allowed_paths.append(path)
|
|
if unexpected:
|
|
raise RuntimeError("工作区已有未预期改动: " + ", ".join(unexpected))
|
|
add = git_run(repo, ["add", "--", *allowed_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)"
|
|
if not git_run(repo, ["remote", "-v"]).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="Read SQLite click-gap data and publish JSON")
|
|
parser.add_argument("--start-date", help="读取开始日期 YYYY-MM-DD")
|
|
parser.add_argument("--end-date", help="读取结束日期 YYYY-MM-DD")
|
|
parser.add_argument("--all", action="store_true", help="读取数据库中所有实际有数据的日期")
|
|
parser.add_argument("--dry-run", action="store_true", help="只读取、校验并生成临时 JSON，不写文件或 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 同时使用")
|
|
if args.all and (args.start_date or args.end_date):
|
|
raise ValueError("--all 不可与 --start-date/--end-date 同时使用")
|
|
if args.all:
|
|
rows = validate_rows(read_rows(args.db_path, None, None), "00010101", "99991231")
|
|
dates = available_dates(rows)
|
|
start = dates[-1] if dates else "无日期"
|
|
end = dates[0] if dates else "无日期"
|
|
else:
|
|
start, end = window(args.start_date, args.end_date)
|
|
rows = validate_rows(read_rows(args.db_path, start, end), start, end)
|
|
first = dt.datetime.strptime(start, "%Y%m%d").date()
|
|
last = dt.datetime.strptime(end, "%Y%m%d").date()
|
|
dates = [(first + dt.timedelta(days=i)).strftime("%Y%m%d") for i in range((last - first).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)
|
|
result = git_publish(repo, start, end, args.no_push)
|
|
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())
|