65 lines
2.6 KiB
Python
65 lines
2.6 KiB
Python
import json
|
|
import os
|
|
import threading
|
|
import time
|
|
from typing import Dict, Optional
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class AccessTracker:
|
|
"""Tracks when video URLs are accessed, storing data in JSON file and keeping it in memory"""
|
|
|
|
def __init__(self, json_file_path: str = "access_times.json"):
|
|
self.json_file_path = json_file_path
|
|
self.access_times: Dict[str, float] = {}
|
|
self.lock = threading.Lock()
|
|
self._load_from_file()
|
|
|
|
def _load_from_file(self) -> None:
|
|
"""Load access times from JSON file into memory"""
|
|
try:
|
|
if os.path.exists(self.json_file_path):
|
|
with open(self.json_file_path, 'r') as f:
|
|
self.access_times = json.load(f)
|
|
logger.info(f"Loaded {len(self.access_times)} access times from {self.json_file_path}")
|
|
else:
|
|
logger.info(f"Access times file {self.json_file_path} does not exist, starting fresh")
|
|
except Exception as e:
|
|
logger.error(f"Error loading access times from {self.json_file_path}: {e}")
|
|
self.access_times = {}
|
|
|
|
def _save_to_file(self) -> None:
|
|
"""Save current access times from memory to JSON file"""
|
|
try:
|
|
with open(self.json_file_path, 'w') as f:
|
|
json.dump(self.access_times, f, indent=2)
|
|
logger.debug(f"Saved {len(self.access_times)} access times to {self.json_file_path}")
|
|
except Exception as e:
|
|
logger.error(f"Error saving access times to {self.json_file_path}: {e}")
|
|
|
|
def record_access(self, video_id: str) -> None:
|
|
"""Record that a video was accessed at the current time"""
|
|
current_time = time.time()
|
|
with self.lock:
|
|
self.access_times[video_id] = current_time
|
|
self._save_to_file()
|
|
logger.debug(f"Recorded access for {video_id} at {current_time}")
|
|
|
|
def get_last_access(self, video_id: str) -> Optional[float]:
|
|
"""Get the last access time for a video (returns None if never accessed)"""
|
|
with self.lock:
|
|
return self.access_times.get(video_id)
|
|
|
|
def get_all_access_times(self) -> Dict[str, float]:
|
|
"""Get a copy of all access times"""
|
|
with self.lock:
|
|
return self.access_times.copy()
|
|
|
|
def remove_access_record(self, video_id: str) -> None:
|
|
"""Remove access record for a video (e.g., when video is deleted)"""
|
|
with self.lock:
|
|
if video_id in self.access_times:
|
|
del self.access_times[video_id]
|
|
self._save_to_file()
|
|
logger.debug(f"Removed access record for {video_id}")
|