ss-tools/backend/src/api/routes/dashboards/_projection.py

# #region DashboardProjection [C:2] [TYPE Module] [SEMANTICS dashboard, api, transform, profile, filter]
# @defgroup Api Module group.
# @BRIEF Dashboard response projection and profile-filter helpers for Dashboard Hub routes.
# @LAYER Infrastructure
# @RELATION DEPENDS_ON -> [SupersetClient]
# @RELATION DEPENDS_ON -> [ProfileService]

from typing import Any

from src.core.logger import logger
from src.core.async_superset_client import AsyncSupersetClient
from src.core.superset_profile_lookup import SupersetAccountLookupAdapter
from src.models.auth import User
from src.services.profile_service import ProfileService


# #region _normalize_actor_alias_token [C:2] [TYPE Function]
# @BRIEF Normalize actor alias token to comparable trim+lower text.
def _normalize_actor_alias_token(value: Any) -> str | None:
    if value is None:
        return None
    normalized = str(value).strip().lower()
    return normalized if normalized else None


# #endregion _normalize_actor_alias_token


# #region _normalize_owner_display_token [C:2] [TYPE Function]
# @BRIEF Project owner payload value into stable display string for API response contracts.
def _normalize_owner_display_token(owner: Any) -> str | None:
    if owner is None:
        return None
    if isinstance(owner, dict):
        for key in ("username", "full_name", "first_name", "email"):
            candidate = owner.get(key)
            if isinstance(candidate, str) and candidate.strip():
                return candidate.strip()
        return None
    if isinstance(owner, str):
        return owner.strip() or None
    return None


# #endregion _normalize_owner_display_token


# #region _normalize_dashboard_owner_values [C:2] [TYPE Function]
# @BRIEF Normalize dashboard owners payload to optional list of display strings.
def _normalize_dashboard_owner_values(owners: Any) -> list[str] | None:
    if owners is None:
        return None
    raw_items: list[Any]
    if isinstance(owners, list):
        raw_items = owners
    else:
        raw_items = [owners]
    normalized: list[str] = []
    for owner in raw_items:
        token = _normalize_owner_display_token(owner)
        if token and token not in normalized:
            normalized.append(token)
    return normalized


# #endregion _normalize_dashboard_owner_values


# #region _project_dashboard_response_items [C:2] [TYPE Function]
# @BRIEF Project dashboard payloads to response-contract-safe shape.
def _project_dashboard_response_items(
    dashboards: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    projected: list[dict[str, Any]] = []
    for dashboard in dashboards:
        projected_dashboard = dict(dashboard)
        projected_dashboard["owners"] = _normalize_dashboard_owner_values(
            projected_dashboard.get("owners")
        )
        projected.append(projected_dashboard)
    return projected


# #endregion _project_dashboard_response_items


# #region _get_profile_filter_binding [C:2] [TYPE Function]
# @BRIEF Resolve dashboard profile-filter binding through current or legacy profile service contracts.
def _get_profile_filter_binding(
    profile_service: Any, current_user: User
) -> dict[str, Any]:

    def _read_optional_string(value: Any) -> str | None:
        return value if isinstance(value, str) else None

    def _read_bool(value: Any, default: bool) -> bool:
        return value if isinstance(value, bool) else default

    if hasattr(profile_service, "get_dashboard_filter_binding"):
        binding = profile_service.get_dashboard_filter_binding(current_user)
        if isinstance(binding, dict):
            return {
                "superset_username": _read_optional_string(
                    binding.get("superset_username")
                ),
                "superset_username_normalized": _read_optional_string(
                    binding.get("superset_username_normalized")
                ),
                "show_only_my_dashboards": _read_bool(
                    binding.get("show_only_my_dashboards"), False
                ),
                "show_only_slug_dashboards": _read_bool(
                    binding.get("show_only_slug_dashboards"), False
                ),
            }
    if hasattr(profile_service, "get_my_preference"):
        response = profile_service.get_my_preference(current_user)
        preference = getattr(response, "preference", None)
        return {
            "superset_username": _read_optional_string(
                getattr(preference, "superset_username", None)
            ),
            "superset_username_normalized": _read_optional_string(
                getattr(preference, "superset_username_normalized", None)
            ),
            "show_only_my_dashboards": _read_bool(
                getattr(preference, "show_only_my_dashboards", False), False
            ),
            "show_only_slug_dashboards": _read_bool(
                getattr(preference, "show_only_slug_dashboards", False), False
            ),
        }
    return {
        "superset_username": None,
        "superset_username_normalized": None,
        "show_only_my_dashboards": False,
        "show_only_slug_dashboards": False,
    }


# #endregion _get_profile_filter_binding


# #region _resolve_profile_actor_aliases [C:2] [TYPE Function]
# @BRIEF Resolve stable actor aliases for profile filtering without per-dashboard detail fan-out.
# @SIDE_EFFECT  Асинхронный HTTP-вызов к Superset API для поиска пользователей.
async def _resolve_profile_actor_aliases(env: Any, bound_username: str) -> list[str]:
    normalized_bound = _normalize_actor_alias_token(bound_username)
    if not normalized_bound:
        return []
    aliases: list[str] = [normalized_bound]
    try:
        client = AsyncSupersetClient(env)
        adapter = SupersetAccountLookupAdapter(
            network_client=client.client,
            environment_id=str(getattr(env, "id", "")),
        )
        lookup_payload = await adapter.get_users_page(
            search=normalized_bound,
            page_index=0,
            page_size=20,
            sort_column="username",
            sort_order="asc",
        )
        lookup_items = (
            lookup_payload.get("items", []) if isinstance(lookup_payload, dict) else []
        )
        matched_item: dict[str, Any] | None = None
        for item in lookup_items:
            if not isinstance(item, dict):
                continue
            if _normalize_actor_alias_token(item.get("username")) == normalized_bound:
                matched_item = item
                break
        if matched_item is None:
            for item in lookup_items:
                if isinstance(item, dict):
                    matched_item = item
                    break
        display_alias = _normalize_actor_alias_token(
            (matched_item or {}).get("display_name")
        )
        if display_alias and display_alias not in aliases:
            aliases.append(display_alias)
        logger.reflect(
            "Resolved profile actor aliases",
            extra={"src": "_resolve_profile_actor_aliases", "payload": {"env": getattr(env, 'id', None), "bound_username": normalized_bound, "lookup_items": len(lookup_items), "aliases": aliases}},
        )
    except Exception as alias_error:
        logger.explore(
            "Failed to resolve profile actor aliases via Superset users lookup",
            extra={"src": "_resolve_profile_actor_aliases", "payload": {"env": getattr(env, 'id', None), "bound_username": normalized_bound}, "error": str(alias_error)},
        )
    return aliases


# #endregion _resolve_profile_actor_aliases


# #region _matches_dashboard_actor_aliases [C:2] [TYPE Function]
# @BRIEF Apply profile actor matching against multiple aliases (username + optional display name).
def _matches_dashboard_actor_aliases(
    profile_service: ProfileService,
    actor_aliases: list[str],
    owners: Any | None,
    modified_by: str | None,
) -> bool:
    for actor_alias in actor_aliases:
        if profile_service.matches_dashboard_actor(
            bound_username=actor_alias,
            owners=owners,
            modified_by=modified_by,
        ):
            return True
    return False


# #endregion _matches_dashboard_actor_aliases


# #region _task_matches_dashboard [C:2] [TYPE Function]
# @BRIEF Checks whether task params are tied to a specific dashboard and environment.
def _task_matches_dashboard(
    task: Any, dashboard_id: int, env_id: str | None
) -> bool:
    plugin_id = getattr(task, "plugin_id", None)
    if plugin_id not in {"superset-backup", "llm_dashboard_validation"}:
        return False
    params = getattr(task, "params", {}) or {}
    dashboard_id_str = str(dashboard_id)
    if plugin_id == "llm_dashboard_validation":
        task_dashboard_id = params.get("dashboard_id")
        if str(task_dashboard_id) != dashboard_id_str:
            return False
        if env_id:
            task_env = params.get("environment_id")
            return str(task_env) == str(env_id)
        return True
    dashboard_ids = params.get("dashboard_ids") or params.get("dashboards") or []
    normalized_ids = {str(item) for item in dashboard_ids}
    if dashboard_id_str not in normalized_ids:
        return False
    if env_id:
        task_env = params.get("environment_id") or params.get("env")
        return str(task_env) == str(env_id)
    return True


# #endregion _task_matches_dashboard
# #endregion DashboardProjection