79 lines
2.3 KiB
Python
79 lines
2.3 KiB
Python
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
|
|
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: list[dict[str, str]], options: ProcessingOptions
|
|
) -> list[dict[str, str]]:
|
|
"""Apply user-selected filters to parsed records."""
|
|
filtered: list[dict[str, str]] = []
|
|
|
|
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
|
|
|
|
filtered.append(record)
|
|
|
|
return filtered
|
|
|
|
|
|
def sort_records(
|
|
records: list[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":
|
|
key_func = _datetime_key
|
|
elif options.sort_by == "severity":
|
|
key_func = _severity_key
|
|
else:
|
|
raise ProcessingError("Unsupported sort field.")
|
|
|
|
return sorted(records, key=key_func, reverse=reverse)
|
|
|
|
|
|
def _datetime_key(record: dict[str, str]) -> tuple[int, datetime]:
|
|
date_value = record.get("v015xxxxdate", "").strip()
|
|
time_value = record.get("time", "").strip()
|
|
if not date_value or not time_value:
|
|
return (1, datetime.max)
|
|
|
|
try:
|
|
parsed = datetime.strptime(f"{date_value} {time_value}", "%Y-%m-%d %H:%M:%S")
|
|
except ValueError:
|
|
return (1, datetime.max)
|
|
return (0, parsed)
|
|
|
|
|
|
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)
|