93 lines
2.9 KiB
Python
93 lines
2.9 KiB
Python
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from typing import Iterable
|
|
|
|
from app.constants import SEVERITY_RANKING
|
|
|
|
|
|
class ProcessingError(ValueError):
|
|
"""Raised when records cannot be processed according to the selected options."""
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ProcessingOptions:
|
|
policy_cs: str
|
|
policy_ci: str
|
|
severity_cs: str
|
|
severity_ci: str
|
|
sort_by: str
|
|
order: str
|
|
mode: str
|
|
|
|
|
|
def filter_records(
|
|
records: Iterable[dict[str, str]], options: ProcessingOptions
|
|
) -> Iterable[dict[str, str]]:
|
|
"""Apply user-selected filters lazily to parsed records."""
|
|
for record in records:
|
|
policy_value = record.get("policy", "")
|
|
severity_value = record.get("severity_level", "")
|
|
|
|
if options.policy_cs and options.policy_cs not in policy_value:
|
|
continue
|
|
if options.policy_ci and options.policy_ci.lower() not in policy_value.lower():
|
|
continue
|
|
if options.severity_cs and options.severity_cs not in severity_value:
|
|
continue
|
|
if options.severity_ci and options.severity_ci.lower() not in severity_value.lower():
|
|
continue
|
|
|
|
yield record
|
|
|
|
|
|
def sort_records(
|
|
records: Iterable[dict[str, str]], options: ProcessingOptions
|
|
) -> list[dict[str, str]]:
|
|
"""Sort records by datetime or severity using the requested order."""
|
|
reverse = options.order == "desc"
|
|
|
|
if options.sort_by == "datetime":
|
|
return _sort_records_by_datetime(records, reverse)
|
|
elif options.sort_by == "severity":
|
|
key_func = _severity_key
|
|
else:
|
|
raise ProcessingError("Unsupported sort field.")
|
|
|
|
return sorted(records, key=key_func, reverse=reverse)
|
|
|
|
|
|
def _sort_records_by_datetime(
|
|
records: Iterable[dict[str, str]], reverse: bool
|
|
) -> list[dict[str, str]]:
|
|
"""Sort valid datetimes normally and always place invalid/missing values last."""
|
|
valid_records: list[tuple[datetime, dict[str, str]]] = []
|
|
invalid_records: list[dict[str, str]] = []
|
|
|
|
for record in records:
|
|
parsed_datetime = _parse_datetime(record)
|
|
if parsed_datetime is None:
|
|
invalid_records.append(record)
|
|
continue
|
|
valid_records.append((parsed_datetime, record))
|
|
|
|
sorted_valid_records = sorted(valid_records, key=lambda item: item[0], reverse=reverse)
|
|
return [record for _parsed, record in sorted_valid_records] + invalid_records
|
|
|
|
|
|
def _parse_datetime(record: dict[str, str]) -> datetime | None:
|
|
date_value = record.get("v015xxxxdate", "").strip()
|
|
time_value = record.get("time", "").strip()
|
|
if not date_value or not time_value:
|
|
return None
|
|
|
|
try:
|
|
return datetime.strptime(f"{date_value} {time_value}", "%Y-%m-%d %H:%M:%S")
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _severity_key(record: dict[str, str]) -> tuple[int, str]:
|
|
raw_value = record.get("severity_level", "").strip().lower()
|
|
rank = SEVERITY_RANKING.get(raw_value, 0)
|
|
return (rank, raw_value)
|