Add log_enhancer/src/log_enhancer/core.py

log_enhancer/src/log_enhancer/core.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import List, Dict, Any
+
+
+class LogEnhancerError(Exception):
+    """Custom exception for Log Enhancer related errors."""
+
+
+class EnhancedLogEntry(dict):
+    """Data model representing an enhanced log entry."""
+
+    required_fields = {"message", "expected_artifact_path", "artifact_key"}
+
+    def __init__(self, message: str, expected_artifact_path: str, artifact_key: str):
+        super().__init__(
+            message=message,
+            expected_artifact_path=expected_artifact_path,
+            artifact_key=artifact_key,
+        )
+
+    @classmethod
+    def validate(cls, entry: Dict[str, Any]) -> "EnhancedLogEntry":
+        if not isinstance(entry, dict):
+            raise LogEnhancerError("Each log entry must be a dictionary.")
+
+        message = entry.get("message")
+        if not isinstance(message, str) or not message.strip():
+            raise LogEnhancerError("Log entry must contain a non-empty 'message' field.")
+
+        expected_path = entry.get("expected_artifact_path") or cls._infer_artifact_path(message)
+        artifact_key = entry.get("artifact_key") or cls._infer_artifact_key(message)
+
+        return cls(message, expected_path, artifact_key)
+
+    @staticmethod
+    def _infer_artifact_path(message: str) -> str:
+        # Simple heuristic: if message mentions a file or artifact id, construct a plausible path.
+        base_dir = "/artifacts/unknown"
+        file_part = "artifact_autogen.log"
+        if "art" in message.lower():
+            file_part = f"{message.lower().replace(' ', '_')[:20]}.log"
+        path = os.path.join(base_dir, file_part)
+        return path
+
+    @staticmethod
+    def _infer_artifact_key(message: str) -> str:
+        # Example heuristic: derive a key hash-like string from message.
+        key = f"key_{abs(hash(message)) % 10**8}"
+        return key
+
+
+def enhance_log_entries(log_data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    """Enhances incomplete log entries by adding expected_artifact_path and artifact_key.
+
+    Args:
+        log_data (list[dict]): List of log dictionaries, each with 'message' and optional fields.
+
+    Returns:
+        list[dict]: List of enhanced log entries with all required fields populated.
+    """
+    if not isinstance(log_data, list):
+        raise LogEnhancerError("Input log_data must be a list of dicts.")
+
+    enhanced_entries: List[Dict[str, Any]] = []
+
+    for entry in log_data:
+        # Defensive validation
+        enhanced_entry = EnhancedLogEntry.validate(entry)
+        enhanced_entries.append(dict(enhanced_entry))
+
+    # CI readiness: ensure structure integrity
+    for enhanced_entry in enhanced_entries:
+        assert all(field in enhanced_entry for field in EnhancedLogEntry.required_fields), (
+            f"Enhanced entry missing required fields: {enhanced_entry}"
+        )
+
+    return enhanced_entries