company-ai-platform/app/application/feishu/handlers/admin.py

from typing import Any

from fastapi import HTTPException
from sqlalchemy.orm import Session

from app.application.feishu.delivery import send_text_if_configured
from app.application.feishu.results import command_result
from app.modules.audit.constants import AuditRiskLevel, AuditSource
from app.modules.audit.schemas import AuditLogCreate
from app.modules.audit.service import AuditService
from app.modules.feishu.constants import FeishuCommandName, FeishuReplyType
from app.modules.feishu.service import FeishuService
from app.modules.feishu_users.constants import (
    FEISHU_USER_TARGET_TYPE,
    FeishuUserAuditAction,
    FeishuUserRole,
    FeishuUserStatus,
)
from app.modules.feishu_users.principal import FeishuMention, FeishuPrincipal
from app.modules.feishu_users.services import (
    FeishuIdentityService,
    FeishuUserManagementService,
)

ADMIN_COMMAND_TITLE = "飞书用户管理"
ADMIN_COMMAND_HELP = (
    "用户管理命令必须使用一个真实的飞书 @用户：\n"
    "设为管理员 @用户\n"
    "设为普通用户 @用户\n"
    "停用用户 @用户\n"
    "启用用户 @用户"
)

_ADMIN_COMMANDS: dict[str, tuple[FeishuCommandName, dict[str, str], str]] = {
    "设为管理员": (
        FeishuCommandName.USER_SET_ADMIN,
        {"role": FeishuUserRole.ADMIN},
        "已设为管理员",
    ),
    "设为普通用户": (
        FeishuCommandName.USER_SET_USER,
        {"role": FeishuUserRole.USER},
        "已设为普通用户",
    ),
    "停用用户": (
        FeishuCommandName.USER_DISABLE,
        {"status": FeishuUserStatus.DISABLED},
        "已停用",
    ),
    "启用用户": (
        FeishuCommandName.USER_ENABLE,
        {"status": FeishuUserStatus.ACTIVE},
        "已启用",
    ),
}


def is_admin_command(command_text: str) -> bool:
    return any(command_text.startswith(prefix) for prefix in _ADMIN_COMMANDS)


def handle_admin_command(
    db: Session,
    feishu: FeishuService,
    *,
    raw_text: str,
    command_text: str,
    principal: FeishuPrincipal,
    auto_reply: bool,
) -> dict[str, Any] | None:
    """Manage a user selected only from verified structured mention metadata."""

    matched = next(
        (
            (prefix, definition)
            for prefix, definition in _ADMIN_COMMANDS.items()
            if command_text.startswith(prefix)
        ),
        None,
    )
    if matched is None:
        return None
    command_prefix, definition = matched
    command, changes, success_text = definition
    target = _target_mention(raw_text, command_prefix, principal.mentions)
    if target is None or not target.open_id:
        _audit_denied(db, principal, "missing_or_ambiguous_structured_mention")
        return _result(
            feishu,
            command,
            ADMIN_COMMAND_HELP,
            principal,
            auto_reply,
        )
    target_tenant = target.tenant_key or principal.tenant_key
    if target_tenant != principal.tenant_key:
        _audit_denied(db, principal, "cross_tenant_target")
        return _result(
            feishu,
            command,
            "只能管理当前租户内通过飞书 @ 提及的用户。",
            principal,
            auto_reply,
        )

    target_principal = FeishuIdentityService(db).resolve_or_register(
        tenant_key=target_tenant,
        open_id=target.open_id,
        union_id=target.union_id,
        user_id=target.user_id,
        actor=principal.user_code,
    )
    try:
        updated = FeishuUserManagementService(db).update_user(
            target_principal.user_code,
            changes=changes,
            actor=principal.user_code,
        )
        display_name = target.name or updated.code
        content = f"{display_name} {success_text}。"
    except HTTPException as exc:
        if exc.status_code == 409:
            content = "操作已拒绝：不能停用或降级最后一个有效管理员。"
        else:
            content = "用户状态未修改，请确认目标用户后重试。"
    return _result(feishu, command, content, principal, auto_reply)


def _target_mention(
    raw_text: str,
    command_text: str,
    mentions: tuple[FeishuMention, ...],
) -> FeishuMention | None:
    command_index = raw_text.find(command_text)
    if command_index >= 0:
        command_tail = raw_text[command_index + len(command_text) :]
        candidates = [
            mention
            for mention in mentions
            if mention.key and mention.key in command_tail
        ]
        if len(candidates) == 1:
            return candidates[0]
        return None
    if len(mentions) == 1:
        return mentions[0]
    return None


def _audit_denied(
    db: Session,
    principal: FeishuPrincipal,
    reason: str,
) -> None:
    AuditService(db).record(
        AuditLogCreate(
            actor=principal.user_code,
            source=AuditSource.FEISHU,
            action=FeishuUserAuditAction.UPDATE_DENIED,
            target_type=FEISHU_USER_TARGET_TYPE,
            risk_level=AuditRiskLevel.HIGH,
            response_payload={"result": "denied", "reason": reason},
            status="denied",
        )
    )
    db.commit()


def _result(
    feishu: FeishuService,
    command: FeishuCommandName,
    content: str,
    principal: FeishuPrincipal,
    auto_reply: bool,
) -> dict[str, Any]:
    response = (
        send_text_if_configured(
            feishu,
            principal.chat_id,
            content,
            principal.user_code,
        )
        if auto_reply
        else None
    )
    return command_result(
        command,
        FeishuReplyType.TEXT,
        ADMIN_COMMAND_TITLE,
        content,
        response,
    )