"""Run DataSteward's repository-maintenance audits as one deterministic command."""
from __future__ import annotations

import argparse
import subprocess
import sys
from pathlib import Path

# Blocking audits are clean on main and act as regression gates.
_AUDITS = (
    "audit_ci_contract.py",
    "audit_docs_contract.py",
    "audit_entrypoints.py",
    "audit_env_example.py",
    "audit_gitignore_contract.py",
    "audit_markdown_links.py",
    "audit_path_case.py",
    "audit_pyproject_metadata.py",
    "audit_python_compile.py",
    "audit_repository_hygiene.py",
    "audit_secret_filenames.py",
    "audit_sensitive.py",
    "audit_source_layout.py",
    "audit_test_layout.py",
    "audit_text_integrity.py",
    "audit_workflow_security.py",
    "audit_json_syntax.py",
    "audit_toml_syntax.py",
    "audit_yaml_syntax.py",
    "audit_unicode_bidi.py",
    "audit_dynamic_code.py",
    "audit_subprocess_shell.py",
    "audit_unsafe_deserialization.py",
    "audit_weak_hash.py",
    "audit_tls_verification.py",
    "audit_tempfile_safety.py",
    "audit_dependency_sources.py",
    "audit_markdown_fences.py",
    "audit_merge_markers.py",
    "audit_yaml_loading.py",
    "audit_unicode_paths.py",
    "audit_symlink_targets.py",
    "audit_bare_except.py",
    "audit_mutable_default.py",
    "audit_wildcard_import.py",
    "audit_debug_calls.py",
    "audit_os_system.py",
    "audit_datetime_utcnow.py",
    "audit_absolute_user_paths.py",
    "audit_duplicate_definitions.py",
    "audit_unsafe_chmod.py",
    "audit_http_timeout.py",
    "audit_urlopen_timeout.py",
    "audit_subprocess_timeout.py",
    "audit_async_blocking_sleep.py",
    "audit_async_subprocess.py",
    "audit_runtime_asserts.py",
    "audit_baseexception_handlers.py",
    "audit_sys_path_mutation.py",
    "audit_interactive_input.py",
    "audit_unverified_ssl_context.py",
    "audit_uuid1.py",
    "audit_socket_timeout.py",
    "audit_path_text_encoding.py",
    "audit_open_text_encoding.py",
    "audit_naive_fromtimestamp.py",
    "audit_builtin_hash.py",
    "audit_tar_extractall.py",
    "audit_unpack_archive.py",
    "audit_os_chdir.py",
    "audit_os_umask.py",
    "audit_locale_mutation.py",
    "audit_warning_suppression.py",
    "audit_socket_default_timeout.py",
    "audit_signal_handlers.py",
    "audit_sys_exit.py",
    "audit_recursion_limit.py",
    "audit_gc_disable.py",
    "audit_random_seed.py",
    "audit_asyncio_run.py",
    "audit_logging_basic_config.py",
    "audit_subprocess_run_check.py",
    "audit_thread_daemon.py",
)

# New or repository-wide rules begin as visible advisories. This keeps the
# signal available on every CI run without turning legacy findings into a red
# main branch. Promote a rule to _AUDITS after its backlog is understood and
# remediated.
_ADVISORY_AUDITS = (
    "audit_naive_datetime_now.py",
    "audit_environ_mutation.py",
    "audit_json_nan.py",
    "audit_sql_interpolation.py",
    "audit_contextlib_suppress.py",
    "audit_tempfile_mktemp.py",
    "audit_unbounded_queue.py",
    "audit_unbounded_deque.py",
    "audit_unbounded_lru_cache.py",
    "audit_os_putenv.py",
    "audit_tzset.py",
    "audit_numpy_global_state.py",
    "audit_resource_limits.py",
    "audit_signal_timers.py",
    "audit_sqlite_timeout.py",
    "audit_httpx_timeout.py",
    "audit_aiohttp_timeout.py",
    "audit_websocket_timeout.py",
    "audit_requests_session.py",
    "audit_asyncio_create_task.py",
    "audit_multiprocessing_daemon.py",
    "audit_shelve_usage.py",
    "audit_zip_extractall.py",
    "audit_fetchall.py",
    "audit_pandas_read_sql.py",
    "audit_destructive_sql.py",
    "audit_atexit_register.py",
    "audit_multiprocessing_start_method.py",
    "audit_multiprocessing_executable.py",
    "audit_multiprocessing_forkserver_preload.py",
    "audit_asyncio_event_loop_policy.py",
    "audit_asyncio_set_event_loop.py",
    "audit_sys_excepthook.py",
    "audit_sys_trace.py",
    "audit_sys_profile.py",
    "audit_sys_int_max_str_digits.py",
    "audit_sys_coroutine_origin_tracking.py",
    "audit_sys_asyncgen_hooks.py",
    "audit_threading_excepthook.py",
    "audit_threading_stack_size.py",
    "audit_threading_trace_all.py",
    "audit_threading_profile_all.py",
    "audit_gc_debug.py",
    "audit_gc_threshold.py",
    "audit_random_setstate.py",
    "audit_tempfile_tempdir.py",
    "audit_urllib_install_opener.py",
    "audit_sqlite_callback_tracebacks.py",
    "audit_sqlite_shared_cache.py",
    "audit_logging_capture_warnings.py",
    "audit_numpy_printoptions.py",
    "audit_decimal_default_context.py",
    "audit_runtime_global_state.py",
)


def _validate_scripts(scripts: tuple[str, ...]) -> tuple[str, ...]:
    if not isinstance(scripts, tuple):
        raise ValueError("scripts must be a tuple of audit filenames")
    seen: set[str] = set()
    for script in scripts:
        if (
            not isinstance(script, str)
            or not script
            or script != script.strip()
            or not script.endswith(".py")
            or "/" in script
            or "\\" in script
        ):
            raise ValueError("audit script names must be simple .py filenames")
        if script in seen:
            raise ValueError("audit script names must be unique")
        seen.add(script)
    return scripts


def _failure_detail(result: object) -> str:
    stdout = getattr(result, "stdout", "") or ""
    stderr = getattr(result, "stderr", "") or ""
    detail = " ".join(f"{stdout}\n{stderr}".split()) or "audit failed"
    return detail[:1000]


def _run_audit(root: Path, script: str) -> str | None:
    path = root / "scripts" / script
    if not path.is_file():
        return "audit script is missing"
    module = f"scripts.{Path(script).stem}"
    try:
        result = subprocess.run(
            [sys.executable, "-m", module, str(root)],
            cwd=root,
            capture_output=True,
            encoding="utf-8",
            errors="replace",
            timeout=30,
            check=False,
        )
    except subprocess.TimeoutExpired:
        return "audit timed out"
    except OSError as exc:
        return f"audit could not start ({exc})"
    return _failure_detail(result) if result.returncode else None


def run_audits(root: Path, *, scripts: tuple[str, ...] = _AUDITS) -> list[str]:
    if not isinstance(root, Path) or not root.is_dir():
        raise ValueError("root must be an existing directory")
    scripts = _validate_scripts(scripts)
    failures: list[str] = []
    for script in scripts:
        detail = _run_audit(root, script)
        if detail:
            failures.append(f"{script}: {detail}")
    return failures


def run_advisory_audits(root: Path, *, scripts: tuple[str, ...] = _ADVISORY_AUDITS) -> list[str]:
    if not isinstance(root, Path) or not root.is_dir():
        raise ValueError("root must be an existing directory")
    scripts = _validate_scripts(scripts)
    findings: list[str] = []
    for script in scripts:
        detail = _run_audit(root, script)
        if detail:
            findings.append(f"{script}: {detail}")
    return findings


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("root", nargs="?", default=".")
    args = parser.parse_args(argv)
    root = Path(args.root)
    failures = run_audits(root)
    advisories = run_advisory_audits(root)
    for finding in advisories:
        print(f"[advisory] {finding}", file=sys.stderr)
    for failure in failures:
        print(failure)
    return 1 if failures else 0


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