from django.contrib.auth.decorators import login_required from django.core.paginator import Paginator from django.http import HttpResponseForbidden from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse from django.views.decorators.http import require_POST from apps.core.audit.models import AuditLog from apps.core.permissions.roles import UserRole def handler403(request, exception=None): return render(request, "core/403.html", status=403) def root_redirect(request): if request.user.is_authenticated: return redirect("core:dashboard") return render(request, "core/landing.html", _landing_context(request)) def _featured_moodle_courses(limit=6): """Featured courses for the public landing page, sourced live from Moodle. Cached (15 min) so the public page never blocks on a Web Services round-trip. Returns ``(featured_list, total_visible_count)``; ``([], 0)`` when Moodle is unconfigured or unreachable. """ from types import SimpleNamespace from django.core.cache import cache cache_key = "landing_featured_moodle_courses" cached = cache.get(cache_key) if cached is not None: return cached from django.urls import reverse from apps.core.nav_context import _moodle_url from apps.mel.tracking.moodle_client import MoodleClient, MoodleError client = MoodleClient() if not client.configured: return [], 0 base = _moodle_url() try: # by_field gives category names + course images in one call. raw = (client.call("core_course_get_courses_by_field") or {}).get("courses") or [] except MoodleError: return [], 0 featured, total, thumb_urls = [], 0, {} for c in raw: cid = c.get("id") if not cid or cid == 1 or not c.get("visible", 1): continue total += 1 if len(featured) >= limit: continue # Course image (uploaded overview file) — a token-protected pluginfile URL, # so it's proxied through Django (see moodle_course_thumbnail) rather than # leaking the WS token into the public page. Falls back to a monogram tile. thumbnail = None for f in c.get("overviewfiles") or []: if f.get("fileurl"): thumb_urls[cid] = f["fileurl"] thumbnail = SimpleNamespace(url=reverse("core:moodle_course_thumbnail", args=[cid])) break featured.append(SimpleNamespace( title=c.get("fullname") or c.get("shortname") or "Course", category=c.get("categoryname") or "Course", description=c.get("summary") or "", url=f"{base}/course/view.php?id={cid}", thumbnail=thumbnail, instructor=None, estimated_hours=None, )) cache.set("landing_course_thumb_urls", thumb_urls, 3600) result = (featured, total) cache.set(cache_key, result, 900) return result def moodle_course_thumbnail(request, course_id): """Proxy a Moodle course overview image for the public landing page. Moodle course images are served from token-protected ``webservice/pluginfile.php`` URLs; fetch them server-side with the WS token (never exposing it to the browser) and cache the bytes. 404s cleanly so the card falls back to its monogram tile. """ from django.conf import settings from django.core.cache import cache from django.http import Http404, HttpResponse img_key = f"moodle_course_thumb_img:{course_id}" cached = cache.get(img_key) if cached is not None: content, ctype = cached resp = HttpResponse(content, content_type=ctype) resp["Cache-Control"] = "public, max-age=3600" return resp fileurl = (cache.get("landing_course_thumb_urls") or {}).get(int(course_id)) if not fileurl: # Mapping cold/stale — look this course's image up directly. from apps.mel.tracking.moodle_client import MoodleClient, MoodleError client = MoodleClient() if client.configured: try: res = client.call("core_course_get_courses_by_field", field="id", value=course_id) for c in (res or {}).get("courses", []): for f in c.get("overviewfiles") or []: if f.get("fileurl"): fileurl = f["fileurl"] break except MoodleError: pass if not fileurl or not settings.MOODLE_WS_TOKEN: raise Http404("no thumbnail") import requests sep = "&" if "?" in fileurl else "?" try: r = requests.get( f"{fileurl}{sep}token={settings.MOODLE_WS_TOKEN}", timeout=8, verify=getattr(settings, "MOODLE_WS_VERIFY", True), ) except requests.RequestException as exc: # pragma: no cover - network guard raise Http404("fetch failed") from exc ctype = r.headers.get("Content-Type", "") if r.status_code != 200 or not ctype.startswith("image/"): raise Http404("not an image") cache.set(img_key, (r.content, ctype), 3600) resp = HttpResponse(r.content, content_type=ctype) resp["Cache-Control"] = "public, max-age=3600" return resp def _landing_context(request): from django.utils import timezone from django.db.models import Q now = timezone.now() context = {} stats = {} try: from apps.rims.grants.models import GrantCall calls_qs = GrantCall.objects.filter( status=GrantCall.Status.PUBLISHED, closes_at__gte=now, ) context["open_calls"] = calls_qs.order_by("-opens_at")[:6] stats["open_calls"] = calls_qs.count() except Exception: context["open_calls"] = [] stats["open_calls"] = 0 featured, course_total = _featured_moodle_courses() context["featured_courses"] = featured stats["courses"] = course_total try: from apps.smehub.marketplace.models import ProductListing products_qs = ProductListing.objects.filter( status=ProductListing.Status.PUBLISHED ) context["featured_products"] = ( products_qs.select_related("business", "entrepreneur__user") .order_by("-created_at")[:6] ) stats["products"] = products_qs.count() except Exception: context["featured_products"] = [] stats["products"] = 0 try: from django.contrib.auth import get_user_model stats["members"] = get_user_model().objects.filter(is_active=True).count() except Exception: stats["members"] = 0 context["landing_stats"] = stats return context @login_required def dashboard_home(request): from apps.core.dashboard.base import ZONE_ORDER from apps.core.dashboard.registry import for_user from apps.core.dashboard.role_groups import RIMS_MANAGEMENT_ROLES, has_role, is_admin from apps.core.dashboard.base import ZONE_KPI widgets = for_user(request.user) zones = {z: [] for z in ZONE_ORDER} kpi_heading = kpi_action_url = kpi_action_label = "" for w in widgets: try: ctx = w.get_context(request.user) except Exception: # noqa: BLE001 — never let one widget break the dashboard import logging logging.getLogger(__name__).exception("Widget %s.get_context raised", w.key) continue zones[w.zone].append({"key": w.key, "template": w.template, "ctx": ctx}) # Let the first KPI-zone widget that declares a heading name the band. if w.zone == ZONE_KPI and not kpi_heading and getattr(w, "zone_heading", ""): kpi_heading = w.zone_heading kpi_action_label = getattr(w, "zone_heading_action_label", "") or "View all" url_name = getattr(w, "zone_heading_url_name", "") or "" if url_name: from django.urls import NoReverseMatch, reverse try: kpi_action_url = reverse(url_name) except NoReverseMatch: kpi_action_url = "" role = getattr(request.user, "role", "") or "" return render( request, "core/dashboard_home.html", { "page_title": "Dashboard", "user_role": role, "is_admin": is_admin(request.user), "is_manager": has_role(request.user, RIMS_MANAGEMENT_ROLES), "zones": zones, "has_widgets": any(zones[z] for z in ZONE_ORDER), "kpi_heading": kpi_heading, "kpi_action_url": kpi_action_url, "kpi_action_label": kpi_action_label, }, ) def _is_admin(user): return user.is_superuser or getattr(user, "role", "") == UserRole.ADMIN @login_required def audit_log_list(request): if not _is_admin(request.user): return HttpResponseForbidden() qs = AuditLog.objects.select_related("actor").all() action = request.GET.get("action", "").strip() model = request.GET.get("model", "").strip() actor_q = request.GET.get("actor", "").strip() method = request.GET.get("method", "").strip() date_from = request.GET.get("date_from", "").strip() date_to = request.GET.get("date_to", "").strip() if action: qs = qs.filter(action=action) if model: qs = qs.filter(target_model=model) if actor_q: qs = qs.filter(actor__email__icontains=actor_q) if method: qs = qs.filter(method=method) if date_from: qs = qs.filter(timestamp__date__gte=date_from) if date_to: qs = qs.filter(timestamp__date__lte=date_to) target_models = ( AuditLog.objects.exclude(target_model="") .values_list("target_model", flat=True) .distinct() .order_by("target_model") ) has_filters = any([action, model, actor_q, method, date_from, date_to]) paginator = Paginator(qs, 25) page = paginator.get_page(request.GET.get("page")) return render( request, "core/audit/log_list.html", { "page_obj": page, "actions": AuditLog.Action.choices, "target_models": target_models, "http_methods": ["GET", "POST", "PUT", "PATCH", "DELETE"], "has_filters": has_filters, "breadcrumbs": [("Dashboard", reverse("core:dashboard")), ("Audit log", "")], }, ) @login_required @require_POST def notification_mark_read(request, pk): from apps.core.notifications.models import Notification n = get_object_or_404(Notification, pk=pk, recipient=request.user) n.is_read = True n.save(update_fields=["is_read"]) return redirect("core:notification_inbox") @login_required def notification_go(request, pk): """Mark notification read and redirect to its action URL (or inbox if none).""" from apps.core.notifications.models import Notification n = get_object_or_404(Notification, pk=pk, recipient=request.user) if not n.is_read: n.is_read = True n.save(update_fields=["is_read"]) if n.action_url: return redirect(n.action_url) return redirect("core:notification_inbox") @login_required @require_POST def notification_mark_all_read(request): from django.http import HttpResponse from apps.core.notifications.models import Notification Notification.objects.filter(recipient=request.user, is_read=False).update(is_read=True) if getattr(request, "htmx", False): return HttpResponse( status=204, headers={"HX-Trigger": "notificationsUpdated"}, ) return redirect("core:notification_inbox") @login_required def notification_inbox(request): from apps.core.notifications.models import Notification qs = Notification.objects.filter(recipient=request.user) paginator = Paginator(qs, 20) page = paginator.get_page(request.GET.get("page")) unread = qs.filter(is_read=False).count() return render( request, "notifications/inbox.html", { "page_obj": page, "unread_count": unread, "breadcrumbs": [("Dashboard", reverse("core:dashboard")), ("Notifications", "")], }, ) @login_required def notification_badge_partial(request): from apps.core.notifications.models import Notification unread = Notification.objects.filter(recipient=request.user, is_read=False).count() return render( request, "notifications/partials/badge.html", {"unread_count": unread}, ) @login_required def notification_stream(request): """Server-Sent Events stream for unread-count + new-notification events. DEPRECATED / DORMANT (2026-05-24): no client opens this anymore. Under sync ``gthread`` gunicorn workers each open EventSource parked one of the 36 worker threads for its whole lifetime (up to ``NOTIFICATION_SSE_MAX_SECONDS``), so a handful of open tabs exhausted the pool and normal page navigations hung ("loads forever, stays on the current page"; nginx logged 502s on this path and 499s on real navigations). The notification badge now refreshes via HTMX polling (``every 45s`` on the badge element) plus an instant ``notificationsUpdated`` event on mark-read. Do NOT re-wire ``EventSource`` to this endpoint without first migrating gunicorn to ASGI/uvicorn (async generator) + nginx ``proxy_buffering off`` — long-lived streams must not run on sync workers. The view + URL are retained only so the existing test (``test_notifications_sse.py``) and middleware exemptions stay valid. Original behaviour: emits the initial unread count immediately, then re-polls the database every ``NOTIFICATION_SSE_INTERVAL_SECONDS`` (default 5s) and emits a ``notification`` event whenever the unread count or last id changes. Heartbeats every 30s keep proxies from closing the connection. Connections cap at ``NOTIFICATION_SSE_MAX_SECONDS`` (default 5 min); the browser auto-reconnects. """ import json import time from django.conf import settings as dj_settings from django.http import StreamingHttpResponse from apps.core.notifications.models import Notification user_id = request.user.pk interval = int(getattr(dj_settings, "NOTIFICATION_SSE_INTERVAL_SECONDS", 5) or 5) max_lifetime = int(getattr(dj_settings, "NOTIFICATION_SSE_MAX_SECONDS", 300) or 300) heartbeat_every = 30 def _snapshot(): qs = Notification.objects.filter(recipient_id=user_id) unread = qs.filter(is_read=False).count() last_id = qs.order_by("-id").values_list("id", flat=True).first() or 0 return unread, last_id def gen(): # Initial state unread, last_id = _snapshot() yield f"event: notification\ndata: {json.dumps({'unread': unread, 'last_id': last_id})}\n\n" prev = (unread, last_id) started = time.monotonic() last_heartbeat = started while time.monotonic() - started < max_lifetime: time.sleep(interval) try: cur = _snapshot() except Exception: # noqa: BLE001 — never crash the stream yield ": error\n\n" continue if cur != prev: yield f"event: notification\ndata: {json.dumps({'unread': cur[0], 'last_id': cur[1]})}\n\n" prev = cur now = time.monotonic() if now - last_heartbeat >= heartbeat_every: yield ": keepalive\n\n" last_heartbeat = now response = StreamingHttpResponse(gen(), content_type="text/event-stream") response["Cache-Control"] = "no-cache, no-transform" response["X-Accel-Buffering"] = "no" # disable nginx response buffering return response @login_required def global_search(request): """Cross-module search — businesses, products, events, programmes, calls, mentors, and repository documents, grouped by module and filtered to what the requesting user may see. """ from django.db.models import Q as _Q query = (request.GET.get("q") or "").strip() sections: list[dict] = [] def add_section(title, items, browse_url="", browse_label=""): if items: sections.append({ "title": title, "items": items, "browse_url": browse_url, "browse_label": browse_label, }) if len(query) >= 2: # ── SME-Hub ──────────────────────────────────────────────── try: from apps.smehub.marketplace.models import ProductListing products = ProductListing.objects.filter( _Q(name__icontains=query) | _Q(description__icontains=query), status=ProductListing.Status.PUBLISHED, ).select_related("business")[:8] add_section( "Marketplace products", [{ "label": p.name, "sublabel": p.business.name if p.business_id else "", "href": reverse("smehub_marketplace:product_detail", args=[p.pk]), } for p in products], reverse("smehub_marketplace:product_catalogue"), "Browse catalogue", ) except Exception: # noqa: BLE001 — a broken module must not kill search pass try: from apps.smehub.showcasing.models import ( InnovationChallenge, ShowcaseEvent, ) events = ShowcaseEvent.objects.filter( _Q(name__icontains=query) | _Q(summary__icontains=query), ).exclude(status=ShowcaseEvent.Status.DRAFT)[:8] add_section( "Showcase events", [{ "label": e.name, "sublabel": e.get_status_display(), "href": reverse("smehub_showcasing:event_detail", args=[e.pk]), } for e in events], reverse("smehub_showcasing:event_list"), "All events", ) challenges = InnovationChallenge.objects.filter( _Q(title__icontains=query) | _Q(summary__icontains=query), ).exclude(status=InnovationChallenge.Status.DRAFT)[:8] add_section( "Innovation challenges", [{ "label": c.title, "sublabel": c.get_status_display(), "href": reverse("smehub_showcasing:challenge_detail", args=[c.pk]), } for c in challenges], reverse("smehub_showcasing:challenge_list"), "All challenges", ) except Exception: # noqa: BLE001 pass try: from apps.smehub.incubation.models import Programme programmes = Programme.objects.filter( _Q(title__icontains=query) | _Q(excerpt__icontains=query), status=Programme.Status.PUBLISHED, )[:8] add_section( "Incubation programmes", [{ "label": p.title, "sublabel": p.get_call_type_display() if hasattr(p, "get_call_type_display") else "", "href": reverse("smehub_incubation:programme_detail", args=[p.slug]), } for p in programmes], reverse("smehub_incubation:programme_list"), "All programmes", ) except Exception: # noqa: BLE001 pass try: from apps.smehub.investment.models import FundingCall calls = FundingCall.objects.filter( _Q(title__icontains=query) | _Q(summary__icontains=query), status=FundingCall.Status.PUBLISHED, )[:8] add_section( "Funding calls", [{ "label": c.title, "sublabel": c.get_call_type_display() if hasattr(c, "get_call_type_display") else "", "href": reverse("smehub_investment:funding_call_detail", args=[c.pk]), } for c in calls], reverse("smehub_investment:funding_calls"), "All funding calls", ) except Exception: # noqa: BLE001 pass try: from apps.smehub.onboarding.models import MentorProfile mentors = MentorProfile.objects.filter( _Q(user__first_name__icontains=query) | _Q(user__last_name__icontains=query) | _Q(organisation__icontains=query), verification_status="verified", ).select_related("user")[:8] add_section( "Mentors", [{ "label": m.user.get_full_name() or m.user.email, "sublabel": getattr(m, "organisation", ""), "href": reverse("smehub_onboarding:mentor_detail", args=[m.pk]), } for m in mentors], reverse("smehub_onboarding:mentor_directory"), "Mentor directory", ) except Exception: # noqa: BLE001 pass # Verified businesses — admin-facing (detail pages are owner/admin gated). if getattr(request.user, "role", "") in ("admin", "system_admin", "sme_admin") or request.user.is_superuser: try: from apps.smehub.onboarding.models import Business businesses = Business.objects.filter( _Q(name__icontains=query) | _Q(description__icontains=query), verification_status="verified", )[:8] add_section( "Businesses", [{ "label": b.name, "sublabel": f"{b.sector} · {b.country}", "href": reverse("smehub_onboarding:business_detail_admin", args=[b.pk]), } for b in businesses], ) except Exception: # noqa: BLE001 pass # ── Repository documents ────────────────────────────────── try: from apps.repository.documents.models import Document docs = Document.objects.filter( _Q(title__icontains=query) | _Q(abstract__icontains=query), status="published", )[:8] add_section( "Repository documents", [{ "label": d.title, "sublabel": str(d.year or ""), "href": reverse("repo_documents:document_detail", args=[d.document_uid]), } for d in docs], reverse("repo_search:search") + f"?q={query}", "Full repository search", ) except Exception: # noqa: BLE001 pass return render( request, "core/global_search.html", { "query": query, "sections": sections, "total": sum(len(s["items"]) for s in sections), "breadcrumbs": [("Dashboard", reverse("core:dashboard")), ("Search", "")], }, )