deer-flow/backend/app/channels/allowed_users.py

"""Shared scalar-or-list parsing for channel ID configuration."""

from __future__ import annotations

import logging
from collections.abc import Callable, Hashable
from typing import Any


def _parse_id_entries(
    value: Any,
    *,
    logger: logging.Logger,
    channel_name: str,
    key_name: str,
    valid_id_name: str,
    warn_on_scalar_separators: bool,
) -> list[Any]:
    """Normalize config shape without splitting scalar IDs or iterating mappings."""
    if value is None or (isinstance(value, str) and not value.strip()):
        return []
    if warn_on_scalar_separators and isinstance(value, str) and ("," in value or any(char.isspace() for char in value.strip())):
        logger.warning(
            "[%s] %s is a scalar string containing separators; treating it as one literal %s. Use a YAML list for multiple IDs",
            channel_name,
            key_name,
            valid_id_name,
        )
    return list(value) if isinstance(value, (list, tuple, set, frozenset)) else [value]


def parse_channel_ids(
    value: Any,
    *,
    logger: logging.Logger,
    channel_name: str,
    key_name: str,
    warn_on_scalar_separators: bool = False,
) -> frozenset[str]:
    """Return trimmed literal channel IDs; empty config grants no exemptions.

    These IDs are compared as strings by the connector. An unusable literal
    never matches, so it leaves the connector's mention requirement in force.
    """
    entries = _parse_id_entries(
        value,
        logger=logger,
        channel_name=channel_name,
        key_name=key_name,
        valid_id_name="channel id",
        warn_on_scalar_separators=warn_on_scalar_separators,
    )
    return frozenset(text for text in (str(entry).strip() for entry in entries) if text)


def parse_allowed_users[UserId: Hashable](
    allowed_users: Any,
    *,
    parse_user_id: Callable[[Any], UserId | None],
    logger: logging.Logger,
    channel_name: str,
    expected_id: str,
    valid_id_name: str = "user ID",
    warn_on_scalar_separators: bool = False,
) -> frozenset[UserId] | None:
    """Return ``None`` for an unset/empty allowlist, or its valid user IDs.

    Scalars are single entries; mappings must not authorize their keys. The
    channel's coercer validates each entry. Invalid entries warn and are dropped,
    but a configured value with no valid entry returns an empty set and logs an
    error, so callers deny everyone instead of silently removing the restriction.
    String-ID channels can opt into a scalar-separator warning without splitting
    the ID; surrounding whitespace alone does not trigger it.
    """
    entries = _parse_id_entries(
        allowed_users,
        logger=logger,
        channel_name=channel_name,
        key_name="allowed_users",
        valid_id_name=valid_id_name,
        warn_on_scalar_separators=warn_on_scalar_separators,
    )
    if not entries:
        return None

    user_ids: set[UserId] = set()
    for entry in entries:
        user_id = parse_user_id(entry)
        if user_id is None:
            logger.warning("[%s] Ignoring allowed_users entry %r: expected %s; list several IDs as a YAML list", channel_name, entry, expected_id)
        else:
            user_ids.add(user_id)
    if not user_ids:
        logger.error("[%s] allowed_users has no valid %s; denying every user until it is fixed", channel_name, valid_id_name)
    return frozenset(user_ids)


def parse_string_or_int_user_id(entry: Any) -> str | None:
    """A user ID from a non-blank string or an integer, else ``None``.

    An unquoted YAML number keeps matching its string form. Quote IDs with
    leading zeros.
    """
    if isinstance(entry, bool):
        # bool is an int subclass: ``true`` must not become the user "True".
        return None
    if isinstance(entry, int):
        return str(entry)
    if isinstance(entry, str):
        return entry.strip() or None
    return None