53 lines
1.5 KiB
Python
53 lines
1.5 KiB
Python
import json
|
|
from datetime import datetime, timedelta
|
|
|
|
import redis
|
|
|
|
|
|
class RoomExport:
|
|
def __init__(self, room_id: str, allowed_again: datetime):
|
|
self.room_id = room_id
|
|
self.allowed_again = allowed_again
|
|
|
|
def to_json(self):
|
|
return json.dumps({
|
|
'room_id': self.room_id,
|
|
'allowed_again': self.allowed_again.isoformat()
|
|
})
|
|
|
|
@classmethod
|
|
def from_json(cls, data):
|
|
data = json.loads(data)
|
|
return cls(data['room_id'], datetime.fromisoformat(data['allowed_again']))
|
|
|
|
|
|
class ExportTracker:
|
|
def __init__(self):
|
|
self._redis = redis.Redis(host='localhost', port=6379, db=3)
|
|
|
|
def add_export(self, room_id: str, seconds_elapsed_allowed: int) -> None:
|
|
allowed_again = datetime.now() + timedelta(seconds=seconds_elapsed_allowed)
|
|
room_export = RoomExport(room_id, allowed_again)
|
|
self._redis.set(room_id, room_export.to_json())
|
|
|
|
def check_allowed(self, room_id):
|
|
room_export = self.get_export(room_id)
|
|
if room_export:
|
|
return datetime.now() >= room_export.allowed_again
|
|
return True
|
|
|
|
def get_export(self, room_id: str) -> RoomExport | None:
|
|
data = self._redis.get(room_id)
|
|
if data:
|
|
return RoomExport.from_json(data.decode('utf-8'))
|
|
return None
|
|
|
|
def all(self):
|
|
keys = self._redis.keys('*')
|
|
exports = []
|
|
for key in keys:
|
|
export = self.get_export(key.decode('utf-8'))
|
|
if export:
|
|
exports.append(export)
|
|
return exports
|