added PDF support and autoupdating on refresh
This commit is contained in:
+248
@@ -0,0 +1,248 @@
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
class DataManager:
|
||||
def __init__(self, participants_file="participants.json", logs_file="scan_logs.json"):
|
||||
self.participants_file = participants_file
|
||||
self.logs_file = logs_file
|
||||
self.init_files()
|
||||
|
||||
def init_files(self):
|
||||
"""Initialize JSON files if they don't exist"""
|
||||
if not os.path.exists(self.participants_file):
|
||||
self.save_participants([])
|
||||
if not os.path.exists(self.logs_file):
|
||||
self.save_logs([])
|
||||
|
||||
def load_participants(self) -> List[Dict]:
|
||||
"""Load participants from JSON file"""
|
||||
try:
|
||||
with open(self.participants_file, 'r') as f:
|
||||
return json.load(f)
|
||||
except Exception as e:
|
||||
print(f"Error loading participants: {e}")
|
||||
return []
|
||||
|
||||
def save_participants(self, participants: List[Dict]):
|
||||
"""Save participants to JSON file"""
|
||||
try:
|
||||
with open(self.participants_file, 'w') as f:
|
||||
json.dump(participants, f, indent=2)
|
||||
except Exception as e:
|
||||
print(f"Error saving participants: {e}")
|
||||
|
||||
def load_logs(self) -> List[Dict]:
|
||||
"""Load scan logs from JSON file"""
|
||||
try:
|
||||
with open(self.logs_file, 'r') as f:
|
||||
return json.load(f)
|
||||
except Exception as e:
|
||||
print(f"Error loading logs: {e}")
|
||||
return []
|
||||
|
||||
def save_logs(self, logs: List[Dict]):
|
||||
"""Save scan logs to JSON file"""
|
||||
try:
|
||||
with open(self.logs_file, 'w') as f:
|
||||
json.dump(logs, f, indent=2)
|
||||
except Exception as e:
|
||||
print(f"Error saving logs: {e}")
|
||||
|
||||
def add_participant(self, name: str, card_uid: str = None, expiration_date: str = None, key_uid: str = None) -> bool:
|
||||
"""Add a new participant"""
|
||||
try:
|
||||
participants = self.load_participants()
|
||||
|
||||
# Check if card UID already exists
|
||||
if card_uid and any(p.get('card_uid') == card_uid for p in participants):
|
||||
return False
|
||||
|
||||
# Check if key UID already exists
|
||||
if key_uid and any(p.get('key_uid') == key_uid for p in participants):
|
||||
return False
|
||||
|
||||
new_participant = {
|
||||
'id': len(participants) + 1,
|
||||
'name': name,
|
||||
'card_uid': card_uid,
|
||||
'key_uid': key_uid,
|
||||
'expiration_date': expiration_date,
|
||||
'date_added': datetime.now().isoformat()
|
||||
}
|
||||
participants.append(new_participant)
|
||||
self.save_participants(participants)
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Error adding participant: {e}")
|
||||
return False
|
||||
|
||||
def get_participant_by_card(self, card_uid: str) -> Optional[Dict]:
|
||||
"""Get participant by card UID or key UID"""
|
||||
try:
|
||||
participants = self.load_participants()
|
||||
for p in participants:
|
||||
if p.get('card_uid') == card_uid or p.get('key_uid') == card_uid:
|
||||
return p
|
||||
return None
|
||||
except Exception as e:
|
||||
print(f"Error getting participant: {e}")
|
||||
return None
|
||||
|
||||
def is_card_expired(self, participant: Dict) -> bool:
|
||||
"""Check if participant's card is expired"""
|
||||
try:
|
||||
expiration_date = participant.get('expiration_date')
|
||||
if not expiration_date:
|
||||
return False
|
||||
|
||||
exp_date = datetime.fromisoformat(expiration_date.replace('Z', '+00:00'))
|
||||
current_date = datetime.now()
|
||||
return current_date > exp_date
|
||||
except Exception as e:
|
||||
print(f"Error checking expiration: {e}")
|
||||
return False
|
||||
|
||||
def get_all_participants(self) -> List[Dict]:
|
||||
"""Get all participants"""
|
||||
return self.load_participants()
|
||||
|
||||
def update_participant(self, participant_id: int, name: str, card_uid: str = None, expiration_date: str = None, key_uid: str = None) -> bool:
|
||||
"""Update participant information"""
|
||||
try:
|
||||
participants = self.load_participants()
|
||||
|
||||
# Find the participant being updated
|
||||
participant_to_update = None
|
||||
for p in participants:
|
||||
if p['id'] == participant_id:
|
||||
participant_to_update = p
|
||||
break
|
||||
|
||||
if not participant_to_update:
|
||||
return False
|
||||
|
||||
# Check if new card_uid is already used by another participant (skip if same as current)
|
||||
if card_uid and card_uid != participant_to_update.get('card_uid'):
|
||||
for p in participants:
|
||||
if p['id'] != participant_id and p.get('card_uid') == card_uid:
|
||||
return False
|
||||
|
||||
# Check if new key_uid is already used by another participant (skip if same as current)
|
||||
if key_uid and key_uid != participant_to_update.get('key_uid'):
|
||||
for p in participants:
|
||||
if p['id'] != participant_id and p.get('key_uid') == key_uid:
|
||||
return False
|
||||
|
||||
# Update name (always)
|
||||
participant_to_update['name'] = name
|
||||
|
||||
# Update UIDs and expiration only if explicitly provided (not None)
|
||||
# card_uid and key_uid can be explicitly set to empty string to clear them
|
||||
if card_uid is not None:
|
||||
participant_to_update['card_uid'] = card_uid if card_uid else None
|
||||
if key_uid is not None:
|
||||
participant_to_update['key_uid'] = key_uid if key_uid else None
|
||||
if expiration_date is not None:
|
||||
participant_to_update['expiration_date'] = expiration_date if expiration_date else None
|
||||
|
||||
self.save_participants(participants)
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Error updating participant: {e}")
|
||||
return False
|
||||
|
||||
def delete_participant(self, participant_id: int) -> bool:
|
||||
"""Delete participant"""
|
||||
try:
|
||||
participants = self.load_participants()
|
||||
participants = [p for p in participants if p['id'] != participant_id]
|
||||
self.save_participants(participants)
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Error deleting participant: {e}")
|
||||
return False
|
||||
|
||||
def log_scan(self, scanned_uid: str, result: str, name: str = "Unknown", reason: str = None, card_uid: str = None, key_uid: str = None):
|
||||
"""Log a card/key scan"""
|
||||
try:
|
||||
logs = self.load_logs()
|
||||
log_entry = {
|
||||
'id': len(logs) + 1,
|
||||
'card_uid': card_uid,
|
||||
'key_uid': key_uid,
|
||||
'name': name,
|
||||
'scan_result': result,
|
||||
'reason': reason,
|
||||
'timestamp': datetime.now().isoformat()
|
||||
}
|
||||
logs.append(log_entry)
|
||||
self.save_logs(logs)
|
||||
except Exception as e:
|
||||
print(f"Error logging scan: {e}")
|
||||
|
||||
def get_scan_history(self, limit: int = 100) -> List[Dict]:
|
||||
"""Get scan history (most recent first)"""
|
||||
try:
|
||||
logs = self.load_logs()
|
||||
return logs[-limit:][::-1] # Return last N items in reverse order
|
||||
except Exception as e:
|
||||
print(f"Error getting scan history: {e}")
|
||||
return []
|
||||
|
||||
def get_statistics(self) -> Dict:
|
||||
"""Get statistics from data"""
|
||||
try:
|
||||
participants = self.load_participants()
|
||||
logs = self.load_logs()
|
||||
|
||||
pass_count = sum(1 for l in logs if l['scan_result'] == 'PASS')
|
||||
fail_count = sum(1 for l in logs if l['scan_result'] == 'FAIL')
|
||||
|
||||
return {
|
||||
'total_participants': len(participants),
|
||||
'total_passes': pass_count,
|
||||
'total_fails': fail_count,
|
||||
'total_scans': len(logs)
|
||||
}
|
||||
except Exception as e:
|
||||
print(f"Error getting statistics: {e}")
|
||||
return {}
|
||||
|
||||
def log_unregistered_user(self, name: str, timestamp: str = None) -> bool:
|
||||
"""Log an unregistered user manually (admin only)"""
|
||||
try:
|
||||
logs = self.load_logs()
|
||||
log_entry = {
|
||||
'id': len(logs) + 1,
|
||||
'card_uid': None,
|
||||
'key_uid': None,
|
||||
'name': name,
|
||||
'scan_result': 'MANUAL',
|
||||
'reason': 'Unregistered',
|
||||
'timestamp': timestamp if timestamp else datetime.now().isoformat()
|
||||
}
|
||||
logs.append(log_entry)
|
||||
self.save_logs(logs)
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Error logging unregistered user: {e}")
|
||||
return False
|
||||
|
||||
def delete_log(self, log_id: int) -> bool:
|
||||
"""Delete a log entry by ID (admin only)"""
|
||||
try:
|
||||
logs = self.load_logs()
|
||||
original_length = len(logs)
|
||||
logs = [log for log in logs if log['id'] != log_id]
|
||||
|
||||
if len(logs) == original_length:
|
||||
# Log ID not found
|
||||
return False
|
||||
|
||||
self.save_logs(logs)
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Error deleting log: {e}")
|
||||
return False
|
||||
Reference in New Issue
Block a user