parse_deployment.py

"""
parse_deployment.py — Salesforce Deployment Data Extractor
------------------------------------------------------------
Converts raw Salesforce CLI and PMD Scanner JSON output into the
input schema expected by the Deployment Failure Analyser.

Usage:
  # Step 1: Run deployment and capture output
  sf project deploy start --json > deploy_result.json

  # Step 2: Run PMD scanner and capture output (optional)
  sf scanner run --json > pmd_result.json

  # Step 3: Parse and combine
  python parse_deployment.py --deploy deploy_result.json
  python parse_deployment.py --deploy deploy_result.json --pmd pmd_result.json

  # Step 4: Analyse with Claude
  python parse_deployment.py --deploy deploy_result.json --pmd pmd_result.json --out input.json
  python main.py --input input.json --live
"""

import argparse
import json
import os
import sys


# ── Parsers ───────────────────────────────────────────────────────────────────

def parse_deploy_result(data: dict) -> dict:
    """
    Extract code coverage and deployment failures from sf project deploy --json output.

    Handles both:
    - CLI v2 (sf): result.deployedSource, result.runTestResult
    - CLI v1 (sfdx): result.details.runTestResult, result.details.componentFailures
    """
    result = data.get("result", data)  # some CLI versions wrap in result{}

    # ── Code coverage ──────────────────────────────────────────────────────
    coverage_pct = 0
    run_test = (
        result.get("runTestResult")
        or result.get("details", {}).get("runTestResult", {})
        or {}
    )
    coverage_records = run_test.get("codeCoverage", [])

    if coverage_records:
        total_covered = sum(r.get("numLocations", 0) - r.get("numLocationsNotCovered", 0)
                            for r in coverage_records)
        total_lines = sum(r.get("numLocations", 0) for r in coverage_records)
        coverage_pct = round((total_covered / total_lines) * 100, 1) if total_lines > 0 else 0
    else:
        # Fall back to org-level coverage if available
        coverage_pct = result.get("numberTestsTotal", 0) and round(
            result.get("numberTestsCompleted", 0) / result.get("numberTestsTotal", 1) * 100, 1
        )

    # ── Deployment failures ────────────────────────────────────────────────
    failures = []

    component_failures = (
        result.get("details", {}).get("componentFailures", [])
        or result.get("componentFailures", [])
    )
    test_failures = run_test.get("failures", [])

    # Group component failures by component name
    failure_map = {}
    for f in component_failures:
        name = f.get("fullName") or f.get("fileName", "Unknown")
        error = f.get("problem") or f.get("message", "Deployment error")
        if name not in failure_map:
            failure_map[name] = {"component": name, "error": error, "failed_tests": 0}

    # Count test failures per component
    for t in test_failures:
        name = t.get("name") or t.get("className", "Unknown")
        if name in failure_map:
            failure_map[name]["failed_tests"] += 1
        else:
            failure_map[name] = {
                "component": name,
                "error": t.get("message") or t.get("type", "Test failure"),
                "failed_tests": 1,
            }

    failures = list(failure_map.values())

    return {
        "code_coverage": coverage_pct,
        "failed_deployments": failures,
    }


def parse_pmd_result(data: dict) -> dict:
    """
    Extract PMD violation counts from sf scanner run --json output.

    Critical severity = PMD severity 1 or 2 (SOQL injection, hardcoded IDs, etc.)
    """
    violations_total = 0
    violations_critical = 0

    results = data.get("result", [])
    if isinstance(results, list):
        for file_result in results:
            for v in file_result.get("violations", []):
                violations_total += 1
                severity = v.get("severity", 5)
                if isinstance(severity, int) and severity <= 2:
                    violations_critical += 1
    elif isinstance(results, dict):
        # Some versions return a flat violations list
        for v in results.get("violations", []):
            violations_total += 1
            if v.get("severity", 5) <= 2:
                violations_critical += 1

    return {
        "code_quality_issues": {
            "pmd_violations": violations_total,
            "critical": violations_critical,
        }
    }


# ── Main ──────────────────────────────────────────────────────────────────────

def parse_args():
    parser = argparse.ArgumentParser(
        prog="parse_deployment.py",
        description="Convert Salesforce CLI + PMD output to Deployment Analyser input schema",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
examples:
  python parse_deployment.py --deploy deploy_result.json
  python parse_deployment.py --deploy deploy_result.json --pmd pmd_result.json
  python parse_deployment.py --deploy deploy_result.json --pmd pmd_result.json --out input.json
  python main.py --input input.json --live
        """
    )
    parser.add_argument("--deploy", metavar="FILE", required=True,
                        help="Path to sf project deploy --json output file")
    parser.add_argument("--pmd", metavar="FILE",
                        help="Path to sf scanner run --json output file (optional)")
    parser.add_argument("--out", metavar="FILE",
                        help="Write result to file instead of stdout")
    return parser.parse_args()


def load_json_file(path: str) -> dict:
    abs_path = os.path.abspath(path)
    if not os.path.exists(abs_path):
        print(f"❌  File not found: {abs_path}", file=sys.stderr)
        sys.exit(1)
    with open(abs_path, "r", encoding="utf-8") as f:
        try:
            return json.load(f)
        except json.JSONDecodeError as e:
            print(f"❌  Invalid JSON in {abs_path}:\n    {e}", file=sys.stderr)
            sys.exit(1)


def main():
    args = parse_args()

    deploy_data = load_json_file(args.deploy)
    result = parse_deploy_result(deploy_data)

    if args.pmd:
        pmd_data = load_json_file(args.pmd)
        result.update(parse_pmd_result(pmd_data))
    else:
        # Default: no PMD data — set zero violations
        result["code_quality_issues"] = {"pmd_violations": 0, "critical": 0}

    output = json.dumps(result, indent=2)

    if args.out:
        out_path = os.path.abspath(args.out)
        with open(out_path, "w", encoding="utf-8") as f:
            f.write(output)
        print(f"✅  Written to {out_path}", file=sys.stderr)
        print(f"    Run: python main.py --input {args.out} --live", file=sys.stderr)
    else:
        print(output)


if __name__ == "__main__":
    main()