""" Webhook Notifier for Health Check Alerts (management command) """ import json import os from pathlib import Path from typing import Dict, List, Optional from django.core.management.base import BaseCommand from django.conf import settings from django.utils.timezone import now import requests class WebhookFormatter: @staticmethod def format_slack(portal: str, health_data: Dict) -> Dict: status = health_data['overall_status'] score = health_data['overall_score'] color_map = { 'healthy': '#36a64f', 'degraded': '#ff9900', 'critical': '#ff0000', 'failing': '#8b0000' } color = color_map.get(status, '#cccccc') fields = [ {'title': 'Overall Score', 'value': f'{score:.1f}/100', 'short': True}, {'title': 'Status', 'value': status.upper(), 'short': True}, ] if health_data.get('critical_issues'): issues_text = '\n'.join([f"• {i['title']}: {i['description']}" for i in health_data['critical_issues'][:3]]) fields.append({'title': f"Critical Issues ({len(health_data['critical_issues'])})", 'value': issues_text, 'short': False}) if health_data.get('recommendations'): fields.append({'title': 'Quick Fix', 'value': health_data['recommendations'][0], 'short': False}) return {'attachments': [{'color': color, 'title': f'🏥 Health Alert: {portal}', 'fields': fields, 'footer': 'Advanced Health Monitor', 'ts': int(now().timestamp())}]} @staticmethod def format_discord(portal: str, health_data: Dict) -> Dict: status = health_data['overall_status'] score = health_data['overall_score'] color_map = {'healthy': 0x36a64f, 'degraded': 0xff9900, 'critical': 0xff0000, 'failing': 0x8b0000} color = color_map.get(status, 0xcccccc) embed = { 'title': f'🏥 Health Alert: {portal}', 'color': color, 'fields': [ {'name': 'Overall Score', 'value': f'{score:.1f}/100', 'inline': True}, {'name': 'Status', 'value': status.upper(), 'inline': True}, ], 'timestamp': now().isoformat(), } if health_data.get('critical_issues'): issues_text = '\n'.join([f"• **{i['title']}**: {i['description']}" for i in health_data['critical_issues'][:3]]) embed['fields'].append({'name': f"Critical Issues ({len(health_data['critical_issues'])})", 'value': issues_text, 'inline': False}) if health_data.get('recommendations'): embed['fields'].append({'name': '💡 Quick Fix', 'value': health_data['recommendations'][0], 'inline': False}) return {'embeds': [embed]} @staticmethod def format_teams(portal: str, health_data: Dict) -> Dict: status = health_data['overall_status'] score = health_data['overall_score'] color_map = {'healthy': '36a64f', 'degraded': 'ff9900', 'critical': 'ff0000', 'failing': '8b0000'} theme_color = color_map.get(status, 'cccccc') sections = [{ 'activityTitle': f'🏥 Health Alert: {portal}', 'activitySubtitle': f'Status: {status.upper()}', 'facts': [{'name': 'Overall Score', 'value': f'{score:.1f}/100'}, {'name': 'Status', 'value': status.upper()}], 'markdown': True, }] if health_data.get('critical_issues'): issues_text = '\n\n'.join([f"**{i['title']}** \n{i['description']}" for i in health_data['critical_issues'][:3]]) sections.append({'title': f"Critical Issues ({len(health_data['critical_issues'])})", 'text': issues_text}) if health_data.get('recommendations'): sections.append({'title': '💡 Quick Fix', 'text': health_data['recommendations'][0]}) return {'@type': 'MessageCard', '@context': 'https://schema.org/extensions', 'themeColor': theme_color, 'summary': f'Health Alert: {portal}', 'sections': sections} class Command(BaseCommand): help = 'Send health check alerts to Slack/Discord/Teams webhooks' def add_arguments(self, parser): parser.add_argument('--json', type=str, help='Path to health report JSON') parser.add_argument('--portal', type=str, help='Check specific portal') parser.add_argument('--all', action='store_true', help='Check all portals') parser.add_argument('--threshold-error', type=float, default=0.15, help='Error rate threshold (unused in simple mode)') parser.add_argument('--threshold-score', type=float, default=70.0, help='Score threshold') parser.add_argument('--min-pages', type=int, default=10, help='Minimum pages to consider (unused in simple mode)') parser.add_argument('--webhook-url', type=str, help='Override webhook URL') parser.add_argument('--webhook-type', choices=['slack', 'discord', 'teams', 'auto'], default='auto') parser.add_argument('--dry-run', action='store_true', help='Print payload without sending') parser.add_argument('--channel', type=str, help='Optional channel/mention') def handle(self, *args, **opts): json_file = opts['json'] portal = opts['portal'] check_all = opts['all'] webhook_url = opts['webhook_url'] or os.getenv('DEBUGGER_WEBHOOK_URL') webhook_type = opts['webhook_type'] dry_run = opts['dry_run'] threshold_score = opts['threshold_score'] channel = opts['channel'] if not webhook_url and not dry_run: self.stderr.write(self.style.ERROR('Set DEBUGGER_WEBHOOK_URL or pass --webhook-url')) raise SystemExit(1) if webhook_type == 'auto' and webhook_url: if 'slack.com' in webhook_url: webhook_type = 'slack' elif 'discord.com' in webhook_url or 'discordapp.com' in webhook_url: webhook_type = 'discord' elif 'office.com' in webhook_url or 'webhook.office' in webhook_url: webhook_type = 'teams' else: webhook_type = 'slack' if json_file: reports = self._load_json_report(json_file) else: # run live checks if not (portal or check_all): self.stderr.write(self.style.ERROR('Must specify --json, --portal, or --all')) raise SystemExit(1) from extractly.models import SourceManual from manual_agregator.management.commands.advanced_health import SimpleHealthManager mgr = SimpleHealthManager() portals = list(SourceManual.objects.filter(enable=True).values_list('name', flat=True)) if check_all else [portal] reports = [self._run_health_check(mgr, p) for p in portals] unhealthy = [r for r in reports if self._is_unhealthy(r, threshold_score)] if not unhealthy: self.stdout.write(self.style.SUCCESS('✅ All portals healthy, no alerts sent')) return sent = 0 for r in unhealthy: if self._send_alert(r, webhook_url, webhook_type, channel, dry_run): sent += 1 self.stdout.write(self.style.WARNING(f'⚠️ Sent {sent} alert(s)')) raise SystemExit(2) def _load_json_report(self, filepath: str) -> List[Dict]: path = Path(filepath) if not path.exists(): self.stderr.write(self.style.ERROR(f'JSON not found: {filepath}')) raise SystemExit(1) with open(path, 'r', encoding='utf-8') as f: data = json.load(f) return data.get('reports', [data]) def _run_health_check(self, mgr, portal: str) -> Dict: r = mgr.run(portal, hours=24) return { 'portal': r.portal, 'overall_status': r.overall_status.value, 'overall_score': r.overall_score, 'timestamp': r.timestamp.isoformat(), 'critical_issues': [ {'title': i.title, 'description': i.description, 'impact': i.impact, 'remediation_steps': i.remediation_steps} for i in r.issues if i.severity.value == 'critical' ], 'recommendations': r.recommendations, 'business_metrics': r.business_metrics, } def _is_unhealthy(self, report: Dict, threshold_score: float) -> bool: return ( report['overall_score'] < threshold_score or bool(report.get('critical_issues')) or report['overall_status'] in ['critical', 'failing'] ) def _send_alert(self, report: Dict, webhook_url: str, webhook_type: str, channel: Optional[str], dry_run: bool) -> bool: portal = report['portal'] fmt = WebhookFormatter() if webhook_type == 'slack': payload = fmt.format_slack(portal, report) if channel: payload['text'] = f'{channel} Health Alert' elif webhook_type == 'discord': payload = fmt.format_discord(portal, report) if channel: payload['content'] = f'{channel} Health Alert' elif webhook_type == 'teams': payload = fmt.format_teams(portal, report) else: self.stderr.write(self.style.ERROR(f'Unknown webhook type: {webhook_type}')) return False if dry_run: self.stdout.write(self.style.WARNING(f'[DRY RUN] Would send to {webhook_type}:')) self.stdout.write(json.dumps(payload, indent=2)) return True try: resp = requests.post(webhook_url, json=payload, headers={'Content-Type': 'application/json'}, timeout=10) if resp.status_code in (200, 204): self.stdout.write(self.style.SUCCESS(f'✓ Alert sent for {portal}')) return True self.stderr.write(self.style.ERROR(f'✗ Failed to send alert for {portal}: {resp.status_code} {resp.text}')) return False except Exception as e: self.stderr.write(self.style.ERROR(f'✗ Error sending alert for {portal}: {e}')) return False