60 lines
2.0 KiB
Python
60 lines
2.0 KiB
Python
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta
|
|
from typing import List, Optional
|
|
import json
|
|
import os
|
|
|
|
@dataclass
|
|
class FeedbackLog:
|
|
transaction_id: str
|
|
original_match: str
|
|
correction: str
|
|
reason: str
|
|
timestamp: datetime
|
|
user_id: str
|
|
|
|
class FeedbackLogger:
|
|
def __init__(self, log_file: str = "feedback_logs.json"):
|
|
self.log_file = log_file
|
|
self.logs: List[FeedbackLog] = self._load_logs()
|
|
|
|
def _load_logs(self) -> List[FeedbackLog]:
|
|
if not os.path.exists(self.log_file):
|
|
return []
|
|
|
|
try:
|
|
with open(self.log_file, 'r') as f:
|
|
data = json.load(f)
|
|
return [FeedbackLog(**log) for log in data]
|
|
except:
|
|
return []
|
|
|
|
def _save_logs(self):
|
|
with open(self.log_file, 'w') as f:
|
|
json.dump([{
|
|
'transaction_id': log.transaction_id,
|
|
'original_match': log.original_match,
|
|
'correction': log.correction,
|
|
'reason': log.reason,
|
|
'timestamp': log.timestamp.isoformat(),
|
|
'user_id': log.user_id
|
|
} for log in self.logs], f, indent=2)
|
|
|
|
def log_override(self, transaction_id: str, original_match: str, correction: str, reason: str, user_id: str):
|
|
log = FeedbackLog(
|
|
transaction_id=transaction_id,
|
|
original_match=original_match,
|
|
correction=correction,
|
|
reason=reason,
|
|
timestamp=datetime.now(),
|
|
user_id=user_id
|
|
)
|
|
self.logs.append(log)
|
|
self._save_logs()
|
|
|
|
def get_logs_by_transaction(self, transaction_id: str) -> List[FeedbackLog]:
|
|
return [log for log in self.logs if log.transaction_id == transaction_id]
|
|
|
|
def get_recent_logs(self, days: int = 30) -> List[FeedbackLog]:
|
|
cutoff = datetime.now() - timedelta(days=days)
|
|
return [log for log in self.logs if log.timestamp > cutoff] |