28 lines
1.4 KiB
Python
28 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
|
|
|
|
class FeedbackStore:
|
|
def __init__(self, path: str = "state/signalscope-feedback.json") -> None:
|
|
self.path = Path(path)
|
|
|
|
def entries(self) -> list[dict[str, object]]:
|
|
try:
|
|
items = json.loads(self.path.read_text(encoding="utf-8"))
|
|
except (FileNotFoundError, json.JSONDecodeError):
|
|
items = []
|
|
now = int(time.time())
|
|
return [item for item in items if not item.get("expires_at") or int(item["expires_at"]) > now]
|
|
|
|
def add(self, item: dict[str, object]) -> dict[str, object]:
|
|
status = str(item.get("status", "")).lower()
|
|
if status not in {"false_positive", "expected", "confirmed"}:
|
|
raise ValueError("invalid feedback status")
|
|
entry = {"stream_id": str(item.get("stream_id", "")), "entity": str(item.get("entity", "")), "field": str(item.get("field", "")), "status": status, "note": str(item.get("note", "")), "created_at": int(time.time()), "expires_at": int(item.get("expires_at", 0) or 0)}
|
|
entries = [value for value in self.entries() if (value.get("stream_id"), value.get("entity"), value.get("field")) != (entry["stream_id"], entry["entity"], entry["field"])]
|
|
entries.append(entry); self.path.parent.mkdir(parents=True, exist_ok=True); self.path.write_text(json.dumps(entries, indent=2), encoding="utf-8")
|
|
return entry
|