diff --git a/admin-frontend/src/analytics/AnalyticsService.ts b/admin-frontend/src/analytics/AnalyticsService.ts index 1ca866d2..19a7c193 100644 --- a/admin-frontend/src/analytics/AnalyticsService.ts +++ b/admin-frontend/src/analytics/AnalyticsService.ts @@ -148,9 +148,16 @@ export default class AnalyticsService { } } - async getSkillGapStats(limit = 10, institution?: string): Promise { + async getSkillGapStats( + limit = 10, + institution?: string, + province?: string, + sector?: string + ): Promise { const params = new URLSearchParams({ limit: String(limit) }); if (institution) params.set("institution", institution); + if (province) params.set("province", province); + if (sector) params.set("sector", sector); const url = `${this.baseUrl}/analytics/skill-gap-stats?${params}`; const errorFactory = getRestAPIErrorFactory(SERVICE_NAME, "getSkillGapStats", "GET", url); const response = await customFetch(url, { @@ -203,9 +210,16 @@ export default class AnalyticsService { } } - async getSkillsSupplyStats(limit = 10, institution?: string): Promise { + async getSkillsSupplyStats( + limit = 10, + institution?: string, + province?: string, + sector?: string + ): Promise { const params = new URLSearchParams({ limit: String(limit) }); if (institution) params.set("institution", institution); + if (province) params.set("province", province); + if (sector) params.set("sector", sector); const url = `${this.baseUrl}/analytics/skills-supply-stats?${params}`; const errorFactory = getRestAPIErrorFactory(SERVICE_NAME, "getSkillsSupplyStats", "GET", url); const response = await customFetch(url, { diff --git a/admin-frontend/src/components/SkillsAnalytics/SkillsAnalytics.tsx b/admin-frontend/src/components/SkillsAnalytics/SkillsAnalytics.tsx index 6b052e77..c7785129 100644 --- a/admin-frontend/src/components/SkillsAnalytics/SkillsAnalytics.tsx +++ b/admin-frontend/src/components/SkillsAnalytics/SkillsAnalytics.tsx @@ -1,10 +1,11 @@ -import React from "react"; +import React, { useState } from "react"; import { Box, Skeleton, Typography, useTheme, Select, MenuItem, LinearProgress } from "@mui/material"; import { useTranslation } from "react-i18next"; import type { SkillsGapSectorData } from "src/types"; import { useSkillGapStats } from "src/hooks/useSkillGapStats"; import { useSkillsSupplyStats } from "src/hooks/useSkillsSupplyStats"; import MetricInfoIcon from "src/components/MetricInfoIcon/MetricInfoIcon"; +import { MODULE_FILTER_LOCATIONS, MODULE_FILTER_SECTORS } from "src/data/moduleFilterOptions"; interface SkillsAnalyticsProps { institution?: string; @@ -13,8 +14,20 @@ interface SkillsAnalyticsProps { const SkillsAnalytics: React.FC = ({ institution }) => { const theme = useTheme(); const { t } = useTranslation(); - const { data: skillGapData, loading: skillGapLoading } = useSkillGapStats(5, institution); - const { data: skillSupplyData, loading: skillSupplyLoading } = useSkillsSupplyStats(5, institution); + const [province, setProvince] = useState(""); + const [sector, setSector] = useState(""); + const { data: skillGapData, loading: skillGapLoading } = useSkillGapStats( + 5, + institution, + province || undefined, + sector || undefined + ); + const { data: skillSupplyData, loading: skillSupplyLoading } = useSkillsSupplyStats( + 5, + institution, + province || undefined, + sector || undefined + ); // Supply: top skills students actually have, as % of students with that skill vs total with any skill const supplyTotal = skillSupplyData?.total_students_with_skills ?? 0; @@ -85,19 +98,31 @@ const SkillsAnalytics: React.FC = ({ institution }) => { diff --git a/admin-frontend/src/data/moduleFilterOptions.ts b/admin-frontend/src/data/moduleFilterOptions.ts index 32353e2d..c8226baf 100644 --- a/admin-frontend/src/data/moduleFilterOptions.ts +++ b/admin-frontend/src/data/moduleFilterOptions.ts @@ -85,6 +85,8 @@ export const MODULE_FILTER_INSTITUTIONS: string[] = [ export const MODULE_FILTER_YEARS: string[] = ["Year 1", "Year 2", "Year 3"]; +export const MODULE_FILTER_SECTORS: string[] = ["Agriculture", "Energy", "Hospitality", "Mining", "Water"]; + export const MODULE_FILTER_PROGRAMMES: string[] = [ "Accountancy", "Accounting and Finance", diff --git a/admin-frontend/src/hooks/useSkillGapStats.ts b/admin-frontend/src/hooks/useSkillGapStats.ts index eec0de10..2b86d9b7 100644 --- a/admin-frontend/src/hooks/useSkillGapStats.ts +++ b/admin-frontend/src/hooks/useSkillGapStats.ts @@ -8,7 +8,12 @@ export interface UseSkillGapStatsResult { error: Error | null; } -export function useSkillGapStats(limit = 10, institution?: string): UseSkillGapStatsResult { +export function useSkillGapStats( + limit = 10, + institution?: string, + province?: string, + sector?: string +): UseSkillGapStatsResult { const [data, setData] = useState(null); const [loading, setLoading] = useState(true); const [error, setError] = useState(null); @@ -17,7 +22,7 @@ export function useSkillGapStats(limit = 10, institution?: string): UseSkillGapS let isMounted = true; setLoading(true); AnalyticsService.getInstance() - .getSkillGapStats(limit, institution) + .getSkillGapStats(limit, institution, province, sector) .then((result) => { if (!isMounted) return; setData(result); @@ -32,7 +37,7 @@ export function useSkillGapStats(limit = 10, institution?: string): UseSkillGapS return () => { isMounted = false; }; - }, [limit, institution]); + }, [limit, institution, province, sector]); return { data, loading, error }; } diff --git a/admin-frontend/src/hooks/useSkillsSupplyStats.ts b/admin-frontend/src/hooks/useSkillsSupplyStats.ts index 4addeb45..ff6c378f 100644 --- a/admin-frontend/src/hooks/useSkillsSupplyStats.ts +++ b/admin-frontend/src/hooks/useSkillsSupplyStats.ts @@ -8,7 +8,12 @@ export interface UseSkillsSupplyStatsResult { error: Error | null; } -export function useSkillsSupplyStats(limit = 10, institution?: string): UseSkillsSupplyStatsResult { +export function useSkillsSupplyStats( + limit = 10, + institution?: string, + province?: string, + sector?: string +): UseSkillsSupplyStatsResult { const [data, setData] = useState(null); const [loading, setLoading] = useState(true); const [error, setError] = useState(null); @@ -17,7 +22,7 @@ export function useSkillsSupplyStats(limit = 10, institution?: string): UseSkill let isMounted = true; setLoading(true); AnalyticsService.getInstance() - .getSkillsSupplyStats(limit, institution) + .getSkillsSupplyStats(limit, institution, province, sector) .then((result) => { if (!isMounted) return; setData(result); @@ -32,7 +37,7 @@ export function useSkillsSupplyStats(limit = 10, institution?: string): UseSkill return () => { isMounted = false; }; - }, [limit, institution]); + }, [limit, institution, province, sector]); return { data, loading, error }; } diff --git a/backend/app/analytics/skill_gap/routes.py b/backend/app/analytics/skill_gap/routes.py index b0319207..ee91e81e 100644 --- a/backend/app/analytics/skill_gap/routes.py +++ b/backend/app/analytics/skill_gap/routes.py @@ -13,9 +13,14 @@ SkillGapAnalyticsRepository, ) from app.analytics.skill_gap.types import SkillGapStatsResponse +from app.analytics.user_filter import ( + resolve_user_ids_for_institution, + resolve_user_ids_for_province, + resolve_user_ids_for_sector, + intersect_user_id_sets, +) from app.constants.errors import HTTPErrorResponse from app.server_dependencies.db_dependencies import CompassDBProvider -from app.server_dependencies.database_collections import Collections from app.users.auth import Authentication from app.users.access_role import AccessRole, get_access_role_dependency, decode_institution_id @@ -28,17 +33,6 @@ async def _get_skill_gap_analytics_repository( return SkillGapAnalyticsRepository(application_db) -async def _resolve_user_ids_for_institution( - institution_name: str, - userdata_db: AsyncIOMotorDatabase, -) -> Optional[list[str]]: - """Return user_ids belonging to a given institution, or None if no filter.""" - docs = await userdata_db.get_collection(Collections.PLAIN_PERSONAL_DATA).find( - {"data.institution_name": institution_name}, {"user_id": 1} - ).to_list(length=None) - return [d["user_id"] for d in docs if d.get("user_id")] - - def add_skill_gap_analytics_routes(router: APIRouter, auth: Authentication) -> None: """Register skill gap analytics routes on the given router.""" @@ -50,20 +44,32 @@ def add_skill_gap_analytics_routes(router: APIRouter, auth: Authentication) -> N }, description=( "Aggregate skill gap statistics across students with pre-computed recommendations. " - "Institution staff are automatically scoped to their own institution." + "Institution staff are automatically scoped to their own institution. " + "Province filter is admin-only. Sector filter applies to all roles." ), ) async def _skill_gap_stats( limit: Annotated[int, Query(ge=1, le=100, description="Maximum number of top skill gaps to return.")] = 10, + province: Optional[str] = Query(None, description="Filter by student province (admin only)."), + sector: Optional[str] = Query(None, description="Filter by programme sector (e.g. Agriculture, Energy)."), access_role: AccessRole = Depends(get_access_role_dependency(auth)), repo: ISkillGapAnalyticsRepository = Depends(_get_skill_gap_analytics_repository), userdata_db: AsyncIOMotorDatabase = Depends(CompassDBProvider.get_userdata_db), ) -> SkillGapStatsResponse: try: - user_ids: Optional[list[str]] = None + user_id_sets: list[list[str]] = [] + if access_role.is_institution_staff and access_role.institution_id: institution_name = decode_institution_id(access_role.institution_id) - user_ids = await _resolve_user_ids_for_institution(institution_name, userdata_db) + user_id_sets.append(await resolve_user_ids_for_institution(institution_name, userdata_db)) + + if province and not access_role.is_institution_staff: + user_id_sets.append(await resolve_user_ids_for_province(province, userdata_db)) + + if sector: + user_id_sets.append(await resolve_user_ids_for_sector(sector, userdata_db)) + + user_ids = intersect_user_id_sets(user_id_sets) return await repo.get_skill_gap_stats(limit, user_ids=user_ids) except Exception as e: logger.exception(e) diff --git a/backend/app/analytics/skills_supply/routes.py b/backend/app/analytics/skills_supply/routes.py index 6ebbf130..a61cff74 100644 --- a/backend/app/analytics/skills_supply/routes.py +++ b/backend/app/analytics/skills_supply/routes.py @@ -13,8 +13,13 @@ SkillsSupplyAnalyticsRepository, ) from app.analytics.skills_supply.types import SkillsSupplyStatsResponse +from app.analytics.user_filter import ( + resolve_user_ids_for_institution, + resolve_user_ids_for_province, + resolve_user_ids_for_sector, + intersect_user_id_sets, +) from app.constants.errors import HTTPErrorResponse -from app.server_dependencies.database_collections import Collections from app.server_dependencies.db_dependencies import CompassDBProvider from app.users.auth import Authentication from app.users.access_role import AccessRole, get_access_role_dependency, decode_institution_id @@ -28,16 +33,6 @@ async def _get_skills_supply_repository( return SkillsSupplyAnalyticsRepository(application_db) -async def _resolve_user_ids_for_institution( - institution_name: str, - userdata_db: AsyncIOMotorDatabase, -) -> Optional[list[str]]: - docs = await userdata_db.get_collection(Collections.PLAIN_PERSONAL_DATA).find( - {"data.institution_name": institution_name}, {"user_id": 1} - ).to_list(length=None) - return [d["user_id"] for d in docs if d.get("user_id")] - - def add_skills_supply_analytics_routes(router: APIRouter, auth: Authentication) -> None: @router.get( path="/skills-supply-stats", @@ -47,20 +42,32 @@ def add_skills_supply_analytics_routes(router: APIRouter, auth: Authentication) }, description=( "Aggregate the most common skills identified by students during skills discovery. " - "Institution staff are automatically scoped to their own institution." + "Institution staff are automatically scoped to their own institution. " + "Province filter is admin-only. Sector filter applies to all roles." ), ) async def _skills_supply_stats( limit: int = Query(default=10, ge=1, le=50, description="Number of top skills to return"), + province: Optional[str] = Query(None, description="Filter by student province (admin only)."), + sector: Optional[str] = Query(None, description="Filter by programme sector (e.g. Agriculture, Energy)."), access_role: AccessRole = Depends(get_access_role_dependency(auth)), repo: ISkillsSupplyAnalyticsRepository = Depends(_get_skills_supply_repository), userdata_db: AsyncIOMotorDatabase = Depends(CompassDBProvider.get_userdata_db), ) -> SkillsSupplyStatsResponse: try: - user_ids: Optional[list[str]] = None + user_id_sets: list[list[str]] = [] + if access_role.is_institution_staff and access_role.institution_id: institution_name = decode_institution_id(access_role.institution_id) - user_ids = await _resolve_user_ids_for_institution(institution_name, userdata_db) + user_id_sets.append(await resolve_user_ids_for_institution(institution_name, userdata_db)) + + if province and not access_role.is_institution_staff: + user_id_sets.append(await resolve_user_ids_for_province(province, userdata_db)) + + if sector: + user_id_sets.append(await resolve_user_ids_for_sector(sector, userdata_db)) + + user_ids = intersect_user_id_sets(user_id_sets) return await repo.get_skills_supply_stats(limit=limit, user_ids=user_ids) except Exception as e: logger.exception(e) diff --git a/backend/app/analytics/user_filter.py b/backend/app/analytics/user_filter.py new file mode 100644 index 00000000..a6fb639a --- /dev/null +++ b/backend/app/analytics/user_filter.py @@ -0,0 +1,64 @@ +""" +Shared utilities for resolving user_id sets by filter dimension (institution, province, sector). +Used by analytics route handlers to scope aggregations to a filtered population. +""" +from typing import Optional + +from motor.motor_asyncio import AsyncIOMotorDatabase + +from app.server_dependencies.database_collections import Collections +from app.teveta.loader import get_data, SECTOR_KEY_MAP + + +async def resolve_user_ids_for_institution(institution_name: str, userdata_db: AsyncIOMotorDatabase) -> list[str]: + docs = await userdata_db.get_collection(Collections.PLAIN_PERSONAL_DATA).find( + {"data.institution_name": institution_name}, {"user_id": 1} + ).to_list(length=None) + return [d["user_id"] for d in docs if d.get("user_id")] + + +async def resolve_user_ids_for_province(province: str, userdata_db: AsyncIOMotorDatabase) -> list[str]: + docs = await userdata_db.get_collection(Collections.PLAIN_PERSONAL_DATA).find( + {"data.province": province}, {"user_id": 1} + ).to_list(length=None) + return [d["user_id"] for d in docs if d.get("user_id")] + + +async def resolve_user_ids_for_sector(sector: str, userdata_db: AsyncIOMotorDatabase) -> list[str]: + """ + Resolve user_ids whose enrolled programme belongs to the given sector. + + Uses the in-memory TEVETA data (loaded at startup) to find programme names + that have priority_sectors[] == True, then matches those names + against PLAIN_PERSONAL_DATA.data.programme_name. + + sector must be one of the hub display names: Agriculture, Energy, Hospitality, Mining, Water. + SECTOR_KEY_MAP translates these to the TEVETA internal keys used in priority_sectors. + """ + teveta_key = SECTOR_KEY_MAP.get(sector) + if not teveta_key: + return [] + programme_names = { + p["name"] + for p in get_data().get("programmes", []) + if p.get("priority_sectors", {}).get(teveta_key) + } + if not programme_names: + return [] + ppd_docs = await userdata_db.get_collection(Collections.PLAIN_PERSONAL_DATA).find( + {"data.programme_name": {"$in": list(programme_names)}}, {"user_id": 1} + ).to_list(length=None) + return [d["user_id"] for d in ppd_docs if d.get("user_id")] + + +def intersect_user_id_sets(sets: list[list[str]]) -> Optional[list[str]]: + """ + Intersect multiple user_id lists. Returns None when no filters are active + (meaning: all users). Returns an empty list when filters yield no overlap. + """ + if not sets: + return None + result: set[str] = set(sets[0]) + for s in sets[1:]: + result &= set(s) + return list(result) diff --git a/backend/scripts/seed_analytics_test_data.py b/backend/scripts/seed_analytics_test_data.py new file mode 100644 index 00000000..5922e4e6 --- /dev/null +++ b/backend/scripts/seed_analytics_test_data.py @@ -0,0 +1,236 @@ +#!/usr/bin/env python3 +""" +Seed / cleanup test data for Skills Analytics filter verification. + +Seeds 6 users spanning all 5 sectors (Agriculture, Energy, Hospitality, Mining, Water) +and 4 provinces (Lusaka, Copperbelt, Northern, Central) so that province and sector +filters can be tested end-to-end in the admin dashboard. + +All test documents share the prefix '_test_analytics_' so cleanup is safe and surgical. + +Usage (from backend/): + poetry run python scripts/seed_analytics_test_data.py # seed + poetry run python scripts/seed_analytics_test_data.py --cleanup # remove + +Reads APPLICATION_MONGODB_URI, APPLICATION_DATABASE_NAME, + USERDATA_MONGODB_URI, USERDATA_DATABASE_NAME from .env or environment. +""" + +import asyncio +import logging +import os +import sys +from datetime import datetime, timezone +from pathlib import Path + +# Allow importing app modules from the backend root +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from dotenv import load_dotenv +from motor.motor_asyncio import AsyncIOMotorClient + +from app.server_dependencies.database_collections import Collections +from app.teveta.loader import SECTOR_KEY_MAP, get_data + +load_dotenv(Path(__file__).parent.parent / ".env") + +logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") +logger = logging.getLogger(__name__) + +TEST_PREFIX = "_test_analytics_" + +# (user_id_suffix, province, sector_hub_name) +# Two Agriculture users across different provinces lets us verify the +# province+sector intersection test case. +TEST_USERS = [ + ("001", "Lusaka", "Agriculture"), + ("002", "Copperbelt", "Energy"), + ("003", "Northern", "Mining"), + ("004", "Lusaka", "Hospitality"), + ("005", "Central", "Water"), + ("006", "Copperbelt", "Agriculture"), # second Agriculture, different province +] + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _pick_programme_for_sector(sector_hub_name: str) -> str: + """Return a programme name from TEVETA data that belongs to the given sector.""" + teveta_key = SECTOR_KEY_MAP[sector_hub_name] + for p in get_data().get("programmes", []): + if p.get("priority_sectors", {}).get(teveta_key): + return p["name"] + raise ValueError(f"No programme found in TEVETA data for sector {sector_hub_name!r}") + + +def _build_docs(user_id: str, province: str, sector_hub_name: str) -> tuple: + programme_name = _pick_programme_for_sector(sector_hub_name) + session_id = f"{user_id}_session" + + plain_personal_data = { + "user_id": user_id, + "created_at": _now(), + "updated_at": _now(), + "data": { + "first_name": "Test", + "last_name": f"User{user_id[-3:]}", + "institution_name": "Test Institution", + "programme_name": programme_name, + "province": province, + "school_year": "Year 1", + }, + } + + user_preferences = { + "user_id": user_id, + "sessions": [session_id], + "accepted_tc": _now(), + } + + skill_gap_recommendations = { + "user_id": user_id, + "skill_gap_recommendations": [ + { + "skill_id": f"skill_{sector_hub_name.lower()}_001", + "skill_label": f"{sector_hub_name} Planning", + "job_unlock_count": 3, + "proximity_score": 0.82, + }, + { + "skill_id": f"skill_{sector_hub_name.lower()}_002", + "skill_label": f"{sector_hub_name} Management", + "job_unlock_count": 5, + "proximity_score": 0.75, + }, + ], + } + + skills_supply_state = { + "session_id": session_id, + "conversation_phase": "DIVE_IN", + "experiences_state": { + "exp_001": { + "dive_in_phase": "PROCESSED", + "experience": { + "top_skills": [ + { + "UUID": f"skill-uuid-{sector_hub_name.lower()}-a", + "preferredLabel": f"{sector_hub_name} Operations", + "score": 0.88, + }, + { + "UUID": f"skill-uuid-{sector_hub_name.lower()}-b", + "preferredLabel": f"{sector_hub_name} Analysis", + "score": 0.72, + }, + ] + }, + } + }, + } + + return plain_personal_data, user_preferences, skill_gap_recommendations, skills_supply_state + + +def _connect(): + app_uri = os.environ.get("APPLICATION_MONGODB_URI", "") + app_db_name = os.environ.get("APPLICATION_DATABASE_NAME", "") + userdata_uri = os.environ.get("USERDATA_MONGODB_URI", "") + userdata_db_name = os.environ.get("USERDATA_DATABASE_NAME", "") + + if not all([app_uri, app_db_name, userdata_uri, userdata_db_name]): + raise SystemExit( + "Missing required env vars. Ensure APPLICATION_MONGODB_URI, " + "APPLICATION_DATABASE_NAME, USERDATA_MONGODB_URI, and " + "USERDATA_DATABASE_NAME are set (or present in backend/.env)." + ) + + app_client = AsyncIOMotorClient(app_uri, tlsAllowInvalidCertificates=True) + userdata_client = AsyncIOMotorClient(userdata_uri, tlsAllowInvalidCertificates=True) + return app_client[app_db_name], userdata_client[userdata_db_name], app_client, userdata_client + + +async def seed(): + app_db, userdata_db, app_client, userdata_client = _connect() + try: + logger.info("Seeding %d test users (prefix=%r)...", len(TEST_USERS), TEST_PREFIX) + for suffix, province, sector in TEST_USERS: + user_id = f"{TEST_PREFIX}{suffix}" + ppd, prefs, gap, supply = _build_docs(user_id, province, sector) + logger.info( + " %-30s province=%-12s sector=%-12s programme=%s", + user_id, province, sector, ppd["data"]["programme_name"], + ) + + await userdata_db[Collections.PLAIN_PERSONAL_DATA].replace_one( + {"user_id": user_id}, ppd, upsert=True + ) + await app_db[Collections.USER_PREFERENCES].replace_one( + {"user_id": user_id}, prefs, upsert=True + ) + await app_db[Collections.USER_RECOMMENDATIONS].replace_one( + {"user_id": user_id}, gap, upsert=True + ) + await app_db[Collections.EXPLORE_EXPERIENCES_DIRECTOR_STATE].replace_one( + {"session_id": supply["session_id"]}, supply, upsert=True + ) + + logger.info("") + logger.info("Seed complete. Expected filter results:") + logger.info(" All filters off → 6 users in both charts") + logger.info(" Province = Lusaka → 2 users (Agriculture + Hospitality)") + logger.info(" Province = Copperbelt → 2 users (Energy + Agriculture)") + logger.info(" Sector = Agriculture → 2 users (Lusaka + Copperbelt)") + logger.info(" Sector = Energy → 1 user (Copperbelt)") + logger.info(" Sector = Hospitality → 1 user (Lusaka)") + logger.info(" Sector = Mining → 1 user (Northern)") + logger.info(" Sector = Water → 1 user (Central)") + logger.info(" Province=Copperbelt + Sector=Agriculture → 1 user (intersection)") + logger.info(" Province=Lusaka + Sector=Mining → 0 users (empty state)") + logger.info("") + logger.info("Run with --cleanup to remove all test data when done.") + finally: + app_client.close() + userdata_client.close() + + +async def cleanup(): + app_db, userdata_db, app_client, userdata_client = _connect() + try: + logger.info("Removing test data with prefix %r ...", TEST_PREFIX) + user_ids = [f"{TEST_PREFIX}{suffix}" for suffix, _, _ in TEST_USERS] + session_ids = [f"{uid}_session" for uid in user_ids] + + r = await userdata_db[Collections.PLAIN_PERSONAL_DATA].delete_many( + {"user_id": {"$in": user_ids}} + ) + logger.info(" plain_personal_data: %d deleted", r.deleted_count) + + r = await app_db[Collections.USER_PREFERENCES].delete_many( + {"user_id": {"$in": user_ids}} + ) + logger.info(" user_preferences: %d deleted", r.deleted_count) + + r = await app_db[Collections.USER_RECOMMENDATIONS].delete_many( + {"user_id": {"$in": user_ids}} + ) + logger.info(" user_recommendations: %d deleted", r.deleted_count) + + r = await app_db[Collections.EXPLORE_EXPERIENCES_DIRECTOR_STATE].delete_many( + {"session_id": {"$in": session_ids}} + ) + logger.info(" explore_experiences_director_state: %d deleted", r.deleted_count) + + logger.info("Cleanup complete.") + finally: + app_client.close() + userdata_client.close() + + +if __name__ == "__main__": + if "--cleanup" in sys.argv: + asyncio.run(cleanup()) + else: + asyncio.run(seed())