63 lines
2.6 KiB
Python
63 lines
2.6 KiB
Python
from django.core.management.base import BaseCommand
|
|
from core.models import Employee, BiometricDevice, AttendanceLog
|
|
from core.utils import sync_all_daily_summaries
|
|
from zk import ZK, const
|
|
from django.utils import timezone
|
|
import datetime
|
|
|
|
class Command(BaseCommand):
|
|
help = 'Fetches attendance logs from ZKTeco biometric devices'
|
|
|
|
def handle(self, *args, **options):
|
|
devices = BiometricDevice.objects.filter(is_active=True)
|
|
if not devices.exists():
|
|
self.stdout.write(self.style.WARNING('No active biometric devices found.'))
|
|
return
|
|
|
|
for device in devices:
|
|
self.stdout.write(f'Connecting to device: {device.name} ({device.ip_address})...')
|
|
zk = ZK(device.ip_address, port=device.port, timeout=5, password=0, force_udp=False, ommit_ping=False)
|
|
conn = None
|
|
try:
|
|
conn = zk.connect()
|
|
# Disable device while fetching data
|
|
conn.disable_device()
|
|
|
|
logs = conn.get_attendance()
|
|
count = 0
|
|
for log in logs:
|
|
try:
|
|
employee = Employee.objects.get(biometric_id=log.user_id)
|
|
# Avoid duplicates using unique_together constraint logic
|
|
obj, created = AttendanceLog.objects.get_or_create(
|
|
employee=employee,
|
|
timestamp=log.timestamp,
|
|
defaults={
|
|
'device': device,
|
|
'uid': log.uid
|
|
}
|
|
)
|
|
if created:
|
|
count += 1
|
|
except Employee.DoesNotExist:
|
|
# self.stdout.write(self.style.WARNING(f'Employee with biometric ID {log.user_id} not found.'))
|
|
continue
|
|
|
|
device.last_sync = timezone.now()
|
|
device.save()
|
|
|
|
self.stdout.write(self.style.SUCCESS(f'Successfully synced {count} new logs from {device.name}.'))
|
|
|
|
# Re-enable device
|
|
conn.enable_device()
|
|
except Exception as e:
|
|
self.stdout.write(self.style.ERROR(f'Failed to connect to {device.name}: {str(e)}'))
|
|
finally:
|
|
if conn:
|
|
conn.disconnect()
|
|
|
|
# After all devices synced, update daily summaries
|
|
self.stdout.write('Updating daily attendance summaries...')
|
|
sync_all_daily_summaries()
|
|
self.stdout.write(self.style.SUCCESS('All summaries updated.'))
|