"""Scheduled + event-driven push to M&EL (PRD REPO-SP8).""" from __future__ import annotations import logging from datetime import date, timedelta from typing import Any import requests from django.conf import settings from django.db.models import Count, Q from django.utils import timezone logger = logging.getLogger(__name__) # FRREP-AR012 — daily-indicator payload keys (see # ``build_daily_indicator_payload``) that can drive an M&EL ``Indicator``. # Each is looked up as ``auto_event_type=f"repository_{key}"`` on an # ``Indicator`` with ``source_module="repository"`` (per # ``apps.mel.indicators.models.Indicator.auto_event_type``: "Event type key # matched when draining the integration outbox"). A key with no matching # active Indicator configured is simply skipped — MEL log-frame setup is out # of Repository's hands. DAILY_INDICATOR_METRIC_KEYS = ( "total_documents", "open_access_rate", "new_publications", "downloads_period", "unique_visitors_period", ) def build_publication_event_payload(document) -> dict[str, Any]: return { "event": "document_published", "document_id": document.pk, "public_document_id": document.public_document_id, "title": document.title, "collection": document.collection.slug, "document_type": document.document_type, "license_type": document.license_type, "open_access": document.is_open_access, "grant_reference": document.grant_reference, "published_at": document.published_at.isoformat() if document.published_at else None, "authors": [ { "full_name": a.full_name, "orcid": a.orcid, # FRREP-QA011 — institutional reach is part of M&EL's reporting. "institution": a.institution.name if a.institution_id else "", } for a in document.authors.all() ], } def build_daily_indicator_payload(on_date: date | None = None) -> dict[str, Any]: from apps.repository.analytics.models import AccessEvent, DownloadEvent from apps.repository.documents.models import Collection, Document on_date = on_date or timezone.now().date() day_start = on_date - timedelta(days=1) docs = Document.objects.filter(status=Document.Status.PUBLISHED) total = docs.count() open_count = docs.filter( Q(visibility=Document.Visibility.PUBLIC_OPEN) | Q( visibility=Document.Visibility.INHERIT, collection__visibility=Collection.Visibility.PUBLIC_OPEN, ) ).count() by_type = list( docs.values("document_type").annotate(total=Count("id")).order_by("-total") ) by_collection = list( docs.values("collection__slug").annotate(total=Count("id")).order_by("-total") ) new_today = docs.filter(published_at__date=on_date).count() downloads_today = DownloadEvent.objects.filter(timestamp__date=on_date).count() unique_visitors_today = ( AccessEvent.objects.filter(timestamp__date=on_date) .values("ip_address") .distinct() .count() ) return { "report_date": on_date.isoformat(), "total_documents": total, "open_access_count": open_count, "open_access_rate": round((open_count / total) * 100, 1) if total else 0.0, "by_type": by_type, "by_collection": by_collection, "new_publications": new_today, "downloads_period": downloads_today, "unique_visitors_period": unique_visitors_today, } def push_daily_indicators() -> int: """Send yesterday's indicators to MEL; also drain any queued event payloads.""" from apps.repository.documents.models import IntegrationOutbox count = IntegrationOutbox.objects.create( target=IntegrationOutbox.Target.MEL, event_type="daily_indicators", payload=build_daily_indicator_payload(), ) return 1 if count else 0 def _post_to_mel(payload: dict[str, Any]) -> None: url = settings.REPOSITORY_MEL_PUSH_URL if not url: raise RuntimeError("REPOSITORY_MEL_PUSH_URL not configured.") token = settings.REPOSITORY_MEL_PUSH_TOKEN headers = {"Content-Type": "application/json"} if token: headers["Authorization"] = f"Bearer {token}" resp = requests.post(url, json=payload, headers=headers, timeout=10) resp.raise_for_status() def deliver(outbox_row) -> None: """Delegate-friendly handler called from the generic outbox dispatcher. FRREP-AR012 — ``daily_indicators`` rows used to require ``REPOSITORY_MEL_PUSH_URL`` (an external HTTP endpoint) to go anywhere; that setting is blank in every env file this project ships, so the daily push silently no-op'd in every environment and M&EL never actually received anything. Repository and M&EL already run in the same Django process (mirroring the same-process pattern ``apps.repository.integration.rep_link.deliver`` uses for REP), so daily indicators are now written directly into ``apps.mel.indicators.DataPoint`` rows in-process instead of requiring a network hop. Other MEL event types (e.g. ``document_published``) are unaffected and keep the HTTP-or-log behaviour. """ if outbox_row.event_type == "daily_indicators": _deliver_daily_indicators_same_process(outbox_row.payload) return if settings.REPOSITORY_MEL_PUSH_URL: _post_to_mel(outbox_row.payload) else: logger.info("mel: REPOSITORY_MEL_PUSH_URL unset; logging payload only: %s", outbox_row.event_type) def _deliver_daily_indicators_same_process(payload: dict[str, Any]) -> None: """Write the daily indicator payload into M&EL ``DataPoint`` rows in-process. Idempotent: ``record_data_point`` dedupes on ``(indicator, source_module, source_object_id)``, so replaying the same ``report_date`` (e.g. a retried Celery task) is a no-op rather than a duplicate data point. """ from apps.repository.documents.audit_helper import audit_repository from apps.mel.indicators.models import Indicator from apps.mel.indicators.services import record_data_point report_date = payload.get("report_date") or timezone.now().date().isoformat() matched = 0 written = 0 for key in DAILY_INDICATOR_METRIC_KEYS: value = payload.get(key) if value is None: continue auto_event_type = f"repository_{key}" indicator = Indicator.objects.filter( source_module="repository", auto_event_type=auto_event_type, is_active=True, ).first() if indicator is None: continue matched += 1 result = record_data_point( indicator=indicator, value=value, period_label=report_date, source_module="repository", source_event="daily_indicators", source_object_id=f"{report_date}:{key}", auto_verify=True, raise_on_invalid=False, ) if result.created: written += 1 # Explicit receipt (log + audit entry) so a misconfigured M&EL side (no # Indicator wired to source_module="repository") is visible instead of a # silent no-op — the same visibility gap that made the old HTTP-push # failure mode go unnoticed in the first place. if matched: logger.info( "mel.daily_indicators: matched=%s written=%s report_date=%s", matched, written, report_date, ) else: logger.warning( "mel.daily_indicators: no active M&EL Indicator wired to " "source_module='repository' for report_date=%s — nothing written", report_date, ) try: audit_repository( actor=None, action="MEL_DAILY_INDICATORS_PUSHED" if matched else "MEL_DAILY_INDICATORS_SKIPPED", target_model="IntegrationOutbox", object_id=report_date, changes={ "report_date": report_date, "matched_indicators": matched, "data_points_written": written, }, ) except Exception: # pragma: no cover — auditing must never break delivery logger.exception("mel.daily_indicators: failed to write audit receipt")