12 lines
1015 B
Python
12 lines
1015 B
Python
from __future__ import annotations
|
|
|
|
from .models import LogEvent
|
|
|
|
|
|
def assess_data_quality(events: list[LogEvent], stream_status: dict[str, object]) -> dict[str, object]:
|
|
total = len(events)
|
|
missing_timestamp = sum(not (event.fields.get("eventtime") or event.fields.get("timestamp") or (event.fields.get("date") and event.fields.get("time"))) for event in events)
|
|
missing_source = sum(not event.src_ip or event.src_ip == "-" for event in events)
|
|
streams = stream_status.get("streams", []) if isinstance(stream_status.get("streams"), list) else []
|
|
return {"events": total, "missing_timestamp": missing_timestamp, "missing_source": missing_source, "timestamp_coverage": round(100 * (total - missing_timestamp) / total, 1) if total else 0, "source_coverage": round(100 * (total - missing_source) / total, 1) if total else 0, "truncated_streams": [item.get("stream_name") or item.get("stream_title") or item.get("stream_id") for item in streams if isinstance(item, dict) and item.get("truncated")]}
|