Python: HTTP health check + alert (JSON output + exit codes)

#!/usr/bin/env python3
"""
HTTP Health Check Script
A reusable monitoring tool for websites and APIs with Discord alerting.

This script performs comprehensive HTTP health checks with configurable validation,
retry logic, and optional Discord webhook notifications. Designed for use in
monitoring pipelines, cron jobs, and operational automation.

Author: ScriptedOps
License: MIT
"""

import argparse
import json
import sys
import time
from datetime import datetime, timezone
from typing import Dict, List, Optional, Tuple, Any

try:
    import requests
except ImportError:
    print("Error: requests library is required. Install with: pip install requests", file=sys.stderr)
    sys.exit(1)


class HealthCheckResult:
    """Container for health check results and validation outcomes."""

    def __init__(self, url: str, method: str):
        self.url = url
        self.method = method
        self.status_code: Optional[int] = None
        self.response_time_ms: Optional[float] = None
        self.final_status = "unknown"
        self.checks_passed: List[str] = []
        self.failure_reasons: List[str] = []
        self.retry_count = 0
        self.timestamp = datetime.now(timezone.utc).isoformat()

    def add_check_passed(self, check: str):
        """Record a successful validation check."""
        self.checks_passed.append(check)

    def add_failure_reason(self, reason: str):
        """Record a failure reason."""
        self.failure_reasons.append(reason)

    def is_healthy(self) -> bool:
        """Return True if no failures occurred."""
        return len(self.failure_reasons) == 0

    def has_retryable_failure(self) -> bool:
        """Return True if the failure might be resolved by retrying."""
        if not self.failure_reasons:
            return False

        # Response time failures are not retryable - a slow server will still be slow
        non_retryable_indicators = ["exceeds threshold"]

        for reason in self.failure_reasons:
            for indicator in non_retryable_indicators:
                if indicator in reason:
                    return False

        # Other failures (network, status code, validation) are retryable
        return True

    def to_dict(self) -> Dict[str, Any]:
        """Convert result to dictionary for JSON output."""
        return {
            "url": self.url,
            "method": self.method,
            "final_status": self.final_status,
            "status_code": self.status_code,
            "response_time_ms": self.response_time_ms,
            "checks_passed": self.checks_passed,
            "failure_reasons": self.failure_reasons,
            "retry_count": self.retry_count,
            "timestamp": self.timestamp
        }


def parse_headers(header_args: List[str]) -> Dict[str, str]:
    """Parse repeatable --header arguments into a dictionary."""
    headers = {}
    for header in header_args:
        if ':' not in header:
            print(f"Warning: Invalid header format '{header}'. Expected 'Key: Value'", file=sys.stderr)
            continue
        key, value = header.split(':', 1)
        headers[key.strip()] = value.strip()
    return headers


def perform_http_request(url: str, method: str, timeout: int, headers: Dict[str, str]) -> Tuple[requests.Response, float]:
    """
    Perform HTTP request and measure response time.

    Returns:
        Tuple of (response object, response_time_ms)

    Raises:
        requests.RequestException: For any request failures
    """
    start_time = time.time()

    response = requests.request(
        method=method,
        url=url,
        headers=headers,
        timeout=timeout,
        allow_redirects=True
    )

    end_time = time.time()
    response_time_ms = (end_time - start_time) * 1000

    return response, response_time_ms


def validate_status_code(response: requests.Response, expected_status: int, result: HealthCheckResult):
    """Validate HTTP status code matches expected value."""
    if response.status_code == expected_status:
        result.add_check_passed(f"status_code={expected_status}")
    else:
        result.add_failure_reason(f"Expected status {expected_status}, got {response.status_code}")


def validate_contains_text(response: requests.Response, contains_text: str, result: HealthCheckResult):
    """Validate response body contains specified text."""
    try:
        response_text = response.text
        if contains_text in response_text:
            result.add_check_passed(f"contains_text='{contains_text}'")
        else:
            result.add_failure_reason(f"Response body does not contain '{contains_text}'")
    except Exception as e:
        result.add_failure_reason(f"Failed to read response text: {str(e)}")


def validate_json_key(response: requests.Response, json_key: str, result: HealthCheckResult):
    """Validate JSON response contains specified key."""
    try:
        json_data = response.json()
        if json_key in json_data:
            result.add_check_passed(f"json_key_exists='{json_key}'")
        else:
            result.add_failure_reason(f"JSON key '{json_key}' not found in response")
    except json.JSONDecodeError:
        result.add_failure_reason("Response is not valid JSON")
    except Exception as e:
        result.add_failure_reason(f"Failed to parse JSON: {str(e)}")


def validate_json_value(response: requests.Response, json_key: str, expected_value: str, result: HealthCheckResult):
    """Validate JSON response key equals expected value."""
    try:
        json_data = response.json()
        if json_key not in json_data:
            result.add_failure_reason(f"JSON key '{json_key}' not found in response")
            return

        actual_value = str(json_data[json_key])
        if actual_value == expected_value:
            result.add_check_passed(f"json_value '{json_key}'='{expected_value}'")
        else:
            result.add_failure_reason(f"JSON key '{json_key}' expected '{expected_value}', got '{actual_value}'")
    except json.JSONDecodeError:
        result.add_failure_reason("Response is not valid JSON")
    except Exception as e:
        result.add_failure_reason(f"Failed to parse JSON: {str(e)}")


def check_response_time(response_time_ms: float, max_response_ms: Optional[int], result: HealthCheckResult):
    """Check if response time exceeds threshold and mark as degraded if so."""
    if max_response_ms is not None:
        if response_time_ms <= max_response_ms:
            result.add_check_passed(f"response_time<={max_response_ms}ms")
        else:
            # This is degraded, not unhealthy - site works but is slow
            result.add_failure_reason(f"Response time {response_time_ms:.1f}ms exceeds threshold {max_response_ms}ms")


def send_discord_alert(webhook_url: str, result: HealthCheckResult):
    """Send Discord webhook alert for unhealthy status."""
    if result.final_status != "unhealthy":
        return  # Only alert on unhealthy, not degraded

    failure_summary = "; ".join(result.failure_reasons[:3])  # Limit to avoid message length issues
    if len(result.failure_reasons) > 3:
        failure_summary += f" (and {len(result.failure_reasons) - 3} more)"

    embed = {
        "title": "🚨 Health Check Failed",
        "color": 0xFF0000,  # Red
        "fields": [
            {"name": "URL", "value": result.url, "inline": True},
            {"name": "Status Code", "value": str(result.status_code) if result.status_code else "N/A", "inline": True},
            {"name": "Response Time", "value": f"{result.response_time_ms:.1f}ms" if result.response_time_ms else "N/A", "inline": True},
            {"name": "Failure Reason", "value": failure_summary, "inline": False}
        ],
        "timestamp": result.timestamp,
        "footer": {"text": "HTTP Health Check"}
    }

    payload = {
        "embeds": [embed]
    }

    try:
        response = requests.post(webhook_url, json=payload, timeout=10)
        response.raise_for_status()
    except requests.RequestException as e:
        print(f"Warning: Failed to send Discord alert: {str(e)}", file=sys.stderr)


def perform_health_check_attempt(args: argparse.Namespace) -> HealthCheckResult:
    """Perform a single health check attempt with all validations."""
    result = HealthCheckResult(args.url, args.method.upper())

    try:
        headers = parse_headers(args.header or [])
        response, response_time_ms = perform_http_request(args.url, args.method, args.timeout, headers)

        result.status_code = response.status_code
        result.response_time_ms = response_time_ms

        # Perform all validations
        validate_status_code(response, args.expected_status, result)

        if args.contains_text:
            validate_contains_text(response, args.contains_text, result)

        if args.expect_json_key:
            validate_json_key(response, args.expect_json_key, result)

        if args.expect_json_value and args.expect_json_key:
            validate_json_value(response, args.expect_json_key, args.expect_json_value, result)

        # Check response time (can mark as degraded)
        check_response_time(response_time_ms, args.max_response_ms, result)

    except requests.exceptions.Timeout:
        result.add_failure_reason(f"Request timed out after {args.timeout} seconds")
    except requests.exceptions.ConnectionError:
        result.add_failure_reason("Connection error - could not reach endpoint")
    except requests.RequestException as e:
        result.add_failure_reason(f"Request failed: {str(e)}")
    except Exception as e:
        result.add_failure_reason(f"Unexpected error: {str(e)}")

    return result


def determine_final_status(result: HealthCheckResult, max_response_ms: Optional[int]) -> str:
    """
    Determine final health status based on validation results.

    Returns:
        - "healthy": All checks passed
        - "degraded": Checks passed but response time exceeded threshold
        - "unhealthy": One or more checks failed
    """
    if not result.failure_reasons:
        return "healthy"

    # If only failure is response time, mark as degraded
    if (len(result.failure_reasons) == 1 and
        max_response_ms is not None and
        "exceeds threshold" in result.failure_reasons[0]):
        return "degraded"

    return "unhealthy"


def main():
    """Main execution function."""
    parser = argparse.ArgumentParser(
        description="HTTP Health Check Script for monitoring websites and APIs",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s --url https://api.example.com/health
  %(prog)s --url https://example.com --expected-status 200 --contains-text "Welcome"
  %(prog)s --url https://api.example.com/status --expect-json-key "status" --expect-json-value "ok"
  %(prog)s --url https://example.com --max-response-ms 1000 --discord-webhook-url https://discord.com/api/webhooks/...
        """
    )

    # Required arguments
    parser.add_argument("--url", required=True, help="URL to check")

    # HTTP options
    parser.add_argument("--method", default="GET", help="HTTP method (default: GET)")
    parser.add_argument("--timeout", type=int, default=30, help="Request timeout in seconds (default: 30)")
    parser.add_argument("--header", action="append", help="Custom header (format: 'Key: Value'), can be used multiple times")

    # Validation options
    parser.add_argument("--expected-status", type=int, default=200, help="Expected HTTP status code (default: 200)")
    parser.add_argument("--contains-text", help="Check if response body contains this text")
    parser.add_argument("--expect-json-key", help="Check if JSON response contains this key")
    parser.add_argument("--expect-json-value", help="Expected value for JSON key (requires --expect-json-key)")
    parser.add_argument("--max-response-ms", type=int, help="Maximum response time in milliseconds (marks as degraded if exceeded)")

    # Retry options
    parser.add_argument("--retries", type=int, default=1, help="Number of retry attempts (default: 1)")
    parser.add_argument("--retry-delay", type=int, default=2, help="Delay between retries in seconds (default: 2)")

    # Output options
    parser.add_argument("--output", choices=["json", "text"], default="json", help="Output format (default: json)")

    # Alerting options
    parser.add_argument("--discord-webhook-url", help="Discord webhook URL for failure alerts")

    args = parser.parse_args()

    # Validate argument combinations
    if args.expect_json_value and not args.expect_json_key:
        print("Error: --expect-json-value requires --expect-json-key", file=sys.stderr)
        sys.exit(1)

    # Perform health check with retries
    last_result = None

    for attempt in range(args.retries + 1):  # +1 because we want retries + initial attempt
        result = perform_health_check_attempt(args)
        result.retry_count = attempt
        last_result = result

        # If successful, no retryable failures, or this is our last attempt, break
        if result.is_healthy() or not result.has_retryable_failure() or attempt == args.retries:
            break

        # Wait before retry (except on last attempt)
        if attempt < args.retries:
            time.sleep(args.retry_delay)

    # Determine final status
    final_status = determine_final_status(last_result, args.max_response_ms)
    last_result.final_status = final_status

    # Send Discord alert if configured and unhealthy
    if args.discord_webhook_url and final_status == "unhealthy":
        send_discord_alert(args.discord_webhook_url, last_result)

    # Output results
    if args.output == "json":
        print(json.dumps(last_result.to_dict(), indent=2))
    else:
        # Text output for human readability
        status_emoji = {"healthy": "✅", "degraded": "⚠️", "unhealthy": "❌"}
        print(f"{status_emoji.get(final_status, '?')} {final_status.upper()}: {last_result.url}")
        print(f"Status Code: {last_result.status_code}")
        print(f"Response Time: {last_result.response_time_ms:.1f}ms" if last_result.response_time_ms else "Response Time: N/A")

        if last_result.checks_passed:
            print("Checks Passed:")
            for check in last_result.checks_passed:
                print(f"  ✓ {check}")

        if last_result.failure_reasons:
            print("Failures:")
            for failure in last_result.failure_reasons:
                print(f"  ✗ {failure}")

        if last_result.retry_count > 0:
            print(f"Retries Used: {last_result.retry_count}")

    # Exit with appropriate code
    exit_codes = {"healthy": 0, "degraded": 2, "unhealthy": 1}
    sys.exit(exit_codes[final_status])


if __name__ == "__main__":
    main()