Some checks failed
Tests / Go Tests (push) Has been cancelled
Tests / Python Tests (push) Has been cancelled
Tests / Integration Tests (push) Has been cancelled
Tests / Go Lint (push) Has been cancelled
Tests / Python Lint (push) Has been cancelled
Tests / Security Scan (push) Has been cancelled
Tests / All Checks Passed (push) Has been cancelled
Security Scanning / Secret Scanning (push) Has been cancelled
Security Scanning / Dependency Vulnerability Scan (push) Has been cancelled
Security Scanning / Go Security Scan (push) Has been cancelled
Security Scanning / Python Security Scan (push) Has been cancelled
Security Scanning / Node.js Security Scan (push) Has been cancelled
Security Scanning / Docker Image Security (push) Has been cancelled
Security Scanning / Security Summary (push) Has been cancelled
CI/CD Pipeline / Go Tests (push) Has been cancelled
CI/CD Pipeline / Python Tests (push) Has been cancelled
CI/CD Pipeline / Website Tests (push) Has been cancelled
CI/CD Pipeline / Linting (push) Has been cancelled
CI/CD Pipeline / Security Scan (push) Has been cancelled
CI/CD Pipeline / Docker Build & Push (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Deploy to Staging (push) Has been cancelled
CI/CD Pipeline / Deploy to Production (push) Has been cancelled
CI/CD Pipeline / CI Summary (push) Has been cancelled
ci/woodpecker/manual/build-ci-image Pipeline was successful
ci/woodpecker/manual/main Pipeline failed
All services: admin-v2, studio-v2, website, ai-compliance-sdk, consent-service, klausur-service, voice-service, and infrastructure. Large PDFs and compiled binaries excluded via .gitignore.
124 lines
3.1 KiB
Python
124 lines
3.1 KiB
Python
"""
|
|
Base Classes für Alert Actions.
|
|
|
|
Definiert das Interface für alle Action-Handler.
|
|
"""
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from typing import Dict, Any, Optional, List
|
|
from enum import Enum
|
|
|
|
|
|
class ActionType(str, Enum):
|
|
"""Verfügbare Aktionstypen."""
|
|
EMAIL = "email"
|
|
WEBHOOK = "webhook"
|
|
SLACK = "slack"
|
|
TEAMS = "teams"
|
|
TAG = "tag"
|
|
ARCHIVE = "archive"
|
|
|
|
|
|
@dataclass
|
|
class ActionResult:
|
|
"""Ergebnis einer ausgeführten Aktion."""
|
|
success: bool
|
|
action_type: ActionType
|
|
message: str
|
|
timestamp: datetime = field(default_factory=datetime.utcnow)
|
|
details: Dict[str, Any] = field(default_factory=dict)
|
|
error: Optional[str] = None
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Konvertiert zu Dict für Logging/Speicherung."""
|
|
return {
|
|
"success": self.success,
|
|
"action_type": self.action_type.value,
|
|
"message": self.message,
|
|
"timestamp": self.timestamp.isoformat(),
|
|
"details": self.details,
|
|
"error": self.error,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class AlertContext:
|
|
"""Kontext für eine Aktion mit Alert-Informationen."""
|
|
alert_id: str
|
|
title: str
|
|
url: str
|
|
snippet: str
|
|
topic_name: str
|
|
relevance_score: Optional[float] = None
|
|
relevance_decision: Optional[str] = None
|
|
matched_rule: Optional[str] = None
|
|
tags: List[str] = field(default_factory=list)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Konvertiert zu Dict für Templates."""
|
|
return {
|
|
"alert_id": self.alert_id,
|
|
"title": self.title,
|
|
"url": self.url,
|
|
"snippet": self.snippet,
|
|
"topic_name": self.topic_name,
|
|
"relevance_score": self.relevance_score,
|
|
"relevance_decision": self.relevance_decision,
|
|
"matched_rule": self.matched_rule,
|
|
"tags": self.tags,
|
|
}
|
|
|
|
|
|
class ActionHandler(ABC):
|
|
"""
|
|
Abstrakte Basisklasse für Action-Handler.
|
|
|
|
Jede Aktion (Email, Webhook, Slack) implementiert diese Schnittstelle.
|
|
"""
|
|
|
|
@property
|
|
@abstractmethod
|
|
def action_type(self) -> ActionType:
|
|
"""Gibt den Aktionstyp zurück."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def execute(
|
|
self,
|
|
context: AlertContext,
|
|
config: Dict[str, Any],
|
|
) -> ActionResult:
|
|
"""
|
|
Führt die Aktion aus.
|
|
|
|
Args:
|
|
context: Alert-Kontext mit allen relevanten Informationen
|
|
config: Aktionsspezifische Konfiguration
|
|
|
|
Returns:
|
|
ActionResult mit Erfolgsstatus und Details
|
|
"""
|
|
pass
|
|
|
|
def validate_config(self, config: Dict[str, Any]) -> bool:
|
|
"""
|
|
Validiert die Aktions-Konfiguration.
|
|
|
|
Args:
|
|
config: Zu validierende Konfiguration
|
|
|
|
Returns:
|
|
True wenn gültig
|
|
"""
|
|
return True
|
|
|
|
def get_required_config_fields(self) -> List[str]:
|
|
"""
|
|
Gibt erforderliche Konfigurationsfelder zurück.
|
|
|
|
Returns:
|
|
Liste von Feldnamen
|
|
"""
|
|
return []
|