369 lines
14 KiB
Python
369 lines
14 KiB
Python
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
|
|
|
|
# ===== STORE =====
|
|
|
|
def load_store(self) -> Dict:
|
|
try:
|
|
store_file = self.participants_file.replace('participants.json', 'store.json')
|
|
with open(store_file, 'r') as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
return {'balance': 0.0, 'items': [], 'payments': []}
|
|
|
|
def save_store(self, store: Dict):
|
|
try:
|
|
store_file = self.participants_file.replace('participants.json', 'store.json')
|
|
with open(store_file, 'w') as f:
|
|
json.dump(store, f, indent=2)
|
|
except Exception as e:
|
|
print(f"Error saving store: {e}")
|
|
|
|
def get_store_items(self) -> List[Dict]:
|
|
return self.load_store().get('items', [])
|
|
|
|
def get_balance(self) -> float:
|
|
return self.load_store().get('balance', 0.0)
|
|
|
|
def set_balance(self, amount: float) -> bool:
|
|
try:
|
|
store = self.load_store()
|
|
store['balance'] = round(amount, 2)
|
|
self.save_store(store)
|
|
return True
|
|
except Exception as e:
|
|
print(f"Error setting balance: {e}")
|
|
return False
|
|
|
|
def add_store_item(self, name: str, price: float, stock: int, image: str = None, item_type: str = 'product', price_member=None, price_non_member=None) -> bool:
|
|
try:
|
|
store = self.load_store()
|
|
items = store.get('items', [])
|
|
new_id = max((i['id'] for i in items), default=0) + 1
|
|
entry = {'id': new_id, 'name': name, 'price': round(price, 2), 'stock': stock, 'image': image, 'type': item_type,
|
|
'price_member': round(price_member, 2) if price_member is not None else None,
|
|
'price_non_member': round(price_non_member, 2) if price_non_member is not None else None}
|
|
items.append(entry)
|
|
store['items'] = items
|
|
self.save_store(store)
|
|
return True
|
|
except Exception as e:
|
|
print(f"Error adding item: {e}")
|
|
return False
|
|
|
|
def update_store_item(self, item_id: int, name: str, price: float, stock: int, image: str = None, item_type: str = None, price_member=None, price_non_member=None) -> bool:
|
|
try:
|
|
store = self.load_store()
|
|
items = store.get('items', [])
|
|
for item in items:
|
|
if item['id'] == item_id:
|
|
item['name'] = name
|
|
item['price'] = round(price, 2)
|
|
item['stock'] = stock
|
|
item['image'] = image
|
|
if item_type is not None:
|
|
item['type'] = item_type
|
|
item['price_member'] = round(price_member, 2) if price_member is not None else None
|
|
item['price_non_member'] = round(price_non_member, 2) if price_non_member is not None else None
|
|
store['items'] = items
|
|
self.save_store(store)
|
|
return True
|
|
return False
|
|
except Exception as e:
|
|
print(f"Error updating item: {e}")
|
|
return False
|
|
|
|
def delete_store_item(self, item_id: int) -> bool:
|
|
try:
|
|
store = self.load_store()
|
|
items = store.get('items', [])
|
|
new_items = [i for i in items if i['id'] != item_id]
|
|
if len(new_items) == len(items):
|
|
return False
|
|
store['items'] = new_items
|
|
self.save_store(store)
|
|
return True
|
|
except Exception as e:
|
|
print(f"Error deleting item: {e}")
|
|
return False
|
|
|
|
def log_payment(self, items: list, total: float, balance_before: float, balance_after: float, seller: str = '') -> bool:
|
|
try:
|
|
store = self.load_store()
|
|
payments = store.get('payments', [])
|
|
new_id = max((p['id'] for p in payments), default=0) + 1
|
|
payments.append({
|
|
'id': new_id,
|
|
'timestamp': datetime.now().isoformat(),
|
|
'seller': seller,
|
|
'items': items,
|
|
'total': round(total, 2),
|
|
'balance_before': round(balance_before, 2),
|
|
'balance_after': round(balance_after, 2)
|
|
})
|
|
store['payments'] = payments
|
|
self.save_store(store)
|
|
return True
|
|
except Exception as e:
|
|
print(f"Error logging payment: {e}")
|
|
return False
|
|
|
|
def get_payments(self) -> list:
|
|
return self.load_store().get('payments', [])
|
|
|
|
def clear_payments(self) -> bool:
|
|
try:
|
|
store = self.load_store()
|
|
store['payments'] = []
|
|
self.save_store(store)
|
|
return True
|
|
except Exception as e:
|
|
print(f"Error clearing payments: {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
|