101 lines
2.8 KiB
Python
101 lines
2.8 KiB
Python
"""Feedback storage utilities.
|
|
|
|
Provides helpers for reading and managing feedback log files.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any, Iterator
|
|
|
|
|
|
class FeedbackStorage:
|
|
"""Utilities for reading feedback logs."""
|
|
|
|
def __init__(self, storage_dir: Path) -> None:
|
|
"""Initialize feedback storage reader.
|
|
|
|
Args:
|
|
storage_dir: Base directory containing feedback logs
|
|
"""
|
|
self.storage_dir = Path(storage_dir)
|
|
|
|
def read_events(
|
|
self,
|
|
event_type: str,
|
|
days: int = 30,
|
|
) -> Iterator[dict[str, Any]]:
|
|
"""Read events from the past N days.
|
|
|
|
Args:
|
|
event_type: Type of event to read (e.g., "corrections", "validations")
|
|
days: Number of days to look back (default: 30)
|
|
|
|
Yields:
|
|
Event dictionaries
|
|
"""
|
|
type_dir = self.storage_dir / event_type
|
|
if not type_dir.exists():
|
|
return
|
|
|
|
# Generate date range
|
|
end_date = datetime.now()
|
|
start_date = end_date - timedelta(days=days)
|
|
|
|
current_date = start_date
|
|
while current_date <= end_date:
|
|
date_str = current_date.strftime("%Y%m%d")
|
|
log_file = type_dir / f"{date_str}.jsonl"
|
|
|
|
if log_file.exists():
|
|
with open(log_file, "r", encoding="utf-8") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if line:
|
|
yield json.loads(line)
|
|
|
|
current_date += timedelta(days=1)
|
|
|
|
def count_events(self, event_type: str, days: int = 30) -> int:
|
|
"""Count total events of a given type in the past N days.
|
|
|
|
Args:
|
|
event_type: Type of event to count
|
|
days: Number of days to look back
|
|
|
|
Returns:
|
|
Total event count
|
|
"""
|
|
return sum(1 for _ in self.read_events(event_type, days))
|
|
|
|
def get_recent_files(self, event_type: str, count: int = 7) -> list[Path]:
|
|
"""Get the N most recent log files for an event type.
|
|
|
|
Args:
|
|
event_type: Type of event
|
|
count: Number of recent files to return
|
|
|
|
Returns:
|
|
List of log file paths, newest first
|
|
"""
|
|
type_dir = self.storage_dir / event_type
|
|
if not type_dir.exists():
|
|
return []
|
|
|
|
files = sorted(
|
|
type_dir.glob("*.jsonl"),
|
|
key=lambda p: p.stem, # stem is YYYYMMDD
|
|
reverse=True,
|
|
)
|
|
return list(files[:count])
|
|
|
|
|
|
def get_default_storage_path() -> Path:
|
|
"""Get the default feedback storage directory path.
|
|
|
|
Returns:
|
|
Path to ~/.cache/crystal/feedback
|
|
"""
|
|
return Path.home() / ".cache" / "crystal" / "feedback"
|