scripts/webhook_server.py

#!/usr/bin/env python3
"""
Webhook服务器启动脚本
独立运行Webhook消息广播服务器
"""

import asyncio
import logging
import sys
from pathlib import Path

# 添加src目录到Python路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root / "src"))

from claude_agent.webhook.server import WebhookServer
from claude_agent.webhook.models import WebhookConfig
from claude_agent.utils.config import get_config_manager


def setup_logging():
    """设置日志"""
    logging.basicConfig(
        level=logging.INFO,
        format='[%(asctime)s] %(levelname)s %(name)s: %(message)s',
        handlers=[
            logging.StreamHandler(),
            logging.FileHandler('webhook_server.log')
        ]
    )


async def main():
    """主函数"""
    try:
        # 设置日志
        setup_logging()
        logger = logging.getLogger(__name__)

        # 获取配置
        config_name = sys.argv[1] if len(sys.argv) > 1 else "default"
        config_manager = get_config_manager(config_name)
        webhook_config_dict = config_manager.get_webhook_config()

        if not webhook_config_dict.get('enabled', False):
            logger.error("Webhook功能未启用，请检查配置文件")
            return

        # 获取服务器配置
        server_config = webhook_config_dict.get('server', {})

        # 创建Webhook配置
        webhook_config = WebhookConfig(
            server_host=server_config.get('host', '0.0.0.0'),
            server_port=server_config.get('port', 8080),
            auth_token=webhook_config_dict.get('auth_token', 'your-webhook-auth-token-here')
        )

        if webhook_config.auth_token == 'your-webhook-auth-token-here':
            logger.error("请配置有效的认证Token")
            return

        # 创建并启动服务器
        server = WebhookServer(webhook_config)

        logger.info(f"启动Webhook服务器: {webhook_config.server_host}:{webhook_config.server_port}")
        logger.info(f"认证Token: {webhook_config.auth_token[:8]}...")

        # 启动服务器
        await server.start_server()

    except KeyboardInterrupt:
        logger.info("收到中断信号，正在关闭服务器...")
    except Exception as e:
        logger.error(f"服务器启动失败: {e}")
        import traceback
        logger.error(traceback.format_exc())


if __name__ == "__main__":
    asyncio.run(main())