mirror of
https://github.com/wahyd4/links.git
synced 2026-08-09 05:06:16 +10:00
90 lines
3.3 KiB
Python
90 lines
3.3 KiB
Python
from django.apps import AppConfig
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class NetscanConfig(AppConfig):
|
|
default_auto_field = 'django.db.models.BigAutoField'
|
|
name = 'netscan'
|
|
|
|
def ready(self):
|
|
import netscan.signals # noqa: F401
|
|
self._register_job_type()
|
|
try:
|
|
from netscan.tasks import schedule_profile
|
|
from netscan.models import ScanProfile
|
|
for profile in ScanProfile.objects.filter(enabled=True):
|
|
schedule_profile(profile)
|
|
logger.info(f'Scheduled netscan profile: {profile.name}')
|
|
except Exception as e:
|
|
logger.warning(f'Could not schedule netscan profiles on startup: {e}')
|
|
|
|
def _register_job_type(self):
|
|
from links import job_registry
|
|
from netscan.models import ScanRun
|
|
from django.urls import reverse
|
|
|
|
def ns_stats():
|
|
return {
|
|
'total': ScanRun.objects.count(),
|
|
'pending': ScanRun.objects.filter(status='pending').count(),
|
|
'processing': ScanRun.objects.filter(status='running').count(),
|
|
'completed': ScanRun.objects.filter(status='success').count(),
|
|
'failed': ScanRun.objects.filter(status='failed').count(),
|
|
}
|
|
|
|
def ns_queryset(sf):
|
|
qs = ScanRun.objects.select_related('profile').order_by('-started_at')
|
|
if sf == 'pending':
|
|
return qs.filter(status='pending')
|
|
if sf == 'processing':
|
|
return qs.filter(status='running')
|
|
if sf == 'completed':
|
|
return qs.filter(status='success')
|
|
if sf == 'failed':
|
|
return qs.filter(status='failed')
|
|
return qs
|
|
|
|
def ns_serialize(obj):
|
|
try:
|
|
detail_url = reverse('netscan-run-detail', args=[obj.pk])
|
|
except Exception:
|
|
detail_url = '#'
|
|
duration = obj.duration_seconds
|
|
return {
|
|
'id': str(obj.id),
|
|
'title': obj.profile.name if obj.profile else f'Run #{obj.pk}',
|
|
'detail_url': detail_url,
|
|
'status': obj.status,
|
|
'retry': None,
|
|
'retry_max': None,
|
|
'error': obj.summary.get('error', '') if isinstance(obj.summary, dict) else '',
|
|
'updated_at': obj.finished_at or obj.started_at,
|
|
'extra': {
|
|
'duration': f'{duration}s' if duration is not None else '',
|
|
},
|
|
}
|
|
|
|
job_registry.register({
|
|
'id': 'netscan',
|
|
'label': 'Netscan',
|
|
'icon_color': 'text-blue-500',
|
|
'icon_path': (
|
|
'M9 3H5a2 2 0 00-2 2v4m6-6h10a2 2 0 012 2v4M9 3v10m0 0h10M9 13H5'
|
|
'm4 0v6m10-6v6m-5-6v6'
|
|
),
|
|
'title_label': 'Profile',
|
|
'status_choices': [
|
|
('all', 'All'), ('pending', 'Pending'), ('processing', 'Running'),
|
|
('completed', 'Success'), ('failed', 'Failed'),
|
|
],
|
|
'columns': ['id', 'title', 'status', 'error', 'updated'],
|
|
'get_stats': ns_stats,
|
|
'get_queryset': ns_queryset,
|
|
'serialize': ns_serialize,
|
|
'bulk_actions': {
|
|
'delete': 'bulk_delete_netscan_runs',
|
|
},
|
|
})
|