Runbook Generator

#!/usr/bin/env python3
"""Runbook Scaffolder — Generate runbook markdown templates from service definitions.

Accepts a JSON service definition (via file or stdin) containing service name,
stack components, dependencies, and endpoints. Produces a production-grade
runbook markdown file with deployment steps, rollback procedures, monitoring
checks, escalation paths, and staleness tracking.

Usage:
    python runbook_scaffolder.py --input service.json
    python runbook_scaffolder.py --input service.json --output runbook.md
    cat service.json | python runbook_scaffolder.py --input -
    python runbook_scaffolder.py --input service.json --json
    python runbook_scaffolder.py --input service.json --type incident
"""

import argparse
import json
import sys
import textwrap
from datetime import date


RUNBOOK_TYPES = ["deployment", "incident", "database", "scaling", "monitoring"]

DEPLOY_COMMANDS = {
    "vercel": "vercel --prod",
    "fly": "fly deploy --app $APP_NAME",
    "kubernetes": "kubectl apply -f k8s/deployment.yaml",
    "aws-ecs": "aws ecs update-service --cluster $CLUSTER --service $SERVICE --force-new-deployment",
    "heroku": "git push heroku main",
    "docker-compose": "docker-compose -f docker-compose.prod.yml up -d",
}

ROLLBACK_COMMANDS = {
    "vercel": "vercel rollback $PREVIOUS_DEPLOYMENT_URL",
    "fly": "fly releases --app $APP_NAME\nfly deploy --image $PREVIOUS_IMAGE",
    "kubernetes": "kubectl rollout undo deployment/$APP_NAME",
    "aws-ecs": "aws ecs update-service --cluster $CLUSTER --service $SERVICE --task-definition $PREV_TASK_DEF",
    "heroku": "heroku rollback --app $APP_NAME",
    "docker-compose": "docker-compose -f docker-compose.prod.yml down\ndocker-compose -f docker-compose.prod.yml up -d --no-build",
}

DB_BACKUP_COMMANDS = {
    "postgresql": 'pg_dump $PROD_DB_URL --format=custom --compress=9 --file="backup-$(date +%Y%m%d-%H%M%S).dump"',
    "mysql": 'mysqldump -h $DB_HOST -u $DB_USER -p$DB_PASS $DB_NAME > "backup-$(date +%Y%m%d-%H%M%S).sql"',
    "mongodb": 'mongodump --uri="$MONGO_URI" --out="backup-$(date +%Y%m%d-%H%M%S)"',
}

SCALE_COMMANDS = {
    "kubernetes": "kubectl scale deployment/$APP_NAME --replicas=$REPLICA_COUNT",
    "fly": "fly scale count $REPLICA_COUNT --app $APP_NAME",
    "aws-ecs": "aws ecs update-service --cluster $CLUSTER --service $SERVICE --desired-count $REPLICA_COUNT",
    "heroku": "heroku ps:scale web=$REPLICA_COUNT --app $APP_NAME",
    "docker-compose": "docker-compose -f docker-compose.prod.yml up -d --scale web=$REPLICA_COUNT",
}


def load_service_definition(input_path):
    """Load and validate a JSON service definition from file or stdin."""
    try:
        if input_path == "-":
            data = json.load(sys.stdin)
        else:
            with open(input_path, "r") as f:
                data = json.load(f)
    except json.JSONDecodeError as e:
        print(f"Error: Invalid JSON — {e}", file=sys.stderr)
        sys.exit(1)
    except FileNotFoundError:
        print(f"Error: File not found — {input_path}", file=sys.stderr)
        sys.exit(1)

    required = ["name"]
    missing = [k for k in required if k not in data]
    if missing:
        print(f"Error: Missing required fields — {', '.join(missing)}", file=sys.stderr)
        sys.exit(1)

    data.setdefault("stack", {})
    data.setdefault("dependencies", [])
    data.setdefault("endpoints", [])
    data.setdefault("contacts", {})
    data["stack"].setdefault("hosting", "kubernetes")
    data["stack"].setdefault("database", "postgresql")
    data["stack"].setdefault("ci_cd", "github-actions")
    data["stack"].setdefault("framework", "unknown")

    return data


def build_deployment_runbook(svc):
    """Build a deployment runbook markdown string."""
    hosting = svc["stack"]["hosting"].lower()
    db = svc["stack"]["database"].lower()
    today = date.today().isoformat()

    deploy_cmd = DEPLOY_COMMANDS.get(hosting, f"# Deploy using {hosting}")
    rollback_cmd = ROLLBACK_COMMANDS.get(hosting, f"# Rollback using {hosting}")

    health_checks = ""
    for ep in svc.get("endpoints", []):
        url = ep if isinstance(ep, str) else ep.get("url", ep.get("path", "/health"))
        health_checks += f"curl -sf {url} | jq .\n"
    if not health_checks:
        health_checks = "curl -sf https://$APP_HOST/api/health | jq .\n"

    contacts_table = _build_contacts_table(svc.get("contacts", {}))
    deps_list = "\n".join(f"- {d}" for d in svc.get("dependencies", [])) or "- None documented"
    config_files = svc.get("config_files", ["vercel.json", ".github/workflows/deploy.yml"])
    staleness_rows = "\n".join(
        f"| {cf} | `git log -1 --format=%ci -- {cf}` | Deploy |"
        for cf in config_files
    )

    return textwrap.dedent(f"""\
    # Deployment Runbook — {svc['name']}

    **Stack:** {svc['stack']['framework']} + {db} + {hosting}
    **Last verified:** {today}
    **Owner:** {svc.get('contacts', {}).get('owner', 'FILL IN')}
    **Estimated total time:** 15-25 minutes

    ---

    ## Staleness Check

    | Config File | Last Modified | Affects Steps |
    |-------------|--------------|---------------|
    {staleness_rows}

    If any config was modified after the "Last verified" date, review affected steps.

    ---

    ## Dependencies

    {deps_list}

    ## Pre-Deployment Checklist

    - [ ] All PRs merged to main
    - [ ] CI passing on main branch
    - [ ] Database migrations tested in staging
    - [ ] Rollback plan confirmed
    - [ ] On-call engineer notified

    ## Step 1: Verify CI Status (2 min)

    ```bash
    gh run list --branch main --limit 3
    ```

    VERIFY: Latest run shows green checkmark. If red, do not proceed.

    ## Step 2: Apply Database Migrations (5 min)

    ```bash
    DATABASE_URL=$STAGING_DB_URL pnpm db:migrate
    DATABASE_URL=$STAGING_DB_URL pnpm db:migrate status
    ```

    VERIFY: Output shows "All migrations applied" with today's date.

    ```bash
    DATABASE_URL=$PROD_DB_URL pnpm db:migrate
    ```

    VERIFY: Same output as staging. If error, see Rollback section.

    ## Step 3: Deploy to Production (5 min)

    ```bash
    {deploy_cmd}
    ```

    VERIFY: Deployment dashboard shows new version in progress.

    ## Step 4: Smoke Test (5 min)

    ```bash
    {health_checks.strip()}
    ```

    VERIFY: Health returns `{{"status": "ok"}}`. Error rate < 1%.

    ## Step 5: Monitor (10 min)

    Watch metrics for 10 minutes:
    - Error rate: < 1%
    - P95 latency: < 200ms
    - DB connections: < 80% of max

    ---

    ## Rollback

    ```bash
    {rollback_cmd}
    ```

    VERIFY: Previous version serving traffic. Re-run smoke tests.

    ---

    ## Escalation

    {contacts_table}
    """)


def build_incident_runbook(svc):
    """Build an incident response runbook."""
    contacts_table = _build_contacts_table(svc.get("contacts", {}))
    return textwrap.dedent(f"""\
    # Incident Response Runbook — {svc['name']}

    **Severity:** P1 (down), P2 (degraded), P3 (minor)
    **Last verified:** {date.today().isoformat()}
    **Owner:** {svc.get('contacts', {}).get('owner', 'FILL IN')}

    ---

    ## Phase 1: Triage (5 min)

    ```bash
    curl -sw "%{{http_code}}" https://$APP_HOST/api/health -o /dev/null
    ```

    VERIFY: 200 = app is up. 5xx or timeout = incident confirmed.

    | Condition | Severity | Action |
    |-----------|----------|--------|
    | Site completely unreachable | P1 | Page L2/L3 immediately |
    | Partial degradation | P2 | Notify team channel |
    | Single feature broken | P3 | Create ticket |

    ## Phase 2: Diagnose (10-15 min)

    ```bash
    git log --oneline -10
    ```

    Check for recent deployments and correlate with incident start time.

    ## Phase 3: Mitigate

    Apply the first applicable fix:
    1. Rollback last deployment
    2. Kill runaway database queries
    3. Scale up replicas
    4. Enable circuit breaker for failing external dependency

    ## Phase 4: Resolve and Postmortem

    Within 24 hours:
    1. Write incident timeline
    2. Identify root cause (5 Whys)
    3. Define action items with owners
    4. Update this runbook

    ---

    ## Escalation

    {contacts_table}
    """)


def build_database_runbook(svc):
    """Build a database maintenance runbook."""
    db = svc["stack"]["database"].lower()
    backup_cmd = DB_BACKUP_COMMANDS.get(db, f"# Backup command for {db}")
    return textwrap.dedent(f"""\
    # Database Maintenance Runbook — {svc['name']}

    **Database:** {db}
    **Schedule:** Weekly vacuum (automated), monthly manual review
    **Last verified:** {date.today().isoformat()}

    ---

    ## Step 1: Backup (5 min)

    ```bash
    {backup_cmd}
    ```

    VERIFY: Backup file exists and size > 0.

    ## Step 2: Check Bloat (3 min)

    ```bash
    psql $PROD_DB_URL -c "
    SELECT tablename,
           pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) AS size,
           n_dead_tup,
           ROUND(n_dead_tup::numeric / NULLIF(n_live_tup, 0) * 100, 1) AS dead_pct
    FROM pg_stat_user_tables
    ORDER BY n_dead_tup DESC
    LIMIT 10;"
    ```

    VERIFY: Identify tables with dead_pct > 5%.

    ## Step 3: Vacuum (5 min)

    ```bash
    psql $PROD_DB_URL -c "VACUUM ANALYZE <table_name>;"
    ```

    VERIFY: dead_pct drops below 5% after vacuum.

    ## Step 4: Reindex (10 min)

    ```bash
    psql $PROD_DB_URL -c "REINDEX INDEX CONCURRENTLY <index_name>;"
    ```

    VERIFY: Index size reduced. Query performance stable.

    ## Rollback

    Restore from backup if vacuum or reindex causes issues:
    ```bash
    pg_restore --dbname=$PROD_DB_URL backup-*.dump
    ```
    """)


def build_scaling_runbook(svc):
    """Build a scaling operations runbook."""
    hosting = svc["stack"]["hosting"].lower()
    scale_cmd = SCALE_COMMANDS.get(hosting, f"# Scale command for {hosting}")
    return textwrap.dedent(f"""\
    # Scaling Runbook — {svc['name']}

    **Hosting:** {hosting}
    **Last verified:** {date.today().isoformat()}

    ---

    ## When to Scale

    - CPU utilization > 70% sustained for 5+ minutes
    - Memory utilization > 80%
    - Request queue depth increasing
    - P95 latency > 500ms

    ## Step 1: Assess Current State (2 min)

    Check current replica count and resource utilization before scaling.

    ## Step 2: Scale Up (3 min)

    ```bash
    {scale_cmd}
    ```

    VERIFY: New replicas are running and receiving traffic.

    ## Step 3: Monitor (10 min)

    Watch metrics for 10 minutes to confirm scaling resolved the issue.

    ## Scale Down

    After the load subsides, scale back to baseline to control costs.

    ```bash
    {scale_cmd.replace('$REPLICA_COUNT', '$BASELINE_COUNT')}
    ```
    """)


def build_monitoring_runbook(svc):
    """Build a monitoring setup runbook."""
    endpoints = svc.get("endpoints", [])
    ep_checks = ""
    for ep in endpoints:
        url = ep if isinstance(ep, str) else ep.get("url", ep.get("path", "/health"))
        ep_checks += f"    - URL: {url}\n"
    if not ep_checks:
        ep_checks = "    - URL: https://$APP_HOST/api/health\n"

    return textwrap.dedent(f"""\
    # Monitoring Runbook — {svc['name']}

    **Last verified:** {date.today().isoformat()}

    ---

    ## Health Check Endpoints

{ep_checks}
    ## Key Metrics

    | Metric | Warning Threshold | Critical Threshold |
    |--------|------------------|-------------------|
    | Error rate | > 1% | > 5% |
    | P95 latency | > 200ms | > 1000ms |
    | CPU utilization | > 70% | > 90% |
    | Memory utilization | > 75% | > 90% |
    | DB connections | > 60% of max | > 80% of max |

    ## Alert Configuration

    Configure alerts for each critical threshold. Ensure PagerDuty integration
    is active for P1-level alerts.

    ## Dashboard Links

    - Application dashboard: FILL IN
    - Database dashboard: FILL IN
    - Infrastructure dashboard: FILL IN

    ## On-Call Rotation

    Ensure the on-call schedule is current. Review monthly.
    """)


def _build_contacts_table(contacts):
    """Build a markdown escalation table from contacts dict."""
    if not contacts:
        return (
            "| Level | Who | When | Contact |\n"
            "|-------|-----|------|---------|\n"
            "| L1 | On-call engineer | First responder | PagerDuty rotation |\n"
            "| L2 | Platform lead | Escalation | FILL IN |\n"
            "| L3 | VP Engineering | Production down > 30 min | FILL IN |"
        )
    rows = ["| Level | Who | When | Contact |", "|-------|-----|------|---------|"]
    level_map = {"l1": "First responder", "l2": "Escalation", "l3": "Production down > 30 min"}
    for level in ["l1", "l2", "l3"]:
        info = contacts.get(level, {})
        if isinstance(info, str):
            rows.append(f"| {level.upper()} | {info} | {level_map.get(level, '')} | FILL IN |")
        elif isinstance(info, dict):
            rows.append(
                f"| {level.upper()} | {info.get('name', 'FILL IN')} "
                f"| {level_map.get(level, '')} | {info.get('contact', 'FILL IN')} |"
            )
        else:
            rows.append(f"| {level.upper()} | FILL IN | {level_map.get(level, '')} | FILL IN |")
    return "\n".join(rows)


BUILDERS = {
    "deployment": build_deployment_runbook,
    "incident": build_incident_runbook,
    "database": build_database_runbook,
    "scaling": build_scaling_runbook,
    "monitoring": build_monitoring_runbook,
}


def main():
    parser = argparse.ArgumentParser(
        description="Generate runbook markdown templates from JSON service definitions.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=textwrap.dedent("""\
            Example service definition JSON:
              {
                "name": "my-api",
                "stack": {"hosting": "kubernetes", "database": "postgresql",
                          "ci_cd": "github-actions", "framework": "FastAPI"},
                "dependencies": ["redis", "stripe-api", "sendgrid"],
                "endpoints": ["/api/health", "/api/v1/users"],
                "contacts": {
                  "owner": "Platform Team",
                  "l1": {"name": "On-call", "contact": "PagerDuty"},
                  "l2": {"name": "Platform Lead", "contact": "#platform-lead"}
                },
                "config_files": ["k8s/deployment.yaml", ".github/workflows/deploy.yml"]
              }
        """),
    )
    parser.add_argument("--input", "-i", required=True, help="Path to JSON service definition (use - for stdin)")
    parser.add_argument("--output", "-o", help="Output file path (default: stdout)")
    parser.add_argument("--type", "-t", default="deployment", choices=RUNBOOK_TYPES,
                        help="Runbook type to generate (default: deployment)")
    parser.add_argument("--json", action="store_true", help="Output as JSON with metadata")

    args = parser.parse_args()
    svc = load_service_definition(args.input)
    builder = BUILDERS[args.type]
    content = builder(svc)

    if args.json:
        result = {
            "service": svc["name"],
            "runbook_type": args.type,
            "generated_date": date.today().isoformat(),
            "stack": svc["stack"],
            "content": content,
        }
        output = json.dumps(result, indent=2)
    else:
        output = content

    if args.output:
        with open(args.output, "w") as f:
            f.write(output)
        print(f"Runbook written to {args.output}", file=sys.stderr)
    else:
        print(output)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Runbook Validator — Validate runbooks for completeness and quality.

Checks runbook markdown files against a set of required sections, contact
information, rollback steps, monitoring links, verification blocks, and
time estimates. Reports missing elements with severity levels.

Usage:
    python runbook_validator.py runbook.md
    python runbook_validator.py runbook.md another.md
    python runbook_validator.py --dir docs/runbooks/
    python runbook_validator.py runbook.md --json
    python runbook_validator.py runbook.md --strict
"""

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


# Required sections (heading text to search for, case-insensitive)
REQUIRED_SECTIONS = {
    "deployment": [
        "pre-deployment checklist",
        "rollback",
        "escalation",
    ],
    "incident": [
        "triage",
        "diagnose",
        "mitigate",
        "escalation",
    ],
    "database": [
        "backup",
        "rollback",
    ],
    "general": [
        "rollback",
        "escalation",
    ],
}

# Patterns that indicate quality markers
QUALITY_PATTERNS = {
    "verify_blocks": {
        "pattern": r"(?i)^VERIFY[:.]",
        "description": "VERIFY block after steps",
        "severity": "error",
        "min_count": 1,
    },
    "time_estimates": {
        "pattern": r"\(\d+\s*(?:min|minutes|hour|hours|sec|seconds)\)",
        "description": "Time estimate in step heading",
        "severity": "warning",
        "min_count": 1,
    },
    "code_blocks": {
        "pattern": r"^```",
        "description": "Code block with commands",
        "severity": "error",
        "min_count": 1,
    },
    "last_verified": {
        "pattern": r"(?i)last\s+verified[:]\s*\d{4}-\d{2}-\d{2}",
        "description": "Last verified date",
        "severity": "error",
        "min_count": 1,
    },
    "owner_field": {
        "pattern": r"(?i)\*\*owner\*\*[:]\s*\S+",
        "description": "Owner field",
        "severity": "error",
        "min_count": 1,
    },
    "env_vars_not_hardcoded": {
        "pattern": r"(?:postgres|mysql|mongodb)://\w+:\w+@[\w.]+",
        "description": "Hardcoded database URL (should use env var)",
        "severity": "error",
        "min_count": 0,  # 0 means we want zero matches (inverted check)
    },
    "checklist_items": {
        "pattern": r"^- \[[ x]\]",
        "description": "Checklist item",
        "severity": "warning",
        "min_count": 1,
    },
    "escalation_table": {
        "pattern": r"\|\s*L[123]\s*\|",
        "description": "Escalation table with L1/L2/L3",
        "severity": "warning",
        "min_count": 1,
    },
    "monitoring_links": {
        "pattern": r"(?i)dashboard[:]\s*(?:https?://\S+|\[.*\]\(.*\)|FILL IN)",
        "description": "Monitoring dashboard link or placeholder",
        "severity": "warning",
        "min_count": 1,
    },
}


def detect_runbook_type(content):
    """Detect the type of runbook from its content."""
    lower = content.lower()
    if "deployment runbook" in lower or "pre-deployment" in lower:
        return "deployment"
    if "incident response" in lower or "triage" in lower:
        return "incident"
    if "database maintenance" in lower or "vacuum" in lower:
        return "database"
    return "general"


def extract_headings(content):
    """Extract all markdown headings from content."""
    headings = []
    for line in content.split("\n"):
        match = re.match(r"^(#{1,6})\s+(.+)", line)
        if match:
            level = len(match.group(1))
            text = match.group(2).strip()
            headings.append({"level": level, "text": text})
    return headings


def count_pattern_matches(content, pattern):
    """Count how many lines match a regex pattern."""
    count = 0
    for line in content.split("\n"):
        if re.search(pattern, line):
            count += 1
    return count


def check_required_sections(content, runbook_type):
    """Check for required sections based on runbook type."""
    findings = []
    headings_lower = [h["text"].lower() for h in extract_headings(content)]
    required = REQUIRED_SECTIONS.get(runbook_type, REQUIRED_SECTIONS["general"])

    for section in required:
        found = any(section in h for h in headings_lower)
        if not found:
            findings.append({
                "check": "required_section",
                "severity": "error",
                "message": f"Missing required section: '{section}'",
                "suggestion": f"Add a '## {section.title()}' section",
            })

    return findings


def check_quality_patterns(content):
    """Check for quality patterns in the runbook."""
    findings = []

    for name, spec in QUALITY_PATTERNS.items():
        count = count_pattern_matches(content, spec["pattern"])

        if spec["min_count"] == 0:
            # Inverted check: we want zero matches
            if count > 0:
                findings.append({
                    "check": name,
                    "severity": spec["severity"],
                    "message": f"Found {count} instance(s) of: {spec['description']}",
                    "suggestion": "Use environment variables instead of hardcoded values",
                })
        else:
            if count < spec["min_count"]:
                findings.append({
                    "check": name,
                    "severity": spec["severity"],
                    "message": f"Missing: {spec['description']} (found {count}, need >= {spec['min_count']})",
                    "suggestion": f"Add at least {spec['min_count']} {spec['description'].lower()}",
                })

    return findings


def check_rollback_coverage(content):
    """Check that destructive steps have corresponding rollback instructions."""
    findings = []
    destructive_keywords = [
        "delete", "drop", "truncate", "migrate", "deploy",
        "scale", "restart", "terminate", "remove",
    ]

    lines = content.split("\n")
    has_rollback_section = any(
        re.match(r"^#{1,4}\s+.*rollback", line, re.IGNORECASE) for line in lines
    )

    destructive_found = False
    for line in lines:
        lower = line.lower()
        if any(kw in lower for kw in destructive_keywords):
            if re.search(r"```|^\s*(#|--|//)", line):
                continue  # skip comments and code fence markers
            if any(kw in lower for kw in destructive_keywords):
                destructive_found = True
                break

    if destructive_found and not has_rollback_section:
        findings.append({
            "check": "rollback_coverage",
            "severity": "error",
            "message": "Runbook contains destructive operations but no rollback section",
            "suggestion": "Add a '## Rollback' section with undo steps for each destructive action",
        })

    return findings


def check_step_numbering(content):
    """Check that steps are numbered consistently."""
    findings = []
    step_headings = re.findall(r"^#{2,3}\s+Step\s+(\d+)", content, re.MULTILINE)

    if not step_headings:
        findings.append({
            "check": "step_numbering",
            "severity": "warning",
            "message": "No numbered steps found (e.g., '## Step 1: ...')",
            "suggestion": "Use numbered step headings for clarity: '## Step 1: Description (X min)'",
        })
    else:
        numbers = [int(n) for n in step_headings]
        expected = list(range(1, len(numbers) + 1))
        if numbers != expected:
            findings.append({
                "check": "step_numbering",
                "severity": "warning",
                "message": f"Step numbering is not sequential: found {numbers}, expected {expected}",
                "suggestion": "Re-number steps sequentially starting from 1",
            })

    return findings


def check_placeholder_commands(content):
    """Check for placeholder values in commands that should be env vars."""
    findings = []
    placeholder_patterns = [
        (r"<[A-Z_]+>", "angle-bracket placeholder"),
        (r"YOUR_[A-Z_]+", "YOUR_ placeholder"),
        (r"CHANGEME", "CHANGEME placeholder"),
        (r"TODO", "TODO marker"),
        (r"xxx+", "xxx placeholder"),
    ]

    in_code_block = False
    for i, line in enumerate(content.split("\n"), 1):
        if line.strip().startswith("```"):
            in_code_block = not in_code_block
            continue
        if in_code_block:
            for pattern, desc in placeholder_patterns:
                if re.search(pattern, line, re.IGNORECASE):
                    findings.append({
                        "check": "placeholder_command",
                        "severity": "warning",
                        "message": f"Line {i}: Found {desc} in command — '{line.strip()}'",
                        "suggestion": "Replace with environment variable reference (e.g., $PROD_DB_URL)",
                    })

    return findings


def validate_runbook(filepath):
    """Run all validation checks on a single runbook file."""
    try:
        with open(filepath, "r") as f:
            content = f.read()
    except FileNotFoundError:
        return {
            "file": str(filepath),
            "valid": False,
            "errors": 1,
            "warnings": 0,
            "findings": [{
                "check": "file_exists",
                "severity": "error",
                "message": f"File not found: {filepath}",
                "suggestion": "Verify the file path",
            }],
        }

    if not content.strip():
        return {
            "file": str(filepath),
            "valid": False,
            "errors": 1,
            "warnings": 0,
            "findings": [{
                "check": "file_empty",
                "severity": "error",
                "message": "Runbook file is empty",
                "suggestion": "Add runbook content or generate using runbook_scaffolder.py",
            }],
        }

    runbook_type = detect_runbook_type(content)
    findings = []
    findings.extend(check_required_sections(content, runbook_type))
    findings.extend(check_quality_patterns(content))
    findings.extend(check_rollback_coverage(content))
    findings.extend(check_step_numbering(content))
    findings.extend(check_placeholder_commands(content))

    errors = sum(1 for f in findings if f["severity"] == "error")
    warnings = sum(1 for f in findings if f["severity"] == "warning")

    return {
        "file": str(filepath),
        "runbook_type": runbook_type,
        "valid": errors == 0,
        "errors": errors,
        "warnings": warnings,
        "findings": findings,
    }


def format_human_output(results):
    """Format validation results for human consumption."""
    lines = []
    total_errors = 0
    total_warnings = 0

    for result in results:
        total_errors += result["errors"]
        total_warnings += result["warnings"]

        status = "PASS" if result["valid"] else "FAIL"
        lines.append(f"\n{'=' * 60}")
        lines.append(f"[{status}] {result['file']}")
        if "runbook_type" in result:
            lines.append(f"  Type: {result['runbook_type']}")
        lines.append(f"  Errors: {result['errors']}  Warnings: {result['warnings']}")
        lines.append(f"{'=' * 60}")

        for finding in result["findings"]:
            icon = "ERROR" if finding["severity"] == "error" else "WARN "
            lines.append(f"  [{icon}] {finding['message']}")
            lines.append(f"         -> {finding['suggestion']}")

        if not result["findings"]:
            lines.append("  All checks passed.")

    lines.append(f"\n{'=' * 60}")
    lines.append(f"SUMMARY: {len(results)} file(s) checked")
    lines.append(f"  Total errors:   {total_errors}")
    lines.append(f"  Total warnings: {total_warnings}")
    passed = sum(1 for r in results if r["valid"])
    lines.append(f"  Passed: {passed}/{len(results)}")
    lines.append(f"{'=' * 60}")

    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Validate runbook markdown files for completeness and quality.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=textwrap.dedent("""\
            Checks performed:
              - Required sections (rollback, escalation, etc.) based on runbook type
              - VERIFY blocks after steps
              - Time estimates in step headings
              - Code blocks with copy-paste commands
              - Last verified date
              - Owner field
              - Hardcoded credentials (flags them as errors)
              - Checklist items
              - Escalation table with L1/L2/L3
              - Monitoring dashboard links
              - Sequential step numbering
              - Placeholder values in commands

            Exit codes:
              0 = all runbooks valid
              1 = one or more errors found
              2 = input error
        """),
    )
    parser.add_argument("files", nargs="*", help="Runbook markdown files to validate")
    parser.add_argument("--dir", "-d", help="Directory containing runbook markdown files")
    parser.add_argument("--json", action="store_true", help="Output results as JSON")
    parser.add_argument("--strict", action="store_true",
                        help="Treat warnings as errors (exit 1 if any warnings)")

    args = parser.parse_args()

    files = list(args.files) if args.files else []

    if args.dir:
        dir_path = Path(args.dir)
        if not dir_path.is_dir():
            print(f"Error: Directory not found — {args.dir}", file=sys.stderr)
            sys.exit(2)
        files.extend(str(p) for p in sorted(dir_path.glob("*.md")))

    if not files:
        parser.print_help()
        print("\nError: No files specified. Provide file paths or use --dir.", file=sys.stderr)
        sys.exit(2)

    results = [validate_runbook(f) for f in files]

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

    has_errors = any(r["errors"] > 0 for r in results)
    has_warnings = any(r["warnings"] > 0 for r in results)

    if has_errors:
        sys.exit(1)
    if args.strict and has_warnings:
        sys.exit(1)
    sys.exit(0)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Staleness Checker — Check runbook freshness against configurable thresholds.

Inspects runbook markdown files for their "Last verified" date and compares
it against configurable staleness thresholds. Also checks whether referenced
config files have been modified more recently than the runbook verification
date. Designed for CI integration to catch stale runbooks before they cause
incident response failures.

Usage:
    python staleness_checker.py docs/runbooks/
    python staleness_checker.py runbook.md --threshold 30
    python staleness_checker.py docs/runbooks/ --repo-root /path/to/repo
    python staleness_checker.py docs/runbooks/ --json
    python staleness_checker.py docs/runbooks/ --config staleness.json
"""

import argparse
import json
import os
import re
import subprocess
import sys
import textwrap
from datetime import date, datetime, timedelta
from pathlib import Path


DEFAULT_THRESHOLDS = {
    "stale_days": 90,
    "warning_days": 60,
    "critical_days": 180,
}

CONFIG_FILE_PATTERN = re.compile(
    r"git\s+log\s+.*?--\s+(\S+)|"
    r"(?:config|file|path)[:=]\s*[`\"]?([a-zA-Z0-9_./-]+\.[a-zA-Z]+)[`\"]?"
)

LAST_VERIFIED_PATTERN = re.compile(
    r"(?i)(?:\*\*)?last\s+verified(?:\*\*)?[:]\s*(\d{4}-\d{2}-\d{2})"
)

LAST_UPDATED_PATTERN = re.compile(
    r"(?i)(?:\*\*)?(?:last\s+updated|updated|date)(?:\*\*)?[:]\s*(\d{4}-\d{2}-\d{2})"
)


def parse_date(date_str):
    """Parse a YYYY-MM-DD date string."""
    try:
        return datetime.strptime(date_str, "%Y-%m-%d").date()
    except (ValueError, TypeError):
        return None


def extract_verified_date(content):
    """Extract the 'Last verified' or 'Last updated' date from runbook content."""
    match = LAST_VERIFIED_PATTERN.search(content)
    if match:
        return parse_date(match.group(1))
    match = LAST_UPDATED_PATTERN.search(content)
    if match:
        return parse_date(match.group(1))
    return None


def extract_referenced_configs(content):
    """Extract config file paths referenced in the runbook."""
    configs = set()
    for match in CONFIG_FILE_PATTERN.finditer(content):
        path = match.group(1) or match.group(2)
        if path:
            path = path.strip("`\"'")
            # Filter out obvious non-file-paths
            if "." in path and not path.startswith("http") and len(path) < 200:
                configs.add(path)
    return sorted(configs)


def get_file_modified_date(filepath):
    """Get the last modification date of a file from the filesystem."""
    try:
        mtime = os.path.getmtime(filepath)
        return datetime.fromtimestamp(mtime).date()
    except (OSError, FileNotFoundError):
        return None


def get_git_modified_date(filepath, repo_root=None):
    """Get the last modification date of a file from git history."""
    try:
        cwd = repo_root if repo_root else os.path.dirname(filepath) or "."
        result = subprocess.run(
            ["git", "log", "-1", "--format=%ci", "--", filepath],
            capture_output=True,
            text=True,
            cwd=cwd,
            timeout=10,
        )
        if result.returncode == 0 and result.stdout.strip():
            date_str = result.stdout.strip().split(" ")[0]
            return parse_date(date_str)
    except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
        pass
    return None


def classify_staleness(days_since, thresholds):
    """Classify the staleness level based on days since last verification."""
    if days_since >= thresholds["critical_days"]:
        return "critical"
    if days_since >= thresholds["stale_days"]:
        return "stale"
    if days_since >= thresholds["warning_days"]:
        return "warning"
    return "fresh"


def check_config_drift(content, repo_root, verified_date):
    """Check if any referenced config files were modified after verification."""
    drifted = []
    configs = extract_referenced_configs(content)

    for config_path in configs:
        full_path = os.path.join(repo_root, config_path) if repo_root else config_path

        mod_date = get_git_modified_date(config_path, repo_root)
        if mod_date is None:
            mod_date = get_file_modified_date(full_path)

        if mod_date is None:
            continue

        if verified_date and mod_date > verified_date:
            drifted.append({
                "config_file": config_path,
                "modified_date": mod_date.isoformat(),
                "verified_date": verified_date.isoformat(),
                "drift_days": (mod_date - verified_date).days,
            })

    return drifted


def check_runbook_staleness(filepath, thresholds, repo_root=None, today=None):
    """Check staleness of a single runbook file."""
    if today is None:
        today = date.today()

    filepath = str(filepath)
    result = {
        "file": filepath,
        "verified_date": None,
        "days_since_verified": None,
        "status": "unknown",
        "config_drift": [],
        "issues": [],
    }

    try:
        with open(filepath, "r") as f:
            content = f.read()
    except FileNotFoundError:
        result["status"] = "error"
        result["issues"].append(f"File not found: {filepath}")
        return result

    if not content.strip():
        result["status"] = "error"
        result["issues"].append("Runbook file is empty")
        return result

    verified_date = extract_verified_date(content)

    if verified_date is None:
        result["status"] = "unknown"
        result["issues"].append(
            "No 'Last verified' or 'Last updated' date found. "
            "Add '**Last verified:** YYYY-MM-DD' to the runbook header."
        )
        # Fall back to git modification date of the runbook itself
        git_date = get_git_modified_date(filepath, repo_root)
        fs_date = get_file_modified_date(filepath)
        fallback = git_date or fs_date
        if fallback:
            verified_date = fallback
            result["issues"].append(
                f"Using file modification date as fallback: {fallback.isoformat()}"
            )

    if verified_date:
        result["verified_date"] = verified_date.isoformat()
        days = (today - verified_date).days
        result["days_since_verified"] = days
        result["status"] = classify_staleness(days, thresholds)

        if result["status"] in ("stale", "critical"):
            result["issues"].append(
                f"Runbook was last verified {days} days ago ({verified_date.isoformat()}). "
                f"Threshold: {thresholds['stale_days']} days."
            )

    # Check config drift
    effective_root = repo_root or os.path.dirname(os.path.abspath(filepath))
    drift = check_config_drift(content, effective_root, verified_date)
    result["config_drift"] = drift
    for d in drift:
        result["issues"].append(
            f"Config drift: {d['config_file']} was modified {d['drift_days']} days "
            f"after the runbook was last verified (modified: {d['modified_date']}, "
            f"verified: {d['verified_date']})"
        )
        if result["status"] == "fresh":
            result["status"] = "drift"

    return result


def load_config(config_path):
    """Load threshold configuration from a JSON file."""
    try:
        with open(config_path, "r") as f:
            data = json.load(f)
        thresholds = dict(DEFAULT_THRESHOLDS)
        for key in DEFAULT_THRESHOLDS:
            if key in data:
                thresholds[key] = int(data[key])
        return thresholds
    except (json.JSONDecodeError, FileNotFoundError, ValueError) as e:
        print(f"Error loading config: {e}", file=sys.stderr)
        sys.exit(2)


def format_human_output(results, thresholds):
    """Format staleness results for human consumption."""
    lines = []
    lines.append(f"Staleness Checker Report — {date.today().isoformat()}")
    lines.append(f"Thresholds: warning={thresholds['warning_days']}d, "
                 f"stale={thresholds['stale_days']}d, "
                 f"critical={thresholds['critical_days']}d")
    lines.append("")

    status_counts = {"fresh": 0, "warning": 0, "stale": 0, "critical": 0,
                     "drift": 0, "unknown": 0, "error": 0}

    for result in results:
        status = result["status"]
        status_counts[status] = status_counts.get(status, 0) + 1

        icon_map = {
            "fresh": "OK   ",
            "warning": "WARN ",
            "stale": "STALE",
            "critical": "CRIT ",
            "drift": "DRIFT",
            "unknown": "?????",
            "error": "ERROR",
        }
        icon = icon_map.get(status, "?????")
        lines.append(f"  [{icon}] {result['file']}")

        if result["verified_date"]:
            days = result["days_since_verified"]
            lines.append(f"         Last verified: {result['verified_date']} ({days} days ago)")

        for issue in result["issues"]:
            lines.append(f"         - {issue}")

        if result["config_drift"]:
            lines.append(f"         Config drift detected in {len(result['config_drift'])} file(s)")

        lines.append("")

    lines.append("=" * 60)
    lines.append("SUMMARY")
    lines.append(f"  Files checked: {len(results)}")
    for status, count in sorted(status_counts.items()):
        if count > 0:
            lines.append(f"  {status.upper():>10}: {count}")
    lines.append("=" * 60)

    stale_count = status_counts["stale"] + status_counts["critical"]
    if stale_count > 0:
        lines.append(f"\n{stale_count} runbook(s) need review.")
    elif status_counts["drift"] > 0:
        lines.append(f"\n{status_counts['drift']} runbook(s) have config drift.")
    else:
        lines.append("\nAll runbooks are up to date.")

    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Check runbook freshness against configurable staleness thresholds.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=textwrap.dedent("""\
            Status levels:
              fresh    — verified within the warning threshold
              warning  — approaching staleness (default: 60+ days)
              stale    — exceeds staleness threshold (default: 90+ days)
              critical — severely stale (default: 180+ days)
              drift    — referenced config files changed after verification
              unknown  — no verification date found in runbook
              error    — file not found or empty

            Config file format (JSON):
              {
                "warning_days": 60,
                "stale_days": 90,
                "critical_days": 180
              }

            Exit codes:
              0 = all runbooks fresh or warning-level
              1 = one or more stale or critical runbooks
              2 = input error
        """),
    )
    parser.add_argument("paths", nargs="*", help="Runbook files or directories to check")
    parser.add_argument("--threshold", "-t", type=int, default=None,
                        help=f"Staleness threshold in days (default: {DEFAULT_THRESHOLDS['stale_days']})")
    parser.add_argument("--warning", "-w", type=int, default=None,
                        help=f"Warning threshold in days (default: {DEFAULT_THRESHOLDS['warning_days']})")
    parser.add_argument("--critical", type=int, default=None,
                        help=f"Critical threshold in days (default: {DEFAULT_THRESHOLDS['critical_days']})")
    parser.add_argument("--config", "-c", help="Path to JSON config file with thresholds")
    parser.add_argument("--repo-root", "-r", help="Repository root for resolving config file paths")
    parser.add_argument("--json", action="store_true", help="Output results as JSON")

    args = parser.parse_args()

    if not args.paths:
        parser.print_help()
        print("\nError: No paths specified.", file=sys.stderr)
        sys.exit(2)

    # Build thresholds
    if args.config:
        thresholds = load_config(args.config)
    else:
        thresholds = dict(DEFAULT_THRESHOLDS)

    if args.threshold is not None:
        thresholds["stale_days"] = args.threshold
    if args.warning is not None:
        thresholds["warning_days"] = args.warning
    if args.critical is not None:
        thresholds["critical_days"] = args.critical

    # Collect files
    files = []
    for path_str in args.paths:
        p = Path(path_str)
        if p.is_dir():
            files.extend(sorted(p.glob("**/*.md")))
        elif p.is_file():
            files.append(p)
        else:
            print(f"Warning: Path not found — {path_str}", file=sys.stderr)

    if not files:
        print("Error: No markdown files found.", file=sys.stderr)
        sys.exit(2)

    # Check each file
    results = [
        check_runbook_staleness(f, thresholds, repo_root=args.repo_root)
        for f in files
    ]

    # Output
    if args.json:
        print(json.dumps(results, indent=2))
    else:
        print(format_human_output(results, thresholds))

    # Exit code
    has_stale = any(r["status"] in ("stale", "critical") for r in results)
    sys.exit(1 if has_stale else 0)


if __name__ == "__main__":
    main()