profile adding

src/fgai/monitor.py

@@ -17,6 +17,7 @@ from .mitigation import parse_allowlist, suggest_block_candidates
 from .policies import audit_policies, read_policies
 from .recommendations import build_recommendations
 from .threat_intel import ThreatIntelClient, enrich_ips, is_public_ip
+from .stream_profiles import parse_profiles
 
 
 def build_status(
@@ -33,6 +34,7 @@ def build_status(
     config_exists = bool(config_store and config_store.path.exists())
     runtime_values = config_store.read() if config_exists and config_store else {}
     runtime_config = config_store.public() if config_store else {}
+    stream_profiles = parse_profiles(runtime_values.get("graylog_stream_profiles", []))
     events = read_events(log_path) if Path(log_path).exists() else []
     mcp_status: dict[str, object] = {"status": "not_configured"}
     if runtime_values.get("log_source") == "graylog_mcp":
@@ -98,6 +100,7 @@ def build_status(
         "baseline": {"enabled": bool(baseline), "sources_ready": len(profiles), "new_events_recorded": baseline_events},
         "capabilities": {"threat_intel": threat_intel_status, "graylog_mcp": mcp_status},
         "configuration": runtime_config,
+        "stream_profiles": [{"stream_id": item.stream_id, "entity_field": item.entity_field, "timestamp_field": item.timestamp_field, "categorical_fields": list(item.categorical_fields), "numeric_fields": list(item.numeric_fields)} for item in stream_profiles.values()],
         "diagnostics": {
             "top_source_ips": top_field_values(events, "srcip", limit=10),
             "top_destination_ips": top_field_values(events, "dstip", limit=10),

src/fgai/stream_profiles.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class StreamProfile:
+    stream_id: str
+    entity_field: str
+    timestamp_field: str
+    categorical_fields: tuple[str, ...] = ()
+    numeric_fields: tuple[str, ...] = ()
+
+
+def parse_profiles(value: object) -> dict[str, StreamProfile]:
+    profiles: dict[str, StreamProfile] = {}
+    for item in value if isinstance(value, list) else []:
+        if not isinstance(item, dict):
+            continue
+        stream_id = str(item.get("stream_id", "")).strip()
+        entity = str(item.get("entity_field", "")).strip()
+        timestamp = str(item.get("timestamp_field", "timestamp")).strip()
+        if not stream_id or not entity:
+            continue
+        profiles[stream_id] = StreamProfile(
+            stream_id, entity, timestamp,
+            tuple(str(field) for field in item.get("categorical_fields", []) if field),
+            tuple(str(field) for field in item.get("numeric_fields", []) if field),
+        )
+    return profiles

tests/test_stream_profiles.py

@@ -0,0 +1,10 @@
+import unittest
+
+from fgai.stream_profiles import parse_profiles
+
+
+class StreamProfileTests(unittest.TestCase):
+    def test_parses_field_level_profile(self):
+        profiles = parse_profiles([{"stream_id": "dns", "entity_field": "IP", "categorical_fields": ["QH"], "numeric_fields": ["Elapsed"]}])
+        self.assertEqual(profiles["dns"].entity_field, "IP")
+        self.assertEqual(profiles["dns"].numeric_fields, ("Elapsed",))