import csv import json from datetime import datetime, timedelta from pathlib import Path from django.conf import settings from django.contrib.auth import get_user_model from django.contrib.auth.models import update_last_login from django.http import HttpResponseNotAllowed from django.http import FileResponse, Http404, JsonResponse from django.shortcuts import redirect, render from django.urls import reverse from django.utils.http import url_has_allowed_host_and_scheme from django.utils import timezone from django.views.decorators.csrf import csrf_exempt from .auth import delete_jwt_cookie, jwt_login_required, set_jwt_cookie from .services.api_client import ApiClientError, get_api_debug_log, reset_api_debug_log from .services.campaign_options_service import get_campaign_options from .services.ga4_service import get_ga4_acquisition_funnel, get_ga4_acquisition_funnel_all from .services.google_ads_service import get_google_ads_performance from .services.meta_ads_service import get_meta_ads_performance from .services.jwt_auth_service import create_access_token from .services.csv_data_service import load_crm_data from .services.csv_filter_service import apply_filters from .services.csv_kpi_service import calculate_kpis def _safe_next_url(request): next_url = request.POST.get("next") or request.GET.get("next") or reverse("dashboard") if url_has_allowed_host_and_scheme( next_url, allowed_hosts={request.get_host()}, require_https=request.is_secure(), ): return next_url return reverse("dashboard") def _authenticate_email_password(email, password): if not email or not password: return None user = ( get_user_model() .objects .filter(email__iexact=email.strip(), is_active=True) .first() ) if user and user.check_password(password): return user return None def _user_payload(user): return { "id": user.pk, "email": user.email, "username": user.get_username(), "is_staff": user.is_staff, } def _json_body(request): if not request.body: return {} try: return json.loads(request.body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): return {} def _issue_login_response(user, response, token=None): token = token or create_access_token(user) update_last_login(None, user) set_jwt_cookie(response, token) return token, response def login_view(request): if request.method == "GET": return render( request, "dashboard/login.html", { "next": _safe_next_url(request), "error": "", "email": "", }, ) if request.method != "POST": return HttpResponseNotAllowed(["GET", "POST"]) email = request.POST.get("email", "") password = request.POST.get("password", "") user = _authenticate_email_password(email, password) if user is None: return render( request, "dashboard/login.html", { "next": _safe_next_url(request), "error": "Email ou mot de passe incorrect.", "email": email, }, status=401, ) response = redirect(_safe_next_url(request)) _issue_login_response(user, response) return response def logout_view(request): response = redirect("login") delete_jwt_cookie(response) return response @csrf_exempt def api_login(request): if request.method != "POST": return JsonResponse( {"success": False, "error": "Methode non autorisee."}, status=405, ) body = _json_body(request) email = body.get("email") or request.POST.get("email", "") password = body.get("password") or request.POST.get("password", "") user = _authenticate_email_password(email, password) if user is None: return JsonResponse( {"success": False, "error": "Email ou mot de passe incorrect."}, status=401, ) token = create_access_token(user) response = JsonResponse( { "success": True, "token_type": "Bearer", "access_token": token, "expires_in": settings.JWT_ACCESS_TOKEN_SECONDS, "user": _user_payload(user), } ) _issue_login_response(user, response, token=token) return response @jwt_login_required def api_me(request): return JsonResponse({"success": True, "user": _user_payload(request.user)}) def _format_number(value): try: number = float(value) except (TypeError, ValueError): return "0" if number.is_integer(): return f"{int(number):,}".replace(",", " ") return f"{number:,.2f}".replace(",", " ") def _parse_percent_number(value): if value in (None, ""): return 0 if isinstance(value, str): value = value.strip().replace("%", "").replace(" ", "").replace(",", ".") try: return float(value) except (TypeError, ValueError): return 0 def _format_percent(value): return f"{_parse_percent_number(value):.2f}%" def _extract_response_json_object(text, markers=("reponse", "data retourner", "data :")): lower_text = text.lower() marker_index = -1 for marker in markers: marker_index = lower_text.find(marker) if marker_index != -1: break search_start = marker_index if marker_index != -1 else 0 start = text.find("{", search_start) if start == -1: return {} decoder = json.JSONDecoder() data, _ = decoder.raw_decode(text[start:]) return data def _load_parcours_acquisition_data(): analytics_data = _load_markdown_json("analyticss.md") if analytics_data: return analytics_data analytics_data = _load_markdown_json("analytics.md") if analytics_data: return analytics_data return _load_markdown_json( "parcours_acquisition.md", markers=("reponse", "exemple de data"), ) def _parse_client_id(value): try: return int(value) except (TypeError, ValueError): return value def _default_filter_dates(): today = timezone.localdate() current_start = today.replace(day=1) previous_end = current_start - timedelta(days=1) previous_start = previous_end.replace(day=1) return { "start_date": current_start.isoformat(), "end_date": today.isoformat(), "previous_start_date": previous_start.isoformat(), "previous_end_date": previous_end.isoformat(), } def _parse_offer_mix_date(value): if not value: return None text = str(value).strip() date_text = text[:10] for date_format in ("%Y-%m-%d", "%d/%m/%Y"): try: return datetime.strptime(date_text, date_format).date() except ValueError: continue return None OFFER_FILTER_MAP = { "start": "FORFAIT_START", "forfait_start": "FORFAIT_START", "dynamic": "FORFAIT_DYN", "dyn": "FORFAIT_DYN", "forfait_dyn": "FORFAIT_DYN", "signa": "FORFAIT_SIG", "sig": "FORFAIT_SIG", "forfait_sig": "FORFAIT_SIG", } def _selected_offer_codes(filters): selected_offer_codes = { OFFER_FILTER_MAP.get(str(forfait).strip().lower()) for forfait in (filters or {}).get("forfaits", []) } selected_offer_codes.discard(None) return selected_offer_codes def _build_offer_mix_context(filters=None): data_dir = Path(settings.BASE_DIR) / "dashboard" / "data" csv_path = data_dir / "mix forfait carte.csv" if not csv_path.exists(): csv_path = data_dir / "mix-forfait-carte.csv" offer_config = [ { "code": "FORFAIT_START", "label": "Start", "id": "bar-mix-start", "value_id": "bar-mix-start-val", "classes": "bg-emerald-500/80 text-white", "label_classes": "text-emerald-500", }, { "code": "FORFAIT_DYN", "label": "Dynamic", "id": "bar-mix-dynamic", "value_id": "bar-mix-dynamic-val", "classes": "bg-sky-500/80 text-white", "label_classes": "text-sky-500", }, { "code": "FORFAIT_SIG", "label": "Signa.", "id": "bar-mix-signa", "value_id": "bar-mix-signa-val", "classes": "bg-indigo-500/80 text-white", "label_classes": "text-indigo-500", }, ] card_config = { "CARD_VISA_CLASSIC": {"label": "Visa", "color": "#435E5A"}, "CARD_VISA_GOLD": {"label": "Gold", "color": "#B19E58"}, "CARD_MC_TITANIUM": {"label": "Titanium", "color": "#1A2436"}, "CARD_VISA_PLATINUM": {"label": "Platinum", "color": "#888889"}, "CARD_VISA_INFINITE": {"label": "Infinite", "color": "#373633"}, } selected_offer_codes = _selected_offer_codes(filters) counts = {offer["code"]: 0 for offer in offer_config} card_counts = {offer["code"]: {} for offer in offer_config} total_rows = 0 start_date = _parse_offer_mix_date((filters or {}).get("start_date")) end_date = _parse_offer_mix_date((filters or {}).get("end_date")) if not csv_path.exists(): return {"total": 0, "segments": [], "card_rows": []} with csv_path.open(newline="", encoding="utf-8-sig") as file: for row in csv.DictReader(file): row_date = _parse_offer_mix_date( row.get("jour_selection") or row.get("jour") ) if (start_date or end_date) and row_date is None: continue if start_date and row_date < start_date: continue if end_date and row_date > end_date: continue offer_code = str(row.get("selected_offer_code", "")).strip().upper() if offer_code not in counts: continue if selected_offer_codes and offer_code not in selected_offer_codes: continue total_rows += 1 counts[offer_code] += 1 card_code = str(row.get("selected_card_code", "")).strip().upper() if card_code: card_counts[offer_code][card_code] = ( card_counts[offer_code].get(card_code, 0) + 1 ) segments = [] card_rows = [] for offer in offer_config: count = counts[offer["code"]] if count == 0: continue percent = ( round((count / total_rows) * 100, 2) if total_rows else 0 ) segments.append( { **offer, "count": count, "percent": percent, "percent_label": _format_percent(percent), } ) cards = [] for card_code, card_count in sorted( card_counts[offer["code"]].items(), key=lambda item: item[1], reverse=True, ): card = card_config.get( card_code, { "label": card_code.replace("CARD_", "").replace("_", " ").title(), "color": "#64748b", }, ) card_percent = round((card_count / count) * 100, 2) if count else 0 cards.append( { "code": card_code, "label": card["label"], "color": card["color"], "count": card_count, "percent": card_percent, "percent_label": _format_percent(card_percent), } ) card_rows.append( { "code": offer["code"], "label": offer["label"], "label_classes": offer["label_classes"], "count": count, "cards": cards, } ) return { "total": total_rows, "segments": segments, "card_rows": card_rows, } def _csv_number(value): try: return float(str(value or "0").strip().replace(",", ".")) except ValueError: return 0 def _build_terminal_systems_context(filters=None): data_dir = Path(settings.BASE_DIR) / "dashboard" / "data" csv_path = data_dir / "taux de conversion.csv" if not csv_path.exists(): csv_path = data_dir / "taux-de-completion.csv" system_config = [ { "key": "ios", "code": "IOS", "label": "iOS", "icon": "ph-fill ph-apple-logo", "icon_classes": "text-slate-800 dark:text-white", "bar_classes": "bg-indigo-500", }, { "key": "android", "code": "ANDROID", "label": "Android", "icon": "ph-fill ph-android-logo", "icon_classes": "text-emerald-500", "bar_classes": "bg-emerald-500", }, { "key": "unknown", "code": "INCONNU", "label": "Inconnu", "icon": "ph-fill ph-question", "icon_classes": "text-slate-500", "bar_classes": "bg-slate-500", }, ] totals = { system["code"]: {"entrant": 0, "sortant": 0, "rate_sum": 0, "rows": 0} for system in system_config } completion_times = _build_terminal_completion_times_context(filters) start_date = _parse_offer_mix_date((filters or {}).get("start_date")) end_date = _parse_offer_mix_date((filters or {}).get("end_date")) if csv_path.exists(): with csv_path.open(newline="", encoding="utf-8-sig") as file: for row in csv.DictReader(file): if str(row.get("transition", "")).strip().upper() != "PARCOURS_COMPLET": continue row_date = _parse_offer_mix_date(row.get("periode") or row.get("jour")) if (start_date or end_date) and row_date is None: continue if start_date and row_date < start_date: continue if end_date and row_date > end_date: continue system_code = str(row.get("systeme", "") or row.get("system", "")).strip().upper() if system_code not in totals: continue totals[system_code]["entrant"] += _csv_number(row.get("volume_entrant")) totals[system_code]["sortant"] += _csv_number(row.get("volume_sortant")) totals[system_code]["rate_sum"] += _csv_number(row.get("taux_conversion_pct")) totals[system_code]["rows"] += 1 rows = [] for system in system_config: total = totals[system["code"]] completion_time = completion_times.get(system["code"], {}) if total["rows"]: conversion = round(total["rate_sum"] / total["rows"], 2) else: conversion = 0 rows.append( { **system, "completion": _format_percent(conversion), "completion_width": min(max(conversion, 0), 100), "volume_label": f"{_format_number(total['sortant'])}/{_format_number(total['entrant'])}", "avg_time": completion_time.get("average", 0), "avg_time_label": _format_minutes(completion_time.get("average", 0)), "avg_time_width": completion_time.get("width", 0), "avg_time_volume_label": _format_number(completion_time.get("count", 0)), } ) return rows def _build_terminal_completion_times_context(filters=None): data_dir = Path(settings.BASE_DIR) / "dashboard" / "data" csv_path = data_dir / "Funnel transition.csv" if not csv_path.exists(): csv_path = data_dir / "funnel transition.csv" totals = { "IOS": {"sum": 0, "count": 0}, "ANDROID": {"sum": 0, "count": 0}, } start_date = _parse_offer_mix_date((filters or {}).get("start_date")) end_date = _parse_offer_mix_date((filters or {}).get("end_date")) selected_devices = _selected_device_codes(filters) if csv_path.exists(): with csv_path.open(newline="", encoding="utf-8-sig") as file: for row in csv.DictReader(file): row_date = _parse_offer_mix_date(row.get("jour")) if (start_date or end_date) and row_date is None: continue if start_date and row_date < start_date: continue if end_date and row_date > end_date: continue system_code = str(row.get("systeme", "") or row.get("system", "")).strip().upper() if system_code not in totals: continue if selected_devices and system_code not in selected_devices: continue raw_completion_time = str(row.get("temps_completion", "")).strip() if not raw_completion_time: continue completion_time = _csv_number(raw_completion_time) totals[system_code]["sum"] += completion_time totals[system_code]["count"] += 1 averages = { system_code: ( round(total["sum"] / total["count"], 2) if total["count"] else 0 ) for system_code, total in totals.items() } max_average = max(averages.values()) if averages else 0 return { system_code: { "average": average, "count": totals[system_code]["count"], "width": round((average / max_average) * 100, 2) if max_average else 0, } for system_code, average in averages.items() } def _format_minutes(value): number = _to_number(value) if number <= 0: return "0 min" if number.is_integer(): return f"{int(number)} min" return f"{number:.2f} min" def _parse_csv_bool(value): return str(value or "").strip().lower() in {"true", "1", "yes", "oui"} def _format_mix_percent(value): number = round(float(value or 0), 2) if number.is_integer(): return f"{int(number)}%" return f"{number:.2f}%" def _build_cdc_mix_context(filters=None, start_key="start_date", end_key="end_date"): data_dir = Path(settings.BASE_DIR) / "dashboard" / "data" csv_path = data_dir / "funnel snapshot.csv" if not csv_path.exists(): csv_path = data_dir / "funnel-snapshot.csv" counts = { "cdc": 0, "non_cdc": 0, } start_date = _parse_offer_mix_date((filters or {}).get(start_key)) end_date = _parse_offer_mix_date((filters or {}).get(end_key)) selected_offer_codes = _selected_offer_codes(filters) if csv_path.exists(): with csv_path.open(newline="", encoding="utf-8-sig") as file: for row in csv.DictReader(file): row_date = _parse_offer_mix_date(row.get("jour")) if (start_date or end_date) and row_date is None: continue if start_date and row_date < start_date: continue if end_date and row_date > end_date: continue offer_code = str(row.get("selected_offer_code", "")).strip().upper() if selected_offer_codes and offer_code not in selected_offer_codes: continue if _parse_csv_bool(row.get("isCDC")): counts["cdc"] += 1 else: counts["non_cdc"] += 1 total = counts["cdc"] + counts["non_cdc"] cdc_percent = round((counts["cdc"] / total) * 100, 2) if total else 0 non_cdc_percent = round((counts["non_cdc"] / total) * 100, 2) if total else 0 return { "total": total, "cdc": { "count": counts["cdc"], "count_label": _format_number(counts["cdc"]), "percent": cdc_percent, "percent_label": _format_mix_percent(cdc_percent), }, "non_cdc": { "count": counts["non_cdc"], "count_label": _format_number(counts["non_cdc"]), "percent": non_cdc_percent, "percent_label": _format_mix_percent(non_cdc_percent), }, } def _build_acquisition_payload(filters): return { "client_id": _parse_client_id(filters["client_id"]), "start_date": filters["start_date"], "end_date": filters["end_date"], "previous_start_date": filters["previous_start_date"], "previous_end_date": filters["previous_end_date"], } def _period_matches(actual, expected): return actual not in (None, "") and str(actual) == str(expected) def _validate_acquisition_period(data, payload): return None def _validate_paid_media_period(data, payload, label): period = data.get("period", {}) current = period.get("current", {}) previous = period.get("previous", {}) mismatches = [] checks = ( ("current since", current.get("since"), payload.get("since")), ("current until", current.get("until"), payload.get("until")), ("previous since", previous.get("since"), payload.get("previous_since")), ("previous until", previous.get("until"), payload.get("previous_until")), ) for name, actual, expected in checks: if not _period_matches(actual, expected): mismatches.append(f"{name} attendu {expected}, recu {actual or 'absent'}") if mismatches: return f"{label} periode differente: {'; '.join(mismatches)}" return None def _add_source_warning(data, warning): if not warning: return data existing_warning = data.get("warning", "") data["warning"] = ( f"{existing_warning} | {warning}" if existing_warning else warning ) return data def _load_acquisition_api_data(filters): payload = _build_acquisition_payload(filters) data = get_ga4_acquisition_funnel(payload) _validate_acquisition_period(data, payload) return data def _load_ga4_all_data(client_id=10, start_date=None, end_date=None, bearer_token=None): payload = {"client_id": _parse_client_id(client_id)} if start_date: payload["start_date"] = start_date if end_date: payload["end_date"] = end_date return get_ga4_acquisition_funnel_all(payload, bearer_token=bearer_token) def _extract_onboarding_start(data, start_date=None, end_date=None): if not isinstance(data, dict): return None event = data.get("event", {}) if event == "onboarding_start": daily_data = data.get("daily_data") if isinstance(daily_data, list): start = _parse_offer_mix_date(start_date) end = _parse_offer_mix_date(end_date) daily_total = 0 for row in daily_data: if not isinstance(row, dict): continue row_date = _parse_offer_mix_date(row.get("date")) if (start or end) and row_date is None: continue if start and row_date < start: continue if end and row_date > end: continue daily_total += _to_number(row.get("count")) return daily_total return _to_optional_number(data.get("total")) if isinstance(event, dict) and event.get("name") == "onboarding_start": return _to_optional_number(event.get("count")) events_detected = data.get("events_detected", {}) if not isinstance(events_detected, dict): return None return _to_optional_number(events_detected.get("onboarding_start")) def _load_markdown_json(filename, markers=("reponse", "data retourner", "data :", "exemple de data")): markdown_path = Path(settings.BASE_DIR) / filename if not markdown_path.exists(): return {} try: return _extract_response_json_object( markdown_path.read_text(encoding="utf-8"), markers=markers, ) except (OSError, json.JSONDecodeError): return {} def _selected_acquisition_key(filters): devices = { device.strip().lower() for device in filters.get("devices", []) if device.strip() } if devices == {"ios"}: return "ios" if devices == {"android"}: return "android" return "total" def _get_acquisition_payload_data(data, filters): response_data = data.get("data", {}) if isinstance(response_data, dict): selected_key = _selected_acquisition_key(filters) selected_by_filter = response_data.get(selected_key) if isinstance(selected_by_filter, dict): return selected_by_filter api_selected_key = data.get("filters", {}).get("selected_key") if api_selected_key and isinstance(response_data.get(api_selected_key), dict): return response_data[api_selected_key] selected = response_data.get("selected") if isinstance(selected, dict): return selected total = response_data.get("total") if isinstance(total, dict): return total return data def _extract_cpl_value(data, period): for key in ("cpl", "cost_per_prospect", "coutParProspect"): value = data.get(key) if isinstance(value, dict): if period in value: return _to_optional_number(value.get(period)) if period == "current" and "value" in value: return _to_optional_number(value.get("value")) if period == "previous" and "previous" in value: return _to_optional_number(value.get("previous")) elif period == "current": number = _to_optional_number(value) if number is not None: return number return None def _funnel_conversion_rate(numerator, denominator): denominator = _to_number(denominator) if denominator == 0: return 0 return round((_to_number(numerator) / denominator) * 100, 2) def _calculate_acquisition_conversion_rates(stage_metrics, rate_config): rates = [] for key, label, numerator_key, denominator_key in rate_config: numerator = stage_metrics.get(numerator_key, {}) denominator = stage_metrics.get(denominator_key, {}) current_rate = _funnel_conversion_rate( numerator.get("current"), denominator.get("current"), ) previous_rate = _funnel_conversion_rate( numerator.get("previous"), denominator.get("previous"), ) variation = _to_number(numerator.get("variation")) rates.append( { "key": key, "label": label, "value": _format_percent(current_rate), "variation": _format_percent(variation), "variation_value": variation, } ) return rates def _source_item_label(item): source_medium = str(item.get("source_medium", "") or "").strip() channel_group = str(item.get("channel_group", "") or "").strip() if source_medium: return f"{channel_group} - {source_medium}" if channel_group else source_medium source = str(item.get("source", "") or "").strip() medium = str(item.get("medium", "") or "").strip() if source and medium: label = f"{source} / {medium}" return f"{channel_group} - {label}" if channel_group else label return source or medium or channel_group or "Source inconnue" def _top_sources_dataset(top_sources, source_type): source_data = top_sources.get(source_type, {}) items = source_data.get("items", []) if isinstance(source_data, dict) else [] rows = [] for item in items[:5]: value = _to_number(item.get("new_accounts")) rows.append( { "label": _source_item_label(item), "value": value, "value_label": _format_number(value), "channel_group": item.get("channel_group", ""), } ) return { "labels": [row["label"] for row in rows], "values": [row["value"] for row in rows], "rows": rows, } def _distribution_segment(distribution, paid_vs_organic, key, label): source = distribution.get(key, {}) if isinstance(distribution, dict) else {} if isinstance(source, dict) and source: value = source.get("value", 0) percent = source.get("percent", 0) else: value = paid_vs_organic.get(key, 0) percent = paid_vs_organic.get(f"{key}_percent", 0) percent = _to_number(percent) value = _to_number(value) return { "type": key, "label": label, "value": value, "value_label": _format_number(value), "percent": percent, "percent_label": _format_percent(percent), } def _new_accounts_distribution_context(selected_data, paid_vs_organic): distribution = selected_data.get("new_accounts_distribution", {}) segments = [ _distribution_segment(distribution, paid_vs_organic, "organic", "Organic"), _distribution_segment(distribution, paid_vs_organic, "paid", "Paid"), _distribution_segment(distribution, paid_vs_organic, "other", "Other"), ] return { "label": distribution.get( "label", "Répartition des nouveaux comptes : Paid vs Organic", ) if isinstance(distribution, dict) else "Répartition des nouveaux comptes : Paid vs Organic", "total": _to_number( distribution.get("total_new_accounts", paid_vs_organic.get("total", 0)) if isinstance(distribution, dict) else paid_vs_organic.get("total", 0) ), "segments": segments, } def _build_acquisition_context( data, filters, onboarding_start_value=None, previous_onboarding_start_value=None, ): selected_data = _get_acquisition_payload_data(data, filters) funnel = selected_data.get("funnel", {}) paid_vs_organic = selected_data.get("paid_vs_organic", {}) top_sources = selected_data.get("top_sources", {}) new_accounts_distribution = _new_accounts_distribution_context( selected_data, paid_vs_organic, ) warning = data.get("warning", {}) if not isinstance(warning, dict): warning = {"cpa": str(warning)} cpa = selected_data.get("cpa", {}) stage_config = [ # ("telechargements", "Telechargements", "h-32", None), # ("onboardings_kpi", "Onboardings", "h-28", "onboarding"), # ("donnees_personnelles", "Donnees personnelles", "h-24", None), # ("donnees_professionnelles", "Donnees professionnelles", "h-20", None), # ("documents_kpi", "Documents", "h-20", ("documents", "document_kpi")), # ( # "nouveaux_comptes_contract_signed", # "Nouveaux comptes", # "h-16", # None, # ), # ( # "nouveaux_comptes_cdc", # "Nouveaux comptes cdc", # "h-16", # "nouveaux_comptes_contract_signed", # ), ("sessions_web", "Sessions web", "h-32", ("sessions_web",)), ("downloads", "Telechargements", "h-28", ("downloads",)), ("installs", "Installations", "h-24", ("onboarding", "onboardings", "onboardings_kpi", "installs")), ( "prospects", "Prospects", "h-20", ("onboarding", "onboardings", "onboardings_kpi", "prospects"), ), ("new_accounts", "Nouveaux comptes", "h-16", ("new_accounts",)), ] rate_config = [ ( "session_to_download", "Sessions web > Telechargements", "downloads", "sessions_web", ), ( "download_to_install", "Telechargements > Installations", "installs", "downloads", ), ( "install_to_prospect", "Installations > Prospects", "prospects", "installs", ), ( "prospect_to_account", "Prospects > Nouveaux comptes", "new_accounts", "prospects", ), ] stages = [] stage_metrics = {} kpis = selected_data.get("kpis", {}) if not isinstance(kpis, dict): kpis = {} for key, label, height_class, lookup_keys in stage_config: if isinstance(lookup_keys, str): lookup_keys = (lookup_keys,) item = {} for lookup_key in lookup_keys: item = funnel.get(lookup_key, {}) or kpis.get(lookup_key, {}) if item: break current_value = _to_number(item.get("value")) previous_value = _to_number(item.get("previous")) raw_variation = item.get("variation_percent") source = item.get("source", "") if key in ("installs", "prospects") and onboarding_start_value is not None: current_value = _to_number(onboarding_start_value) source = "GA4 event: onboarding_start" if previous_onboarding_start_value is not None: previous_value = _to_number(previous_onboarding_start_value) raw_variation = None variation = ( _parse_percent_number(raw_variation) if raw_variation not in (None, "") else _variation(current_value, previous_value) ) stage_metrics[key] = { "current": current_value, "previous": previous_value, "variation": variation, } stages.append( { "key": key, "label": label if key in ("prospects", "nouveaux_comptes_cdc") else item.get("label") or label, "height_class": height_class, "value": _format_number(current_value), "previous": _format_number(previous_value), "variation": _format_percent(variation), "variation_value": variation, "source": source, } ) stage_values = {stage["key"]: stage["value"] for stage in stages} stage_rows = [ stages[:6], stages[6:], ] rates = _calculate_acquisition_conversion_rates(stage_metrics, rate_config) paid_percent = paid_vs_organic.get("paid_percent", 0) or 0 organic_percent = paid_vs_organic.get("organic_percent", 0) or 0 other_percent = paid_vs_organic.get("other_percent", 0) or 0 return { "raw": data, "label": selected_data.get("label", ""), "selected_key": _selected_acquisition_key(filters), "prospectsAcquisition": stage_values.get("prospects", "0"), "nouveauxComptesAcquisition": stage_values.get("new_accounts", "0"), "metrics": stage_metrics, "cpl": { "current": _extract_cpl_value(selected_data, "current"), "previous": _extract_cpl_value(selected_data, "previous"), }, "cpa": { "value": "0 MAD", "rawValue": 0, "available": False, }, "stages": stages, "stage_rows": stage_rows, "rates": rates, "paid_vs_organic": { "paid": _format_number(paid_vs_organic.get("paid", 0)), "organic": _format_number(paid_vs_organic.get("organic", 0)), "other": _format_number(paid_vs_organic.get("other", 0)), "paid_percent": paid_percent, "organic_percent": organic_percent, "other_percent": other_percent, "paid_percent_label": _format_percent(paid_percent), "organic_percent_label": _format_percent(organic_percent), "other_percent_label": _format_percent(other_percent), }, "new_accounts_distribution": new_accounts_distribution, "top_sources_charts": { "comptes": { "paidTitle": "Top 5 des sources Paid", "organicTitle": "Top 5 des sources organique", "paid": _top_sources_dataset(top_sources, "paid"), "organic": _top_sources_dataset(top_sources, "organic"), }, "cdc": { "paidTitle": "Top 5 des sources Paid CDC", "organicTitle": "Top 5 des sources organique CDC", "paid": {"labels": [], "values": [], "rows": []}, "organic": {"labels": [], "values": [], "rows": []}, }, }, "cpa_warning": warning.get("cpa", "") or cpa.get("source", ""), "period_warning": warning.get("period", ""), } def _to_number(value): try: return float(value or 0) except (TypeError, ValueError): return 0 def _to_optional_number(value): if value in (None, ""): return None try: return float(value) except (TypeError, ValueError): return None def _average_available_values(values): available_values = [value for value in values if value is not None] if not available_values: return 0 return round(sum(available_values) / len(available_values), 2) def _variation(current, previous): current = _to_number(current) previous = _to_number(previous) if previous == 0: return 0 return round(((current - previous) / previous) * 100, 2) def _format_money(value, currency="MAD"): return f"{_format_number(value)} {currency}" def _format_compact_number(value): number = _to_number(value) abs_number = abs(number) if abs_number >= 1_000_000: compact = number / 1_000_000 suffix = "M" elif abs_number >= 1_000: compact = number / 1_000 suffix = "K" else: return _format_number(number) if float(compact).is_integer(): return f"{int(compact)}{suffix}" return f"{compact:.1f}{suffix}" def _selected_values(request, *keys): values = [] for key in keys: values.extend(request.GET.getlist(key)) normalized_values = [] for value in values: normalized_values.extend(str(value).split(",")) return [value.strip() for value in normalized_values if value.strip()] def _campaign_source_rows(data): rows = data.get("campaigns") or data.get("top_5_campaigns_by_accounts") or [] normalized = [] for campaign in rows: current = dict(campaign.get("current", campaign) or {}) previous = dict(campaign.get("previous", {}) or {}) variation = campaign.get("variation", {}) campaign_id = campaign.get("campaign_id") or current.get("campaign_id") campaign_name = campaign.get("campaign_name") or current.get("campaign_name") for metrics in (current, previous): if "cost" not in metrics and "spend" in metrics: metrics["cost"] = metrics["spend"] if "spend" not in metrics and "cost" in metrics: metrics["spend"] = metrics["cost"] normalized.append( { "campaign_id": str(campaign_id), "campaign_name": campaign_name or str(campaign_id), "current": current, "previous": previous, "variation": variation, } ) return normalized def _parse_campaign_id(value): try: return int(value) except (TypeError, ValueError): return value def _campaign_ids_for_channel(filters, channel): ids = [] for value in filters["campaign_ids"]: if value.startswith(f"{channel}:"): ids.append(_parse_campaign_id(value.split(":", 1)[1])) return ids def _build_paid_media_payload(filters, channel): payload = { "client_id": _parse_client_id(filters["client_id"]), "since": filters["start_date"], "until": filters["end_date"], "previous_since": filters["previous_start_date"], "previous_until": filters["previous_end_date"], } campaign_ids = _campaign_ids_for_channel(filters, channel) if campaign_ids: payload["campaign_ids"] = campaign_ids return payload def _empty_paid_source(label, error=None): return { "label": label, "data": { "totals": {}, "campaigns": [], "top_5_campaigns_by_accounts": [], "error": error or "", }, } def _load_paid_media_sources(filters): selected_channels = filters["channels"] show_all_channels = not selected_channels sources = {} if show_all_channels or "google_ads" in selected_channels: try: google_payload = _build_paid_media_payload(filters, "google_ads") google_data = get_google_ads_performance(google_payload) _add_source_warning( google_data, _validate_paid_media_period(google_data, google_payload, "Google Ads"), ) sources["google_ads"] = {"label": "Google Ads", "data": google_data} except ApiClientError as error: sources["google_ads"] = _empty_paid_source("Google Ads", str(error)) if show_all_channels or "meta" in selected_channels: try: meta_payload = _build_paid_media_payload(filters, "meta") meta_data = get_meta_ads_performance(meta_payload) _add_source_warning( meta_data, _validate_paid_media_period(meta_data, meta_payload, "Meta"), ) sources["meta"] = {"label": "Meta", "data": meta_data} except ApiClientError as error: sources["meta"] = _empty_paid_source("Meta", str(error)) return sources def _build_filter_context(request, sources): selected_channels = _selected_values(request, "channels") selected_campaigns = _selected_values(request, "campaign_ids") selected_devices = _selected_values(request, "devices", "devices[]", "crm_devices") selected_forfaits = _selected_values(request, "forfaits", "forfaits[]", "crm_forfaits") default_dates = _default_filter_dates() if selected_channels: selected_campaigns = [ campaign for campaign in selected_campaigns if campaign.split(":", 1)[0] in selected_channels ] return { "client_id": request.GET.get("client_id", "10"), "start_date": request.GET.get( "start_date", request.GET.get("since", default_dates["start_date"]), ), "end_date": request.GET.get( "end_date", request.GET.get("until", default_dates["end_date"]), ), "previous_start_date": request.GET.get( "previous_start_date", request.GET.get("previous_since", default_dates["previous_start_date"]), ), "previous_end_date": request.GET.get( "previous_end_date", request.GET.get("previous_until", default_dates["previous_end_date"]), ), "channels": selected_channels, "campaign_ids": selected_campaigns, "devices": selected_devices, "forfaits": selected_forfaits, "forfait_options": [ "Start", "Dynamic", "Signa", ], "campaign_options": [], } def _empty_metrics(): return { "impressions": 0, "clicks": 0, "downloads": 0, "prospects": 0, "accounts": 0, "cost": 0, "spend": 0, "ctr": 0, "cpc": 0, "cpl": 0, "cpa": 0, } def _paid_metric_value(source, key): if key == "cost": return _to_number(source.get("cost", source.get("spend", 0))) if key == "spend": return _to_number(source.get("spend", source.get("cost", 0))) return _to_number(source.get(key)) def _sum_metric(target, source, keys): for key in keys: target[key] += _paid_metric_value(source, key) def _recalculate_ratios(metrics): impressions = metrics["impressions"] clicks = metrics["clicks"] prospects = metrics["prospects"] accounts = metrics["accounts"] cost = metrics["cost"] metrics["spend"] = cost metrics["ctr"] = round((clicks / impressions) * 100, 2) if impressions else 0 metrics["cpc"] = round(cost / clicks, 2) if clicks else 0 metrics["cpl"] = round(cost / prospects, 2) if prospects else 0 metrics["cpa"] = round(cost / accounts, 2) if accounts else 0 return metrics def _source_cpl(source, period, selected_ids=None): if selected_ids is not None and not selected_ids: return None totals = source["data"].get("totals", {}) return _to_optional_number(totals.get(period, {}).get("cpl")) def _build_paid_media_context(sources, filters, acquisition_context=None): selected_channels = filters["channels"] selected_campaigns = filters["campaign_ids"] show_all_channels = not selected_channels has_campaign_filter = bool(selected_campaigns) current = _empty_metrics() previous = _empty_metrics() campaign_rows = [] source_cpls = { "google_ads": {"current": None, "previous": None}, "meta": {"current": None, "previous": None}, "google_analytics": {"current": None, "previous": None}, } google_analytics_cpl = (acquisition_context or {}).get("cpl", {}) source_cpls["google_analytics"] = { "current": google_analytics_cpl.get("current"), "previous": google_analytics_cpl.get("previous"), } errors = [ f"{source['label']}: {source['data'].get('error')}" for source in sources.values() if source["data"].get("error") ] warnings = [ f"{source['label']}: {source['data'].get('warning')}" for source in sources.values() if source["data"].get("warning") ] for channel, source in sources.items(): if not show_all_channels and channel not in selected_channels: continue channel_campaigns = _campaign_source_rows(source["data"]) selected_ids = { value.split(":", 1)[1] for value in selected_campaigns if value.startswith(f"{channel}:") } source_cpls[channel] = { "current": _source_cpl( source, "current", selected_ids if has_campaign_filter else None, ), "previous": _source_cpl( source, "previous", selected_ids if has_campaign_filter else None, ), } if has_campaign_filter: channel_campaigns = [ campaign for campaign in channel_campaigns if campaign["campaign_id"] in selected_ids ] for campaign in channel_campaigns: campaign_cost = campaign["current"].get("cost", 0) campaign_accounts = campaign["current"].get("accounts", 0) campaign_rows.append( { "platform": source["label"], "name": campaign["campaign_name"], "accounts": _to_number(campaign_accounts), "accounts_label": _format_number(campaign_accounts), "budget": _format_money(campaign_cost), "cpc": _format_money(campaign["current"].get("cpc", 0)), "cpl": _format_money(campaign["current"].get("cpl", 0)), "cpa": _format_money(campaign["current"].get("cpa", 0)), "cdc_cpa": _format_money(0), "cdc": _format_number(0), } ) if has_campaign_filter: for campaign in channel_campaigns: _sum_metric( current, campaign["current"], ("impressions", "clicks", "downloads", "prospects", "accounts", "cost", "spend"), ) _sum_metric( previous, campaign["previous"], ("impressions", "clicks", "downloads", "prospects", "accounts", "cost", "spend"), ) else: totals = source["data"].get("totals", {}) _sum_metric( current, totals.get("current", {}), ("impressions", "clicks", "downloads", "prospects", "accounts", "cost", "spend"), ) _sum_metric( previous, totals.get("previous", {}), ("impressions", "clicks", "downloads", "prospects", "accounts", "cost", "spend"), ) current["accounts"] = _count_snapshot_rows_by_jour(filters) previous["accounts"] = _count_snapshot_rows_by_jour( filters, start_key="previous_start_date", end_key="previous_end_date", ) current = _recalculate_ratios(current) previous = _recalculate_ratios(previous) current["accounts_cdc"] = _count_snapshot_cdc_rows(filters) previous["accounts_cdc"] = _count_snapshot_cdc_rows( filters, start_key="previous_start_date", end_key="previous_end_date", ) current["cpa_cdc"] = round(current["cost"] / current["accounts_cdc"], 2) if current["accounts_cdc"] else 0 previous["cpa_cdc"] = round(previous["cost"] / previous["accounts_cdc"], 2) if previous["accounts_cdc"] else 0 variation = { key: _variation(current.get(key), previous.get(key)) for key in current } stage_config = [ ("impressions", "Impressions", "h-32", "bg-indigo-500/80", "border-indigo-400/50"), ("clicks", "Clics", "h-28", "bg-indigo-500/70", "border-indigo-400/40"), ("downloads", "Telechargements", "h-24", "bg-indigo-500/60", "border-indigo-400/30"), ("prospects", "Prospects", "h-20", "bg-indigo-500/50", "border-indigo-400/20"), ("accounts", "Comptes", "h-16", "bg-indigo-500/40", "border-indigo-400/20"), ("accounts_cdc", "Comptes CDC", "h-12", "bg-indigo-500/30", "border-indigo-400/10"), ] campaign_rows = sorted(campaign_rows, key=lambda row: row["accounts"], reverse=True) top_campaigns = campaign_rows[:5] current_cpl = _average_available_values( source_cpl["current"] for source_cpl in source_cpls.values() ) previous_cpl = _average_available_values( source_cpl["previous"] for source_cpl in source_cpls.values() ) conversion_config = [ ("clicks", "impressions", current["cpc"]), ("downloads", "clicks", None), ("prospects", "downloads", current_cpl), ("accounts", "prospects", current["cpa"]), ("accounts_cdc", "accounts", current["cpa_cdc"]), ] conversion_rows = [] for numerator_key, denominator_key, average_cost in conversion_config: current_rate = _funnel_conversion_rate( current.get(numerator_key), current.get(denominator_key), ) previous_rate = _funnel_conversion_rate( previous.get(numerator_key), previous.get(denominator_key), ) conversion_rows.append( { "rate": _format_percent(current_rate), "variation": _format_percent(_variation(current_rate, previous_rate)), "variation_value": _variation(current_rate, previous_rate), "cost": _format_money(average_cost) if average_cost is not None else "-", } ) cdc_target = 17_000 cdc_unit_cost_target = 1_400 budget_target = cdc_target * cdc_unit_cost_target budget_progress = _funnel_conversion_rate(current["cost"], budget_target) cdc_progress = _funnel_conversion_rate(current["accounts_cdc"], cdc_target) return { "stages": [ { "key": key, "label": label, "height_class": height_class, "bg_class": bg_class, "border_class": border_class, "value": _format_number(current[key]), "variation": _format_percent(variation[key]), "variation_value": variation[key], } for key, label, height_class, bg_class, border_class in stage_config ], "conversion_rows": conversion_rows, "totals": { "cost": _format_money(current["cost"]), "spend": _format_money(current["spend"]), "ctr": _format_percent(current["ctr"]), "cpc": _format_money(current["cpc"]), "cpa": _format_money(current["cpa"]), "cpa_cdc": _format_money(current["cpa_cdc"]), "cpa_cdc_variation": _format_percent(variation["cpa_cdc"]), "cpa_cdc_variation_value": variation["cpa_cdc"], "cost_variation": _format_percent(variation["cost"]), "cost_variation_value": variation["cost"], }, "budget_tracking": { "spent": _format_compact_number(current["cost"]), "spent_progress": _format_percent(budget_progress), "spent_width": min(max(budget_progress, 0), 100), "cdc_generated": _format_number(current["accounts_cdc"]), "cdc_progress": _format_percent(cdc_progress), "cdc_width": min(max(cdc_progress, 0), 100), "target_cdc": _format_number(cdc_target), "target_budget": _format_money(budget_target), "target_unit_cost": _format_money(cdc_unit_cost_target), }, "coutParProspect": _format_money(current_cpl), "metrics": { "coutParProspect": { "current": current_cpl, "previous": previous_cpl, }, "cpa": { "current": current["cpa"], "previous": previous["cpa"], "cost": current["cost"], "spend": current["spend"], "accounts": current["accounts"], }, "cpa_cdc": { "current": current["cpa_cdc"], "previous": previous["cpa_cdc"], "cost": current["cost"], "spend": current["spend"], "accounts_cdc": current["accounts_cdc"], }, }, "campaigns": campaign_rows[:8], "chart_labels": [campaign["name"] for campaign in top_campaigns], "chart_values": [campaign["accounts"] for campaign in top_campaigns], "errors": errors, "warnings": warnings, } def _apply_paid_media_cpa_to_acquisition(acquisition_context, paid_media_context): cpa_metrics = paid_media_context.get("metrics", {}).get("cpa", {}) has_paid_media_issue = bool(paid_media_context.get("errors")) if has_paid_media_issue: acquisition_context["cpa"] = { "value": "0 MAD", "rawValue": 0, "available": False, } acquisition_context["cpa_warning"] = ( "CPA indisponible: donnees Google Ads / Meta Ads manquantes ou periode invalide." ) return acquisition_context acquisition_context["cpa"] = { "value": paid_media_context.get("totals", {}).get("cpa", "0 MAD"), "rawValue": cpa_metrics.get("current", 0), "available": True, "cost": cpa_metrics.get("cost", 0), "accounts": cpa_metrics.get("accounts", 0), } acquisition_context["cpa_warning"] = "" return acquisition_context def _period_label(start_date, end_date, fallback): if start_date and end_date: return f"{start_date} au {end_date}" return fallback def _display_number(value): normalized = "".join( char for char in str(value or "") if char.isdigit() or char in ".,-" ).replace(",", ".") try: return float(normalized) if normalized else 0 except ValueError: return 0 def _okr_progression(current_value, objective_value): objective = _display_number(objective_value) if objective <= 0: return { "progression": "0%", "progression_value": 0, "progression_width": 0, } progression = round((_display_number(current_value) / objective) * 100, 2) return { "progression": f"{progression:.0f}%", "progression_value": progression, "progression_width": min(max(progression, 0), 100), } def _build_okr_rows( acquisition_context, paid_media_context, onboarding_start_value=None, previous_onboarding_start_value=None, cdc_mix=None, previous_cdc_mix=None, ): acquisition_metrics = acquisition_context.get("metrics", {}) prospects = acquisition_metrics.get("prospects", {}) prospects_value = ( _format_number(onboarding_start_value) if onboarding_start_value is not None else acquisition_context.get("prospectsAcquisition", "0") ) previous_prospects_value = ( previous_onboarding_start_value if previous_onboarding_start_value is not None else _to_number(prospects.get("previous")) ) current_prospects_value = _display_number(prospects_value) current_cdc_count = (cdc_mix or {}).get("cdc", {}).get("count", 0) previous_cdc_count = (previous_cdc_mix or {}).get("cdc", {}).get("count", 0) new_accounts_value = _format_number(current_cdc_count) prospects_objective = "85 000" new_accounts_objective = "17 000" cost_prospect_objective = "280 MAD" cdc_cost_objective = "1 400 MAD" return [ { "key": "prospectsEer", "chartField": "prospects", "label": "Nombre de prospects ayant demarre parcours EER digital", "objective": prospects_objective, "value": prospects_value, "rawValue": current_prospects_value, "previousValue": _format_number(previous_prospects_value), "previousRawValue": previous_prospects_value, **_okr_progression(prospects_value, prospects_objective), "variation": _format_percent(_variation(current_prospects_value, previous_prospects_value)), "variation_value": _variation(current_prospects_value, previous_prospects_value), }, { "key": "clientsCdc", "chartField": "clients", "label": "Nombre de nouveaux clients CDC acquis via digital parcours EER termine", "objective": new_accounts_objective, "value": new_accounts_value, "rawValue": _display_number(new_accounts_value), "previousValue": _format_number(previous_cdc_count), "previousRawValue": previous_cdc_count, **_okr_progression(new_accounts_value, new_accounts_objective), "variation": _format_percent(_variation(current_cdc_count, previous_cdc_count)), "variation_value": _variation(current_cdc_count, previous_cdc_count), }, { "key": "coutProspect", "chartField": "cost_per_prospect", "label": "Cout par prospect", "objective": cost_prospect_objective, "value": "0 MAD", "rawValue": 0, "previousValue": "0 MAD", "previousRawValue": 0, **_okr_progression(0, cost_prospect_objective), "variation": "+0.00%", "variation_value": 0, }, { "key": "coutCdc", "label": "Cout par acquisition CDC", "objective": cdc_cost_objective, "value": "0 MAD", "rawValue": 0, "previousValue": "0 MAD", "previousRawValue": 0, **_okr_progression(0, cdc_cost_objective), "variation": "+0.00%", "variation_value": 0, }, ] def _format_okr_chart_value(chart_field, value): number = _to_number(value) if chart_field in ("prospects", "clients"): return int(round(number)) return round(number, 2) def _parse_okr_daily_date(value): if not value: return None try: return datetime.strptime(str(value)[:10], "%Y-%m-%d").date() except ValueError: return None def _okr_week_start(day): return day - timedelta(days=day.weekday()) def _format_okr_bucket_label(day, granularity): if granularity == "day": return day.strftime("%d/%m") if granularity == "week": return f"Sem. du {day.strftime('%d/%m')}" return day.strftime("%m/%Y") def _empty_okr_grouped_values(): return { "day": {}, "week": {}, "month": {}, } def _add_okr_grouped_value(grouped, day, count): keys = { "day": day, "week": _okr_week_start(day), "month": day.replace(day=1), } for granularity, key in keys.items(): grouped[granularity][key] = grouped[granularity].get(key, 0) + count def _build_onboarding_start_grouped_values(data, start_date=None, end_date=None): grouped = _empty_okr_grouped_values() if not isinstance(data, dict): return grouped daily_data = data.get("daily_data") if not isinstance(daily_data, list): return grouped start = _parse_offer_mix_date(start_date) end = _parse_offer_mix_date(end_date) for row in daily_data: if not isinstance(row, dict): continue event_name = row.get("event_name") or data.get("event") if event_name != "onboarding_start": continue day = _parse_okr_daily_date(row.get("date")) if day is None: continue if start and day < start: continue if end and day > end: continue _add_okr_grouped_value(grouped, day, _to_number(row.get("count"))) return grouped def _build_cdc_clients_grouped_values(filters): grouped = _empty_okr_grouped_values() data_dir = Path(settings.BASE_DIR) / "dashboard" / "data" csv_path = data_dir / "funnel snapshot.csv" if not csv_path.exists(): csv_path = data_dir / "funnel-snapshot.csv" if not csv_path.exists(): return grouped start = _parse_offer_mix_date((filters or {}).get("start_date")) end = _parse_offer_mix_date((filters or {}).get("end_date")) selected_offer_codes = _selected_offer_codes(filters) with csv_path.open(newline="", encoding="utf-8-sig") as file: for row in csv.DictReader(file): day = _parse_offer_mix_date(row.get("jour")) if (start or end) and day is None: continue if start and day < start: continue if end and day > end: continue offer_code = str(row.get("selected_offer_code", "")).strip().upper() if selected_offer_codes and offer_code not in selected_offer_codes: continue if not _parse_csv_bool(row.get("isCDC")): continue _add_okr_grouped_value(grouped, day, 1) return grouped def _build_okr_granularities(prospect_grouped, client_grouped): granularities = {} for granularity in ("day", "week", "month"): prospect_values = prospect_grouped.get(granularity, {}) client_values = client_grouped.get(granularity, {}) buckets = sorted(set(prospect_values) | set(client_values)) if not buckets: continue granularities[granularity] = { "labels": [ _format_okr_bucket_label(bucket, granularity) for bucket in buckets ], "prospects": [ int(round(prospect_values[bucket])) if bucket in prospect_values else None for bucket in buckets ], "clients": [ int(round(client_values[bucket])) if bucket in client_values else None for bucket in buckets ], "cost_per_prospect": [0 for _ in buckets], } return granularities def _build_onboarding_start_granularities(data, filters): prospect_grouped = _build_onboarding_start_grouped_values( data, start_date=(filters or {}).get("start_date"), end_date=(filters or {}).get("end_date"), ) client_grouped = _build_cdc_clients_grouped_values(filters) return _build_okr_granularities(prospect_grouped, client_grouped) def _build_okr_period_labels(filters, okr_rows=None, onboarding_start_data=None): payload = { "labels": [ _period_label( filters.get("previous_start_date"), filters.get("previous_end_date"), "Periode precedente", ), _period_label( filters.get("start_date"), filters.get("end_date"), "Periode actuelle", ), ], } granularities = _build_onboarding_start_granularities( onboarding_start_data, filters, ) if granularities: payload["granularities"] = granularities for row in okr_rows or []: chart_field = row.get("chartField") if not chart_field: continue payload[chart_field] = [ _format_okr_chart_value(chart_field, row.get("previousRawValue")), _format_okr_chart_value(chart_field, row.get("rawValue")), ] return payload def _filter_crm_data_by_dates(df, start_date, end_date): if not start_date or not end_date or "Date_Contact" not in df.columns: return df try: start = df["Date_Contact"].dtype.type(start_date) end = df["Date_Contact"].dtype.type(end_date) except (TypeError, ValueError): return df return df[(df["Date_Contact"] >= start) & (df["Date_Contact"] <= end)] def _filter_crm_data_by_devices(df, devices): if not devices or "OS_Mobile" not in df.columns: return df normalized_devices = {device.strip().lower() for device in devices} return df[df["OS_Mobile"].fillna("").astype(str).str.lower().isin(normalized_devices)] def _filter_crm_data_by_forfaits(df, forfaits): if not forfaits or "Offre_Souhaitee" not in df.columns: return df return df[df["Offre_Souhaitee"].isin(forfaits)] def _crm_current_data(request): df = load_crm_data() if request.GET.get("start_date") and request.GET.get("end_date"): return apply_filters(df, request) return df def _crm_previous_data(request): if not request.GET.get("previous_start_date") or not request.GET.get("previous_end_date"): return None df = load_crm_data() df = _filter_crm_data_by_dates( df, request.GET.get("previous_start_date"), request.GET.get("previous_end_date"), ) df = _filter_crm_data_by_devices(df, _selected_values(request, "devices", "devices[]", "crm_devices")) return _filter_crm_data_by_forfaits( df, _selected_values(request, "forfaits", "forfaits[]", "crm_forfaits"), ) def _crm_kpis_with_previous(request): current_data = _crm_current_data(request) previous_data = _crm_previous_data(request) return ( calculate_kpis(current_data), calculate_kpis(previous_data) if previous_data is not None else None, ) def _percentage(part, total): total = _to_number(total) if total == 0: return 0 return round((_to_number(part) / total) * 100, 2) def _variation_from_maps(current, previous, key): if not previous: return 0 return _variation(current.get(key), previous.get(key)) def _selected_device_codes(filters): return { str(device).strip().upper().replace("IOS", "IOS") for device in (filters or {}).get("devices", []) if str(device).strip() } def _selected_skipped_offer_card_values(filters): value_map = { "true": True, "1": True, "yes": True, "oui": True, "skipped": True, "skip": True, "sans forfait": True, "false": False, "0": False, "no": False, "non": False, "not_skipped": False, "not skipped": False, "avec forfait": False, "start": False, "forfait_start": False, "dynamic": False, "dyn": False, "forfait_dyn": False, "signa": False, "sig": False, "forfait_sig": False, } values = { value_map[str(forfait).strip().lower()] for forfait in (filters or {}).get("forfaits", []) if str(forfait).strip().lower() in value_map } return values def _count_snapshot_completed_flags(filters, columns, start_key="start_date", end_key="end_date"): return _count_snapshot_matching_rows( filters, lambda row: all(_parse_csv_bool(row.get(column)) for column in columns), start_key=start_key, end_key=end_key, ) def _count_snapshot_known_salary_ranges(filters, start_key="start_date", end_key="end_date"): return _count_snapshot_matching_rows( filters, lambda row: str(row.get("salary_range_label", "")).strip().upper() not in ("", "INCONNU"), start_key=start_key, end_key=end_key, ) def _count_snapshot_offer_card_not_skipped(filters, start_key="start_date", end_key="end_date"): return _count_snapshot_matching_rows( filters, lambda row: not _parse_csv_bool(row.get("skipped_offer_card")), start_key=start_key, end_key=end_key, ) def _count_snapshot_offer_card_skipped(filters, start_key="start_date", end_key="end_date"): return _count_snapshot_matching_rows( filters, lambda row: _parse_csv_bool(row.get("skipped_offer_card")), start_key=start_key, end_key=end_key, ) def _count_snapshot_known_offer_codes(filters, start_key="start_date", end_key="end_date"): return _count_snapshot_matching_rows( filters, lambda row: str(row.get("selected_offer_code", "")).strip().upper() not in ("", "INCONNU"), start_key=start_key, end_key=end_key, ) def _count_snapshot_known_card_codes(filters, start_key="start_date", end_key="end_date"): return _count_snapshot_matching_rows( filters, lambda row: str(row.get("selected_card_code", "")).strip().upper() not in ("", "INCONNU"), start_key=start_key, end_key=end_key, ) def _has_selected_offer_card(row): return not _parse_csv_bool(row.get("skipped_offer_card")) def _count_snapshot_with_selected_offer_card(filters, row_matches, start_key="start_date", end_key="end_date"): return _count_snapshot_matching_rows( filters, lambda row: _has_selected_offer_card(row) and row_matches(row), start_key=start_key, end_key=end_key, ) def _count_snapshot_with_skipped_offer_card(filters, row_matches, start_key="start_date", end_key="end_date"): return _count_snapshot_matching_rows( filters, lambda row: _parse_csv_bool(row.get("skipped_offer_card")) and row_matches(row), start_key=start_key, end_key=end_key, ) def _count_snapshot_mdm_rows(filters, row_matches, start_key="start_date", end_key="end_date"): return _count_snapshot_matching_rows( filters, lambda row: str(row.get("prospect_category", "")).strip().upper() == "MRE" and row_matches(row), start_key=start_key, end_key=end_key, apply_skipped_offer_filter=False, ) def _is_escalated_crc_journey(row): return str(row.get("journey_status", "")).strip().upper() == "ESCALATED_CRC" def _count_snapshot_escalated_crc_journeys(filters, start_key="start_date", end_key="end_date"): return _count_snapshot_matching_rows( filters, _is_escalated_crc_journey, start_key=start_key, end_key=end_key, ) def _count_snapshot_finalized_journeys(filters, start_key="start_date", end_key="end_date"): return _count_snapshot_matching_rows( filters, lambda row: str(row.get("journey_status", "")).strip().upper() == "FINALISE", start_key=start_key, end_key=end_key, ) def _count_snapshot_cdc_rows(filters, start_key="start_date", end_key="end_date"): return _count_snapshot_matching_rows( filters, lambda row: _parse_csv_bool(row.get("isCDC")), start_key=start_key, end_key=end_key, ) def _count_snapshot_rows_by_jour(filters, start_key="start_date", end_key="end_date"): data_dir = Path(settings.BASE_DIR) / "dashboard" / "data" csv_path = data_dir / "funnel snapshot.csv" if not csv_path.exists(): csv_path = data_dir / "funnel-snapshot.csv" if not csv_path.exists(): return 0 start = _parse_offer_mix_date((filters or {}).get(start_key)) end = _parse_offer_mix_date((filters or {}).get(end_key)) count = 0 with csv_path.open(newline="", encoding="utf-8-sig") as file: for row in csv.DictReader(file): row_date = _parse_offer_mix_date(row.get("jour")) if (start or end) and row_date is None: continue if start and row_date < start: continue if end and row_date > end: continue count += 1 return count def _count_snapshot_matching_rows( filters, row_matches, start_key="start_date", end_key="end_date", apply_skipped_offer_filter=True, ): data_dir = Path(settings.BASE_DIR) / "dashboard" / "data" csv_path = data_dir / "funnel snapshot.csv" if not csv_path.exists(): csv_path = data_dir / "funnel-snapshot.csv" if not csv_path.exists(): return 0 start = _parse_offer_mix_date((filters or {}).get(start_key)) end = _parse_offer_mix_date((filters or {}).get(end_key)) selected_devices = _selected_device_codes(filters) selected_offer_codes = _selected_offer_codes(filters) selected_skipped_values = ( _selected_skipped_offer_card_values(filters) if apply_skipped_offer_filter else set() ) count = 0 with csv_path.open(newline="", encoding="utf-8-sig") as file: for row in csv.DictReader(file): row_date = _parse_offer_mix_date(row.get("jour")) if (start or end) and row_date is None: continue if start and row_date < start: continue if end and row_date > end: continue system_code = str(row.get("systeme", "")).strip().upper() if selected_devices and system_code not in selected_devices: continue offer_code = str(row.get("selected_offer_code", "")).strip().upper() if selected_offer_codes and offer_code not in selected_offer_codes: continue skipped_offer_card = _parse_csv_bool(row.get("skipped_offer_card")) if selected_skipped_values and skipped_offer_card not in selected_skipped_values: continue if row_matches(row): count += 1 return count def _apply_eer_snapshot_overrides(current_kpis, previous_kpis=None, filters=None): current = dict(current_kpis) has_previous_period = bool( (filters or {}).get("previous_start_date") and (filters or {}).get("previous_end_date") ) previous = dict(previous_kpis) if previous_kpis else ({} if has_previous_period else None) current["telephone_count"] = _count_snapshot_completed_flags( filters, ("a_mail", "a_telephone"), ) current["email_count"] = _count_snapshot_completed_flags(filters, ("a_cin",)) current["biometric_count"] = _count_snapshot_known_salary_ranges(filters) current["personal_info_count"] = _count_snapshot_escalated_crc_journeys(filters) current["contracts_count"] = _count_snapshot_finalized_journeys(filters) if previous is not None: previous["telephone_count"] = _count_snapshot_completed_flags( filters, ("a_mail", "a_telephone"), start_key="previous_start_date", end_key="previous_end_date", ) previous["email_count"] = _count_snapshot_completed_flags( filters, ("a_cin",), start_key="previous_start_date", end_key="previous_end_date", ) previous["biometric_count"] = _count_snapshot_known_salary_ranges( filters, start_key="previous_start_date", end_key="previous_end_date", ) previous["personal_info_count"] = _count_snapshot_escalated_crc_journeys( filters, start_key="previous_start_date", end_key="previous_end_date", ) previous["contracts_count"] = _count_snapshot_finalized_journeys( filters, start_key="previous_start_date", end_key="previous_end_date", ) return current, previous def _snapshot_variation(current, previous): return _variation(current, previous) if previous is not None else 0 def _build_snapshot_conversion_rows(stages): rows = [] for index in range(len(stages) - 1): current_rate = _percentage( stages[index + 1].get("raw_value"), stages[index].get("raw_value"), ) previous_rate = _percentage( stages[index + 1].get("previous_raw_value"), stages[index].get("previous_raw_value"), ) variation = _variation(current_rate, previous_rate) rows.append( { "rate": _format_percent(current_rate), "variation": _format_percent(variation), "variation_value": variation, "time": "0", } ) return rows def _build_detailed_with_offer_context(filters): stage_config = [ ("validation_contact", "Validation Tél & Email", "h-32", lambda start_key, end_key: _count_snapshot_with_selected_offer_card(filters, lambda row: _parse_csv_bool(row.get("a_mail")) and _parse_csv_bool(row.get("a_telephone")), start_key, end_key)), ("biometric_id", "Id. Biométrique", "h-28", lambda start_key, end_key: _count_snapshot_with_selected_offer_card(filters, lambda row: _parse_csv_bool(row.get("a_cin")), start_key, end_key)), ("personal_info", "Collecte Infos", "h-24", lambda start_key, end_key: _count_snapshot_with_selected_offer_card(filters, lambda row: str(row.get("salary_range_label", "")).strip().upper() not in ("", "INCONNU"), start_key, end_key)), ("offer_choice", "Choix du Forfait", "h-20", lambda start_key, end_key: _count_snapshot_with_selected_offer_card(filters, lambda row: str(row.get("selected_offer_code", "")).strip().upper() not in ("", "INCONNU"), start_key, end_key)), ("card_choice", "Choix de la Carte", "h-16", lambda start_key, end_key: _count_snapshot_with_selected_offer_card(filters, lambda row: str(row.get("selected_card_code", "")).strip().upper() not in ("", "INCONNU"), start_key, end_key)), ("agency_choice", "Choix Agence", "h-14", lambda start_key, end_key: _count_snapshot_with_selected_offer_card(filters, _is_escalated_crc_journey, start_key, end_key)), ("contract_signature", "Signature Contrats", "h-12", lambda start_key, end_key: _count_snapshot_with_selected_offer_card(filters, lambda row: str(row.get("journey_status", "")).strip().upper() == "FINALISE", start_key, end_key)), ] stages = [] for key, label, height_class, counter in stage_config: current = counter("start_date", "end_date") previous = counter("previous_start_date", "previous_end_date") variation = _snapshot_variation(current, previous) stages.append( { "key": key, "label": label, "height_class": height_class, "raw_value": current, "previous_raw_value": previous, "value": _format_number(current), "previous_value": _format_number(previous), "variation": _format_percent(variation), "variation_value": variation, } ) return { "stages": stages, "conversion_rows": _build_snapshot_conversion_rows(stages), } def _build_detailed_without_offer_context(filters): stage_config = [ ("validation_contact", "Validation Tél & Email", "h-32", lambda start_key, end_key: _count_snapshot_with_skipped_offer_card(filters, lambda row: _parse_csv_bool(row.get("a_mail")) and _parse_csv_bool(row.get("a_telephone")), start_key, end_key)), ("biometric_id", "Id. Biométrique", "h-28", lambda start_key, end_key: _count_snapshot_with_skipped_offer_card(filters, lambda row: _parse_csv_bool(row.get("a_cin")), start_key, end_key)), ("personal_info", "Collecte Infos", "h-24", lambda start_key, end_key: _count_snapshot_with_skipped_offer_card(filters, lambda row: str(row.get("salary_range_label", "")).strip().upper() not in ("", "INCONNU"), start_key, end_key)), ("subscription_refusal", "Refus Souscription", "h-20", lambda start_key, end_key: _count_snapshot_with_skipped_offer_card(filters, lambda row: True, start_key, end_key)), ("agency_choice", "Choix Agence", "h-16", lambda start_key, end_key: _count_snapshot_with_skipped_offer_card(filters, _is_escalated_crc_journey, start_key, end_key)), ("contract_signature", "Signature Contrats", "h-14", lambda start_key, end_key: _count_snapshot_with_skipped_offer_card(filters, lambda row: str(row.get("journey_status", "")).strip().upper() == "FINALISE", start_key, end_key)), ] stages = [] for key, label, height_class, counter in stage_config: current = counter("start_date", "end_date") previous = counter("previous_start_date", "previous_end_date") variation = _snapshot_variation(current, previous) stages.append( { "key": key, "label": label, "height_class": height_class, "raw_value": current, "previous_raw_value": previous, "value": _format_number(current), "previous_value": _format_number(previous), "variation": _format_percent(variation), "variation_value": variation, } ) return { "stages": stages, "conversion_rows": _build_snapshot_conversion_rows(stages), } def _build_detailed_mdm_context(filters): stage_config = [ ("validation_contact", "Validation Tél & Email", "h-32", lambda start_key, end_key: _count_snapshot_mdm_rows(filters, lambda row: _parse_csv_bool(row.get("a_mail")) and _parse_csv_bool(row.get("a_telephone")), start_key, end_key)), ("biometric_id", "Id. Biométrique", "h-28", lambda start_key, end_key: _count_snapshot_mdm_rows(filters, lambda row: _parse_csv_bool(row.get("a_cin")), start_key, end_key)), ("personal_info", "Collecte Infos", "h-24", lambda start_key, end_key: _count_snapshot_mdm_rows(filters, lambda row: str(row.get("salary_range_label", "")).strip().upper() not in ("", "INCONNU"), start_key, end_key)), ("agency_choice", "Choix Agence", "h-20", lambda start_key, end_key: _count_snapshot_mdm_rows(filters, _is_escalated_crc_journey, start_key, end_key)), ("contract_signature", "Signature Contrats", "h-16", lambda start_key, end_key: _count_snapshot_mdm_rows(filters, lambda row: str(row.get("journey_status", "")).strip().upper() == "FINALISE", start_key, end_key)), ] stages = [] for key, label, height_class, counter in stage_config: current = counter("start_date", "end_date") previous = counter("previous_start_date", "previous_end_date") variation = _snapshot_variation(current, previous) stages.append( { "key": key, "label": label, "height_class": height_class, "raw_value": current, "previous_raw_value": previous, "value": _format_number(current), "previous_value": _format_number(previous), "variation": _format_percent(variation), "variation_value": variation, } ) return { "stages": stages, "conversion_rows": _build_snapshot_conversion_rows(stages), } def _build_eer_context(current_kpis, previous_kpis=None, filters=None): current_kpis, previous_kpis = _apply_eer_snapshot_overrides( current_kpis, previous_kpis, filters, ) stage_config = [ ("telephone_count", "Validation Téléphone & Email", "h-32", "bg-indigo-500/90", "border-indigo-400/50"), ("email_count", "Identification Biométrique", "h-28", "bg-indigo-500/80", "border-indigo-400/50"), ("biometric_count", "Collecte Informations Personnelles", "h-24", "bg-indigo-500/70", "border-indigo-400/50"), ("personal_info_count", "Choix de l'agence", "h-20", "bg-indigo-500/60", "border-indigo-400/50"), ("contracts_count", "Signature Contrats", "h-16", "bg-indigo-500/50", "border-indigo-400/50"), ] friction_config = [ ("telephone_count", "Telephone", "total_leads"), ("email_count", "Email", "telephone_count"), ("biometric_count", "Biometrie", "email_count"), ("contracts_count", "Contrat", "personal_info_count"), ] os_rates = current_kpis.get("os_completion_rates", {}) os_days = current_kpis.get("os_avg_completion_days", {}) previous_os_rates = previous_kpis.get("os_completion_rates", {}) if previous_kpis else {} previous_os_days = previous_kpis.get("os_avg_completion_days", {}) if previous_kpis else {} conversion_rows = [] for index in range(len(stage_config) - 1): from_key = stage_config[index][0] to_key = stage_config[index + 1][0] current_rate = _percentage(current_kpis.get(to_key), current_kpis.get(from_key)) previous_rate = ( _percentage(previous_kpis.get(to_key), previous_kpis.get(from_key)) if previous_kpis else 0 ) variation = _variation(current_rate, previous_rate) if previous_kpis else 0 conversion_rows.append( { "rate": _format_percent(current_rate), "variation": _format_percent(variation), "variation_value": variation, "time": "0", } ) frictions = [] for key, label, previous_key in friction_config: error_rate = 100 - _percentage(current_kpis.get(key), current_kpis.get(previous_key)) previous_error_rate = None if previous_kpis: previous_error_rate = 100 - _percentage( previous_kpis.get(key), previous_kpis.get(previous_key), ) frictions.append( { "label": label, "error_rate": _format_percent(max(error_rate, 0)), "variation": _format_percent( _variation(error_rate, previous_error_rate) if previous_error_rate is not None else 0 ), "variation_value": ( _variation(error_rate, previous_error_rate) if previous_error_rate is not None else 0 ), } ) return { "stages": [ { "key": key, "label": label, "height_class": height_class, "value": _format_number(current_kpis.get(key, 0)), "variation": _format_percent(_variation_from_maps(current_kpis, previous_kpis, key)), "variation_value": _variation_from_maps(current_kpis, previous_kpis, key), "bg_class": bg_class, "border_class": border_class, } for key, label, height_class, bg_class, border_class in stage_config ], "conversion_rows": conversion_rows, "detailed_with_offer": _build_detailed_with_offer_context(filters), "detailed_without_offer": _build_detailed_without_offer_context(filters), "detailed_mdm": _build_detailed_mdm_context(filters), "frictions": frictions, "systems": [ { "key": "ios", "icon": "ph-fill ph-apple-logo", "completion": _format_percent(os_rates.get("ios", 0)), "completion_width": min(os_rates.get("ios", 0), 100), "completion_variation": _format_percent( _variation(os_rates.get("ios", 0), previous_os_rates.get("ios", 0)) if previous_kpis else 0 ), "completion_variation_value": ( _variation(os_rates.get("ios", 0), previous_os_rates.get("ios", 0)) if previous_kpis else 0 ), "completion_days": f"{os_days.get('ios', 0)} j", "days_variation": _format_percent( _variation(os_days.get("ios", 0), previous_os_days.get("ios", 0)) if previous_kpis else 0 ), "days_variation_value": ( _variation(os_days.get("ios", 0), previous_os_days.get("ios", 0)) if previous_kpis else 0 ), }, { "key": "android", "icon": "ph-fill ph-android-logo", "completion": _format_percent(os_rates.get("android", 0)), "completion_width": min(os_rates.get("android", 0), 100), "completion_variation": _format_percent( _variation(os_rates.get("android", 0), previous_os_rates.get("android", 0)) if previous_kpis else 0 ), "completion_variation_value": ( _variation(os_rates.get("android", 0), previous_os_rates.get("android", 0)) if previous_kpis else 0 ), "completion_days": f"{os_days.get('android', 0)} j", "days_variation": _format_percent( _variation(os_days.get("android", 0), previous_os_days.get("android", 0)) if previous_kpis else 0 ), "days_variation_value": ( _variation(os_days.get("android", 0), previous_os_days.get("android", 0)) if previous_kpis else 0 ), }, ], "offer_mix": _build_offer_mix_context(filters), "terminal_systems": _build_terminal_systems_context(filters), } def _build_eer_api_variations(current_kpis, previous_kpis=None): metric_keys = ( "total_leads", "telephone_count", "email_count", "biometric_count", "personal_info_count", "app_installs_count", "contracts_count", ) variations = { key: _variation_from_maps(current_kpis, previous_kpis, key) for key in metric_keys } current_os_rates = current_kpis.get("os_completion_rates", {}) previous_os_rates = previous_kpis.get("os_completion_rates", {}) if previous_kpis else {} current_os_days = current_kpis.get("os_avg_completion_days", {}) previous_os_days = previous_kpis.get("os_avg_completion_days", {}) if previous_kpis else {} variations["os_completion_rates"] = { key: _variation(current_os_rates.get(key, 0), previous_os_rates.get(key, 0)) for key in ("ios", "android") } variations["os_avg_completion_days"] = { key: _variation(current_os_days.get(key, 0), previous_os_days.get(key, 0)) for key in ("ios", "android") } return variations @jwt_login_required def index(request): reset_api_debug_log() filters = _build_filter_context(request, None) acquisition_error = "" okr_error = "" onboarding_start_value = None previous_onboarding_start_value = None ga4_all_data = {} previous_ga4_all_data = {} try: acquisition_data = _load_acquisition_api_data(filters) except ApiClientError as error: acquisition_data = _load_parcours_acquisition_data() if acquisition_data: warning = acquisition_data.get("warning", {}) if not isinstance(warning, dict): warning = {"cpa": str(warning)} warning["cpa"] = ( f"Erreur chargement API acquisition: {error}. " "Données locales analytics.md utilisées." ) acquisition_data["warning"] = warning else: acquisition_data = { "warning": { "cpa": f"Erreur chargement API acquisition: {error}", } } acquisition_error = str(error) try: ga4_all_data = _load_ga4_all_data( client_id=10, start_date=filters.get("start_date"), end_date=filters.get("end_date"), ) onboarding_start_value = _extract_onboarding_start( ga4_all_data, start_date=filters.get("start_date"), end_date=filters.get("end_date"), ) except ApiClientError as error: okr_error = f"Erreur chargement API GA4 all periode actuelle: {error}" try: previous_ga4_all_data = _load_ga4_all_data( client_id=10, start_date=filters.get("previous_start_date"), end_date=filters.get("previous_end_date"), ) previous_onboarding_start_value = _extract_onboarding_start( previous_ga4_all_data, start_date=filters.get("previous_start_date"), end_date=filters.get("previous_end_date"), ) except ApiClientError as error: okr_error = ( f"{okr_error} | " if okr_error else "" ) + f"Erreur chargement API GA4 all periode precedente: {error}" paid_sources = _load_paid_media_sources(filters) crm_kpis, previous_crm_kpis = _crm_kpis_with_previous(request) eer_context = _build_eer_context( crm_kpis, previous_crm_kpis, filters, ) acquisition_context = _build_acquisition_context( acquisition_data, filters, onboarding_start_value=onboarding_start_value, previous_onboarding_start_value=previous_onboarding_start_value, ) paid_media_context = _build_paid_media_context(paid_sources, filters, acquisition_context) acquisition_context = _apply_paid_media_cpa_to_acquisition( acquisition_context, paid_media_context, ) cdc_mix = _build_cdc_mix_context(filters) previous_cdc_mix = _build_cdc_mix_context( filters, start_key="previous_start_date", end_key="previous_end_date", ) okr_rows = _build_okr_rows( acquisition_context, paid_media_context, onboarding_start_value=onboarding_start_value, previous_onboarding_start_value=previous_onboarding_start_value, cdc_mix=cdc_mix, previous_cdc_mix=previous_cdc_mix, ) return render( request, "dashboard/index.html", { "acquisition": acquisition_context, "acquisition_error": acquisition_error, "okr_error": okr_error, "filters": filters, "paid_media": paid_media_context, "okr_rows": okr_rows, "cdc_mix": cdc_mix, "okr_period_labels": _build_okr_period_labels( filters, okr_rows, onboarding_start_data=ga4_all_data, ), "eer": eer_context, "api_debug_log": get_api_debug_log(), }, ) @jwt_login_required def campaign_options(request): reset_api_debug_log() channels = _selected_values(request, "channels") client_id = request.GET.get("client_id", "10") authorization = request.headers.get("Authorization", "") bearer_token = authorization.removeprefix("Bearer ").strip() if authorization else None if not channels: channels = ["google_ads", "meta"] try: campaigns = get_campaign_options(channels, client_id, bearer_token=bearer_token) except ApiClientError as error: return JsonResponse( { "campaigns": [], "error": str(error), "status_code": error.status_code, "response_body": error.response_body, "api_debug_log": get_api_debug_log(), }, status=502, ) return JsonResponse({"campaigns": campaigns, "api_debug_log": get_api_debug_log()}) @jwt_login_required def okr_csv(request): csv_path = Path(settings.BASE_DIR) / "dashboard" / "data" / "OKR.csv" if not csv_path.exists(): raise Http404("OKR.csv introuvable") return FileResponse( csv_path.open("rb"), content_type="text/csv; charset=utf-8", ) # New view for CSV-based KPIs @jwt_login_required def dashboard_data(request): current_kpis, previous_kpis = _crm_kpis_with_previous(request) filters = _build_filter_context(request, None) current_kpis, previous_kpis = _apply_eer_snapshot_overrides( current_kpis, previous_kpis, filters, ) return JsonResponse({ "success": True, "from_csv": True, "data": current_kpis, "previous_data": previous_kpis, "variations": _build_eer_api_variations(current_kpis, previous_kpis), })