query-risk-user-features/scripts/run_pipeline.py

#!/usr/bin/env python3
"""Run behavior features, social features, and final aggregation in sequence."""

from __future__ import annotations

import argparse
import subprocess
import sys
from pathlib import Path


PROJECT_DIR = Path(__file__).resolve().parents[1]
SCRIPTS_DIR = PROJECT_DIR / "scripts"


def run(command: list[str]) -> None:
    print("+ " + " ".join(command), flush=True)
    subprocess.run(command, check=True)


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--input", type=Path, required=True)
    parser.add_argument("--run-name", required=True)
    parser.add_argument("--lookback-days", type=int, default=180)
    parser.add_argument("--skip-behavior", action="store_true")
    parser.add_argument("--skip-social", action="store_true")
    parser.add_argument("--sql-only", action="store_true")
    args = parser.parse_args()

    behavior_detail = PROJECT_DIR / "output" / args.run_name / "behavior" / "user_detail.csv"
    social_dir = PROJECT_DIR / "output" / args.run_name / "social"
    if not args.skip_behavior:
        command = [
            sys.executable,
            str(SCRIPTS_DIR / "run_behavior_features.py"),
            "--input", str(args.input),
            "--run-name", args.run_name,
            "--lookback-days", str(args.lookback_days),
        ]
        if args.sql_only:
            command.append("--sql-only")
        run(command)
    if args.sql_only:
        if behavior_detail.exists() and not args.skip_social:
            run(
                [
                    sys.executable,
                    str(SCRIPTS_DIR / "run_social_features.py"),
                    "--input", str(args.input),
                    "--behavior-detail", str(behavior_detail),
                    "--run-name", args.run_name,
                    "--sql-only",
                ]
            )
        return
    if not args.skip_social:
        run(
            [
                sys.executable,
                str(SCRIPTS_DIR / "run_social_features.py"),
                "--input", str(args.input),
                "--behavior-detail", str(behavior_detail),
                "--run-name", args.run_name,
                "--lookback-days", str(args.lookback_days),
            ]
        )
    report = PROJECT_DIR / "output" / args.run_name / "risk_user_feature_report.xlsx"
    command = [
        sys.executable,
        str(SCRIPTS_DIR / "build_report.py"),
        "--behavior-detail", str(behavior_detail),
        "--output", str(report),
    ]
    if social_dir.exists():
        command.extend(["--social-dir", str(social_dir)])
    run(command)


if __name__ == "__main__":
    main()