SenSu/services/api_service.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import logging
import asyncio
from typing import Dict, List, Optional
from dataclasses import dataclass
from aiohttp import web
import json

logger = logging.getLogger(__name__)

@dataclass
class APIEndpoint:
    """API端点数据类"""
    path: str
    methods: List[str]
    handler: callable
    plugin_name: str
    require_auth: bool
    admin_only: bool

class APIService:
    """API服务 - 管理插件的API端点"""
    
    def __init__(self, internet_service, auth_service, permission_service):
        self.internet_service = internet_service
        self.auth_service = auth_service
        self.permission_service = permission_service
        self.endpoints: Dict[str, APIEndpoint] = {}
        self.plugin_endpoints: Dict[str, List[str]] = {}
        logger.debug("APIService初始化开始")
    
    async def register_endpoint(self, plugin_name: str, path: str, methods: List[str], 
                              handler: callable, require_auth: bool = True, 
                              admin_only: bool = False) -> bool:
        """注册API端点"""
        try:
            logger.debug(f"注册API端点: {plugin_name} -> {path} {methods}")
            
            # 权限检查 - 只有admin插件可以操作接口
            if not self.permission_service.check_plugin_permission(plugin_name, "admin"):
                logger.error(f"插件 {plugin_name} 无权限注册API端点")
                return False
            
            # 创建端点键
            endpoint_key = f"{plugin_name}:{path}"
            
            # 检查端点是否已存在
            if endpoint_key in self.endpoints:
                logger.warning(f"API端点已存在: {endpoint_key}")
                return False
            
            # 创建端点实例
            endpoint = APIEndpoint(
                path=path,
                methods=methods,
                handler=handler,
                plugin_name=plugin_name,
                require_auth=require_auth,
                admin_only=admin_only
            )
            
            # 注册到互联网服务
            for method in methods:
                internet_endpoint_key = f"{method}:{path}"
                self.internet_service.endpoints[internet_endpoint_key] = endpoint
            
            # 保存端点信息
            self.endpoints[endpoint_key] = endpoint
            
            # 更新插件端点映射
            if plugin_name not in self.plugin_endpoints:
                self.plugin_endpoints[plugin_name] = []
            self.plugin_endpoints[plugin_name].append(endpoint_key)
            
            logger.info(f"API端点注册成功: {endpoint_key}")
            return True
            
        except Exception as e:
            logger.error(f"注册API端点时出错: {str(e)}", exc_info=True)
            return False
    
    async def unregister_endpoint(self, plugin_name: str, path: str) -> bool:
        """注销API端点"""
        try:
            logger.debug(f"注销API端点: {plugin_name} -> {path}")
            
            endpoint_key = f"{plugin_name}:{path}"
            
            if endpoint_key not in self.endpoints:
                logger.warning(f"API端点不存在: {endpoint_key}")
                return False
            
            endpoint = self.endpoints[endpoint_key]
            
            # 从互联网服务中移除
            for method in endpoint.methods:
                internet_endpoint_key = f"{method}:{path}"
                if internet_endpoint_key in self.internet_service.endpoints:
                    del self.internet_service.endpoints[internet_endpoint_key]
            
            # 从端点映射中移除
            del self.endpoints[endpoint_key]
            
            # 从插件端点列表中移除
            if plugin_name in self.plugin_endpoints:
                if endpoint_key in self.plugin_endpoints[plugin_name]:
                    self.plugin_endpoints[plugin_name].remove(endpoint_key)
            
            logger.info(f"API端点注销成功: {endpoint_key}")
            return True
            
        except Exception as e:
            logger.error(f"注销API端点时出错: {str(e)}", exc_info=True)
            return False
    
    async def unregister_all_plugin_endpoints(self, plugin_name: str) -> bool:
        """注销插件的所有API端点"""
        try:
            logger.debug(f"注销插件所有API端点: {plugin_name}")
            
            if plugin_name not in self.plugin_endpoints:
                logger.debug(f"插件无注册的API端点: {plugin_name}")
                return True
            
            endpoints_to_remove = self.plugin_endpoints[plugin_name][:]
            success_count = 0
            
            for endpoint_key in endpoints_to_remove:
                # 从endpoint_key中提取path
                parts = endpoint_key.split(':', 1)
                if len(parts) == 2:
                    path = parts[1]
                    success = await self.unregister_endpoint(plugin_name, path)
                    if success:
                        success_count += 1
            
            logger.info(f"插件API端点清理完成: {plugin_name} -> 成功 {success_count}/{len(endpoints_to_remove)}")
            return success_count == len(endpoints_to_remove)
            
        except Exception as e:
            logger.error(f"注销插件所有API端点时出错: {str(e)}", exc_info=True)
            return False
    
    def get_plugin_endpoints(self, plugin_name: str) -> List[Dict]:
        """获取插件的API端点列表"""
        try:
            if plugin_name not in self.plugin_endpoints:
                return []
            
            endpoints_info = []
            for endpoint_key in self.plugin_endpoints[plugin_name]:
                if endpoint_key in self.endpoints:
                    endpoint = self.endpoints[endpoint_key]
                    endpoints_info.append({
                        'path': endpoint.path,
                        'methods': endpoint.methods,
                        'require_auth': endpoint.require_auth,
                        'admin_only': endpoint.admin_only
                    })
            
            logger.debug(f"获取插件API端点列表: {plugin_name} -> {len(endpoints_info)} 个")
            return endpoints_info
            
        except Exception as e:
            logger.error(f"获取插件API端点列表时出错: {str(e)}", exc_info=True)
            return []
    
    def get_all_endpoints(self) -> List[Dict]:
        """获取所有API端点"""
        try:
            all_endpoints = []
            for endpoint_key, endpoint in self.endpoints.items():
                all_endpoints.append({
                    'plugin': endpoint.plugin_name,
                    'path': endpoint.path,
                    'methods': endpoint.methods,
                    'require_auth': endpoint.require_auth,
                    'admin_only': endpoint.admin_only
                })
            
            logger.debug(f"获取所有API端点: {len(all_endpoints)} 个")
            return all_endpoints
            
        except Exception as e:
            logger.error(f"获取所有API端点时出错: {str(e)}", exc_info=True)
            return []
    
    async def shutdown(self):
        """关闭API服务"""
        try:
            logger.info("关闭API服务")
            
            # 注销所有端点
            all_plugins = list(self.plugin_endpoints.keys())
            for plugin_name in all_plugins:
                await self.unregister_all_plugin_endpoints(plugin_name)
            
            self.endpoints.clear()
            self.plugin_endpoints.clear()
            
            logger.debug("API服务关闭完成")
            
        except Exception as e:
            logger.error(f"关闭API服务时出错: {str(e)}", exc_info=True)