202 lines
8.0 KiB
Python
202 lines
8.0 KiB
Python
"""
|
||
LyngSat related views for data synchronization and linking.
|
||
"""
|
||
from django.contrib import messages
|
||
from django.contrib.auth.mixins import LoginRequiredMixin
|
||
from django.shortcuts import redirect, render
|
||
from django.urls import reverse_lazy
|
||
from django.views import View
|
||
from django.views.generic import FormView
|
||
|
||
from ..forms import FillLyngsatDataForm, LinkLyngsatForm
|
||
from ..mixins import FormMessageMixin
|
||
from ..models import ObjItem
|
||
|
||
|
||
class LinkLyngsatSourcesView(LoginRequiredMixin, FormMessageMixin, FormView):
|
||
"""View for linking LyngSat sources to objects."""
|
||
|
||
template_name = "mainapp/link_lyngsat.html"
|
||
form_class = LinkLyngsatForm
|
||
success_message = "Привязка источников LyngSat завершена"
|
||
error_message = "Ошибка при привязке источников"
|
||
|
||
def form_valid(self, form):
|
||
from lyngsatapp.models import LyngSat
|
||
|
||
satellites = form.cleaned_data.get("satellites")
|
||
frequency_tolerance = form.cleaned_data.get("frequency_tolerance", 0.5)
|
||
|
||
# If satellites not selected, process all
|
||
if satellites:
|
||
objitems = ObjItem.objects.filter(
|
||
parameter_obj__id_satellite__in=satellites
|
||
).select_related('parameter_obj', 'parameter_obj__polarization')
|
||
else:
|
||
objitems = ObjItem.objects.filter(
|
||
parameter_obj__isnull=False
|
||
).select_related('parameter_obj', 'parameter_obj__polarization')
|
||
|
||
linked_count = 0
|
||
total_count = objitems.count()
|
||
|
||
for objitem in objitems:
|
||
if not hasattr(objitem, 'parameter_obj') or not objitem.parameter_obj:
|
||
continue
|
||
|
||
param = objitem.parameter_obj
|
||
|
||
# Round object frequency to 1 decimal place
|
||
if param.frequency:
|
||
rounded_freq = round(param.frequency, 1) # Round to 1 decimal place
|
||
|
||
# Find matching LyngSat source
|
||
# Compare by rounded frequency (with tolerance) and polarization
|
||
# LyngSat frequencies are also rounded to 1 decimal place for comparison
|
||
lyngsat_sources = LyngSat.objects.filter(
|
||
id_satellite=param.id_satellite,
|
||
polarization=param.polarization
|
||
).select_related('id_satellite', 'polarization')
|
||
|
||
# Filter by rounded frequency with tolerance
|
||
matching_sources = []
|
||
for lyngsat in lyngsat_sources:
|
||
if lyngsat.frequency:
|
||
rounded_lyngsat_freq = round(lyngsat.frequency, 1)
|
||
if abs(rounded_lyngsat_freq - rounded_freq) <= frequency_tolerance:
|
||
matching_sources.append(lyngsat)
|
||
|
||
if matching_sources:
|
||
# Take first matching source (sorted by frequency difference)
|
||
matching_sources.sort(key=lambda x: abs(round(x.frequency, 1) - rounded_freq))
|
||
objitem.lyngsat_source = matching_sources[0]
|
||
objitem.save(update_fields=['lyngsat_source'])
|
||
linked_count += 1
|
||
|
||
messages.success(
|
||
self.request,
|
||
f"Привязано {linked_count} из {total_count} объектов к источникам LyngSat"
|
||
)
|
||
return redirect("mainapp:link_lyngsat")
|
||
|
||
def form_invalid(self, form):
|
||
return self.render_to_response(self.get_context_data(form=form))
|
||
|
||
|
||
class FillLyngsatDataView(LoginRequiredMixin, FormMessageMixin, FormView):
|
||
"""
|
||
View for filling data from Lyngsat.
|
||
|
||
Allows selecting satellites and regions for parsing data from Lyngsat website.
|
||
Starts asynchronous Celery task for processing.
|
||
"""
|
||
template_name = "mainapp/fill_lyngsat_data.html"
|
||
form_class = FillLyngsatDataForm
|
||
success_url = reverse_lazy("mainapp:lyngsat_task_status")
|
||
error_message = "Форма заполнена некорректно"
|
||
|
||
def form_valid(self, form):
|
||
satellites = form.cleaned_data["satellites"]
|
||
regions = form.cleaned_data["regions"]
|
||
use_cache = form.cleaned_data.get("use_cache", True)
|
||
force_refresh = form.cleaned_data.get("force_refresh", False)
|
||
|
||
# Get satellite names
|
||
target_sats = [sat.name for sat in satellites]
|
||
|
||
try:
|
||
from lyngsatapp.tasks import fill_lyngsat_data_task
|
||
|
||
# Start asynchronous task with caching parameters
|
||
task = fill_lyngsat_data_task.delay(
|
||
target_sats,
|
||
regions,
|
||
force_refresh=force_refresh,
|
||
use_cache=use_cache
|
||
)
|
||
|
||
cache_status = "без кеша" if not use_cache else ("с обновлением кеша" if force_refresh else "с кешированием")
|
||
|
||
messages.success(
|
||
self.request,
|
||
f"Задача запущена ({cache_status})! ID задачи: {task.id}. "
|
||
"Вы будете перенаправлены на страницу отслеживания прогресса."
|
||
)
|
||
|
||
# Redirect to task status page
|
||
return redirect('mainapp:lyngsat_task_status', task_id=task.id)
|
||
|
||
except Exception as e:
|
||
messages.error(self.request, f"Ошибка при запуске задачи: {str(e)}")
|
||
return redirect("mainapp:fill_lyngsat_data")
|
||
|
||
|
||
class LyngsatTaskStatusView(LoginRequiredMixin, View):
|
||
"""View for tracking Lyngsat data filling task status."""
|
||
|
||
template_name = "mainapp/lyngsat_task_status.html"
|
||
|
||
def get(self, request, task_id=None):
|
||
context = {
|
||
'task_id': task_id
|
||
}
|
||
return render(request, self.template_name, context)
|
||
|
||
|
||
class ClearLyngsatCacheView(LoginRequiredMixin, View):
|
||
"""View for clearing LyngSat cache."""
|
||
|
||
def post(self, request):
|
||
from lyngsatapp.tasks import clear_cache_task
|
||
|
||
cache_type = request.POST.get('cache_type', 'all')
|
||
|
||
try:
|
||
# Start cache clearing task
|
||
task = clear_cache_task.delay(cache_type)
|
||
|
||
messages.success(
|
||
request,
|
||
f"Задача очистки кеша ({cache_type}) запущена! ID задачи: {task.id}"
|
||
)
|
||
except Exception as e:
|
||
messages.error(request, f"Ошибка при запуске задачи очистки кеша: {str(e)}")
|
||
|
||
return redirect(request.META.get('HTTP_REFERER', 'mainapp:home'))
|
||
|
||
def get(self, request):
|
||
"""Cache management page."""
|
||
return render(request, 'mainapp/clear_lyngsat_cache.html')
|
||
|
||
|
||
class UnlinkAllLyngsatSourcesView(LoginRequiredMixin, View):
|
||
"""View for unlinking all LyngSat sources from ObjItems."""
|
||
|
||
def post(self, request):
|
||
"""Unlink all LyngSat sources."""
|
||
try:
|
||
# Count objects with LyngSat sources before unlinking
|
||
linked_count = ObjItem.objects.filter(lyngsat_source__isnull=False).count()
|
||
|
||
# Unlink all LyngSat sources
|
||
ObjItem.objects.filter(lyngsat_source__isnull=False).update(lyngsat_source=None)
|
||
|
||
messages.success(
|
||
request,
|
||
f"Успешно отвязано {linked_count} объектов от источников LyngSat"
|
||
)
|
||
except Exception as e:
|
||
messages.error(request, f"Ошибка при отвязке источников: {str(e)}")
|
||
|
||
return redirect('mainapp:actions')
|
||
|
||
def get(self, request):
|
||
"""Show confirmation page."""
|
||
# Count objects with LyngSat sources
|
||
linked_count = ObjItem.objects.filter(lyngsat_source__isnull=False).count()
|
||
|
||
context = {
|
||
'linked_count': linked_count
|
||
}
|
||
return render(request, 'mainapp/unlink_lyngsat_confirm.html', context)
|