import json from django.shortcuts import render,redirect from django.views import View from django.views.decorators.cache import cache_control from django.utils.decorators import method_decorator from constants.general_const import ActiveStatus from core.models import AppSettings from core.utils import get_api_message from customermanagement.models import Customer from customermanagement.views import HasCustomerProfile from mydevicemanagement.models import CustomerDevice from mydevicemanagement.serializers import CustomerDeviceReadSerializer from products.models import Product from results.models import Result from usermanagement.models import UserProfile from codesofy.custom_config import set_user_profile,get_months_dic,get_privilleges,is_authorized,set_menu_items,get_global_master_details,get_local_date from django.utils import timezone from rest_framework import permissions, status from rest_framework.response import Response from django.db.models import Q,F from django.db.models import Sum,Count from django.db.models.functions import Coalesce from codesofy.master_details import InvoiceStatus,PaymentMethod,ProductType,PaymentStatus,SRNStatus from django.db import transaction, IntegrityError from django.contrib import messages from django.db.models.functions import ExtractYear, ExtractMonth import calendar from django.db.models import DecimalField, ExpressionWrapper,FloatField from django.utils.safestring import mark_safe from django.db.models.functions import Cast from django.db.models.functions import TruncMonth from django.utils.timezone import now from rest_framework.views import APIView from functools import lru_cache from decimal import Decimal from typing import Optional, Dict, Set, Any # add this line import logging app_logger = logging.getLogger("application-log") try: from geopy.geocoders import Nominatim _GEOPY_AVAILABLE = True _GEOCODER = Nominatim(user_agent="wile_dashboard") except Exception: _GEOPY_AVAILABLE = False _GEOCODER = None # Create your views here. menu_item = "dashboard" def set_sub_menu_item(sub_menu_item,context): context = set_menu_items(menu_item,sub_menu_item,context) return context def get_local_master_details(): context = get_global_master_details() return context @lru_cache(maxsize=4096) def city_from_latlon(lat: float, lon: float) -> Optional[str]: """ Reverse-geocode best-effort city/town/village. Falls back further if needed. Returns None if geocoding is unavailable or fails. """ if not _GEOPY_AVAILABLE or _GEOCODER is None: return None try: loc = _GEOCODER.reverse((lat, lon), language="en", exactly_one=True) if not loc or not loc.raw: return None addr = loc.raw.get("address", {}) or {} return ( addr.get("city") or addr.get("town") or addr.get("village") or addr.get("municipality") or addr.get("county") or addr.get("state") or addr.get("country") ) except Exception: return None @method_decorator(cache_control(no_cache=True, must_revalidate=True,no_store=True), name='dispatch') class WelcomeView(View): user_dashboard = "user_dashboard" back_office_dashboard = "back_office_dashboard" def get(self, request): context = get_local_master_details() user_profile = set_user_profile(request,context) if user_profile==None: return redirect('login') get_privilleges(user_profile,context) if is_authorized(user_profile,self.user_dashboard): return redirect('user-dashboard') elif is_authorized(user_profile,self.back_office_dashboard): return redirect('back-office-dashboard') else: return render(request, 'welcome.html',context) @method_decorator(cache_control(no_cache=True, must_revalidate=True, no_store=True), name='dispatch') class UserDashboardView(View): this_feature = "user_dashboard" sub_menu_item = "user_dashboard" def get(self, request): today = get_local_date() context = get_local_master_details() user_profile = set_user_profile(request, context) if user_profile is None: return redirect("login") get_privilleges(user_profile, context) if not is_authorized(user_profile, self.this_feature): return redirect("unauthorized-access") set_sub_menu_item(self.sub_menu_item, context) # Selected year (default = current year) year = int(request.GET.get("year", now().year)) # KPIs active_customer = Customer.objects.filter( customer_status=ActiveStatus.ACTIVE.value ).count() total_results = Result.objects.all().count() customerdevices = CustomerDevice.objects.all().count() # Customers per month (Python counting) customer_dates = CustomerDevice.objects.filter( created_at__year=year ).exclude(created_at__isnull=True).values_list('created_at', flat=True) customer_data = [0]*12 for dt in customer_dates: customer_data[dt.month - 1] += 1 # Results per month result_dates = Result.objects.filter( created_at__year=year ).exclude(created_at__isnull=True).values_list('created_at', flat=True) result_data = [0]*12 for dt in result_dates: result_data[dt.month - 1] += 1 # Months names months = [calendar.month_name[m] for m in range(1, 13)] # Years list current_year = now().year years = list(range(current_year - 5, current_year + 2)) #Bar Chart Data product_data = CustomerDevice.objects.values('product__name').annotate( total_count=Count('id') ).order_by('product__name') product_labels = [] product_counts = [] for p in product_data: product_labels.append(p['product__name']) product_counts.append(int(p['total_count'])) # ensure integers # ---------- City-wise unique devices + avg coordinates + uncategorized ---------- # pull minimal columns for speed rows = ( Result.objects .filter(created_at__year=year) .values("device_id", "gps_lat", "gps_lon") ) # data holders # city -> { "device_ids": set(), "sum_lat": float, "sum_lon": float, "n": int, "has_named_city": bool } city_data: dict[str, dict] = {} uncategorized_results_count = 0 for r in rows: device_id = r["device_id"] lat = r["gps_lat"] lon = r["gps_lon"] # no gps → uncategorized if lat is None or lon is None: uncategorized_results_count += 1 continue # decimals -> floats if isinstance(lat, Decimal): lat = float(lat) if isinstance(lon, Decimal): lon = float(lon) # normalize a bit to improve cache hit rate lat_r = round(lat, 5) lon_r = round(lon, 5) # try city name; if not available, we'll temporarily bucket by "lat,lon" city_label = city_from_latlon(lat_r, lon_r) # if geocoder not available or failed, we fallback to coordinate label if not city_label: city_label = f"{lat_r:.5f},{lon_r:.5f}" has_named_city = False else: has_named_city = True bucket = city_data.get(city_label) if not bucket: bucket = { "device_ids": set(), "sum_lat": 0.0, "sum_lon": 0.0, "n": 0, "has_named_city": has_named_city, } city_data[city_label] = bucket if device_id is not None: bucket["device_ids"].add(device_id) bucket["sum_lat"] += lat_r bucket["sum_lon"] += lon_r bucket["n"] += 1 # preserve if *any* sample resolved to a real city name bucket["has_named_city"] = bucket["has_named_city"] or has_named_city # build clean list for the template city_map_points = [] for label, b in city_data.items(): n = max(b["n"], 1) avg_lat = b["sum_lat"] / n avg_lon = b["sum_lon"] / n devices_count = len(b["device_ids"]) # if label is a fallback "lat,lon" but we *do* have a name in that bucket, re-resolve city_name = label if "," in label and b["has_named_city"]: # attempt one more time for a human-friendly name named = city_from_latlon(round(avg_lat, 5), round(avg_lon, 5)) if named: city_name = named city_map_points.append({ "city": city_name, "devices": devices_count, "lat": round(avg_lat, 5), "lon": round(avg_lon, 5), }) # sort by devices desc, then city asc city_map_points.sort(key=lambda x: (-x["devices"], x["city"])) # into context context["city_map_points"] = city_map_points context["uncategorized_results_count"] = uncategorized_results_count proucts = Product.objects.filter(status=ActiveStatus.ACTIVE.value).count() crop_counts = ( Result.objects .values('crop_type') # Group by crop_type .annotate(count=Count('crop_type')) # Count occurrences .order_by('-count') # Order descending ) # Get the most used crop_type most_used = crop_counts.first() # Returns the first item or None context.update({ 'product_labels': mark_safe(json.dumps(product_labels)), 'product_counts': mark_safe(json.dumps(product_counts)), 'most_used_crop': most_used, "today": today, "proucts":proucts, "active_customer": active_customer, "total_results": total_results, "customerdevices": customerdevices, "months": months, "customer_data": customer_data, "result_data": result_data, "years": years, "year": year, }) return render(request, "user-dashboard.html", context) @method_decorator(cache_control(no_cache=True, must_revalidate=True, no_store=True), name='dispatch') class AppDashboardView(APIView): """ Returns: - Current customer's name (first, last) - Total number of devices - Default device details (from AppSettings) """ permission_classes = [permissions.IsAuthenticated, HasCustomerProfile] def get(self, request): customer = request.user.customer first_name = customer.first_name last_name = customer.last_name try: total_devices = CustomerDevice.objects.filter(customer=customer).count() app_settings = AppSettings.objects.filter(customer_user=customer).first() default_device_data = None device = None test_mode = None if app_settings: test_mode = app_settings.test_mode if app_settings.default_device: device_info = app_settings.default_device device = CustomerDevice.objects.filter( id=device_info.get("device_id"), customer=customer ).select_related("product").first() if device: default_device_data = CustomerDeviceReadSerializer( device, context={"request": request} ).data else: # fallback: use stored JSON if device no longer exists default_device_data = device_info is_quick_test_device_set = bool(device and total_devices > 0) data = { "customer": { "first_name": first_name, "last_name": last_name, }, "total_devices": total_devices, "default_device": default_device_data or { "detail": get_api_message("NO_DEFAULT_DEVICE_SET", request) }, "test_mode": test_mode, "is_quick_test_device_set": is_quick_test_device_set, } app_logger.info("Dashboard data is extracted and sent") return Response(data, status=status.HTTP_200_OK) except Exception as e: app_logger.error(e) return Response( {"detail": "Something went wrong."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR )