callback_handler.py

# callback_handler.py
import asyncio
import datetime
import logging
from enum import Enum
from typing import Dict, Optional, Callable, Union, List, Coroutine
import httpx
from pydantic import BaseModel
from config_manager import config_manager

# 配置常量
MAX_RETRIES = 3
RETRY_DELAY = 1.0
TIMEOUT = 10.0


class EventType(Enum):
    """回调事件类型枚举"""
    LOGIN_REDIRECT = "login_redirect"
    PAGE_NAVIGATED = "page_navigated"
    PAGE_LOADED = "page_loaded"
    STATUS_UPDATE = "status_update"
    REALTIME_MESSAGE = "realtime_message"
    LOGIN_CONFIRM = "login_confirm"
    CLOSE_PAGE = "close_page"
    NEW_MESSAGE = "new_message"
    MOVE_CONVERSATION = "move_conversation"
    LOGOUT = "logout"
    ERROR = "error"


class CallbackConfig(BaseModel):
    """回调配置模型"""
    endpoint: List[str] = []
    timeout: float = TIMEOUT
    retries: int = MAX_RETRIES
    headers: Dict[str, str] = {"Content-Type": "application/json"}

    def __init__(self):
        super().__init__()
        try:
            self.endpoint = config_manager.get_config("callback.endpoint", [])
            self.timeout = config_manager.get_config("callback.timeout", TIMEOUT)
            self.retries = config_manager.get_config("callback.max_retries", MAX_RETRIES)
        except KeyError as e:
            logging.error(f"Configuration key not found: {e}")
            # 使用默认值继续运行
        except ValueError as e:
            logging.error(f"Invalid configuration value: {e}")
            # 使用默认值继续运行
        except Exception as e:
            logging.error(f"Error loading config: {e}")
            raise e



class CallbackRequest(BaseModel):
    """回调请求数据模型"""
    event: EventType
    workstation_id: str
    data: Dict
    timestamp: datetime.datetime
    retry_count: int = 0


class CallbackHandler:
    """异步回调处理器"""

    def __init__(
            self,
            config: CallbackConfig,
            logger: Optional[logging.Logger] = None
    ):
        self._processing_task = None
        self.config = config
        self.logger = logger or logging.getLogger(__name__)
        self._queue = asyncio.Queue()
        self._active_tasks: Dict[str, asyncio.Task] = {}
        self._custom_handlers: Dict[EventType, List[Callable]] = {}
        self._running = True


    def add_callback_endpoint(self, new_endpoint: str):
        """
        动态设置回调地址
        :param new_endpoint: 新的回调地址
        """
        self.config.endpoint.append(new_endpoint)
        config_manager.set_config("callback.endpoint", self.config.endpoint)
        self.logger.info(f"Callback endpoint updated to: {new_endpoint}")
        return self.config.endpoint


    def delete_callback_endpoint(self, endpoint: str):
        """
        删除回调地址
        :param endpoint: 要删除的回调地址
        """
        if endpoint in self.config.endpoint:
            self.config.endpoint.remove(endpoint)
            config_manager.set_config("callback.endpoint", self.config.endpoint)
            self.logger.info(f"Callback endpoint deleted: {endpoint}")
            return True
        else:
            self.logger.warning(f"Callback endpoint not found: {endpoint}")
            return False


    async def start(self):
        """启动后台处理任务"""
        self._processing_task = asyncio.create_task(self._process_queue())
        self.logger.info("Callback handler started")

    async def stop(self):
        """停止处理器并等待任务完成"""
        self._running = False
        await self._queue.join()
        self._processing_task.cancel()
        self.logger.info("Callback handler stopped")

    def register_handler(
            self,
            event_type: EventType,
            handler: Callable[[CallbackRequest], Union[None, Coroutine]]
    ):
        """注册自定义事件处理器"""
        if event_type not in self._custom_handlers:
            self._custom_handlers[event_type] = []
        self._custom_handlers[event_type].append(handler)
        self.logger.debug(f"Registered custom handler for {event_type}")

    async def trigger(
            self,
            event: EventType,
            workstation_id: str,
            data: Dict,
            immediate: bool = False
    ):
        """
        触发回调事件
        :param data:
        :param immediate: 是否立即发送（跳过队列）
        """
        request = CallbackRequest(
            event=event,
            workstation_id=workstation_id,
            data=data,
            timestamp=datetime.datetime.now()
        )

        # 先执行本地注册的处理器
        await self._run_local_handlers(request)

        # HTTP回调加入队列处理
        if immediate:
            await self._send_request(request)
        else:
            await self._queue.put(request)
            self.logger.debug(f"Queued callback: {event.value} for {workstation_id}")

    async def _run_local_handlers(self, request: CallbackRequest):
        """执行本地注册的事件处理器"""
        if handlers := self._custom_handlers.get(request.event):
            for handler in handlers:
                try:
                    if asyncio.iscoroutinefunction(handler):
                        await handler(request)
                    else:
                        handler(request)
                except Exception as e:
                    self.logger.error(
                        f"Local handler failed for {request.event}: {str(e)}",
                        exc_info=True
                    )

    async def _process_queue(self):
        """处理队列中的回调请求"""
        while self._running:
            request = await self._queue.get()
            task_id = f"{request.workstation_id}-{request.event.value}-{id(request)}"

            try:
                self._active_tasks[task_id] = asyncio.create_task(
                    self._send_with_retry(request)
                )
                await self._active_tasks[task_id]
            except Exception as e:
                self.logger.error(f"Failed processing task {task_id}: {str(e)}")
            finally:
                self._queue.task_done()
                await self._active_tasks.pop(task_id, None)

    async def _send_with_retry(self, request: CallbackRequest):
        """带重试机制的发送逻辑"""
        for attempt in range(self.config.retries + 1):
            try:
                return await self._send_request(request)
            except (httpx.RequestError, httpx.HTTPStatusError) as e:
                if attempt == self.config.retries:
                    error_msg = f"Callback failed after {attempt} retries: {str(e)}"
                    await self._handle_failure(request, error_msg)
                    break

                delay = RETRY_DELAY * (2 ** attempt)
                self.logger.warning(
                    f"Retrying {request.event} in {delay}s (attempt {attempt + 1})"
                )
                await asyncio.sleep(delay)
                request.retry_count += 1

    async def _send_request(self, request: CallbackRequest):
        """实际发送HTTP请求"""
        async with httpx.AsyncClient() as client:
            response = None
            for endpoint in self.config.endpoint:
                response = await client.post(
                    endpoint,
                    content=request.json(),
                    headers=self.config.headers,
                    timeout=self.config.timeout
                )
                response.raise_for_status()
                self.logger.info(
                    f"Callback succeeded: {request.event.value} "
                    f"to {self.config.endpoint} (status {response.status_code})"
                )
            return response

    async def _handle_failure(self, request: CallbackRequest, error_msg: str):
        """失败处理"""
        self.logger.error(error_msg)
        # 触发错误回调
        error_data = {
            "original_event": request.dict(),
            "error": error_msg
        }
        await self.trigger(
            EventType.ERROR,
            request.workstation_id,
            error_data,
            immediate=True
        )


class CallbackMetrics:
    """回调监控指标"""

    def __init__(self):
        self.total_events = 0
        self.success_count = 0
        self.failure_count = 0
        self.retry_count = 0

    def increment(self, success: bool):
        self.total_events += 1
        if success:
            self.success_count += 1
        else:
            self.failure_count += 1

    def get_stats(self) -> Dict:
        return {
            "total": self.total_events,
            "success_rate": self.success_count / self.total_events if self.total_events else 0,
            "avg_retries": self.retry_count / self.total_events if self.total_events else 0
        }