Ci Cd Pipeline Builder

#!/usr/bin/env python3
"""Analyze CI/CD pipeline configs and suggest caching improvements.

Reads GitHub Actions or GitLab CI YAML files and identifies missing
caches, suboptimal cache keys, redundant installs across jobs, and
opportunities to use built-in caching features.

Usage:
    python cache_optimizer.py .github/workflows/ci.yml
    python cache_optimizer.py .gitlab-ci.yml --json
    python cache_optimizer.py --dir .github/workflows/
"""

import argparse
import json
import os
import re
import sys
from pathlib import Path


# ---------------------------------------------------------------------------
# Known cache strategies
# ---------------------------------------------------------------------------

GITHUB_CACHE_STRATEGIES = {
    "npm": {
        "preferred": "cache: 'npm' in actions/setup-node",
        "cache_path": "~/.npm",
        "key_pattern": "hashFiles('**/package-lock.json')",
        "lockfile": "package-lock.json",
    },
    "pnpm": {
        "preferred": "cache: 'pnpm' in actions/setup-node (with pnpm/action-setup)",
        "cache_path": "detected by setup-node",
        "key_pattern": "hashFiles('**/pnpm-lock.yaml')",
        "lockfile": "pnpm-lock.yaml",
    },
    "yarn": {
        "preferred": "cache: 'yarn' in actions/setup-node",
        "cache_path": "~/.cache/yarn",
        "key_pattern": "hashFiles('**/yarn.lock')",
        "lockfile": "yarn.lock",
    },
    "pip": {
        "preferred": "cache: 'pip' in actions/setup-python",
        "cache_path": "~/.cache/pip",
        "key_pattern": "hashFiles('**/requirements*.txt')",
        "lockfile": "requirements.txt",
    },
    "uv": {
        "preferred": "Built-in caching via astral-sh/setup-uv",
        "cache_path": "~/.cache/uv",
        "key_pattern": "hashFiles('**/uv.lock')",
        "lockfile": "uv.lock",
    },
    "go": {
        "preferred": "Built-in caching via actions/setup-go (cache: true)",
        "cache_path": "~/go/pkg/mod",
        "key_pattern": "hashFiles('**/go.sum')",
        "lockfile": "go.sum",
    },
    "cargo": {
        "preferred": "actions/cache with Cargo.lock hash key",
        "cache_path": "~/.cargo/registry, ~/.cargo/git, target/",
        "key_pattern": "hashFiles('**/Cargo.lock')",
        "lockfile": "Cargo.lock",
    },
}

# ---------------------------------------------------------------------------
# Analysis functions
# ---------------------------------------------------------------------------

SEVERITY_HIGH = "high"
SEVERITY_MEDIUM = "medium"
SEVERITY_LOW = "low"


def detect_platform(filepath, text):
    """Determine CI platform from path and content."""
    if ".github" in filepath or "runs-on:" in text:
        return "github-actions"
    if ".gitlab-ci" in os.path.basename(filepath) or ("stages:" in text and "script:" in text):
        return "gitlab-ci"
    return "unknown"


def detect_package_managers(text):
    """Detect which package managers are referenced in the pipeline."""
    found = set()
    indicators = {
        "npm": [r'\bnpm\s+(ci|install)\b', r'package-lock\.json'],
        "pnpm": [r'\bpnpm\s+install\b', r'pnpm-lock\.yaml'],
        "yarn": [r'\byarn\s+install\b', r'yarn\.lock'],
        "pip": [r'\bpip\s+install\b', r'requirements.*\.txt'],
        "uv": [r'\buv\s+(sync|install)\b', r'uv\.lock'],
        "poetry": [r'\bpoetry\s+install\b', r'poetry\.lock'],
        "go": [r'\bgo\s+(build|test|vet)\b', r'go\.sum'],
        "cargo": [r'\bcargo\s+(build|test)\b', r'Cargo\.lock'],
    }
    for pm, patterns in indicators.items():
        for pat in patterns:
            if re.search(pat, text):
                found.add(pm)
                break
    return found


def check_missing_cache(text, platform, package_managers):
    """Identify package managers used without any caching."""
    suggestions = []
    for pm in package_managers:
        strategy = GITHUB_CACHE_STRATEGIES.get(pm)
        if not strategy:
            continue

        has_cache = False
        # Check for setup-node/setup-python cache param
        if pm in ("npm", "pnpm", "yarn") and re.search(r"cache:\s*['\"]?" + pm, text):
            has_cache = True
        if pm == "pip" and re.search(r"cache:\s*['\"]?pip", text):
            has_cache = True
        if pm == "uv" and "setup-uv" in text:
            has_cache = True
        if pm == "go" and "setup-go" in text:
            has_cache = True
        # Check for explicit actions/cache
        if re.search(r"actions/cache", text) and strategy["lockfile"] in text:
            has_cache = True
        # GitLab cache block
        if platform == "gitlab-ci" and "cache:" in text:
            has_cache = True

        if not has_cache:
            suggestions.append({
                "type": "missing-cache",
                "severity": SEVERITY_HIGH,
                "package_manager": pm,
                "message": f"No caching detected for {pm}. Dependencies are re-downloaded on every run.",
                "recommendation": strategy["preferred"],
                "estimated_savings": "30-120 seconds per job",
            })

    return suggestions


def check_cache_key_quality(text, platform):
    """Check for suboptimal cache key patterns."""
    suggestions = []

    if platform == "github-actions":
        # Check for cache keys without hashFiles
        cache_key_lines = re.findall(r'key:\s*(.+)', text)
        for key_expr in cache_key_lines:
            if "hashFiles" not in key_expr and "github.sha" in key_expr:
                suggestions.append({
                    "type": "volatile-cache-key",
                    "severity": SEVERITY_HIGH,
                    "message": "Cache key uses github.sha which changes every commit, causing 0% hit rate.",
                    "recommendation": "Use hashFiles('**/lockfile') for dependency caches.",
                    "estimated_savings": "30-120 seconds per job",
                })
            if "hashFiles" not in key_expr and "${{" not in key_expr and key_expr.strip().strip("'\""):
                # Static string key - always same, never invalidated
                suggestions.append({
                    "type": "static-cache-key",
                    "severity": SEVERITY_MEDIUM,
                    "message": f"Cache key appears static: {key_expr.strip()[:60]}. Cache never invalidates.",
                    "recommendation": "Include hashFiles() of your lockfile so cache refreshes on dependency changes.",
                    "estimated_savings": "Prevents stale dependency bugs",
                })

    if platform == "gitlab-ci":
        cache_key_lines = re.findall(r'key:\s*(.+)', text)
        for key_expr in cache_key_lines:
            if "$CI_COMMIT_SHA" in key_expr:
                suggestions.append({
                    "type": "volatile-cache-key",
                    "severity": SEVERITY_HIGH,
                    "message": "Cache key uses $CI_COMMIT_SHA which changes every commit.",
                    "recommendation": "Use $CI_COMMIT_REF_SLUG or Files(['lockfile']) for dependency caches.",
                    "estimated_savings": "30-120 seconds per job",
                })

    return suggestions


def check_redundant_installs(text, platform):
    """Detect when multiple jobs repeat the same install step without shared cache."""
    suggestions = []

    if platform == "github-actions":
        install_patterns = [
            (r'npm ci', "npm ci"),
            (r'pnpm install', "pnpm install"),
            (r'yarn install', "yarn install"),
            (r'pip install -r', "pip install"),
            (r'uv sync', "uv sync"),
        ]
        for pat, label in install_patterns:
            matches = re.findall(pat, text)
            if len(matches) > 2:
                suggestions.append({
                    "type": "redundant-install",
                    "severity": SEVERITY_MEDIUM,
                    "message": f"'{label}' appears {len(matches)} times across jobs.",
                    "recommendation": (
                        "Ensure caching is enabled so repeated installs are fast. "
                        "Consider a dedicated setup job with artifact passing for large dependency trees."
                    ),
                    "estimated_savings": f"{(len(matches) - 1) * 20}-{(len(matches) - 1) * 60} seconds total",
                })

    return suggestions


def check_docker_cache(text, platform):
    """Check if Docker builds use layer caching."""
    suggestions = []
    if "docker" not in text.lower() and "buildx" not in text.lower():
        return suggestions

    has_docker_build = bool(re.search(r'docker\s+build\b', text))
    has_buildx = "docker/build-push-action" in text or "buildx" in text
    has_cache_from = "cache-from" in text

    if has_docker_build and not has_buildx:
        suggestions.append({
            "type": "no-buildx",
            "severity": SEVERITY_MEDIUM,
            "message": "Using 'docker build' without BuildKit/Buildx. No layer caching between runs.",
            "recommendation": "Switch to docker/build-push-action with cache-from: type=gha, cache-to: type=gha,mode=max",
            "estimated_savings": "1-10 minutes per build depending on image size",
        })
    elif has_buildx and not has_cache_from:
        suggestions.append({
            "type": "buildx-no-cache",
            "severity": SEVERITY_HIGH,
            "message": "Using Buildx but no cache-from/cache-to configured. Docker layers rebuild from scratch.",
            "recommendation": "Add cache-from: type=gha and cache-to: type=gha,mode=max to build-push-action",
            "estimated_savings": "1-10 minutes per build depending on image size",
        })

    return suggestions


def check_missing_restore_keys(text, platform):
    """Suggest restore-keys for graceful cache fallback."""
    suggestions = []
    if platform != "github-actions":
        return suggestions

    if "actions/cache" in text and "restore-keys" not in text:
        suggestions.append({
            "type": "missing-restore-keys",
            "severity": SEVERITY_LOW,
            "message": "actions/cache used without restore-keys. Cache miss when lockfile changes means full re-download.",
            "recommendation": (
                "Add restore-keys with a prefix fallback, e.g.:\n"
                "  restore-keys: |\n"
                "    ${{ runner.os }}-npm-\n"
                "This allows partial cache hits when only some deps change."
            ),
            "estimated_savings": "10-60 seconds on lockfile changes",
        })

    return suggestions


# ---------------------------------------------------------------------------
# Main analysis
# ---------------------------------------------------------------------------

ALL_CHECKS = [
    check_missing_cache,
    check_cache_key_quality,
    check_redundant_installs,
    check_docker_cache,
    check_missing_restore_keys,
]


def analyze_file(filepath):
    """Run all cache optimization checks on a single pipeline file."""
    filepath = str(filepath)
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            text = f.read()
    except (OSError, UnicodeDecodeError) as exc:
        return {
            "file": filepath,
            "platform": "unknown",
            "package_managers": [],
            "suggestions": [{
                "type": "file-read-error",
                "severity": SEVERITY_HIGH,
                "message": str(exc),
            }],
        }

    platform = detect_platform(filepath, text)
    package_managers = sorted(detect_package_managers(text))
    suggestions = []

    for check_fn in ALL_CHECKS:
        if check_fn == check_missing_cache:
            suggestions.extend(check_fn(text, platform, package_managers))
        elif check_fn in (check_cache_key_quality, check_redundant_installs,
                          check_docker_cache, check_missing_restore_keys):
            suggestions.extend(check_fn(text, platform))

    # Sort: high first
    severity_order = {SEVERITY_HIGH: 0, SEVERITY_MEDIUM: 1, SEVERITY_LOW: 2}
    suggestions.sort(key=lambda s: severity_order.get(s["severity"], 9))

    return {
        "file": filepath,
        "platform": platform,
        "package_managers": package_managers,
        "suggestions": suggestions,
    }


def collect_files(path):
    """Collect YAML files from a file path or directory."""
    p = Path(path)
    if p.is_file():
        return [p]
    if p.is_dir():
        return sorted(p.glob("**/*.yml")) + sorted(p.glob("**/*.yaml"))
    return []


# ---------------------------------------------------------------------------
# Output formatting
# ---------------------------------------------------------------------------

_SEVERITY_BADGE = {
    SEVERITY_HIGH: "HIGH",
    SEVERITY_MEDIUM: "MED ",
    SEVERITY_LOW: "LOW ",
}


def format_human(results):
    """Return human-readable optimization report."""
    lines = []
    total_suggestions = 0

    for result in results:
        lines.append(f"\n=== {result['file']} ===")
        lines.append(f"Platform: {result['platform']}")
        lines.append(f"Package managers detected: {', '.join(result['package_managers']) or 'none'}")

        if not result["suggestions"]:
            lines.append("\n  No optimization suggestions. Caching looks good!")
            continue

        lines.append("")
        for idx, s in enumerate(result["suggestions"], 1):
            badge = _SEVERITY_BADGE.get(s["severity"], "????")
            lines.append(f"  [{badge}] {idx}. {s['type']}")
            lines.append(f"         {s['message']}")
            if "recommendation" in s:
                for rec_line in s["recommendation"].split("\n"):
                    lines.append(f"         -> {rec_line}")
            if "estimated_savings" in s:
                lines.append(f"         Estimated savings: {s['estimated_savings']}")
            lines.append("")
            total_suggestions += 1

    lines.append(f"Total suggestions: {total_suggestions}")
    high_count = sum(
        1 for r in results for s in r["suggestions"] if s["severity"] == SEVERITY_HIGH
    )
    if high_count:
        lines.append(f"High-priority items: {high_count}")

    return "\n".join(lines)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description="Analyze CI/CD pipeline configs and suggest caching improvements.",
        epilog="Examples:\n"
               "  %(prog)s .github/workflows/ci.yml\n"
               "  %(prog)s --dir .github/workflows/ --json\n"
               "  %(prog)s .gitlab-ci.yml --severity high",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("files", nargs="*", help="Pipeline YAML files to analyze")
    parser.add_argument("--dir", help="Directory to scan recursively for YAML files")
    parser.add_argument("--json", action="store_true", dest="json_output",
                        help="Output results as JSON")
    parser.add_argument("--severity", choices=["high", "medium", "low"],
                        default="low",
                        help="Minimum severity to report (default: low)")
    args = parser.parse_args()

    files_to_analyze = []
    for f in (args.files or []):
        files_to_analyze.extend(collect_files(f))
    if args.dir:
        files_to_analyze.extend(collect_files(args.dir))

    if not files_to_analyze:
        parser.error("No files provided. Pass YAML files or use --dir.")

    severity_rank = {"high": 3, "medium": 2, "low": 1}
    min_rank = severity_rank[args.severity]

    results = []
    for fpath in files_to_analyze:
        result = analyze_file(fpath)
        result["suggestions"] = [
            s for s in result["suggestions"]
            if severity_rank.get(s["severity"], 0) >= min_rank
        ]
        results.append(result)

    if args.json_output:
        print(json.dumps(results, indent=2))
    else:
        print(format_human(results))

    has_high = any(
        s["severity"] == SEVERITY_HIGH
        for r in results for s in r["suggestions"]
    )
    sys.exit(1 if has_high else 0)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Generate CI/CD pipeline YAML from detected project stack signals.

Scans a project directory for lockfiles, manifests, Dockerfiles, and
framework configs, then generates an appropriate GitHub Actions or
GitLab CI pipeline with caching, testing, and deployment stages.

Usage:
    python pipeline_generator.py /path/to/project
    python pipeline_generator.py . --platform gitlab
    python pipeline_generator.py . --platform github --deploy --json
"""

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


# ---------------------------------------------------------------------------
# Stack detection
# ---------------------------------------------------------------------------

STACK_SIGNALS = [
    # (file_or_glob, language, package_manager, framework_hint)
    ("package-lock.json",   "node",   "npm",    None),
    ("pnpm-lock.yaml",      "node",   "pnpm",   None),
    ("yarn.lock",            "node",   "yarn",   None),
    ("bun.lockb",            "node",   "bun",    None),
    ("requirements.txt",     "python", "pip",    None),
    ("Pipfile",              "python", "pipenv", None),
    ("poetry.lock",          "python", "poetry", None),
    ("uv.lock",             "python", "uv",     None),
    ("pyproject.toml",       "python", None,     None),
    ("go.mod",               "go",     "go",     None),
    ("Cargo.lock",           "rust",   "cargo",  None),
    ("Gemfile.lock",         "ruby",   "bundler", None),
    ("composer.lock",        "php",    "composer", None),
]

FRAMEWORK_SIGNALS = [
    ("next.config.js",      "nextjs"),
    ("next.config.ts",      "nextjs"),
    ("next.config.mjs",     "nextjs"),
    ("nuxt.config.ts",      "nuxt"),
    ("nuxt.config.js",      "nuxt"),
    ("angular.json",        "angular"),
    ("svelte.config.js",    "sveltekit"),
    ("vite.config.ts",      "vite"),
    ("vite.config.js",      "vite"),
]

INFRA_SIGNALS = [
    ("Dockerfile",          "docker"),
    ("docker-compose.yml",  "docker-compose"),
    ("docker-compose.yaml", "docker-compose"),
]


def detect_stack(project_dir):
    """Detect project stack from filesystem signals."""
    root = Path(project_dir)
    detected = {
        "language": None,
        "package_manager": None,
        "framework": None,
        "has_docker": False,
        "has_docker_compose": False,
        "has_tests": False,
    }

    for filename, lang, pm, _fw in STACK_SIGNALS:
        if (root / filename).exists():
            if detected["language"] is None:
                detected["language"] = lang
            if pm and detected["package_manager"] is None:
                detected["package_manager"] = pm

    for filename, fw in FRAMEWORK_SIGNALS:
        if (root / filename).exists():
            detected["framework"] = fw
            break

    for filename, kind in INFRA_SIGNALS:
        if (root / filename).exists():
            if kind == "docker":
                detected["has_docker"] = True
            elif kind == "docker-compose":
                detected["has_docker_compose"] = True

    # Detect test presence
    test_indicators = [
        "tests", "test", "__tests__", "spec",
        "pytest.ini", "jest.config.js", "jest.config.ts",
        "vitest.config.ts", "vitest.config.js",
    ]
    for indicator in test_indicators:
        if (root / indicator).exists():
            detected["has_tests"] = True
            break

    return detected


# ---------------------------------------------------------------------------
# Pipeline templates (GitHub Actions)
# ---------------------------------------------------------------------------

def _github_node(stack, deploy):
    pm = stack["package_manager"] or "npm"
    install_cmd = {
        "npm": "npm ci",
        "pnpm": "pnpm install --frozen-lockfile",
        "yarn": "yarn install --frozen-lockfile",
        "bun": "bun install --frozen-lockfile",
    }.get(pm, "npm ci")

    setup_steps = "      - uses: actions/checkout@v4\n"
    if pm == "pnpm":
        setup_steps += "      - uses: pnpm/action-setup@v4\n        with:\n          version: '9'\n"
    setup_steps += (
        "      - uses: actions/setup-node@v4\n"
        "        with:\n"
        f"          node-version: '20'\n"
    )
    if pm in ("npm", "pnpm", "yarn"):
        setup_steps += f"          cache: '{pm}'\n"
    setup_steps += f"      - run: {install_cmd}\n"

    jobs = f"""  lint:
    runs-on: ubuntu-latest
    timeout-minutes: 10
    steps:
{setup_steps}      - run: npx eslint . || true

  test:
    runs-on: ubuntu-latest
    timeout-minutes: 15
    needs: lint
    steps:
{setup_steps}      - run: {"npx vitest run" if stack.get("framework") in ("vite", "nextjs", "sveltekit") else "npm test"}
"""

    if stack["has_docker"]:
        jobs += f"""
  build-image:
    runs-on: ubuntu-latest
    timeout-minutes: 20
    needs: test
    permissions:
      contents: read
      packages: write
    steps:
      - uses: actions/checkout@v4
      - uses: docker/setup-buildx-action@v3
      - uses: docker/build-push-action@v6
        with:
          context: .
          push: false
          tags: app:${{{{ github.sha }}}}
          cache-from: type=gha
          cache-to: type=gha,mode=max
"""
    else:
        jobs += f"""
  build:
    runs-on: ubuntu-latest
    timeout-minutes: 15
    needs: test
    steps:
{setup_steps}      - run: npm run build
      - uses: actions/upload-artifact@v4
        with:
          name: build-output
          path: dist/
          retention-days: 3
"""

    if deploy:
        jobs += """
  deploy-staging:
    if: github.ref == 'refs/heads/main' && github.event_name == 'push'
    needs: [build]
    runs-on: ubuntu-latest
    timeout-minutes: 10
    environment:
      name: staging
    steps:
      - uses: actions/checkout@v4
      - name: Deploy to staging
        run: echo "Add your staging deploy command here"
        env:
          DEPLOY_TOKEN: ${{ secrets.STAGING_DEPLOY_TOKEN }}

  deploy-production:
    if: github.ref == 'refs/heads/main' && github.event_name == 'push'
    needs: deploy-staging
    runs-on: ubuntu-latest
    timeout-minutes: 10
    environment:
      name: production
    steps:
      - uses: actions/checkout@v4
      - name: Deploy to production
        run: echo "Add your production deploy command here"
        env:
          DEPLOY_TOKEN: ${{ secrets.PROD_DEPLOY_TOKEN }}
"""

    return _github_wrapper(jobs)


def _github_python(stack, deploy):
    pm = stack["package_manager"] or "pip"
    install_block = {
        "pip": "      - run: pip install -r requirements.txt",
        "poetry": "      - run: pip install poetry && poetry install --no-interaction",
        "uv": "      - uses: astral-sh/setup-uv@v4\n      - run: uv sync --frozen",
        "pipenv": "      - run: pip install pipenv && pipenv install --deploy",
    }.get(pm, "      - run: pip install -r requirements.txt")

    setup_steps = (
        "      - uses: actions/checkout@v4\n"
        "      - uses: actions/setup-python@v5\n"
        "        with:\n"
        "          python-version: '3.12'\n"
    )
    if pm != "uv":
        setup_steps += install_block + "\n"
    else:
        setup_steps = "      - uses: actions/checkout@v4\n" + install_block + "\n"

    test_cmd = "uv run pytest -v" if pm == "uv" else "pytest -v"
    lint_cmd = "uv run ruff check ." if pm == "uv" else "ruff check . || true"

    jobs = f"""  lint:
    runs-on: ubuntu-latest
    timeout-minutes: 10
    steps:
{setup_steps}      - run: {lint_cmd}

  test:
    runs-on: ubuntu-latest
    timeout-minutes: 15
    needs: lint
    strategy:
      matrix:
        python-version: ['3.11', '3.12', '3.13']
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: ${{{{ matrix.python-version }}}}
{install_block}
      - run: {test_cmd}
"""

    if stack["has_docker"]:
        jobs += """
  build-image:
    runs-on: ubuntu-latest
    timeout-minutes: 20
    needs: test
    permissions:
      contents: read
      packages: write
    steps:
      - uses: actions/checkout@v4
      - uses: docker/setup-buildx-action@v3
      - uses: docker/build-push-action@v6
        with:
          context: .
          push: false
          tags: app:${{ github.sha }}
          cache-from: type=gha
          cache-to: type=gha,mode=max
"""

    if deploy:
        jobs += """
  deploy-staging:
    if: github.ref == 'refs/heads/main' && github.event_name == 'push'
    needs: [test]
    runs-on: ubuntu-latest
    timeout-minutes: 10
    environment:
      name: staging
    steps:
      - uses: actions/checkout@v4
      - name: Deploy to staging
        run: echo "Add your staging deploy command here"

  deploy-production:
    if: github.ref == 'refs/heads/main' && github.event_name == 'push'
    needs: deploy-staging
    runs-on: ubuntu-latest
    timeout-minutes: 10
    environment:
      name: production
    steps:
      - uses: actions/checkout@v4
      - name: Deploy to production
        run: echo "Add your production deploy command here"
"""

    return _github_wrapper(jobs)


def _github_go(stack, deploy):
    jobs = """  lint:
    runs-on: ubuntu-latest
    timeout-minutes: 10
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-go@v5
        with:
          go-version-file: go.mod
      - run: go vet ./...

  test:
    runs-on: ubuntu-latest
    timeout-minutes: 15
    needs: lint
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-go@v5
        with:
          go-version-file: go.mod
      - run: go test -race -coverprofile=coverage.out ./...

  build:
    runs-on: ubuntu-latest
    timeout-minutes: 10
    needs: test
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-go@v5
        with:
          go-version-file: go.mod
      - run: go build -o bin/app ./...
      - uses: actions/upload-artifact@v4
        with:
          name: binary
          path: bin/
          retention-days: 3
"""
    return _github_wrapper(jobs)


def _github_wrapper(jobs_block):
    return f"""name: CI/CD

on:
  push:
    branches: [main, dev]
  pull_request:
    branches: [main, dev]

concurrency:
  group: ${{{{ github.workflow }}}}-${{{{ github.ref }}}}
  cancel-in-progress: true

permissions:
  contents: read

jobs:
{jobs_block}"""


def _github_generic():
    return _github_wrapper("""  build:
    runs-on: ubuntu-latest
    timeout-minutes: 15
    steps:
      - uses: actions/checkout@v4
      - name: Build
        run: echo "Configure your build steps here"
      - name: Test
        run: echo "Configure your test steps here"
""")


# ---------------------------------------------------------------------------
# Pipeline templates (GitLab CI)
# ---------------------------------------------------------------------------

def _gitlab_node(stack, deploy):
    pm = stack["package_manager"] or "npm"
    install = {"npm": "npm ci", "pnpm": "corepack enable && pnpm install --frozen-lockfile",
               "yarn": "yarn install --frozen-lockfile", "bun": "bun install"}.get(pm, "npm ci")
    cache_path = {"npm": "node_modules/", "pnpm": ".pnpm-store/\n      - node_modules/",
                  "yarn": "node_modules/", "bun": "node_modules/"}.get(pm, "node_modules/")

    deploy_block = ""
    if deploy:
        deploy_block = """
deploy_staging:
  stage: deploy
  environment:
    name: staging
  rules:
    - if: $CI_COMMIT_BRANCH == "main"
  script:
    - echo "Add staging deploy command"

deploy_production:
  stage: deploy
  environment:
    name: production
  rules:
    - if: $CI_COMMIT_BRANCH == "main"
      when: manual
  needs: [deploy_staging]
  script:
    - echo "Add production deploy command"
"""

    return f"""stages:
  - validate
  - test
  - build
  - deploy

default:
  image: node:20

cache:
  key: ${{CI_COMMIT_REF_SLUG}}
  paths:
    - {cache_path}

lint:
  stage: validate
  before_script:
    - {install}
  script:
    - npx eslint . || true
  timeout: 10m

test:
  stage: test
  before_script:
    - {install}
  script:
    - npm test
  timeout: 15m

build:
  stage: build
  before_script:
    - {install}
  script:
    - npm run build
  artifacts:
    paths:
      - dist/
    expire_in: 1 hour
{deploy_block}"""


def _gitlab_python(stack, deploy):
    pm = stack["package_manager"] or "pip"
    install = {"pip": "pip install -r requirements.txt",
               "uv": "pip install uv && uv sync --frozen",
               "poetry": "pip install poetry && poetry install"}.get(pm, "pip install -r requirements.txt")

    return f"""stages:
  - validate
  - test
  - build
  - deploy

default:
  image: python:3.12

cache:
  key: ${{CI_COMMIT_REF_SLUG}}
  paths:
    - .cache/pip/

lint:
  stage: validate
  before_script:
    - {install}
  script:
    - ruff check . || true
  timeout: 10m

test:
  stage: test
  before_script:
    - {install}
  script:
    - pytest -v
  timeout: 15m
  parallel:
    matrix:
      - PYTHON_VERSION: ["3.11", "3.12", "3.13"]
"""


# ---------------------------------------------------------------------------
# Generator dispatch
# ---------------------------------------------------------------------------

GENERATORS = {
    ("github", "node"):   _github_node,
    ("github", "python"): _github_python,
    ("github", "go"):     _github_go,
    ("gitlab", "node"):   _gitlab_node,
    ("gitlab", "python"): _gitlab_python,
}


def generate_pipeline(project_dir, platform="github", deploy=False):
    """Generate a CI/CD pipeline for the given project."""
    stack = detect_stack(project_dir)
    lang = stack["language"]

    gen_fn = GENERATORS.get((platform, lang))
    if gen_fn:
        yaml_content = gen_fn(stack, deploy)
    elif platform == "github":
        yaml_content = _github_generic()
    else:
        yaml_content = "# Could not detect stack. Add pipeline configuration manually.\n"

    return {
        "stack": stack,
        "platform": platform,
        "yaml": yaml_content,
    }


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description="Generate CI/CD pipeline YAML from project stack detection.",
        epilog="Examples:\n"
               "  %(prog)s /path/to/project\n"
               "  %(prog)s . --platform gitlab --deploy\n"
               "  %(prog)s . --json",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("project_dir", nargs="?", default=".",
                        help="Project directory to scan (default: current directory)")
    parser.add_argument("--platform", choices=["github", "gitlab"],
                        default="github", help="Target CI platform (default: github)")
    parser.add_argument("--deploy", action="store_true",
                        help="Include deployment stages (staging + production)")
    parser.add_argument("--json", action="store_true", dest="json_output",
                        help="Output results as JSON (includes detected stack info)")
    parser.add_argument("--detect-only", action="store_true",
                        help="Only detect stack, do not generate pipeline")
    args = parser.parse_args()

    project_dir = os.path.abspath(args.project_dir)
    if not os.path.isdir(project_dir):
        print(f"Error: '{project_dir}' is not a valid directory.", file=sys.stderr)
        sys.exit(1)

    if args.detect_only:
        stack = detect_stack(project_dir)
        if args.json_output:
            print(json.dumps({"project_dir": project_dir, "stack": stack}, indent=2))
        else:
            print(f"Project: {project_dir}")
            print(f"Language:        {stack['language'] or 'unknown'}")
            print(f"Package Manager: {stack['package_manager'] or 'unknown'}")
            print(f"Framework:       {stack['framework'] or 'none'}")
            print(f"Docker:          {'yes' if stack['has_docker'] else 'no'}")
            print(f"Tests detected:  {'yes' if stack['has_tests'] else 'no'}")
        sys.exit(0)

    result = generate_pipeline(project_dir, platform=args.platform, deploy=args.deploy)

    if args.json_output:
        print(json.dumps({
            "project_dir": project_dir,
            "stack": result["stack"],
            "platform": result["platform"],
            "yaml": result["yaml"],
        }, indent=2))
    else:
        print(f"# Detected stack: {result['stack']['language'] or 'unknown'}"
              f" / {result['stack']['package_manager'] or 'unknown'}"
              f" / {result['stack']['framework'] or 'none'}")
        print(f"# Platform: {result['platform']}")
        print(f"# Docker: {'yes' if result['stack']['has_docker'] else 'no'}")
        print()
        print(result["yaml"])

    sys.exit(0)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Lint GitHub Actions and GitLab CI YAML files for common CI/CD issues.

Checks for missing permissions, hardcoded secrets, missing timeouts,
unpinned actions, missing concurrency controls, and more.

Usage:
    python pipeline_linter.py .github/workflows/ci.yml
    python pipeline_linter.py .gitlab-ci.yml --json
    python pipeline_linter.py --dir .github/workflows/
"""

import argparse
import json
import os
import re
import sys
from pathlib import Path


# ---------------------------------------------------------------------------
# Rule definitions
# ---------------------------------------------------------------------------

SEVERITY_ERROR = "error"
SEVERITY_WARNING = "warning"
SEVERITY_INFO = "info"


def _lines_containing(text, pattern):
    """Return list of (line_number, line_text) tuples matching *pattern*."""
    results = []
    for idx, line in enumerate(text.splitlines(), start=1):
        if re.search(pattern, line):
            results.append((idx, line.strip()))
    return results


def check_hardcoded_secrets(text, _platform):
    """Detect potential hardcoded secrets or tokens in the YAML."""
    findings = []
    patterns = [
        (r'(?i)(password|secret|token|api[_-]?key)\s*[:=]\s*["\']?[A-Za-z0-9+/=]{16,}',
         "Possible hardcoded secret value"),
        (r'(?i)(ghp_|gho_|github_pat_|sk-|AKIA)[A-Za-z0-9]{10,}',
         "Possible hardcoded API token"),
        (r'(?i)BEGIN\s+(RSA|DSA|EC|OPENSSH)\s+PRIVATE\s+KEY',
         "Private key embedded in pipeline file"),
    ]
    for pat, msg in patterns:
        for lineno, line in _lines_containing(text, pat):
            findings.append({
                "rule": "no-hardcoded-secrets",
                "severity": SEVERITY_ERROR,
                "line": lineno,
                "message": f"{msg}: {line[:80]}",
            })
    return findings


def check_unpinned_actions(text, platform):
    """Warn when GitHub Actions use @main or @master instead of a pinned SHA/tag."""
    if platform != "github-actions":
        return []
    findings = []
    pat = r'uses:\s+([^@\s]+)@(main|master)\s*$'
    for lineno, line in _lines_containing(text, pat):
        findings.append({
            "rule": "pin-action-versions",
            "severity": SEVERITY_WARNING,
            "line": lineno,
            "message": f"Action pinned to mutable branch: {line}",
        })
    return findings


def check_missing_timeout(text, platform):
    """Flag jobs that lack a timeout setting."""
    findings = []
    if platform == "github-actions":
        # Heuristic: look for 'runs-on' (marks a job) without a nearby timeout-minutes
        in_job = False
        job_name = ""
        job_line = 0
        has_timeout = False
        for idx, line in enumerate(text.splitlines(), start=1):
            stripped = line.strip()
            if re.match(r'^[a-zA-Z0-9_-]+:\s*$', stripped) or re.match(r'^[a-zA-Z0-9_-]+:$', stripped):
                if in_job and not has_timeout:
                    findings.append({
                        "rule": "require-timeout",
                        "severity": SEVERITY_WARNING,
                        "line": job_line,
                        "message": f"Job '{job_name}' has no timeout-minutes (default is 6 hours)",
                    })
                in_job = False
                has_timeout = False
            if "runs-on:" in stripped:
                in_job = True
                job_name = stripped
                job_line = idx
            if "timeout-minutes:" in stripped:
                has_timeout = True
        if in_job and not has_timeout:
            findings.append({
                "rule": "require-timeout",
                "severity": SEVERITY_WARNING,
                "line": job_line,
                "message": f"Job '{job_name}' has no timeout-minutes (default is 6 hours)",
            })
    elif platform == "gitlab-ci":
        if "timeout:" not in text:
            findings.append({
                "rule": "require-timeout",
                "severity": SEVERITY_INFO,
                "line": 1,
                "message": "No global or per-job timeout set (GitLab default is 1 hour)",
            })
    return findings


def check_missing_concurrency(text, platform):
    """Flag GitHub Actions workflows missing a concurrency group."""
    if platform != "github-actions":
        return []
    if "concurrency:" not in text:
        return [{
            "rule": "require-concurrency",
            "severity": SEVERITY_WARNING,
            "line": 1,
            "message": "Workflow has no concurrency group; duplicate runs can waste resources",
        }]
    return []


def check_missing_permissions(text, platform):
    """Flag GitHub Actions workflows without explicit permissions."""
    if platform != "github-actions":
        return []
    if "permissions:" not in text:
        return [{
            "rule": "require-permissions",
            "severity": SEVERITY_WARNING,
            "line": 1,
            "message": "No explicit permissions block; workflow runs with default (often broad) token scope",
        }]
    return []


def check_missing_path_filters(text, platform):
    """Info-level hint when pushes trigger on all paths."""
    if platform != "github-actions":
        return []
    if re.search(r'on:\s*\n\s+push:', text) and "paths" not in text:
        return [{
            "rule": "suggest-path-filters",
            "severity": SEVERITY_INFO,
            "line": 1,
            "message": "No path filters; documentation-only changes will trigger full CI",
        }]
    return []


def check_artifact_retention(text, platform):
    """Warn when upload-artifact has no retention-days."""
    findings = []
    if platform == "github-actions":
        in_upload = False
        upload_line = 0
        has_retention = False
        for idx, line in enumerate(text.splitlines(), start=1):
            stripped = line.strip()
            if "actions/upload-artifact" in stripped:
                if in_upload and not has_retention:
                    findings.append({
                        "rule": "set-artifact-retention",
                        "severity": SEVERITY_WARNING,
                        "line": upload_line,
                        "message": "upload-artifact without retention-days; artifacts kept for 90 days by default",
                    })
                in_upload = True
                upload_line = idx
                has_retention = False
            if in_upload and "retention-days:" in stripped:
                has_retention = True
            if in_upload and stripped.startswith("- ") and "upload-artifact" not in stripped:
                if not has_retention:
                    findings.append({
                        "rule": "set-artifact-retention",
                        "severity": SEVERITY_WARNING,
                        "line": upload_line,
                        "message": "upload-artifact without retention-days; artifacts kept for 90 days by default",
                    })
                in_upload = False
                has_retention = False
    elif platform == "gitlab-ci":
        if "artifacts:" in text and "expire_in:" not in text:
            findings.append({
                "rule": "set-artifact-retention",
                "severity": SEVERITY_WARNING,
                "line": 1,
                "message": "Artifacts defined without expire_in; storage costs grow indefinitely",
            })
    return findings


def check_deploy_without_gate(text, platform):
    """Flag deploy jobs that lack environment protection or branch guards."""
    findings = []
    if platform == "github-actions":
        for idx, line in enumerate(text.splitlines(), start=1):
            if re.search(r'deploy.*production', line, re.IGNORECASE):
                # Check next ~15 lines for environment or if guard
                block = "\n".join(text.splitlines()[idx:idx + 15])
                if "environment:" not in block and 'if:' not in block:
                    findings.append({
                        "rule": "gate-production-deploy",
                        "severity": SEVERITY_ERROR,
                        "line": idx,
                        "message": "Production deploy job lacks environment gate or branch condition",
                    })
    elif platform == "gitlab-ci":
        for idx, line in enumerate(text.splitlines(), start=1):
            if re.search(r'deploy.*production', line, re.IGNORECASE):
                block = "\n".join(text.splitlines()[idx:idx + 15])
                if "when: manual" not in block and "rules:" not in block:
                    findings.append({
                        "rule": "gate-production-deploy",
                        "severity": SEVERITY_ERROR,
                        "line": idx,
                        "message": "Production deploy job lacks manual gate or rules guard",
                    })
    return findings


# ---------------------------------------------------------------------------
# Platform detection & orchestration
# ---------------------------------------------------------------------------

ALL_CHECKS = [
    check_hardcoded_secrets,
    check_unpinned_actions,
    check_missing_timeout,
    check_missing_concurrency,
    check_missing_permissions,
    check_missing_path_filters,
    check_artifact_retention,
    check_deploy_without_gate,
]


def detect_platform(filepath, text):
    """Determine CI platform from path and content."""
    name = os.path.basename(filepath)
    parent = os.path.basename(os.path.dirname(filepath))
    if parent == "workflows" or ".github" in filepath:
        return "github-actions"
    if name == ".gitlab-ci.yml" or "stages:" in text:
        return "gitlab-ci"
    # Fallback heuristic
    if "runs-on:" in text:
        return "github-actions"
    if "image:" in text and "script:" in text:
        return "gitlab-ci"
    return "unknown"


def lint_file(filepath):
    """Run all checks on a single file. Returns dict with findings."""
    filepath = str(filepath)
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            text = f.read()
    except (OSError, UnicodeDecodeError) as exc:
        return {
            "file": filepath,
            "platform": "unknown",
            "findings": [{
                "rule": "file-read-error",
                "severity": SEVERITY_ERROR,
                "line": 0,
                "message": str(exc),
            }],
        }

    platform = detect_platform(filepath, text)
    findings = []
    for check_fn in ALL_CHECKS:
        findings.extend(check_fn(text, platform))

    findings.sort(key=lambda f: (f["line"], f["severity"]))
    return {"file": filepath, "platform": platform, "findings": findings}


def collect_files(path):
    """Collect YAML files from a file path or directory."""
    p = Path(path)
    if p.is_file():
        return [p]
    if p.is_dir():
        yamls = sorted(p.glob("**/*.yml")) + sorted(p.glob("**/*.yaml"))
        return yamls
    return []


# ---------------------------------------------------------------------------
# Output formatting
# ---------------------------------------------------------------------------

_SEVERITY_SYMBOL = {
    SEVERITY_ERROR: "E",
    SEVERITY_WARNING: "W",
    SEVERITY_INFO: "I",
}


def format_human(results):
    """Return human-readable report string."""
    lines = []
    total_errors = 0
    total_warnings = 0
    for result in results:
        lines.append(f"\n--- {result['file']} (platform: {result['platform']}) ---")
        if not result["findings"]:
            lines.append("  No issues found.")
            continue
        for f in result["findings"]:
            sym = _SEVERITY_SYMBOL.get(f["severity"], "?")
            lines.append(f"  [{sym}] L{f['line']:>4d}  {f['rule']}: {f['message']}")
            if f["severity"] == SEVERITY_ERROR:
                total_errors += 1
            elif f["severity"] == SEVERITY_WARNING:
                total_warnings += 1
    lines.append(f"\nSummary: {total_errors} error(s), {total_warnings} warning(s)")
    return "\n".join(lines)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description="Lint CI/CD pipeline YAML files for common issues.",
        epilog="Examples:\n"
               "  %(prog)s .github/workflows/ci.yml\n"
               "  %(prog)s --dir .github/workflows/ --json\n"
               "  %(prog)s .gitlab-ci.yml --severity warning",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("files", nargs="*", help="Pipeline YAML files to lint")
    parser.add_argument("--dir", help="Directory to scan recursively for YAML files")
    parser.add_argument("--json", action="store_true", dest="json_output",
                        help="Output results as JSON")
    parser.add_argument("--severity", choices=["error", "warning", "info"],
                        default="info",
                        help="Minimum severity to report (default: info)")
    args = parser.parse_args()

    files_to_lint = []
    for f in (args.files or []):
        files_to_lint.extend(collect_files(f))
    if args.dir:
        files_to_lint.extend(collect_files(args.dir))

    if not files_to_lint:
        parser.error("No files provided. Pass YAML files or use --dir.")

    severity_rank = {"error": 3, "warning": 2, "info": 1}
    min_rank = severity_rank[args.severity]

    results = []
    for fpath in files_to_lint:
        result = lint_file(fpath)
        result["findings"] = [
            f for f in result["findings"]
            if severity_rank.get(f["severity"], 0) >= min_rank
        ]
        results.append(result)

    if args.json_output:
        print(json.dumps(results, indent=2))
    else:
        print(format_human(results))

    has_errors = any(
        f["severity"] == SEVERITY_ERROR
        for r in results for f in r["findings"]
    )
    sys.exit(1 if has_errors else 0)


if __name__ == "__main__":
    main()