wxauto_api/plugins/builtin.py

# ///
# builtin.py
# 描述：内置插件实现
# 作者：AI Generated
# 创建日期：2026-04-05
# 更新日期：2026-04-06
# ///

import httpx
from typing import Any, Dict, List
from plugins.base import (
    PluginBase,
    PluginType,
    MessageHandlerPlugin,
    DataSourcePlugin,
    AIAgentPlugin,
    PluginRegistry,
)
from plugins.http_scheduled_sender import HTTPScheduledSenderPlugin


class HTTPDataSourcePlugin(DataSourcePlugin):
    plugin_name = "http_data_source"
    plugin_version = "1.0.0"
    plugin_description = "HTTP数据源插件，从外部API获取数据"
    plugin_author = "AI Generated"

    def __init__(self):
        super().__init__()
        self.base_url = ""
        self.timeout = 30
        self.headers = {}

    def initialize(self, config: Dict[str, Any]) -> bool:
        self.config = config
        self.base_url = config.get("base_url", "")
        self.timeout = config.get("timeout", 30)
        self.headers = config.get("headers", {})
        return True

    def fetch_data(self, query: str, params: Dict[str, Any]) -> List[Dict[str, Any]]:
        if not self.base_url:
            return []

        try:
            response = httpx.get(
                f"{self.base_url}{query}",
                params=params,
                headers=self.headers,
                timeout=self.timeout
            )
            response.raise_for_status()
            data = response.json()
            if isinstance(data, list):
                return data
            elif isinstance(data, dict) and "data" in data:
                return data["data"]
            return [data] if data else []
        except Exception:
            return []

    def execute(self, params: Dict[str, Any]) -> Dict[str, Any]:
        query = params.get("query", "/")
        query_params = params.get("params", {})
        data = self.fetch_data(query, query_params)
        return {"success": True, "data": data, "count": len(data)}


class DatabaseQueryPlugin(DataSourcePlugin):
    plugin_name = "database_query"
    plugin_version = "1.0.0"
    plugin_description = "数据库查询插件，通过SQL或API查询数据"
    plugin_author = "AI Generated"

    def __init__(self):
        super().__init__()
        self.connection_string = ""

    def initialize(self, config: Dict[str, Any]) -> bool:
        self.config = config
        self.connection_string = config.get("connection_string", "")
        return True

    def fetch_data(self, query: str, params: Dict[str, Any]) -> List[Dict[str, Any]]:
        return []

    def execute(self, params: Dict[str, Any]) -> Dict[str, Any]:
        source_type = params.get("source_type", "api")
        if source_type == "api":
            return {"success": True, "data": [], "message": "Use http_data_source plugin"}
        return {"success": False, "data": [], "message": "Unknown source type"}


class WebhookTriggerPlugin(PluginBase):
    plugin_name = "webhook_trigger"
    plugin_version = "1.0.0"
    plugin_description = "Webhook触发器，接收外部webhook并触发相应动作"
    plugin_author = "AI Generated"

    def __init__(self):
        super().__init__()
        self.triggers: Dict[str, str] = {}

    def initialize(self, config: Dict[str, Any]) -> bool:
        self.config = config
        self.triggers = config.get("triggers", {})
        return True

    def execute(self, params: Dict[str, Any]) -> Dict[str, Any]:
        event = params.get("event", "")
        callback_url = self.triggers.get(event, "")
        if not callback_url:
            return {"success": False, "message": f"No trigger for event: {event}"}

        try:
            response = httpx.post(callback_url, json=params.get("data", {}), timeout=10)
            return {"success": True, "status_code": response.status_code}
        except Exception as e:
            return {"success": False, "message": str(e)}

    def add_trigger(self, event: str, callback_url: str):
        self.triggers[event] = callback_url

    def remove_trigger(self, event: str) -> bool:
        if event in self.triggers:
            del self.triggers[event]
            return True
        return False


class SimpleAIAgentPlugin(AIAgentPlugin):
    plugin_name = "simple_ai_agent"
    plugin_version = "1.0.0"
    plugin_description = "简单AI智能体，支持调用外部LLM API"
    plugin_author = "AI Generated"

    def __init__(self):
        super().__init__()
        self.api_url = ""
        self.api_key = ""
        self.model = "gpt-3.5-turbo"

    def initialize(self, config: Dict[str, Any]) -> bool:
        self.config = config
        self.api_url = config.get("api_url", "")
        self.api_key = config.get("api_key", "")
        self.model = config.get("model", "gpt-3.5-turbo")
        return True

    def process(self, input_text: str, context: Dict[str, Any]) -> str:
        messages = [{"role": "user", "content": input_text}]
        return self.get_response(messages)

    def get_response(self, messages: List[Dict[str, str]]) -> str:
        if not self.api_url or not self.api_key:
            return "AI agent not configured"

        try:
            response = httpx.post(
                self.api_url,
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": self.model,
                    "messages": messages
                },
                timeout=60
            )
            result = response.json()
            return result.get("choices", [{}])[0].get("message", {}).get("content", "")
        except Exception as e:
            return f"Error: {str(e)}"

    def execute(self, params: Dict[str, Any]) -> Dict[str, Any]:
        input_text = params.get("input", "")
        context = params.get("context", {})
        result = self.process(input_text, context)
        return {"success": True, "result": result}


def register_builtin_plugins():
    from plugins.advanced_sender_plugin import AdvancedSenderPlugin
    builtin_plugins = [
        HTTPDataSourcePlugin(),
        DatabaseQueryPlugin(),
        WebhookTriggerPlugin(),
        SimpleAIAgentPlugin(),
        HTTPScheduledSenderPlugin(),
        AdvancedSenderPlugin(),
    ]

    for plugin in builtin_plugins:
        PluginRegistry.register(plugin)