#!/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()