Implemented the feedback foundation.
This commit is contained in:
27
src/fgai/feedback.py
Normal file
27
src/fgai/feedback.py
Normal file
@@ -0,0 +1,27 @@
|
||||
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
|
||||
Reference in New Issue
Block a user