Code Hardcode Audit

id: hardcoded-numeric-arg
language: python
severity: warning
message: >-
  Bare numeric literal in function call argument — extract to named constant.
note: >-
  Numeric literals in function calls (timeout=30, retries=3) are magic numbers
  that should be named constants or config values for maintainability.
rule:
  kind: keyword_argument
  has:
    kind: integer
    field: value

id: hardcoded-path-string
language: python
severity: warning
message: >-
  Hardcoded filesystem path — use pathlib or configuration.
note: >-
  Absolute paths break across machines and deployments. Use Path objects,
  environment variables, or platformdirs for portable paths.
rule:
  kind: string
  regex: "^[\"']/(tmp|var|etc|home|Users|opt)/"

id: hardcoded-sleep
language: python
severity: warning
message: >-
  Hardcoded sleep duration — extract to named constant for tuning.
note: >-
  Sleep durations are operational parameters that often need tuning per environment.
  Create SLEEP_* or DELAY_* constants backed by env vars.
rule:
  any:
    - pattern: time.sleep($N)
    - pattern: asyncio.sleep($N)
    - pattern: await asyncio.sleep($N)
constraints:
  N:
    kind: integer

id: hardcoded-url-string
language: python
severity: warning
message: >-
  Hardcoded URL string — extract to constant or configuration.
note: >-
  URLs change across environments (dev/staging/prod). Move to a constants
  module or BaseSettings class with env-var override.
rule:
  kind: string
  regex: "^[\"']https?://"

ruleDirs:
  - rules

rules:
  # Rule 1: Hardcoded URLs
  - id: hardcoded-url
    pattern-either:
      - pattern: '"http://$..."'
      - pattern: '"https://$..."'
      - pattern: "'http://$...'"
      - pattern: "'https://$...'"
    message: "Hardcoded URL detected. Consider extracting to a constant or configuration."
    severity: WARNING
    languages: [python, javascript, typescript, go, java]
    metadata:
      category: hardcode
      subcategory: url
      suggested_fix: "Move to constants module or environment variable"

  # Rule 2: Hardcoded Ports (simplified)
  - id: hardcoded-port
    pattern-either:
      - pattern: "port=8123"
      - pattern: "port=8443"
      - pattern: "port=9000"
      - pattern: "port=9440"
      - pattern: "port=3000"
      - pattern: "port=5000"
      - pattern: "port=5432"
      - pattern: "port=3306"
      - pattern: "port=6379"
      - pattern: "port=27017"
    message: "Hardcoded port number. Consider extracting to a named constant."
    severity: WARNING
    languages: [python, javascript, typescript, go, java]
    metadata:
      category: hardcode
      subcategory: port
      suggested_fix: "Create PORT_* constant in constants module"

  # Rule 3: Hardcoded Timeframe Strings
  - id: hardcoded-timeframe
    pattern-either:
      - pattern: '"1h"'
      - pattern: '"4h"'
      - pattern: '"1d"'
      - pattern: '"1m"'
      - pattern: '"15m"'
      - pattern: '"30m"'
      - pattern: '"1w"'
      - pattern: "'1h'"
      - pattern: "'4h'"
      - pattern: "'1d'"
      - pattern: "'1m'"
      - pattern: "'15m'"
      - pattern: "'30m'"
      - pattern: "'1w'"
    message: "Hardcoded timeframe string. Consider using a Timeframe type or constant."
    severity: INFO
    languages: [python, javascript, typescript]
    metadata:
      category: hardcode
      subcategory: timeframe
      suggested_fix: "Use TIMEFRAME_* constant or Timeframe enum"

  # Rule 4: Hardcoded File Paths
  - id: hardcoded-path
    pattern-either:
      - pattern: '"/tmp/$..."'
      - pattern: '"/var/$..."'
      - pattern: '"/etc/$..."'
      - pattern: '"/home/$..."'
      - pattern: '"/Users/$..."'
      - pattern: "'/tmp/$...'"
      - pattern: "'/var/$...'"
      - pattern: "'/etc/$...'"
      - pattern: "'/home/$...'"
      - pattern: "'/Users/$...'"
    message: "Hardcoded file path detected. Consider using pathlib or configuration."
    severity: WARNING
    languages: [python, javascript, typescript, go]
    metadata:
      category: hardcode
      subcategory: path
      suggested_fix: "Use Path objects or environment-based configuration"

  # Rule 5: Potential Hardcoded Credentials
  - id: hardcoded-credential
    patterns:
      - pattern-either:
          - pattern: 'password="$VALUE"'
          - pattern: "password='$VALUE'"
          - pattern: 'api_key="$VALUE"'
          - pattern: "api_key='$VALUE'"
          - pattern: 'secret="$VALUE"'
          - pattern: "secret='$VALUE'"
          - pattern: 'token="$VALUE"'
          - pattern: "token='$VALUE'"
      - pattern-not: '$VAR=""'
      - pattern-not: "$VAR=''"
    message: "Potential hardcoded credential. Use environment variables or secrets manager."
    severity: ERROR
    languages: [python, javascript, typescript]
    metadata:
      category: security
      subcategory: credential
      suggested_fix: "Use os.environ, Doppler, or secrets manager"

  # Rule 6: Hardcoded Retry Configuration (simplified)
  - id: hardcoded-retry-config
    pattern-either:
      - pattern: "max_retries=3"
      - pattern: "max_retries=5"
      - pattern: "retry_count=3"
      - pattern: "max_attempts=3"
      - pattern: "max_attempts=5"
      - pattern: "timeout=30"
      - pattern: "timeout=60"
      - pattern: "timeout=120"
      - pattern: "delay=1"
      - pattern: "delay=5"
    message: "Hardcoded retry/timeout configuration. Consider extracting to constants."
    severity: INFO
    languages: [python, javascript, typescript]
    metadata:
      category: hardcode
      subcategory: retry
      suggested_fix: "Create RETRY_* or TIMEOUT_* constants"

  # Rule 7: Hardcoded API Limits (simplified)
  - id: hardcoded-api-limit
    pattern-either:
      - pattern: "limit=100"
      - pattern: "limit=1000"
      - pattern: "page_size=100"
      - pattern: "batch_size=100"
      - pattern: "batch_size=1000"
      - pattern: "chunk_size=1000"
      - pattern: "max_results=100"
      - pattern: "max_results=1000"
    message: "Hardcoded API limit. Consider extracting to a named constant."
    severity: INFO
    languages: [python, javascript, typescript]
    metadata:
      category: hardcode
      subcategory: api_limit
      suggested_fix: "Create API_CHUNK_SIZE or similar constant"

  # Rule 8: Hardcoded Sleep Durations
  - id: hardcoded-sleep-duration
    pattern-either:
      - pattern: time.sleep($N)
      - pattern: asyncio.sleep($N)
      - pattern: await asyncio.sleep($N)
    message: "Hardcoded sleep duration. Extract to a named SLEEP_* or DELAY_* constant."
    severity: INFO
    languages: [python]
    metadata:
      category: hardcode
      subcategory: sleep
      suggested_fix: "Create SLEEP_* or DELAY_* constant backed by env var"

  # Rule 9: Hardcoded Circuit Breaker Config
  - id: hardcoded-circuit-breaker
    pattern-either:
      - pattern: CircuitBreaker(failure_threshold=$N, ...)
      - pattern: CircuitBreaker(..., recovery_timeout=$N, ...)
    message: "Hardcoded circuit breaker config. Extract to constants or BaseSettings."
    severity: WARNING
    languages: [python]
    metadata:
      category: hardcode
      subcategory: circuit_breaker
      suggested_fix: "Create CB_FAILURE_THRESHOLD, CB_RECOVERY_TIMEOUT constants"

  # Rule 10: Hardcoded Stamina Retry
  - id: hardcoded-stamina-retry
    pattern-either:
      - pattern: stamina.retry(attempts=$N, ...)
      - pattern: "@stamina.retry(attempts=$N, ...)"
    message: "Hardcoded stamina retry attempts. Extract to configuration."
    severity: INFO
    languages: [python]
    metadata:
      category: hardcode
      subcategory: retry
      suggested_fix: "Create RETRY_ATTEMPTS constant or add to BaseSettings"

  # Rule 11: Hardcoded Lock/Redis Timeout
  - id: hardcoded-lock-timeout
    pattern-either:
      - pattern: blocking_timeout=$N
      - pattern: lock_timeout=$N
      - pattern: "redis.lock($KEY, timeout=$N, ...)"
      - pattern: "redis.lock($KEY, ..., blocking_timeout=$N, ...)"
    message: "Hardcoded lock/Redis timeout. Extract to configuration."
    severity: WARNING
    languages: [python]
    metadata:
      category: hardcode
      subcategory: lock_timeout
      suggested_fix: "Create LOCK_TIMEOUT constant or add to BaseSettings"

  # Rule 12: Hardcoded Pydantic Field Default
  - id: hardcoded-pydantic-field-default
    pattern-either:
      - pattern: "Field(default=$N, ...)"
      - pattern: "Field($N, ...)"
    message: "Hardcoded Pydantic Field default. Consider BaseSettings with env-var override."
    severity: INFO
    languages: [python]
    metadata:
      category: hardcode
      subcategory: pydantic_default
      suggested_fix: "Use BaseSettings with env-var override or Field(default_factory=...)"

# /// script
# requires-python = ">=3.14"
# dependencies = []
# ///
"""Env-var coverage audit: cross-reference pydantic BaseSettings vs bare constants.

Usage:
    uv run --python 3.14 --script audit_env_coverage.py -- <path> [--output {json,text}]

Finds module-level constants and inline literals that lack env-var backing
via pydantic BaseSettings. Reports coverage gaps.
"""

import argparse
import ast
import json
import re
import sys
from dataclasses import dataclass, field
from pathlib import Path

# Patterns for inline keyword args that should be configurable
CONFIGURABLE_KWARGS = frozenset({
    "timeout", "max_retries", "retries", "attempts", "max_attempts",
    "delay", "backoff", "interval", "port", "batch_size", "chunk_size",
    "limit", "max_results", "page_size", "threshold", "max_failures",
    "workers", "max_workers", "ttl", "budget", "cooldown",
    "blocking_timeout", "lock_timeout", "recovery_timeout",
    "failure_threshold",
})

ALL_CAPS_RE = re.compile(r"^[A-Z][A-Z0-9_]{2,}$")


@dataclass
class SettingsField:
    """A field from a pydantic BaseSettings class."""

    class_name: str
    field_name: str
    file: str
    line: int


@dataclass
class BareConstant:
    """A module-level ALL_CAPS constant with a literal value."""

    name: str
    value: str
    file: str
    line: int


@dataclass
class InlineLiteral:
    """A numeric literal in a configurable keyword argument."""

    kwarg: str
    value: str
    file: str
    line: int
    func_name: str = ""


def _extract_settings_fields(tree: ast.Module, filepath: str) -> list[SettingsField]:
    """Find all fields in BaseSettings subclasses."""
    fields = []
    for node in ast.walk(tree):
        if not isinstance(node, ast.ClassDef):
            continue
        # Check if any base is BaseSettings (simple name check)
        is_settings = any(
            (isinstance(b, ast.Name) and b.id == "BaseSettings")
            or (isinstance(b, ast.Attribute) and b.attr == "BaseSettings")
            for b in node.bases
        )
        if not is_settings:
            continue
        for item in node.body:
            if isinstance(item, ast.AnnAssign) and isinstance(item.target, ast.Name):
                fields.append(SettingsField(
                    class_name=node.name,
                    field_name=item.target.id,
                    file=filepath,
                    line=item.lineno,
                ))
            elif isinstance(item, ast.Assign):
                for target in item.targets:
                    if isinstance(target, ast.Name):
                        fields.append(SettingsField(
                            class_name=node.name,
                            field_name=target.id,
                            file=filepath,
                            line=item.lineno,
                        ))
    return fields


def _extract_bare_constants(tree: ast.Module, filepath: str) -> list[BareConstant]:
    """Find module-level ALL_CAPS = literal assignments."""
    constants = []
    for node in tree.body:
        if not isinstance(node, ast.Assign):
            continue
        for target in node.targets:
            if not isinstance(target, ast.Name):
                continue
            if not ALL_CAPS_RE.match(target.id):
                continue
            if isinstance(node.value, ast.Constant):
                constants.append(BareConstant(
                    name=target.id,
                    value=repr(node.value.value),
                    file=filepath,
                    line=node.lineno,
                ))
    return constants


def _extract_inline_literals(tree: ast.Module, filepath: str) -> list[InlineLiteral]:
    """Find numeric literals in configurable keyword arguments."""
    literals = []
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        func_name = ""
        if isinstance(node.func, ast.Name):
            func_name = node.func.id
        elif isinstance(node.func, ast.Attribute):
            func_name = node.func.attr

        for kw in node.keywords:
            if kw.arg and kw.arg.lower() in CONFIGURABLE_KWARGS:
                if isinstance(kw.value, ast.Constant) and isinstance(kw.value.value, (int, float)):
                    literals.append(InlineLiteral(
                        kwarg=kw.arg,
                        value=repr(kw.value.value),
                        file=filepath,
                        line=kw.value.lineno,
                        func_name=func_name,
                    ))
    return literals


def audit_env_coverage(target: Path, output_format: str = "text") -> int:
    """Scan Python files and report env-var coverage gaps."""
    all_settings: list[SettingsField] = []
    all_constants: list[BareConstant] = []
    all_literals: list[InlineLiteral] = []

    # Collect all .py files
    if target.is_file():
        py_files = [target] if target.suffix == ".py" else []
    else:
        py_files = sorted(target.rglob("*.py"))

    for py_file in py_files:
        try:
            source = py_file.read_text(encoding="utf-8")
            tree = ast.parse(source, filename=str(py_file))
        except (SyntaxError, UnicodeDecodeError):
            continue

        rel = str(py_file)
        all_settings.extend(_extract_settings_fields(tree, rel))
        all_constants.extend(_extract_bare_constants(tree, rel))
        all_literals.extend(_extract_inline_literals(tree, rel))

    # Build set of env-backed field names (case-insensitive)
    env_backed = {f.field_name.lower() for f in all_settings}

    # Classify constants
    findings = []
    finding_id = 0

    for const in all_constants:
        name_lower = const.name.lower()
        # Strip common prefixes for matching (_DEFAULT_, _MAX_, etc.)
        stripped = re.sub(r"^_?(DEFAULT_|MAX_|MIN_)?", "", name_lower)
        has_backing = name_lower in env_backed or stripped in env_backed
        if not has_backing:
            finding_id += 1
            findings.append({
                "id": f"ENVCOV-{finding_id:03d}",
                "tool": "env-coverage",
                "rule": "constant-no-env-backing",
                "file": const.file,
                "line": const.line,
                "column": 0,
                "end_line": None,
                "message": f"Constant {const.name}={const.value} has no BaseSettings env-var override",
                "severity": "medium",
                "suggested_fix": f"Add {const.name.lower()} field to a BaseSettings class",
            })

    for lit in all_literals:
        finding_id += 1
        ctx = f" in {lit.func_name}()" if lit.func_name else ""
        findings.append({
            "id": f"ENVCOV-{finding_id:03d}",
            "tool": "env-coverage",
            "rule": "inline-literal",
            "file": lit.file,
            "line": lit.line,
            "column": 0,
            "end_line": None,
            "message": f"Inline literal {lit.kwarg}={lit.value}{ctx} — extract to named constant",
            "severity": "high",
            "suggested_fix": f"Create {lit.kwarg.upper()} constant or add to BaseSettings",
        })

    # Sort by file, line
    findings.sort(key=lambda f: (f["file"], f["line"]))

    if output_format == "json":
        by_rule: dict[str, int] = {}
        for f in findings:
            by_rule[f["rule"]] = by_rule.get(f["rule"], 0) + 1
        output = {
            "tool": "env-coverage",
            "summary": {
                "total_settings_classes": len({f.class_name for f in all_settings}),
                "total_settings_fields": len(all_settings),
                "total_bare_constants": len(all_constants),
                "total_inline_literals": len(all_literals),
                "total_findings": len(findings),
                "by_rule": by_rule,
            },
            "findings": findings,
        }
        print(json.dumps(output, indent=2))
    else:
        if not findings:
            print("No env-var coverage gaps detected")
            print(f"  BaseSettings classes: {len({f.class_name for f in all_settings})}")
            print(f"  Env-backed fields: {len(all_settings)}")
            print(f"  Module constants: {len(all_constants)}")
        else:
            for f in findings:
                print(f"{f['file']}:{f['line']}: {f['rule']} {f['message']} [env-coverage]")
            print(f"\nSummary: {len(findings)} coverage gap(s)")
            print(f"  BaseSettings fields: {len(all_settings)}")
            print(f"  Bare constants without env backing: {sum(1 for f in findings if f['rule'] == 'constant-no-env-backing')}")
            print(f"  Inline literals: {sum(1 for f in findings if f['rule'] == 'inline-literal')}")

    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description="Audit env-var coverage for Python constants")
    parser.add_argument("path", type=Path, help="Path to audit")
    parser.add_argument(
        "--output",
        choices=["json", "text"],
        default="text",
        help="Output format",
    )
    # uv run --script <file> -- <args> passes literal '--' to the script
    argv = [a for a in sys.argv[1:] if a != "--"]
    args = parser.parse_args(argv)

    if not args.path.exists():
        print(f"Error: Path does not exist: {args.path}", file=sys.stderr)
        return 1

    return audit_env_coverage(args.path, args.output)


if __name__ == "__main__":
    sys.exit(main())

# /// script
# requires-python = ">=3.12"
# dependencies = []
# ///
"""Orchestrator for code hardcode audit combining 9 tools.

Usage:
    uv run --script audit_hardcodes.py -- <path> [options]

Options:
    --output {json,text,both}  Output format (default: both)
    --tools {all,ruff,semgrep,jscpd,gitleaks,ast-grep,env-coverage}  Tools to run
    --severity {all,high,medium,low}  Filter by severity (default: all)
    --exclude PATTERN  Glob pattern to exclude (repeatable)
    --no-parallel  Disable parallel execution
    --skip-preflight  Skip tool availability check
"""

import argparse
import json
import os
import subprocess
import sys
import tempfile
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

# ADR: 2025-12-08-mise-env-centralized-config
# Configuration via environment variables with defaults for backward compatibility
AUDIT_PARALLEL_WORKERS = int(os.environ.get("AUDIT_PARALLEL_WORKERS", "4"))
AUDIT_JSCPD_TIMEOUT = int(os.environ.get("AUDIT_JSCPD_TIMEOUT", "300"))
AUDIT_GITLEAKS_TIMEOUT = int(os.environ.get("AUDIT_GITLEAKS_TIMEOUT", "120"))
AUDIT_ASTGREP_TIMEOUT = int(os.environ.get("AUDIT_ASTGREP_TIMEOUT", "60"))
AUDIT_BANDIT_TIMEOUT = int(os.environ.get("AUDIT_BANDIT_TIMEOUT", "120"))
AUDIT_TRUFFLEHOG_TIMEOUT = int(os.environ.get("AUDIT_TRUFFLEHOG_TIMEOUT", "300"))
AUDIT_WHISPERS_TIMEOUT = int(os.environ.get("AUDIT_WHISPERS_TIMEOUT", "120"))


@dataclass
class Finding:
    """A single finding from any tool."""

    id: str
    tool: str
    rule: str
    file: str
    line: int
    column: int = 0
    end_line: int | None = None
    message: str = ""
    severity: str = "medium"
    suggested_fix: str = ""

    def to_dict(self) -> dict[str, Any]:
        return {
            "id": self.id,
            "tool": self.tool,
            "rule": self.rule,
            "file": self.file,
            "line": self.line,
            "column": self.column,
            "end_line": self.end_line,
            "message": self.message,
            "severity": self.severity,
            "suggested_fix": self.suggested_fix,
        }

    def to_text(self) -> str:
        loc = f"{self.file}:{self.line}"
        if self.column:
            loc += f":{self.column}"
        return f"{loc}: {self.rule} {self.message} [{self.tool}]"


@dataclass
class AuditResult:
    """Aggregated results from all tools."""

    findings: list[Finding] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)

    def add_finding(self, finding: Finding) -> None:
        self.findings.append(finding)

    def add_error(self, error: str) -> None:
        self.errors.append(error)

    def summary(self) -> dict[str, Any]:
        by_tool: dict[str, int] = {}
        by_severity: dict[str, int] = {}

        for f in self.findings:
            by_tool[f.tool] = by_tool.get(f.tool, 0) + 1
            by_severity[f.severity] = by_severity.get(f.severity, 0) + 1

        return {
            "total_findings": len(self.findings),
            "by_tool": by_tool,
            "by_severity": by_severity,
        }

    def to_json(self) -> str:
        return json.dumps(
            {
                "summary": self.summary(),
                "findings": [f.to_dict() for f in self.findings],
                "errors": self.errors,
            },
            indent=2,
        )

    def to_text(self) -> str:
        lines = [f.to_text() for f in self.findings]
        summary = self.summary()
        tool_counts = ", ".join(f"{k}: {v}" for k, v in summary["by_tool"].items())
        lines.append("")
        lines.append(f"Summary: {summary['total_findings']} findings ({tool_counts})")
        if self.errors:
            lines.append(f"Errors: {len(self.errors)}")
            for e in self.errors:
                lines.append(f"  - {e}")
        return "\n".join(lines)


def run_ruff(target: Path, excludes: list[str]) -> list[Finding]:
    """Run Ruff PLR2004 check and return findings."""
    cmd = ["ruff", "check", "--select", "PLR2004", "--output-format", "json", str(target)]
    for pattern in excludes:
        cmd.extend(["--exclude", pattern])

    try:
        result = subprocess.run(cmd, capture_output=True, text=True)
        # Ruff returns exit code 1 when findings exist
        if result.stdout:
            data = json.loads(result.stdout)
            findings = []
            for i, item in enumerate(data):
                findings.append(
                    Finding(
                        id=f"RUFF-{i + 1:03d}",
                        tool="ruff",
                        rule=item.get("code", "PLR2004"),
                        file=item.get("filename", ""),
                        line=item.get("location", {}).get("row", 0),
                        column=item.get("location", {}).get("column", 0),
                        message=item.get("message", ""),
                        severity="medium",
                        suggested_fix="Extract to named constant",
                    )
                )
            return findings
    except (json.JSONDecodeError, FileNotFoundError) as e:
        print(f"ruff error: {e}", file=sys.stderr)
    return []


def _extract_rule_id(check_id: str) -> str:
    """Extract clean rule ID from Semgrep's path-based check_id.

    Semgrep generates check_ids like:
    'Users.terryli..claude.skills.code-hardcode-audit.assets.hardcoded-timeframe'

    This extracts just 'hardcoded-timeframe'.
    """
    # Split by dots and find the actual rule name (after 'assets' or last segment)
    parts = check_id.split(".")
    # Look for 'assets' marker, take everything after it
    if "assets" in parts:
        idx = parts.index("assets")
        return ".".join(parts[idx + 1 :]) if idx + 1 < len(parts) else check_id
    # Fallback: return last segment
    return parts[-1] if parts else check_id


def run_semgrep(target: Path, excludes: list[str]) -> list[Finding]:
    """Run Semgrep with custom rules and return findings."""
    rules_path = Path(__file__).parent.parent / "assets" / "semgrep-hardcode-rules.yaml"

    if not rules_path.exists():
        print(f"semgrep rules not found: {rules_path}", file=sys.stderr)
        return []

    cmd = ["semgrep", "--config", str(rules_path), "--json", str(target)]
    for pattern in excludes:
        cmd.extend(["--exclude", pattern])

    try:
        result = subprocess.run(cmd, capture_output=True, text=True)
        if result.stdout:
            data = json.loads(result.stdout)
            findings = []
            for i, item in enumerate(data.get("results", [])):
                severity_map = {"ERROR": "high", "WARNING": "medium", "INFO": "low"}
                raw_check_id = item.get("check_id", "")
                findings.append(
                    Finding(
                        id=f"SGRP-{i + 1:03d}",
                        tool="semgrep",
                        rule=_extract_rule_id(raw_check_id),
                        file=item.get("path", ""),
                        line=item.get("start", {}).get("line", 0),
                        column=item.get("start", {}).get("col", 0),
                        end_line=item.get("end", {}).get("line"),
                        message=item.get("extra", {}).get("message", ""),
                        severity=severity_map.get(
                            item.get("extra", {}).get("severity", "WARNING"), "medium"
                        ),
                        suggested_fix=item.get("extra", {})
                        .get("metadata", {})
                        .get("suggested_fix", ""),
                    )
                )
            return findings
    except (json.JSONDecodeError, FileNotFoundError) as e:
        print(f"semgrep error: {e}", file=sys.stderr)
    return []


def run_jscpd(target: Path, excludes: list[str]) -> list[Finding]:
    """Run jscpd via npx and return findings."""
    with tempfile.TemporaryDirectory() as tmpdir:
        output_dir = Path(tmpdir)
        cmd = ["npx", "jscpd", "--reporters", "json", "--output", str(output_dir), str(target)]
        for pattern in excludes:
            cmd.extend(["--ignore", pattern])

        try:
            subprocess.run(cmd, capture_output=True, text=True, timeout=AUDIT_JSCPD_TIMEOUT)
            report_path = output_dir / "jscpd-report.json"
            if report_path.exists():
                data = json.loads(report_path.read_text())
                findings = []
                for i, dup in enumerate(data.get("duplicates", [])):
                    first = dup.get("firstFile", {})
                    second = dup.get("secondFile", {})
                    findings.append(
                        Finding(
                            id=f"JSCPD-{i + 1:03d}",
                            tool="jscpd",
                            rule="duplicate-code",
                            file=first.get("name", ""),
                            line=first.get("startLoc", {}).get("line", 0),
                            end_line=first.get("endLoc", {}).get("line"),
                            message=f"Clone detected with {second.get('name', '')} "
                            f"({dup.get('lines', 0)} lines, "
                            f"{dup.get('fragment', '')})",
                            severity="low",
                            suggested_fix="Extract to shared function or module",
                        )
                    )
                return findings
        except (json.JSONDecodeError, FileNotFoundError, subprocess.TimeoutExpired) as e:
            print(f"jscpd error: {e}", file=sys.stderr)
    return []


def run_gitleaks(target: Path, excludes: list[str]) -> list[Finding]:
    """Run gitleaks for secret detection and return findings.

    Uses modern 'dir' command for directory scanning (v8.19.0+).
    Exit codes: 0 = clean, 1 = secrets found (expected), 2+ = error.
    """
    cmd = [
        "gitleaks",
        "dir",
        str(target),
        "--report-format",
        "json",
        "--report-path",
        "/dev/stdout",
    ]

    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=AUDIT_GITLEAKS_TIMEOUT)

        # Exit code 1 = secrets found (expected, not an error)
        if result.returncode in (0, 1) and result.stdout.strip():
            try:
                data = json.loads(result.stdout)
                findings = []
                for i, item in enumerate(data):
                    findings.append(
                        Finding(
                            id=f"GITLEAKS-{i + 1:03d}",
                            tool="gitleaks",
                            rule=item.get("RuleID", "secret"),
                            file=item.get("File", ""),
                            line=item.get("StartLine", 0),
                            end_line=item.get("EndLine"),
                            message=f"Secret detected: {item.get('Match', '')[:30]}...",
                            severity="high",  # Secrets are always high severity
                            suggested_fix="Remove secret and rotate credentials",
                        )
                    )
                return findings
            except json.JSONDecodeError:
                pass
        return []
    except FileNotFoundError:
        print("gitleaks not found. Install with: mise use --global gitleaks", file=sys.stderr)
        return []
    except subprocess.TimeoutExpired:
        print(f"gitleaks timed out after {AUDIT_GITLEAKS_TIMEOUT} seconds", file=sys.stderr)
        return []


def run_ast_grep(target: Path, excludes: list[str]) -> list[Finding]:
    """Run ast-grep with hardcode detection rules and return findings."""
    import shutil

    rules_dir = Path(__file__).parent.parent / "assets" / "ast-grep-hardcode"
    if not rules_dir.exists():
        print(f"ast-grep rules not found: {rules_dir}", file=sys.stderr)
        return []

    sg = None
    for name in ("sg", "ast-grep"):
        if shutil.which(name):
            sg = name
            break
    if not sg:
        print("ast-grep not found. Install with: cargo install ast-grep", file=sys.stderr)
        return []

    cmd = [sg, "scan", str(target.resolve()), "--json=stream"]

    try:
        result = subprocess.run(
            cmd, capture_output=True, text=True,
            timeout=AUDIT_ASTGREP_TIMEOUT, cwd=str(rules_dir),
        )
        findings = []
        severity_map = {"error": "high", "warning": "medium", "hint": "low", "info": "low"}
        if result.stdout.strip():
            for line in result.stdout.strip().splitlines():
                if not line.strip():
                    continue
                try:
                    item = json.loads(line)
                except json.JSONDecodeError:
                    continue
                findings.append(Finding(
                    id=f"ASTGREP-{len(findings) + 1:03d}",
                    tool="ast-grep",
                    rule=item.get("ruleId", "unknown"),
                    file=item.get("file", ""),
                    line=item.get("range", {}).get("start", {}).get("line", 0),
                    column=item.get("range", {}).get("start", {}).get("column", 0),
                    end_line=item.get("range", {}).get("end", {}).get("line"),
                    message=item.get("message", ""),
                    severity=severity_map.get(item.get("severity", "warning"), "medium"),
                    suggested_fix=item.get("note", ""),
                ))
        return findings
    except (FileNotFoundError, subprocess.TimeoutExpired) as e:
        print(f"ast-grep error: {e}", file=sys.stderr)
        return []


def run_env_coverage(target: Path, excludes: list[str]) -> list[Finding]:
    """Run env-coverage audit via subprocess and return findings."""
    script = Path(__file__).parent / "audit_env_coverage.py"
    if not script.exists():
        print(f"env-coverage script not found: {script}", file=sys.stderr)
        return []

    cmd = ["uv", "run", "--python", "3.14", "--script", str(script), "--", str(target), "--output", "json"]

    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
        if result.stdout.strip():
            data = json.loads(result.stdout)
            findings = []
            for item in data.get("findings", []):
                findings.append(Finding(
                    id=item.get("id", f"ENVCOV-{len(findings) + 1:03d}"),
                    tool="env-coverage",
                    rule=item.get("rule", ""),
                    file=item.get("file", ""),
                    line=item.get("line", 0),
                    column=item.get("column", 0),
                    end_line=item.get("end_line"),
                    message=item.get("message", ""),
                    severity=item.get("severity", "medium"),
                    suggested_fix=item.get("suggested_fix", ""),
                ))
            return findings
    except (json.JSONDecodeError, FileNotFoundError, subprocess.TimeoutExpired) as e:
        print(f"env-coverage error: {e}", file=sys.stderr)
    return []


def _gitignore_excludes(target: Path, regex_safe: bool = False) -> list[str]:
    """Read .gitignore from target and return directory names to exclude.

    Args:
        regex_safe: If True, skip glob patterns with *, ?, [, { (for trufflehog regex mode).
    """
    excludes = []
    gitignore = target / ".gitignore"
    if gitignore.exists():
        for line in gitignore.read_text().splitlines():
            line = line.strip()
            if not line or line.startswith("#") or line.startswith("!"):
                continue
            clean = line.rstrip("/")
            if regex_safe and any(c in clean for c in ("*", "?", "[", "{")):
                continue
            if clean:
                excludes.append(clean)
    # Always exclude .venv even if not in .gitignore
    if ".venv" not in excludes:
        excludes.append(".venv")
    return excludes


def run_bandit(target: Path, excludes: list[str]) -> list[Finding]:
    """Run Bandit B105/B106/B107 for hardcoded credential detection."""
    gi_excludes = _gitignore_excludes(target)
    bandit_excludes = [str(target / e) for e in gi_excludes]
    cmd = [
        "bandit", "-r", str(target),
        "-t", "B105,B106,B107",
        "-f", "json",
        "--exclude", ",".join(bandit_excludes),
    ]

    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=AUDIT_BANDIT_TIMEOUT)
        raw = result.stdout or ""
        brace_index = raw.find("{")
        if brace_index == -1:
            return []

        data = json.loads(raw[brace_index:])
        findings = []
        for i, item in enumerate(data.get("results", [])):
            findings.append(Finding(
                id=f"BANDIT-{i + 1:03d}",
                tool="bandit",
                rule=item.get("test_id", "B105"),
                file=item.get("filename", ""),
                line=item.get("line_number", 0),
                column=item.get("col_offset", 0),
                message=item.get("issue_text", ""),
                severity="high",  # Credential findings are always high
                suggested_fix="Move to environment variable or secret manager",
            ))
        return findings
    except (json.JSONDecodeError, FileNotFoundError, subprocess.TimeoutExpired) as e:
        print(f"bandit error: {e}", file=sys.stderr)
        return []


def run_trufflehog(target: Path, excludes: list[str]) -> list[Finding]:
    """Run TruffleHog for entropy-based secret detection with .gitignore awareness."""
    gi_excludes = _gitignore_excludes(target, regex_safe=True)
    gi_excludes.append(".git")

    # trufflehog --exclude-paths expects a file with patterns
    exclude_file = tempfile.NamedTemporaryFile(
        mode="w", suffix=".txt", delete=False, prefix="trufflehog-exclude-"
    )
    try:
        exclude_file.write("\n".join(gi_excludes) + "\n")
        exclude_file.close()

        cmd = [
            "trufflehog", "filesystem", str(target),
            "--json", "--no-update",
            "--exclude-paths", exclude_file.name,
        ]
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=AUDIT_TRUFFLEHOG_TIMEOUT)

        findings = []
        if result.returncode == 0 and result.stdout.strip():
            for line in result.stdout.strip().splitlines():
                if not line.strip():
                    continue
                try:
                    obj = json.loads(line)
                except json.JSONDecodeError:
                    continue
                fs = obj.get("SourceMetadata", {}).get("Data", {}).get("Filesystem", {})
                verified = obj.get("Verified", False)
                findings.append(Finding(
                    id=f"TRUFFLEHOG-{len(findings) + 1:03d}",
                    tool="trufflehog",
                    rule=obj.get("DetectorName", "unknown"),
                    file=fs.get("file", ""),
                    line=fs.get("line", 0),
                    message=f"{obj.get('DetectorName', '')} ({'verified' if verified else 'unverified'}): {obj.get('Raw', '')[:30]}...",
                    severity="high" if verified else "medium",
                    suggested_fix="Remove secret and rotate credentials",
                ))
        return findings
    except (FileNotFoundError, subprocess.TimeoutExpired) as e:
        print(f"trufflehog error: {e}", file=sys.stderr)
        return []
    finally:
        Path(exclude_file.name).unlink(missing_ok=True)


def run_whispers(target: Path, excludes: list[str]) -> list[Finding]:
    """Run Whispers for config-file secret detection with .gitignore awareness."""
    gi_excludes = _gitignore_excludes(target)
    gi_excludes.append(".git")

    # Build whispers config YAML
    config_lines = ["exclude:", "  files:"]
    for p in gi_excludes:
        config_lines.append(f'    - "{p}/**"')
    config_yaml = "\n".join(config_lines) + "\n"

    cfg_file = tempfile.NamedTemporaryFile(
        mode="w", suffix=".yml", delete=False, prefix="whispers-config-"
    )
    try:
        cfg_file.write(config_yaml)
        cfg_file.close()

        cmd = ["whispers", str(target), "-j", "-c", cfg_file.name]
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=AUDIT_WHISPERS_TIMEOUT)

        findings = []
        if result.stdout and result.stdout.strip():
            data = json.loads(result.stdout)
            for i, item in enumerate(data):
                findings.append(Finding(
                    id=f"WHISPERS-{i + 1:03d}",
                    tool="whispers",
                    rule=item.get("rule_id", "unknown"),
                    file=item.get("file", ""),
                    line=item.get("line", 0),
                    message=f"{item.get('key', '')}: {item.get('message', '')}",
                    severity={"Critical": "high", "High": "high", "Medium": "medium", "Low": "low"}.get(
                        item.get("severity", "Low"), "low"
                    ),
                    suggested_fix="Move to environment variable or secret manager",
                ))
        return findings
    except (json.JSONDecodeError, FileNotFoundError, subprocess.TimeoutExpired) as e:
        print(f"whispers error: {e}", file=sys.stderr)
        return []
    finally:
        Path(cfg_file.name).unlink(missing_ok=True)


def run_preflight(target: Path) -> bool:
    """Run preflight check and return True if all tools available."""
    script = Path(__file__).parent / "preflight.py"
    if not script.exists():
        return True  # Skip if preflight script not available

    cmd = ["uv", "run", "--python", "3.14", "--script", str(script), "--", str(target), "--output", "text"]
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
        print(result.stdout, end="")
        if result.stderr:
            print(result.stderr, end="", file=sys.stderr)
        return result.returncode == 0
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return True  # Don't block on preflight failures


def run_audit(
    target: Path,
    tools: list[str],
    excludes: list[str],
    parallel: bool = True,
) -> AuditResult:
    """Run selected tools and aggregate results."""
    result = AuditResult()

    tool_funcs = {
        "ruff": run_ruff,
        "semgrep": run_semgrep,
        "jscpd": run_jscpd,
        "gitleaks": run_gitleaks,
        "ast-grep": run_ast_grep,
        "env-coverage": run_env_coverage,
        "bandit": run_bandit,
        "trufflehog": run_trufflehog,
        "whispers": run_whispers,
    }

    selected = tools if tools != ["all"] else list(tool_funcs.keys())

    if parallel:
        with ThreadPoolExecutor(max_workers=AUDIT_PARALLEL_WORKERS) as executor:
            futures = {
                executor.submit(tool_funcs[name], target, excludes): name
                for name in selected
                if name in tool_funcs
            }
            for future in as_completed(futures):
                name = futures[future]
                try:
                    for finding in future.result():
                        result.add_finding(finding)
                except Exception as e:
                    result.add_error(f"{name}: {e}")
    else:
        for name in selected:
            if name in tool_funcs:
                try:
                    for finding in tool_funcs[name](target, excludes):
                        result.add_finding(finding)
                except Exception as e:
                    result.add_error(f"{name}: {e}")

    # Sort by file, then line
    result.findings.sort(key=lambda f: (f.file, f.line))
    return result


def filter_by_severity(result: AuditResult, severity: str) -> AuditResult:
    """Filter findings by severity level."""
    if severity == "all":
        return result

    severity_order = {"high": 0, "medium": 1, "low": 2}
    threshold = severity_order.get(severity, 2)

    filtered = AuditResult(errors=result.errors)
    for f in result.findings:
        if severity_order.get(f.severity, 2) <= threshold:
            filtered.add_finding(f)
    return filtered


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Audit code for hardcoded values using 9 detection tools"
    )
    parser.add_argument("path", type=Path, help="Path to audit")
    parser.add_argument(
        "--output",
        choices=["json", "text", "both"],
        default="both",
        help="Output format",
    )
    parser.add_argument(
        "--tools",
        choices=["all", "ruff", "semgrep", "jscpd", "gitleaks", "ast-grep", "env-coverage",
                 "bandit", "trufflehog", "whispers"],
        nargs="+",
        default=["all"],
        help="Tools to run",
    )
    parser.add_argument(
        "--severity",
        choices=["all", "high", "medium", "low"],
        default="all",
        help="Filter by minimum severity",
    )
    parser.add_argument(
        "--exclude",
        action="append",
        default=[],
        help="Glob pattern to exclude",
    )
    parser.add_argument(
        "--no-parallel",
        action="store_true",
        help="Disable parallel execution",
    )
    parser.add_argument(
        "--skip-preflight",
        action="store_true",
        help="Skip tool availability preflight check",
    )

    # uv run --script <file> -- <args> passes literal '--' to the script
    argv = [a for a in sys.argv[1:] if a != "--"]
    args = parser.parse_args(argv)

    if not args.path.exists():
        print(f"Error: Path does not exist: {args.path}", file=sys.stderr)
        return 1

    if not args.skip_preflight:
        run_preflight(args.path)
        print()  # Separator between preflight and audit output

    result = run_audit(
        target=args.path,
        tools=args.tools,
        excludes=args.exclude,
        parallel=not args.no_parallel,
    )

    result = filter_by_severity(result, args.severity)

    if args.output == "json":
        print(result.to_json())
    elif args.output == "text":
        print(result.to_text())
    else:  # both
        print(result.to_text())
        print("\n--- JSON Output ---")
        print(result.to_json())

    return 0 if not result.errors else 1


if __name__ == "__main__":
    sys.exit(main())

# /// script
# requires-python = ">=3.14"
# dependencies = []
# ///
"""Preflight check for hardcode audit tools.

Usage:
    uv run --python 3.14 --script preflight.py -- <path> [--output {json,text}]

Verifies all audit tools are installed and properly configured.
Detects silent misconfigurations like globally suppressed Ruff PLR2004.
"""

import argparse
import json
import shutil
import subprocess
import sys
import tomllib
from dataclasses import dataclass, field
from pathlib import Path


@dataclass
class ToolCheck:
    """Result of checking a single tool."""

    name: str
    installed: bool
    version: str | None = None
    issues: list[str] = field(default_factory=list)
    install_cmd: str = ""

    def to_dict(self) -> dict:
        d = {"name": self.name, "installed": self.installed, "version": self.version, "issues": self.issues}
        if not self.installed:
            d["install_cmd"] = self.install_cmd
        return d


def _get_version(cmd: list[str]) -> str | None:
    """Run a version command and extract the version string."""
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=5)
        if result.returncode == 0:
            # Take first line, strip common prefixes
            line = result.stdout.strip().splitlines()[0]
            # Extract version-like substring
            for part in line.split():
                if part[0].isdigit():
                    return part
            return line
    except (FileNotFoundError, subprocess.TimeoutExpired, IndexError):
        pass
    return None


def _find_ruff_config(target: Path) -> Path | None:
    """Walk up from target to find ruff config file."""
    search = target.resolve()
    if search.is_file():
        search = search.parent

    while True:
        for name in ("ruff.toml", ".ruff.toml", "pyproject.toml"):
            candidate = search / name
            if candidate.is_file():
                return candidate
        parent = search.parent
        if parent == search:
            break
        search = parent
    return None


def _check_plr2004_suppressed(config_path: Path) -> list[str]:
    """Check if PLR2004 is suppressed in a ruff config file."""
    issues = []
    try:
        data = tomllib.loads(config_path.read_text())
    except Exception:
        return issues

    # pyproject.toml nests under [tool.ruff]
    ruff_cfg = data
    if config_path.name == "pyproject.toml":
        ruff_cfg = data.get("tool", {}).get("ruff", {})

    lint = ruff_cfg.get("lint", {})

    # Check global ignore list
    ignore = lint.get("ignore", [])
    for rule in ignore:
        if rule in ("PLR2004", "PLR"):
            issues.append(f"PLR2004 globally suppressed in {config_path.name} [tool.ruff.lint.ignore]")
            break

    # Check per-file-ignores
    per_file = lint.get("per-file-ignores", {})
    for pattern, rules in per_file.items():
        if pattern in ("*", "*.py", "**/*.py"):
            for rule in rules:
                if rule in ("PLR2004", "PLR"):
                    issues.append(f"PLR2004 suppressed for '{pattern}' in {config_path.name} [tool.ruff.lint.per-file-ignores]")
                    break

    # Check select — if select is specified but doesn't include PLR/PLR2004
    select = lint.get("select", None)
    if select is not None and "ALL" not in select:
        has_plr = any(r in select for r in ("PLR", "PLR2004", "PL"))
        if not has_plr:
            issues.append(f"PLR2004 not in [tool.ruff.lint.select] in {config_path.name} — rule won't run")

    return issues


def check_ruff(target: Path) -> ToolCheck:
    """Check ruff installation and PLR2004 configuration."""
    check = ToolCheck(name="ruff", installed=False, install_cmd="uv tool install ruff")
    path = shutil.which("ruff")
    if not path:
        return check

    check.installed = True
    check.version = _get_version(["ruff", "--version"])

    # Check PLR2004 suppression in project config
    config = _find_ruff_config(target)
    if config:
        check.issues.extend(_check_plr2004_suppressed(config))

    return check


def check_semgrep() -> ToolCheck:
    """Check semgrep installation."""
    check = ToolCheck(name="semgrep", installed=False, install_cmd="brew install semgrep")
    if shutil.which("semgrep"):
        check.installed = True
        check.version = _get_version(["semgrep", "--version"])
    return check


def check_jscpd() -> ToolCheck:
    """Check jscpd availability via npx."""
    check = ToolCheck(name="jscpd", installed=False, install_cmd="npm install -g jscpd")
    if shutil.which("npx"):
        check.installed = True
        check.version = "via npx"
    return check


def check_gitleaks() -> ToolCheck:
    """Check gitleaks installation."""
    check = ToolCheck(name="gitleaks", installed=False, install_cmd="mise use --global gitleaks")
    if shutil.which("gitleaks"):
        check.installed = True
        check.version = _get_version(["gitleaks", "version"])
    return check


def check_ast_grep() -> ToolCheck:
    """Check ast-grep installation (binary name: sg or ast-grep)."""
    check = ToolCheck(name="ast-grep", installed=False, install_cmd="cargo install ast-grep")
    for binary in ("sg", "ast-grep"):
        if shutil.which(binary):
            check.installed = True
            check.version = _get_version([binary, "--version"])
            break
    return check


def check_bandit() -> ToolCheck:
    """Check bandit installation."""
    check = ToolCheck(name="bandit", installed=False, install_cmd="uv tool install bandit")
    if shutil.which("bandit"):
        check.installed = True
        check.version = _get_version(["bandit", "--version"])
    return check


def check_trufflehog() -> ToolCheck:
    """Check trufflehog installation."""
    check = ToolCheck(name="trufflehog", installed=False, install_cmd="brew install trufflehog")
    if shutil.which("trufflehog"):
        check.installed = True
        check.version = _get_version(["trufflehog", "--version"])
    return check


def check_whispers() -> ToolCheck:
    """Check whispers installation."""
    check = ToolCheck(name="whispers", installed=False, install_cmd="uv tool install whispers")
    if shutil.which("whispers"):
        check.installed = True
        check.version = _get_version(["whispers", "--version"])
    return check


def run_preflight(target: Path, output_format: str = "text") -> int:
    """Run all preflight checks and report results."""
    checks = [
        check_ruff(target),
        check_semgrep(),
        check_jscpd(),
        check_gitleaks(),
        check_ast_grep(),
        check_bandit(),
        check_trufflehog(),
        check_whispers(),
    ]

    has_missing = any(not c.installed for c in checks)
    has_issues = any(c.issues for c in checks)

    if has_missing:
        status = "fail"
    elif has_issues:
        status = "warn"
    else:
        status = "pass"

    if output_format == "json":
        output = {
            "status": status,
            "tools": [c.to_dict() for c in checks],
        }
        print(json.dumps(output, indent=2))
    else:
        print("=== Hardcode Audit Preflight ===\n")
        for c in checks:
            if c.installed:
                icon = "✓" if not c.issues else "⚠"
                ver = f" ({c.version})" if c.version else ""
                print(f"  {icon} {c.name}{ver}")
                for issue in c.issues:
                    print(f"    ⚠ {issue}")
            else:
                print(f"  ✗ {c.name} — not found")
                print(f"    Install: {c.install_cmd}")

        print()
        if status == "pass":
            print("Status: PASS — all tools available and configured")
        elif status == "warn":
            print("Status: WARN — tools available but configuration issues detected")
        else:
            print("Status: FAIL — missing required tools")

    return 0 if status in ("pass", "warn") else 1


def main() -> int:
    parser = argparse.ArgumentParser(description="Preflight check for hardcode audit tools")
    parser.add_argument("path", type=Path, help="Target path (used to find ruff config)")
    parser.add_argument(
        "--output",
        choices=["json", "text"],
        default="text",
        help="Output format",
    )
    # uv run --script <file> -- <args> passes literal '--' to the script
    argv = [a for a in sys.argv[1:] if a != "--"]
    args = parser.parse_args(argv)

    if not args.path.exists():
        print(f"Error: Path does not exist: {args.path}", file=sys.stderr)
        return 1

    return run_preflight(args.path, args.output)


if __name__ == "__main__":
    sys.exit(main())

# /// script
# requires-python = ">=3.14"
# dependencies = []
# ///
"""ast-grep wrapper for AST-based hardcode detection.

Usage:
    uv run --python 3.14 --script run_ast_grep.py -- <path> [--output {json,text}]

Detects hardcoded literals in assignments, function arguments, return values,
URLs, and file paths using AST pattern matching (not regex).
"""

import argparse
import json
import os
import shutil
import subprocess
import sys
from pathlib import Path

RULES_DIR = Path(__file__).parent.parent / "assets" / "ast-grep-hardcode"
TIMEOUT = int(os.environ.get("AUDIT_ASTGREP_TIMEOUT", "60"))


def _find_sg_binary() -> str | None:
    """Find ast-grep binary (sg or ast-grep)."""
    for name in ("sg", "ast-grep"):
        if shutil.which(name):
            return name
    return None


def run_ast_grep(target: Path, output_format: str = "text") -> int:
    """Run ast-grep with hardcode detection rules."""
    sg = _find_sg_binary()
    if not sg:
        print("Error: ast-grep not found. Install with: cargo install ast-grep", file=sys.stderr)
        return 1

    if not RULES_DIR.exists():
        print(f"Error: ast-grep rules not found: {RULES_DIR}", file=sys.stderr)
        return 1

    cmd = [sg, "scan", str(target.resolve()), "--json=stream"]

    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=TIMEOUT,
            cwd=str(RULES_DIR),
        )

        # Parse NDJSON (one JSON object per line)
        findings = []
        if result.stdout.strip():
            for line in result.stdout.strip().splitlines():
                line = line.strip()
                if not line:
                    continue
                try:
                    item = json.loads(line)
                    findings.append(item)
                except json.JSONDecodeError:
                    continue

        if output_format == "json":
            severity_map = {"error": "high", "warning": "medium", "hint": "low", "info": "low"}
            output_findings = []
            for i, item in enumerate(findings):
                output_findings.append({
                    "id": f"ASTGREP-{i + 1:03d}",
                    "tool": "ast-grep",
                    "rule": item.get("ruleId", "unknown"),
                    "file": item.get("file", ""),
                    "line": item.get("range", {}).get("start", {}).get("line", 0),
                    "column": item.get("range", {}).get("start", {}).get("column", 0),
                    "end_line": item.get("range", {}).get("end", {}).get("line"),
                    "message": item.get("message", ""),
                    "severity": severity_map.get(item.get("severity", "warning"), "medium"),
                    "suggested_fix": item.get("note", ""),
                })
            output = {
                "tool": "ast-grep",
                "rules_dir": str(RULES_DIR),
                "total_findings": len(output_findings),
                "findings": output_findings,
            }
            print(json.dumps(output, indent=2))
        else:
            if not findings:
                print("No hardcode patterns detected (ast-grep)")
            else:
                for item in findings:
                    file = item.get("file", "?")
                    line = item.get("range", {}).get("start", {}).get("line", 0)
                    rule = item.get("ruleId", "unknown")
                    msg = item.get("message", "")
                    print(f"{file}:{line}: {rule} {msg} [ast-grep]")
                print(f"\nTotal: {len(findings)} finding(s)")

        return 0

    except FileNotFoundError:
        print("Error: ast-grep not found. Install with: cargo install ast-grep", file=sys.stderr)
        return 1
    except subprocess.TimeoutExpired:
        print(f"Error: ast-grep timed out after {TIMEOUT} seconds", file=sys.stderr)
        return 1
    except json.JSONDecodeError as e:
        print(f"Error parsing ast-grep output: {e}", file=sys.stderr)
        return 1


def main() -> int:
    parser = argparse.ArgumentParser(description="Run ast-grep AST-based hardcode detection")
    parser.add_argument("path", type=Path, help="Path to check")
    parser.add_argument(
        "--output",
        choices=["json", "text"],
        default="text",
        help="Output format",
    )
    # uv run --script <file> -- <args> passes literal '--' to the script
    argv = [a for a in sys.argv[1:] if a != "--"]
    args = parser.parse_args(argv)

    if not args.path.exists():
        print(f"Error: Path does not exist: {args.path}", file=sys.stderr)
        return 1

    return run_ast_grep(args.path, args.output)


if __name__ == "__main__":
    sys.exit(main())