batch_analyze.py

from __future__ import annotations

import argparse
import subprocess
from datetime import datetime
from pathlib import Path

from pipeline.cli_common import build_tool_env, repo_root_from_script, resolve_repo_path, tool_dir, venv_python


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Run SliSE over a directory of Solidity contracts.")
    parser.add_argument("--input-dir", default="Datasets/DB1/All_contract")
    parser.add_argument("--output-dir", default="SliSE_tool/detectiion_output")
    parser.add_argument("--file-pattern", default="*.sol")
    parser.add_argument("--max-files", type=int, default=0)
    parser.add_argument("--resume", action="store_true")
    parser.add_argument("--enable-priority", action="store_true")
    parser.add_argument("--topk", type=int, default=0)
    parser.add_argument("--budget-mode", choices=("fixed", "adaptive"), default="fixed")
    parser.add_argument("--adaptive-budget-policy", default="heuristic")
    parser.add_argument(
        "--adaptive-budget-error-policy",
        choices=("heuristic", "fixed_10", "fail"),
        default="heuristic",
    )
    parser.add_argument("--budget-model-path", default="")
    return parser.parse_args(argv)


def load_completed_contracts(stage_file: Path) -> set[str]:
    completed: set[str] = set()
    if not stage_file.exists():
        return completed
    for line in stage_file.read_text(encoding="utf-8", errors="ignore").splitlines():
        if not line.strip():
            continue
        completed.add(line.split(",", 1)[0])
    return completed


def write_summary(
    summary_file: Path,
    *,
    input_dir: Path,
    output_dir: Path,
    file_pattern: str,
    total_files: int,
    success_count: int,
    fail_count: int,
) -> None:
    success_rate = round((success_count / total_files) * 100, 2) if total_files else 0.0
    summary_text = "\n".join(
        [
            "Batch Analysis Summary",
            f"Generated: {datetime.now():%Y-%m-%d %H:%M:%S}",
            "==========================================",
            f"Input directory: {input_dir}",
            f"Output directory: {output_dir}",
            f"File pattern: {file_pattern}",
            f"Total files: {total_files}",
            f"Succeeded: {success_count}",
            f"Failed: {fail_count}",
            f"Success rate: {success_rate} pct",
            "==========================================",
            "",
        ]
    )
    summary_file.write_text(summary_text, encoding="utf-8")


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    repo_root = repo_root_from_script(__file__)
    input_dir = resolve_repo_path(repo_root, args.input_dir, "Datasets/DB1/All_contract")
    output_dir = resolve_repo_path(repo_root, args.output_dir, "SliSE_tool/detectiion_output")
    budget_model_path = (
        resolve_repo_path(repo_root, args.budget_model_path, args.budget_model_path)
        if args.budget_model_path.strip()
        else None
    )
    slise_path = tool_dir(repo_root)
    python_executable = venv_python(repo_root)

    if not input_dir.exists():
        raise FileNotFoundError(f"Input directory not found: {input_dir}")

    output_dir.mkdir(parents=True, exist_ok=True)
    files = sorted(input_dir.glob(args.file_pattern))
    files = [path for path in files if path.is_file()]
    source_file_count = len(files)

    if args.max_files > 0:
        files = files[: args.max_files]

    if args.resume:
        completed = load_completed_contracts(output_dir / "stage1&2_detection.txt")
        files = [path for path in files if str(path) not in completed]

    total_files = len(files)
    success_count = 0
    fail_count = 0

    print("=" * 80)
    print("SliSE Batch Analysis")
    print("=" * 80)
    print(f"Input directory: {input_dir}")
    print(f"Output directory: {output_dir}")
    print(f"Files to process: {total_files} / {source_file_count}")
    print(f"Resume mode: {args.resume}")
    print(f"Priority enabled: {args.enable_priority}")
    print(f"Priority TopK: {args.topk}")
    print(f"Budget mode: {args.budget_mode}")
    print(f"Adaptive budget policy: {args.adaptive_budget_policy}")
    print(f"Adaptive budget error policy: {args.adaptive_budget_error_policy}")
    print(f"Budget model path: {budget_model_path or ''}")
    print("=" * 80)
    print()

    if total_files == 0:
        print("No pending files. Exiting.")
        return 0

    env = build_tool_env(repo_root, output_dir=output_dir)

    for index, file_path in enumerate(files, start=1):
        percentage = round((index / total_files) * 100, 2)
        print(f"[{index}/{total_files}] ({percentage} pct) Analyze: {file_path.name}")
        try:
            slise_args = [str(python_executable), "SliSE.py", "-s", str(file_path), "-vt", "reentrancy"]
            if args.enable_priority:
                slise_args.extend(
                    [
                        "--priority-enable",
                        "--priority-topk",
                        str(args.topk),
                        "--budget-mode",
                        args.budget_mode,
                        "--adaptive-budget-policy",
                        args.adaptive_budget_policy,
                        "--adaptive-budget-error-policy",
                        args.adaptive_budget_error_policy,
                    ]
                )
                if budget_model_path is not None:
                    slise_args.extend(["--budget-model-path", str(budget_model_path)])

            completed = subprocess.run(
                slise_args,
                cwd=slise_path,
                env=env,
                capture_output=True,
                text=True,
                encoding="utf-8",
                errors="replace",
                check=False,
            )

            if completed.returncode == 0:
                success_count += 1
                print("  Success")
            else:
                fail_count += 1
                print(f"  Failed (exit code: {completed.returncode})")
                error_file = output_dir / f"errors_{file_path.stem}.txt"
                error_file.write_text((completed.stdout or "") + (completed.stderr or ""), encoding="utf-8")
        except Exception as exc:  # pragma: no cover - defensive wrapper around subprocess
            fail_count += 1
            print(f"  Exception: {exc}")
        print()

    success_rate = round((success_count / total_files) * 100, 2) if total_files else 0.0
    print("=" * 80)
    print("Analysis completed.")
    print("=" * 80)
    print(f"Total files: {total_files}")
    print(f"Succeeded: {success_count}")
    print(f"Failed: {fail_count}")
    print(f"Success rate: {success_rate} pct")
    print(f"Results saved to: {output_dir}")
    print("=" * 80)

    summary_file = output_dir / "batch_summary.txt"
    write_summary(
        summary_file,
        input_dir=input_dir,
        output_dir=output_dir,
        file_pattern=args.file_pattern,
        total_files=total_files,
        success_count=success_count,
        fail_count=fail_count,
    )
    print(f"Summary report saved to: {summary_file}")
    return 0 if fail_count == 0 else 1


if __name__ == "__main__":
    raise SystemExit(main())