ForcePilot/backend/package/yuxi/channels/adapters/zalo_user/session.py

from __future__ import annotations

from typing import Any

from .constants import TARGET_PREFIXES
from .normalize import (
    implicit_mention,
    was_explicitly_mentioned,
)


def resolve_agent_route(
    channel_id: str,
    chat_id: str,
    user_id: str,
    chat_type: str,
    config: dict[str, Any],
) -> str:
    default_agent_id = config.get("default_agent_id", "default")

    if chat_type == "direct":
        return f"agent:{default_agent_id}:{channel_id}:direct:{user_id}"
    elif chat_type == "group":
        agent_id = _resolve_group_agent(chat_id, config, default_agent_id)
        return f"agent:{agent_id}:{channel_id}:group:{chat_id}"
    else:
        return f"agent:{default_agent_id}:{channel_id}:unknown:{chat_id}"


def _resolve_group_agent(chat_id: str, config: dict[str, Any], default_agent_id: str) -> str:
    groups_config = config.get("groups", {})
    chat_config = groups_config.get(chat_id, {})
    return chat_config.get("agent_id", default_agent_id)


def check_dm_policy(user_id: str, config: dict[str, Any], paired_users: set[str]) -> bool:
    dm_policy = config.get("dm_policy", "pairing")

    match dm_policy:
        case "open":
            return True
        case "disabled":
            return False
        case "pairing":
            return f"zalo:{user_id}" in paired_users
        case "allowlist":
            allow_from = config.get("allow_from", [])
            return f"zalo:{user_id}" in allow_from
        case _:
            return False


def check_group_policy(chat_id: str, user_id: str, config: dict[str, Any]) -> bool:
    group_policy = config.get("group_policy", "allowlist")

    match group_policy:
        case "open":
            return True
        case "disabled":
            return False
        case "allowlist":
            group_allow = config.get("group_allow_from", [])
            if f"zalo:{user_id}" in group_allow:
                return True
            groups_config = config.get("groups", {})
            chat_config = groups_config.get(chat_id, {})
            per_group_allow = chat_config.get("allow_from", [])
            return f"zalo:{user_id}" in per_group_allow
        case _:
            return False


def check_mention_required(
    chat_id: str,
    content: str,
    config: dict[str, Any],
    bot_names: str | list[str] = "",
) -> bool:
    groups_config = config.get("groups", {})
    chat_config = groups_config.get(chat_id, {})
    wildcard_config = groups_config.get("*", {})

    require_mention = chat_config.get(
        "require_mention",
        wildcard_config.get("require_mention", False),
    )

    if not require_mention:
        return True

    if isinstance(bot_names, str):
        bot_names = [bot_names] if bot_names else []

    if not bot_names:
        return True

    if was_explicitly_mentioned(content, bot_names):
        return True

    if implicit_mention(content, bot_names):
        return True

    return False


def resolve_chat_id(target: str, channel_id: str = "zalo_user") -> tuple[str | None, str | None]:
    normalized = target.strip()

    for prefix, chat_type in TARGET_PREFIXES.items():
        if normalized.lower().startswith(prefix):
            chat_id = normalized[len(prefix) :]
            return chat_id, chat_type

    if normalized.startswith("zalo:"):
        chat_id = normalized[len("zalo:") :]
        return chat_id, None

    return normalized, None


def looks_like_id(target: str) -> bool:
    if not target:
        return False
    stripped = target.strip()
    if stripped.isdigit():
        return True
    for prefix in TARGET_PREFIXES:
        if stripped.lower().startswith(prefix):
            suffix = stripped[len(prefix) :]
            if suffix.isdigit():
                return True
    return False


def parse_outbound_target(target: str) -> dict[str, Any]:
    chat_id, chat_type = resolve_chat_id(target)
    is_group = chat_type == "group" if chat_type is not None else False
    return {
        "thread_id": chat_id or target,
        "is_group": is_group,
        "chat_type": chat_type or ("group" if is_group else "direct"),
    }


def resolve_zalo_allow_from_entries(
    entries: list[str],
    friends: list[dict[str, Any]],
) -> list[str]:
    resolved: list[str] = []
    friend_map: dict[str, str] = {}
    for f in friends:
        uid = str(f.get("user_id") or f.get("id") or "")
        name = (f.get("display_name") or f.get("name") or "").lower()
        if uid:
            friend_map[name] = uid
            friend_map[uid] = uid

    for entry in entries:
        entry_lower = entry.lower()
        if entry_lower in friend_map:
            resolved.append(friend_map[entry_lower])
        elif entry_lower.isdigit():
            resolved.append(entry_lower)
        else:
            resolved.append(entry)

    return resolved


def check_group_tool_policy(
    chat_id: str,
    tool_name: str,
    config: dict[str, Any],
) -> bool:
    groups_config = config.get("groups", {})
    chat_config = groups_config.get(chat_id, {})
    tool_policy = chat_config.get("tools", {})

    if not tool_policy:
        return True

    deny_list = tool_policy.get("deny", [])
    if tool_name in deny_list:
        return False

    allow_list = tool_policy.get("allow", [])
    if allow_list and tool_name not in allow_list:
        return False

    return True


def _find_by_name(query: str, items: list[dict[str, Any]], *name_keys: str) -> dict[str, Any] | None:
    query_lower = query.lower()
    exact_matches: list[dict[str, Any]] = []
    partial_matches: list[dict[str, Any]] = []

    for item in items:
        for key in name_keys:
            name = str(item.get(key, "")).lower()
            if not name:
                continue
            if name == query_lower:
                exact_matches.append(item)
            elif query_lower in name:
                partial_matches.append(item)
            break

    if exact_matches:
        return exact_matches[0]
    if partial_matches:
        return partial_matches[0]
    return None


async def resolve_targets(
    targets: list[str],
    friends: list[dict[str, Any]],
    groups: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    results: list[dict[str, Any]] = []
    for target in targets:
        if not target:
            results.append({"input": target, "resolved_id": None, "resolved_type": None})
            continue

        chat_id, chat_type = resolve_chat_id(target)
        if chat_type is not None:
            results.append({"input": target, "resolved_id": chat_id, "resolved_type": chat_type})
            continue

        if target.strip().isdigit():
            results.append({"input": target, "resolved_id": target.strip(), "resolved_type": "direct"})
            continue

        target_lower = target.strip().lower()
        matched_friend = _find_by_name(target_lower, friends, "display_name", "name")
        if matched_friend:
            uid = str(matched_friend.get("user_id") or matched_friend.get("id") or "")
            results.append(
                {
                    "input": target,
                    "resolved_id": uid,
                    "resolved_type": "user",
                    "display_name": matched_friend.get("display_name", target),
                }
            )
            continue

        matched_group = _find_by_name(target_lower, groups, "name", "group_name")
        if matched_group:
            gid = str(matched_group.get("group_id") or matched_group.get("id") or "")
            results.append(
                {
                    "input": target,
                    "resolved_id": gid,
                    "resolved_type": "group",
                    "display_name": matched_group.get("name", target),
                }
            )
            continue

        results.append(
            {
                "input": target,
                "resolved_id": target.strip(),
                "resolved_type": None,
                "error": f"Could not resolve target '{target}'",
            }
        )

    return results