Web Scraping for DevOps

1import requests
2from requests.adapters import HTTPAdapter
3from requests.packages.urllib3.util.retry import Retry
4from typing import Dict, Optional, List
5import time
6import logging
7from datetime import datetime
8
9logging.basicConfig(level=logging.INFO)
10logger = logging.getLogger(__name__)
11
12
13class ServiceHealthChecker:
14    """
15    Production-grade HTTP health checker with retry logic and timeout handling
16    """
17
18    def __init__(
19        self,
20        timeout: int = 10,
21        max_retries: int = 3,
22        backoff_factor: float = 0.3
23    ):
24        """
25        Initialize health checker with retry strategy
26
27        Args:
28            timeout: Request timeout in seconds
29            max_retries: Maximum retry attempts
30            backoff_factor: Exponential backoff multiplier
31        """
32        self.timeout = timeout
33
34        # Configure retry strategy
35        retry_strategy = Retry(
36            total=max_retries,
37            backoff_factor=backoff_factor,
38            status_forcelist=[429, 500, 502, 503, 504],
39            allowed_methods=["HEAD", "GET", "OPTIONS"]
40        )
41
42        # Create session with retry adapter
43        adapter = HTTPAdapter(max_retries=retry_strategy)
44        self.session = requests.Session()
45        self.session.mount("http://", adapter)
46        self.session.mount("https://", adapter)
47
48        logger.info(f"Health checker initialized: {max_retries} retries, {timeout}s timeout")
49
50    def check_endpoint(
51        self,
52        url: str,
53        method: str = 'GET',
54        expected_status: int = 200,
55        verify_ssl: bool = True
56    ) -> Dict:
57        """
58        Check endpoint health
59
60        Args:
61            url: Endpoint URL to check
62            method: HTTP method (GET, POST, HEAD)
63            expected_status: Expected HTTP status code
64            verify_ssl: Whether to verify SSL certificates
65
66        Returns:
67            Health check result dictionary
68        """
69        start_time = time.time()
70        result = {
71            'url': url,
72            'timestamp': datetime.now().isoformat(),
73            'status': 'unknown',
74            'response_time_ms': 0,
75            'status_code': None,
76            'error': None
77        }
78
79        try:
80            logger.info(f"Checking endpoint: {url}")
81
82            # Make HTTP request
83            response = self.session.request(
84                method=method,
85                url=url,
86                timeout=self.timeout,
87                verify=verify_ssl
88            )
89
90            # Calculate response time
91            response_time = (time.time() - start_time) * 1000
92            result['response_time_ms'] = round(response_time, 2)
93            result['status_code'] = response.status_code
94
95            # Check if status matches expected
96            if response.status_code == expected_status:
97                result['status'] = 'healthy'
98                logger.info(f"✓ Endpoint healthy: {url} ({response_time:.0f}ms)")
99            else:
100                result['status'] = 'unhealthy'
101                result['error'] = f"Unexpected status code: {response.status_code}"
102                logger.warning(f"✗ Endpoint unhealthy: {url} - {result['error']}")
103
104            # Extract response headers
105            result['headers'] = dict(response.headers)
106
107        except requests.exceptions.Timeout:
108            result['status'] = 'timeout'
109            result['error'] = f"Request timeout after {self.timeout}s"
110            result['response_time_ms'] = self.timeout * 1000
111            logger.error(f"✗ Timeout: {url}")
112
113        except requests.exceptions.ConnectionError as e:
114            result['status'] = 'connection_error'
115            result['error'] = str(e)
116            logger.error(f"✗ Connection error: {url}")
117
118        except requests.exceptions.SSLError as e:
119            result['status'] = 'ssl_error'
120            result['error'] = str(e)
121            logger.error(f"✗ SSL error: {url}")
122
123        except Exception as e:
124            result['status'] = 'error'
125            result['error'] = str(e)
126            logger.error(f"✗ Unexpected error: {url} - {e}")
127
128        return result
129
130    def check_multiple_endpoints(
131        self,
132        endpoints: List[Dict]
133    ) -> List[Dict]:
134        """
135        Check multiple endpoints sequentially
136
137        Args:
138            endpoints: List of endpoint configurations
139                      [{url, method, expected_status}, ...]
140
141        Returns:
142            List of health check results
143        """
144        results = []
145
146        logger.info(f"Checking {len(endpoints)} endpoints...")
147
148        for endpoint_config in endpoints:
149            url = endpoint_config.get('url')
150            method = endpoint_config.get('method', 'GET')
151            expected_status = endpoint_config.get('expected_status', 200)
152
153            result = self.check_endpoint(
154                url=url,
155                method=method,
156                expected_status=expected_status
157            )
158
159            # Add endpoint name if provided
160            if 'name' in endpoint_config:
161                result['name'] = endpoint_config['name']
162
163            results.append(result)
164
165        return results
166
167    def close(self):
168        """Close session and cleanup resources"""
169        self.session.close()
170        logger.info("Session closed")
171
172
173# Usage example
174health_checker = ServiceHealthChecker(
175    timeout=10,
176    max_retries=3,
177    backoff_factor=0.5
178)
179
180# Check single endpoint
181result = health_checker.check_endpoint(
182    url='https://api.github.com/status',
183    method='GET',
184    expected_status=200
185)
186
187print(f"Status: {result['status']}")
188print(f"Response Time: {result['response_time_ms']}ms")
189print(f"Status Code: {result['status_code']}")
190
191# Check multiple endpoints
192endpoints = [
193    {
194        'name': 'GitHub API',
195        'url': 'https://api.github.com/status',
196        'method': 'GET',
197        'expected_status': 200
198    },
199    {
200        'name': 'Google DNS',
201        'url': 'https://dns.google',
202        'method': 'GET',
203        'expected_status': 200
204    },
205    {
206        'name': 'Cloudflare',
207        'url': 'https://www.cloudflare.com',
208        'method': 'GET',
209        'expected_status': 200
210    }
211]
212
213results = health_checker.check_multiple_endpoints(endpoints)
214
215# Print summary
216healthy_count = sum(1 for r in results if r['status'] == 'healthy')
217print(f"\nHealth Summary: {healthy_count}/{len(results)} endpoints healthy")
218
219for result in results:
220    status_symbol = "✓" if result['status'] == 'healthy' else "✗"
221    print(f"{status_symbol} {result.get('name', result['url'])}: "
222          f"{result['status']} ({result['response_time_ms']}ms)")
223
224# Cleanup
225health_checker.close()

1import requests
2import json
3from typing import Dict, Optional
4from datetime import datetime, timedelta
5import hashlib
6
7class AdvancedServiceMonitor:
8    """
9    Advanced service monitoring with authentication and content validation
10    """
11
12    def __init__(self, base_url: str):
13        """
14        Initialize monitor
15
16        Args:
17            base_url: Base URL for service
18        """
19        self.base_url = base_url.rstrip('/')
20        self.session = requests.Session()
21
22    def monitor_with_auth(
23        self,
24        endpoint: str,
25        auth_token: str,
26        auth_type: str = 'bearer'
27    ) -> Dict:
28        """
29        Monitor endpoint with authentication
30
31        Args:
32            endpoint: API endpoint path
33            auth_token: Authentication token
34            auth_type: Authentication type (bearer, basic, api_key)
35
36        Returns:
37            Monitoring result
38        """
39        url = f"{self.base_url}{endpoint}"
40        headers = {}
41
42        # Set authentication header
43        if auth_type == 'bearer':
44            headers['Authorization'] = f"Bearer {auth_token}"
45        elif auth_type == 'api_key':
46            headers['X-API-Key'] = auth_token
47
48        try:
49            response = self.session.get(url, headers=headers, timeout=10)
50
51            return {
52                'url': url,
53                'status_code': response.status_code,
54                'response_time': response.elapsed.total_seconds(),
55                'content_length': len(response.content),
56                'authenticated': response.status_code != 401,
57                'timestamp': datetime.now().isoformat()
58            }
59
60        except Exception as e:
61            return {
62                'url': url,
63                'error': str(e),
64                'timestamp': datetime.now().isoformat()
65            }
66
67    def validate_response_content(
68        self,
69        endpoint: str,
70        expected_keys: List[str],
71        expected_values: Optional[Dict] = None
72    ) -> Dict:
73        """
74        Validate API response structure and content
75
76        Args:
77            endpoint: API endpoint path
78            expected_keys: Keys that should exist in response
79            expected_values: Optional dict of key-value pairs to validate
80
81        Returns:
82            Validation result
83        """
84        url = f"{self.base_url}{endpoint}"
85
86        try:
87            response = self.session.get(url, timeout=10)
88            response.raise_for_status()
89
90            data = response.json()
91
92            # Validate keys exist
93            missing_keys = [key for key in expected_keys if key not in data]
94            has_all_keys = len(missing_keys) == 0
95
96            # Validate specific values if provided
97            value_matches = {}
98            if expected_values:
99                for key, expected_value in expected_values.items():
100                    actual_value = data.get(key)
101                    value_matches[key] = actual_value == expected_value
102
103            return {
104                'url': url,
105                'valid': has_all_keys and all(value_matches.values()) if expected_values else has_all_keys,
106                'missing_keys': missing_keys,
107                'value_matches': value_matches,
108                'timestamp': datetime.now().isoformat()
109            }
110
111        except Exception as e:
112            return {
113                'url': url,
114                'valid': False,
115                'error': str(e),
116                'timestamp': datetime.now().isoformat()
117            }
118
119    def check_content_freshness(
120        self,
121        endpoint: str,
122        max_age_seconds: int = 300
123    ) -> Dict:
124        """
125        Check if content has been recently updated
126
127        Args:
128            endpoint: API endpoint path
129            max_age_seconds: Maximum acceptable age in seconds
130
131        Returns:
132            Freshness check result
133        """
134        url = f"{self.base_url}{endpoint}"
135
136        try:
137            response = self.session.get(url, timeout=10)
138            response.raise_for_status()
139
140            # Check Last-Modified header
141            last_modified = response.headers.get('Last-Modified')
142            etag = response.headers.get('ETag')
143
144            result = {
145                'url': url,
146                'last_modified': last_modified,
147                'etag': etag,
148                'fresh': True,
149                'timestamp': datetime.now().isoformat()
150            }
151
152            if last_modified:
153                # Parse Last-Modified header
154                from email.utils import parsedate_to_datetime
155                modified_date = parsedate_to_datetime(last_modified)
156                age_seconds = (datetime.now(modified_date.tzinfo) - modified_date).total_seconds()
157
158                result['age_seconds'] = age_seconds
159                result['fresh'] = age_seconds <= max_age_seconds
160
161            return result
162
163        except Exception as e:
164            return {
165                'url': url,
166                'fresh': False,
167                'error': str(e),
168                'timestamp': datetime.now().isoformat()
169            }
170
171    def calculate_response_hash(
172        self,
173        endpoint: str
174    ) -> Dict:
175        """
176        Calculate hash of response content for change detection
177
178        Args:
179            endpoint: API endpoint path
180
181        Returns:
182            Hash result
183        """
184        url = f"{self.base_url}{endpoint}"
185
186        try:
187            response = self.session.get(url, timeout=10)
188            response.raise_for_status()
189
190            # Calculate SHA256 hash
191            content_hash = hashlib.sha256(response.content).hexdigest()
192
193            return {
194                'url': url,
195                'hash': content_hash,
196                'content_length': len(response.content),
197                'timestamp': datetime.now().isoformat()
198            }
199
200        except Exception as e:
201            return {
202                'url': url,
203                'error': str(e),
204                'timestamp': datetime.now().isoformat()
205            }
206
207
208# Usage example
209monitor = AdvancedServiceMonitor(base_url='https://api.github.com')
210
211# Monitor with authentication
212auth_result = monitor.monitor_with_auth(
213    endpoint='/user',
214    auth_token='your_token_here',
215    auth_type='bearer'
216)
217
218# Validate response structure
219validation_result = monitor.validate_response_content(
220    endpoint='/status',
221    expected_keys=['status', 'message'],
222    expected_values={'status': 'ok'}
223)
224
225# Check content freshness
226freshness_result = monitor.check_content_freshness(
227    endpoint='/status',
228    max_age_seconds=300
229)
230
231# Calculate content hash for change detection
232hash_result = monitor.calculate_response_hash(endpoint='/status')

1import requests
2from bs4 import BeautifulSoup
3from typing import Dict, List, Optional
4import logging
5from datetime import datetime
6import re
7
8logger = logging.getLogger(__name__)
9
10
11class WebPageScraper:
12    """
13    Production-grade web scraper with BeautifulSoup4
14    """
15
16    def __init__(self, user_agent: Optional[str] = None):
17        """
18        Initialize scraper
19
20        Args:
21            user_agent: Custom User-Agent string
22        """
23        self.session = requests.Session()
24
25        # Set default User-Agent
26        if user_agent:
27            self.session.headers['User-Agent'] = user_agent
28        else:
29            self.session.headers['User-Agent'] = (
30                'Mozilla/5.0 (compatible; ServiceMonitor/1.0; +https://example.com/bot)'
31            )
32
33        logger.info("Web scraper initialized")
34
35    def fetch_page(self, url: str, parser: str = 'html.parser') -> Optional[BeautifulSoup]:
36        """
37        Fetch and parse web page
38
39        Args:
40            url: Page URL
41            parser: HTML parser ('html.parser', 'lxml', 'html5lib')
42
43        Returns:
44            BeautifulSoup object or None on error
45        """
46        try:
47            logger.info(f"Fetching page: {url}")
48
49            response = self.session.get(url, timeout=10)
50            response.raise_for_status()
51
52            # Parse HTML
53            soup = BeautifulSoup(response.content, parser)
54            logger.info(f"Page parsed successfully: {url}")
55
56            return soup
57
58        except Exception as e:
59            logger.error(f"Failed to fetch page: {url} - {e}")
60            return None
61
62    def extract_status_info(self, soup: BeautifulSoup) -> Dict:
63        """
64        Extract service status information from page
65
66        Args:
67            soup: BeautifulSoup object
68
69        Returns:
70            Status information dictionary
71        """
72        status_info = {
73            'timestamp': datetime.now().isoformat(),
74            'status': 'unknown',
75            'incidents': [],
76            'components': []
77        }
78
79        try:
80            # Find status indicators (common patterns)
81            status_element = soup.find(['div', 'span', 'p'], class_=re.compile(r'status|state'))
82
83            if status_element:
84                status_text = status_element.get_text(strip=True).lower()
85
86                # Determine status from text
87                if 'operational' in status_text or 'healthy' in status_text:
88                    status_info['status'] = 'operational'
89                elif 'degraded' in status_text or 'issues' in status_text:
90                    status_info['status'] = 'degraded'
91                elif 'outage' in status_text or 'down' in status_text:
92                    status_info['status'] = 'outage'
93
94            # Extract incidents
95            incident_elements = soup.find_all(['div', 'article'], class_=re.compile(r'incident'))
96
97            for incident in incident_elements[:5]:  # Limit to 5 most recent
98                title_elem = incident.find(['h2', 'h3', 'h4'])
99                time_elem = incident.find('time')
100
101                if title_elem:
102                    status_info['incidents'].append({
103                        'title': title_elem.get_text(strip=True),
104                        'time': time_elem.get('datetime') if time_elem else None
105                    })
106
107            # Extract component statuses
108            component_elements = soup.find_all(['div', 'li'], class_=re.compile(r'component'))
109
110            for component in component_elements[:10]:  # Limit to 10
111                name_elem = component.find(['span', 'p'], class_=re.compile(r'name'))
112                status_elem = component.find(['span', 'p'], class_=re.compile(r'status'))
113
114                if name_elem and status_elem:
115                    status_info['components'].append({
116                        'name': name_elem.get_text(strip=True),
117                        'status': status_elem.get_text(strip=True)
118                    })
119
120            logger.info(f"Extracted status: {status_info['status']}")
121
122        except Exception as e:
123            logger.error(f"Failed to extract status info: {e}")
124            status_info['error'] = str(e)
125
126        return status_info
127
128    def extract_metrics(self, soup: BeautifulSoup) -> Dict:
129        """
130        Extract metrics and statistics from page
131
132        Args:
133            soup: BeautifulSoup object
134
135        Returns:
136            Metrics dictionary
137        """
138        metrics = {
139            'timestamp': datetime.now().isoformat(),
140            'values': {}
141        }
142
143        try:
144            # Find metric elements (common patterns)
145            metric_elements = soup.find_all(['div', 'span'], class_=re.compile(r'metric|stat'))
146
147            for element in metric_elements:
148                # Try to find label and value
149                label_elem = element.find(['span', 'label'], class_=re.compile(r'label|name'))
150                value_elem = element.find(['span', 'strong'], class_=re.compile(r'value|number'))
151
152                if label_elem and value_elem:
153                    label = label_elem.get_text(strip=True)
154                    value = value_elem.get_text(strip=True)
155
156                    # Try to parse numeric value
157                    try:
158                        # Remove common non-numeric characters
159                        numeric_value = re.sub(r'[^0-9.-]', '', value)
160                        metrics['values'][label] = float(numeric_value)
161                    except:
162                        metrics['values'][label] = value
163
164            logger.info(f"Extracted {len(metrics['values'])} metrics")
165
166        except Exception as e:
167            logger.error(f"Failed to extract metrics: {e}")
168            metrics['error'] = str(e)
169
170        return metrics
171
172    def extract_links(
173        self,
174        soup: BeautifulSoup,
175        base_url: str,
176        filter_pattern: Optional[str] = None
177    ) -> List[str]:
178        """
179        Extract all links from page
180
181        Args:
182            soup: BeautifulSoup object
183            base_url: Base URL for resolving relative links
184            filter_pattern: Optional regex pattern to filter links
185
186        Returns:
187            List of URLs
188        """
189        links = []
190
191        try:
192            # Find all anchor tags
193            anchor_tags = soup.find_all('a', href=True)
194
195            for tag in anchor_tags:
196                href = tag['href']
197
198                # Resolve relative URLs
199                if href.startswith('/'):
200                    href = base_url.rstrip('/') + href
201                elif not href.startswith('http'):
202                    continue
203
204                # Apply filter if provided
205                if filter_pattern and not re.search(filter_pattern, href):
206                    continue
207
208                links.append(href)
209
210            # Remove duplicates
211            links = list(set(links))
212            logger.info(f"Extracted {len(links)} unique links")
213
214        except Exception as e:
215            logger.error(f"Failed to extract links: {e}")
216
217        return links
218
219    def extract_table_data(
220        self,
221        soup: BeautifulSoup,
222        table_id: Optional[str] = None
223    ) -> List[Dict]:
224        """
225        Extract data from HTML table
226
227        Args:
228            soup: BeautifulSoup object
229            table_id: Optional table ID to target specific table
230
231        Returns:
232            List of row dictionaries
233        """
234        table_data = []
235
236        try:
237            # Find table
238            if table_id:
239                table = soup.find('table', id=table_id)
240            else:
241                table = soup.find('table')
242
243            if not table:
244                logger.warning("No table found")
245                return table_data
246
247            # Extract headers
248            headers = []
249            header_row = table.find('thead')
250            if header_row:
251                headers = [th.get_text(strip=True) for th in header_row.find_all('th')]
252
253            # Extract rows
254            body = table.find('tbody') or table
255            rows = body.find_all('tr')
256
257            for row in rows:
258                cells = row.find_all(['td', 'th'])
259                if not cells:
260                    continue
261
262                row_data = {}
263                for i, cell in enumerate(cells):
264                    key = headers[i] if i < len(headers) else f'column_{i}'
265                    row_data[key] = cell.get_text(strip=True)
266
267                table_data.append(row_data)
268
269            logger.info(f"Extracted {len(table_data)} rows from table")
270
271        except Exception as e:
272            logger.error(f"Failed to extract table data: {e}")
273
274        return table_data
275
276
277# Usage example
278scraper = WebPageScraper(
279    user_agent='ServiceMonitor/1.0 (DevOps; [email protected])'
280)
281
282# Fetch and parse status page
283soup = scraper.fetch_page('https://www.githubstatus.com')
284
285if soup:
286    # Extract status information
287    status_info = scraper.extract_status_info(soup)
288    print(f"Service Status: {status_info['status']}")
289    print(f"Active Incidents: {len(status_info['incidents'])}")
290
291    # Extract metrics
292    metrics = scraper.extract_metrics(soup)
293    print(f"Metrics Collected: {len(metrics['values'])}")
294
295    # Extract links
296    links = scraper.extract_links(
297        soup,
298        base_url='https://www.githubstatus.com',
299        filter_pattern=r'/incidents/'
300    )
301    print(f"Incident Links Found: {len(links)}")
302
303    # Extract table data
304    table_data = scraper.extract_table_data(soup)
305    print(f"Table Rows: {len(table_data)}")

1from bs4 import BeautifulSoup, Comment
2import json
3from typing import Dict, Any
4
5class AdvancedHTMLParser:
6    """
7    Advanced HTML parsing techniques
8    """
9
10    @staticmethod
11    def extract_json_ld(soup: BeautifulSoup) -> List[Dict]:
12        """
13        Extract JSON-LD structured data
14
15        Args:
16            soup: BeautifulSoup object
17
18        Returns:
19            List of JSON-LD objects
20        """
21        json_ld_data = []
22
23        # Find all JSON-LD script tags
24        scripts = soup.find_all('script', type='application/ld+json')
25
26        for script in scripts:
27            try:
28                data = json.loads(script.string)
29                json_ld_data.append(data)
30            except:
31                pass
32
33        return json_ld_data
34
35    @staticmethod
36    def extract_meta_tags(soup: BeautifulSoup) -> Dict[str, str]:
37        """
38        Extract meta tags
39
40        Args:
41            soup: BeautifulSoup object
42
43        Returns:
44            Dictionary of meta tag content
45        """
46        meta_data = {}
47
48        # Extract meta tags
49        for meta in soup.find_all('meta'):
50            name = meta.get('name') or meta.get('property')
51            content = meta.get('content')
52
53            if name and content:
54                meta_data[name] = content
55
56        return meta_data
57
58    @staticmethod
59    def extract_comments(soup: BeautifulSoup) -> List[str]:
60        """
61        Extract HTML comments (sometimes contain useful debug info)
62
63        Args:
64            soup: BeautifulSoup object
65
66        Returns:
67            List of comment strings
68        """
69        comments = soup.find_all(string=lambda text: isinstance(text, Comment))
70        return [comment.strip() for comment in comments]
71
72    @staticmethod
73    def extract_embedded_json(soup: BeautifulSoup) -> List[Dict]:
74        """
75        Extract JSON embedded in script tags
76
77        Args:
78            soup: BeautifulSoup object
79
80        Returns:
81            List of JSON objects
82        """
83        json_objects = []
84
85        scripts = soup.find_all('script')
86
87        for script in scripts:
88            if not script.string:
89                continue
90
91            # Try to find JSON patterns
92            import re
93            json_pattern = r'\{[^{}]*\}'
94
95            matches = re.finditer(json_pattern, script.string)
96
97            for match in matches:
98                try:
99                    data = json.loads(match.group())
100                    json_objects.append(data)
101                except:
102                    pass
103
104        return json_objects
105
106
107# Usage
108parser = AdvancedHTMLParser()
109
110soup = BeautifulSoup(html_content, 'html.parser')
111
112# Extract JSON-LD
113json_ld = parser.extract_json_ld(soup)
114
115# Extract meta tags
116meta_tags = parser.extract_meta_tags(soup)
117
118# Extract comments
119comments = parser.extract_comments(soup)
120
121# Extract embedded JSON
122embedded_json = parser.extract_embedded_json(soup)

1import time
2import json
3from pathlib import Path
4from datetime import datetime, timedelta
5from typing import Dict, List, Optional
6from collections import deque
7import statistics
8import logging
9
10logger = logging.getLogger(__name__)
11
12
13class UptimeTracker:
14    """
15    Production-grade uptime tracking with historical data
16    """
17
18    def __init__(
19        self,
20        service_name: str,
21        data_dir: Path = Path('./monitoring_data')
22    ):
23        """
24        Initialize uptime tracker
25
26        Args:
27            service_name: Name of service to track
28            data_dir: Directory for storing tracking data
29        """
30        self.service_name = service_name
31        self.data_dir = data_dir
32        self.data_dir.mkdir(parents=True, exist_ok=True)
33
34        self.data_file = self.data_dir / f"{service_name}_uptime.json"
35
36        # Load existing data
37        self.data = self._load_data()
38
39        # Recent checks (for quick statistics)
40        self.recent_checks = deque(maxlen=100)
41
42        logger.info(f"Uptime tracker initialized for: {service_name}")
43
44    def _load_data(self) -> Dict:
45        """Load historical tracking data"""
46        if self.data_file.exists():
47            try:
48                with open(self.data_file, 'r') as f:
49                    return json.load(f)
50            except:
51                pass
52
53        return {
54            'service_name': self.service_name,
55            'first_check': None,
56            'last_check': None,
57            'total_checks': 0,
58            'successful_checks': 0,
59            'failed_checks': 0,
60            'checks': [],
61            'incidents': []
62        }
63
64    def _save_data(self):
65        """Save tracking data to disk"""
66        try:
67            with open(self.data_file, 'w') as f:
68                json.dump(self.data, f, indent=2)
69        except Exception as e:
70            logger.error(f"Failed to save data: {e}")
71
72    def record_check(
73        self,
74        status: str,
75        response_time_ms: float,
76        status_code: Optional[int] = None,
77        error: Optional[str] = None
78    ):
79        """
80        Record health check result
81
82        Args:
83            status: Check status ('healthy', 'unhealthy', 'timeout', etc.)
84            response_time_ms: Response time in milliseconds
85            status_code: HTTP status code if applicable
86            error: Error message if failed
87        """
88        timestamp = datetime.now()
89
90        check_result = {
91            'timestamp': timestamp.isoformat(),
92            'status': status,
93            'response_time_ms': response_time_ms,
94            'status_code': status_code,
95            'error': error
96        }
97
98        # Update statistics
99        self.data['total_checks'] += 1
100        self.data['last_check'] = check_result['timestamp']
101
102        if not self.data['first_check']:
103            self.data['first_check'] = check_result['timestamp']
104
105        if status == 'healthy':
106            self.data['successful_checks'] += 1
107        else:
108            self.data['failed_checks'] += 1
109
110        # Store check result (keep last 1000)
111        self.data['checks'].append(check_result)
112        if len(self.data['checks']) > 1000:
113            self.data['checks'] = self.data['checks'][-1000:]
114
115        # Add to recent checks
116        self.recent_checks.append(check_result)
117
118        # Check for incident
119        self._check_for_incident(check_result)
120
121        # Save data
122        self._save_data()
123
124        logger.debug(f"Recorded check: {status} ({response_time_ms:.0f}ms)")
125
126    def _check_for_incident(self, check_result: Dict):
127        """
128        Check if current result indicates an incident
129
130        Args:
131            check_result: Latest check result
132        """
133        # Get last 5 checks
134        recent = list(self.recent_checks)[-5:]
135
136        if len(recent) < 5:
137            return
138
139        # Check if last 3 checks failed
140        failed_count = sum(1 for c in recent[-3:] if c['status'] != 'healthy')
141
142        if failed_count >= 3:
143            # Check if incident already active
144            active_incidents = [i for i in self.data['incidents'] if not i.get('resolved')]
145
146            if not active_incidents:
147                # Create new incident
148                incident = {
149                    'start_time': recent[-3]['timestamp'],
150                    'status': 'active',
151                    'failed_checks': 3,
152                    'resolved': False
153                }
154
155                self.data['incidents'].append(incident)
156                logger.warning(f"⚠️  New incident detected for {self.service_name}")
157
158        # Check if incident resolved
159        else:
160            active_incidents = [i for i in self.data['incidents'] if not i.get('resolved')]
161
162            for incident in active_incidents:
163                # Check if last 3 checks succeeded
164                success_count = sum(1 for c in recent[-3:] if c['status'] == 'healthy')
165
166                if success_count >= 3:
167                    incident['resolved'] = True
168                    incident['end_time'] = check_result['timestamp']
169                    incident['duration_minutes'] = self._calculate_duration(
170                        incident['start_time'],
171                        incident['end_time']
172                    )
173
174                    logger.info(f"✓ Incident resolved for {self.service_name}")
175
176    def _calculate_duration(self, start_time: str, end_time: str) -> float:
177        """Calculate duration between timestamps in minutes"""
178        start = datetime.fromisoformat(start_time)
179        end = datetime.fromisoformat(end_time)
180        return (end - start).total_seconds() / 60
181
182    def get_uptime_percentage(self, period_hours: Optional[int] = None) -> float:
183        """
184        Calculate uptime percentage
185
186        Args:
187            period_hours: Optional time period to calculate for
188
189        Returns:
190            Uptime percentage (0-100)
191        """
192        checks = self.data['checks']
193
194        if period_hours:
195            # Filter checks within period
196            cutoff = datetime.now() - timedelta(hours=period_hours)
197            checks = [
198                c for c in checks
199                if datetime.fromisoformat(c['timestamp']) > cutoff
200            ]
201
202        if not checks:
203            return 0.0
204
205        successful = sum(1 for c in checks if c['status'] == 'healthy')
206        return (successful / len(checks)) * 100
207
208    def get_average_response_time(self, period_hours: Optional[int] = None) -> float:
209        """
210        Calculate average response time
211
212        Args:
213            period_hours: Optional time period to calculate for
214
215        Returns:
216            Average response time in milliseconds
217        """
218        checks = self.data['checks']
219
220        if period_hours:
221            cutoff = datetime.now() - timedelta(hours=period_hours)
222            checks = [
223                c for c in checks
224                if datetime.fromisoformat(c['timestamp']) > cutoff
225            ]
226
227        if not checks:
228            return 0.0
229
230        response_times = [c['response_time_ms'] for c in checks if c['response_time_ms']]
231
232        return statistics.mean(response_times) if response_times else 0.0
233
234    def get_statistics(self, period_hours: Optional[int] = None) -> Dict:
235        """
236        Get comprehensive statistics
237
238        Args:
239            period_hours: Optional time period
240
241        Returns:
242            Statistics dictionary
243        """
244        uptime = self.get_uptime_percentage(period_hours)
245        avg_response = self.get_average_response_time(period_hours)
246
247        checks = self.data['checks']
248        if period_hours:
249            cutoff = datetime.now() - timedelta(hours=period_hours)
250            checks = [
251                c for c in checks
252                if datetime.fromisoformat(c['timestamp']) > cutoff
253            ]
254
255        response_times = [c['response_time_ms'] for c in checks if c['response_time_ms']]
256
257        stats = {
258            'service_name': self.service_name,
259            'uptime_percentage': round(uptime, 2),
260            'average_response_time_ms': round(avg_response, 2),
261            'total_checks': len(checks),
262            'successful_checks': sum(1 for c in checks if c['status'] == 'healthy'),
263            'failed_checks': sum(1 for c in checks if c['status'] != 'healthy')
264        }
265
266        if response_times:
267            stats['min_response_time_ms'] = round(min(response_times), 2)
268            stats['max_response_time_ms'] = round(max(response_times), 2)
269            stats['median_response_time_ms'] = round(statistics.median(response_times), 2)
270
271        # Count incidents
272        incidents = self.data['incidents']
273        if period_hours:
274            cutoff = datetime.now() - timedelta(hours=period_hours)
275            incidents = [
276                i for i in incidents
277                if datetime.fromisoformat(i['start_time']) > cutoff
278            ]
279
280        stats['total_incidents'] = len(incidents)
281        stats['active_incidents'] = sum(1 for i in incidents if not i.get('resolved'))
282
283        return stats
284
285    def get_incident_history(self, limit: int = 10) -> List[Dict]:
286        """
287        Get recent incidents
288
289        Args:
290            limit: Maximum number of incidents to return
291
292        Returns:
293            List of incident dictionaries
294        """
295        incidents = sorted(
296            self.data['incidents'],
297            key=lambda x: x['start_time'],
298            reverse=True
299        )
300
301        return incidents[:limit]
302
303
304# Usage example
305tracker = UptimeTracker(service_name='api.example.com')
306
307# Record checks
308tracker.record_check(
309    status='healthy',
310    response_time_ms=145.5,
311    status_code=200
312)
313
314tracker.record_check(
315    status='timeout',
316    response_time_ms=10000.0,
317    error='Connection timeout'
318)
319
320# Get statistics
321stats_24h = tracker.get_statistics(period_hours=24)
322print(f"24h Uptime: {stats_24h['uptime_percentage']}%")
323print(f"Average Response Time: {stats_24h['average_response_time_ms']}ms")
324print(f"Total Incidents: {stats_24h['total_incidents']}")
325
326# Get all-time statistics
327stats_all = tracker.get_statistics()
328print(f"All-Time Uptime: {stats_all['uptime_percentage']}%")
329
330# Get incident history
331incidents = tracker.get_incident_history(limit=5)
332for incident in incidents:
333    status = "Active" if not incident['resolved'] else "Resolved"
334    print(f"Incident: {status} - Started: {incident['start_time']}")