feat(app): 重构outlook邮箱服务

src/web/routes/settings.py

@@ -734,3 +734,37 @@ async def test_cpa_connection(request: CPATestRequest):
         "success": success,
         "message": message
     }
+
+
+# ============== Outlook 设置 ==============
+
+class OutlookSettings(BaseModel):
+    """Outlook 设置"""
+    default_client_id: Optional[str] = None
+
+
+@router.get("/outlook")
+async def get_outlook_settings():
+    """获取 Outlook 设置"""
+    settings = get_settings()
+
+    return {
+        "default_client_id": settings.outlook_default_client_id,
+        "provider_priority": settings.outlook_provider_priority,
+        "health_failure_threshold": settings.outlook_health_failure_threshold,
+        "health_disable_duration": settings.outlook_health_disable_duration,
+    }
+
+
+@router.post("/outlook")
+async def update_outlook_settings(request: OutlookSettings):
+    """更新 Outlook 设置"""
+    update_dict = {}
+
+    if request.default_client_id is not None:
+        update_dict["outlook_default_client_id"] = request.default_client_id
+
+    if update_dict:
+        update_settings(**update_dict)
+
+    return {"success": True, "message": "Outlook 设置已更新"}

src/web/routes/websocket.py

@@ -0,0 +1,170 @@
+"""
+WebSocket 路由
+提供实时日志推送和任务状态更新
+"""
+
+import asyncio
+import logging
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect
+
+from ..task_manager import task_manager
+
+logger = logging.getLogger(__name__)
+router = APIRouter()
+
+
+@router.websocket("/ws/task/{task_uuid}")
+async def task_websocket(websocket: WebSocket, task_uuid: str):
+    """
+    任务日志 WebSocket
+
+    消息格式：
+    - 服务端发送: {"type": "log", "task_uuid": "xxx", "message": "...", "timestamp": "..."}
+    - 服务端发送: {"type": "status", "task_uuid": "xxx", "status": "running|completed|failed|cancelled", ...}
+    - 客户端发送: {"type": "ping"} - 心跳
+    - 客户端发送: {"type": "cancel"} - 取消任务
+    """
+    await websocket.accept()
+
+    # 注册连接（会记录当前日志数量，避免重复发送历史日志）
+    task_manager.register_websocket(task_uuid, websocket)
+    logger.info(f"WebSocket 连接已建立: {task_uuid}")
+
+    try:
+        # 发送当前状态
+        status = task_manager.get_status(task_uuid)
+        if status:
+            await websocket.send_json({
+                "type": "status",
+                "task_uuid": task_uuid,
+                **status
+            })
+
+        # 发送历史日志（只发送注册时已存在的日志，避免与实时推送重复）
+        history_logs = task_manager.get_unsent_logs(task_uuid, websocket)
+        for log in history_logs:
+            await websocket.send_json({
+                "type": "log",
+                "task_uuid": task_uuid,
+                "message": log
+            })
+
+        # 保持连接，等待客户端消息
+        while True:
+            try:
+                # 使用 wait_for 实现超时，但不是断开连接
+                # 而是发送心跳检测
+                data = await asyncio.wait_for(
+                    websocket.receive_json(),
+                    timeout=30.0  # 30秒超时
+                )
+
+                # 处理心跳
+                if data.get("type") == "ping":
+                    await websocket.send_json({"type": "pong"})
+
+                # 处理取消请求
+                elif data.get("type") == "cancel":
+                    task_manager.cancel_task(task_uuid)
+                    await websocket.send_json({
+                        "type": "status",
+                        "task_uuid": task_uuid,
+                        "status": "cancelling",
+                        "message": "取消请求已提交"
+                    })
+
+            except asyncio.TimeoutError:
+                # 超时，发送心跳检测
+                try:
+                    await websocket.send_json({"type": "ping"})
+                except Exception:
+                    # 发送失败，可能是连接断开
+                    logger.info(f"WebSocket 心跳检测失败: {task_uuid}")
+                    break
+
+    except WebSocketDisconnect:
+        logger.info(f"WebSocket 断开: {task_uuid}")
+
+    except Exception as e:
+        logger.error(f"WebSocket 错误: {e}")
+
+    finally:
+        task_manager.unregister_websocket(task_uuid, websocket)
+
+
+@router.websocket("/ws/batch/{batch_id}")
+async def batch_websocket(websocket: WebSocket, batch_id: str):
+    """
+    批量任务 WebSocket
+
+    用于批量注册任务的实时状态更新
+
+    消息格式：
+    - 服务端发送: {"type": "log", "batch_id": "xxx", "message": "...", "timestamp": "..."}
+    - 服务端发送: {"type": "status", "batch_id": "xxx", "status": "running|completed|cancelled", ...}
+    - 客户端发送: {"type": "ping"} - 心跳
+    - 客户端发送: {"type": "cancel"} - 取消批量任务
+    """
+    await websocket.accept()
+
+    # 注册连接（会记录当前日志数量，避免重复发送历史日志）
+    task_manager.register_batch_websocket(batch_id, websocket)
+    logger.info(f"批量任务 WebSocket 连接已建立: {batch_id}")
+
+    try:
+        # 发送当前状态
+        status = task_manager.get_batch_status(batch_id)
+        if status:
+            await websocket.send_json({
+                "type": "status",
+                "batch_id": batch_id,
+                **status
+            })
+
+        # 发送历史日志（只发送注册时已存在的日志，避免与实时推送重复）
+        history_logs = task_manager.get_unsent_batch_logs(batch_id, websocket)
+        for log in history_logs:
+            await websocket.send_json({
+                "type": "log",
+                "batch_id": batch_id,
+                "message": log
+            })
+
+        # 保持连接，等待客户端消息
+        while True:
+            try:
+                data = await asyncio.wait_for(
+                    websocket.receive_json(),
+                    timeout=30.0
+                )
+
+                # 处理心跳
+                if data.get("type") == "ping":
+                    await websocket.send_json({"type": "pong"})
+
+                # 处理取消请求
+                elif data.get("type") == "cancel":
+                    task_manager.cancel_batch(batch_id)
+                    await websocket.send_json({
+                        "type": "status",
+                        "batch_id": batch_id,
+                        "status": "cancelling",
+                        "message": "取消请求已提交"
+                    })
+
+            except asyncio.TimeoutError:
+                # 超时，发送心跳检测
+                try:
+                    await websocket.send_json({"type": "ping"})
+                except Exception:
+                    logger.info(f"批量任务 WebSocket 心跳检测失败: {batch_id}")
+                    break
+
+    except WebSocketDisconnect:
+        logger.info(f"批量任务 WebSocket 断开: {batch_id}")
+
+    except Exception as e:
+        logger.error(f"批量任务 WebSocket 错误: {e}")
+
+    finally:
+        task_manager.unregister_batch_websocket(batch_id, websocket)

src/web/task_manager.py

@@ -0,0 +1,361 @@
+"""
+任务管理器
+负责管理后台任务、日志队列和 WebSocket 推送
+"""
+
+import asyncio
+import logging
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from typing import Dict, Optional, List, Callable, Any
+from collections import defaultdict
+from datetime import datetime
+
+logger = logging.getLogger(__name__)
+
+# 全局线程池
+_executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="reg_worker")
+
+# 任务日志队列 (task_uuid -> list of logs)
+_log_queues: Dict[str, List[str]] = defaultdict(list)
+_log_locks: Dict[str, threading.Lock] = defaultdict(threading.Lock)
+
+# WebSocket 连接管理 (task_uuid -> list of websockets)
+_ws_connections: Dict[str, List] = defaultdict(list)
+_ws_lock = threading.Lock()
+
+# WebSocket 已发送日志索引 (task_uuid -> {websocket: sent_count})
+_ws_sent_index: Dict[str, Dict] = defaultdict(dict)
+
+# 任务状态
+_task_status: Dict[str, dict] = {}
+
+# 任务取消标志
+_task_cancelled: Dict[str, bool] = {}
+
+# 批量任务状态 (batch_id -> dict)
+_batch_status: Dict[str, dict] = {}
+_batch_logs: Dict[str, List[str]] = defaultdict(list)
+_batch_locks: Dict[str, threading.Lock] = defaultdict(threading.Lock)
+
+
+class TaskManager:
+    """任务管理器"""
+
+    def __init__(self):
+        self.executor = _executor
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
+
+    def set_loop(self, loop: asyncio.AbstractEventLoop):
+        """设置事件循环（在 FastAPI 启动时调用）"""
+        self._loop = loop
+
+    def get_loop(self) -> Optional[asyncio.AbstractEventLoop]:
+        """获取事件循环"""
+        return self._loop
+
+    def is_cancelled(self, task_uuid: str) -> bool:
+        """检查任务是否已取消"""
+        return _task_cancelled.get(task_uuid, False)
+
+    def cancel_task(self, task_uuid: str):
+        """取消任务"""
+        _task_cancelled[task_uuid] = True
+        logger.info(f"任务 {task_uuid} 已标记为取消")
+
+    def add_log(self, task_uuid: str, log_message: str):
+        """添加日志并推送到 WebSocket（线程安全）"""
+        # 先广播到 WebSocket，确保实时推送
+        # 然后再添加到队列，这样 get_unsent_logs 不会获取到这条日志
+        if self._loop and self._loop.is_running():
+            try:
+                asyncio.run_coroutine_threadsafe(
+                    self._broadcast_log(task_uuid, log_message),
+                    self._loop
+                )
+            except Exception as e:
+                logger.warning(f"推送日志到 WebSocket 失败: {e}")
+
+        # 广播后再添加到队列
+        with _log_locks[task_uuid]:
+            _log_queues[task_uuid].append(log_message)
+
+    async def _broadcast_log(self, task_uuid: str, log_message: str):
+        """广播日志到所有 WebSocket 连接"""
+        with _ws_lock:
+            connections = _ws_connections.get(task_uuid, []).copy()
+            # 注意：不在这里更新 sent_index，因为日志已经通过 add_log 添加到队列
+            # sent_index 应该只在 get_unsent_logs 或发送历史日志时更新
+            # 这样可以避免竞态条件
+
+        for ws in connections:
+            try:
+                await ws.send_json({
+                    "type": "log",
+                    "task_uuid": task_uuid,
+                    "message": log_message,
+                    "timestamp": datetime.utcnow().isoformat()
+                })
+                # 发送成功后更新 sent_index
+                with _ws_lock:
+                    ws_id = id(ws)
+                    if task_uuid in _ws_sent_index and ws_id in _ws_sent_index[task_uuid]:
+                        _ws_sent_index[task_uuid][ws_id] += 1
+            except Exception as e:
+                logger.warning(f"WebSocket 发送失败: {e}")
+
+    async def broadcast_status(self, task_uuid: str, status: str, **kwargs):
+        """广播任务状态更新"""
+        with _ws_lock:
+            connections = _ws_connections.get(task_uuid, []).copy()
+
+        message = {
+            "type": "status",
+            "task_uuid": task_uuid,
+            "status": status,
+            "timestamp": datetime.utcnow().isoformat(),
+            **kwargs
+        }
+
+        for ws in connections:
+            try:
+                await ws.send_json(message)
+            except Exception as e:
+                logger.warning(f"WebSocket 发送状态失败: {e}")
+
+    def register_websocket(self, task_uuid: str, websocket):
+        """注册 WebSocket 连接"""
+        with _ws_lock:
+            if task_uuid not in _ws_connections:
+                _ws_connections[task_uuid] = []
+            # 避免重复注册同一个连接
+            if websocket not in _ws_connections[task_uuid]:
+                _ws_connections[task_uuid].append(websocket)
+                # 记录已发送的日志数量，用于发送历史日志时避免重复
+                with _log_locks[task_uuid]:
+                    _ws_sent_index[task_uuid][id(websocket)] = len(_log_queues.get(task_uuid, []))
+                logger.info(f"WebSocket 连接已注册: {task_uuid}")
+            else:
+                logger.warning(f"WebSocket 连接已存在，跳过重复注册: {task_uuid}")
+
+    def get_unsent_logs(self, task_uuid: str, websocket) -> List[str]:
+        """获取未发送给该 WebSocket 的日志"""
+        with _ws_lock:
+            ws_id = id(websocket)
+            sent_count = _ws_sent_index.get(task_uuid, {}).get(ws_id, 0)
+
+        with _log_locks[task_uuid]:
+            all_logs = _log_queues.get(task_uuid, [])
+            unsent_logs = all_logs[sent_count:]
+            # 更新已发送索引
+            _ws_sent_index[task_uuid][ws_id] = len(all_logs)
+            return unsent_logs
+
+    def unregister_websocket(self, task_uuid: str, websocket):
+        """注销 WebSocket 连接"""
+        with _ws_lock:
+            if task_uuid in _ws_connections:
+                try:
+                    _ws_connections[task_uuid].remove(websocket)
+                except ValueError:
+                    pass
+            # 清理已发送索引
+            if task_uuid in _ws_sent_index:
+                _ws_sent_index[task_uuid].pop(id(websocket), None)
+        logger.info(f"WebSocket 连接已注销: {task_uuid}")
+
+    def get_logs(self, task_uuid: str) -> List[str]:
+        """获取任务的所有日志"""
+        with _log_locks[task_uuid]:
+            return _log_queues.get(task_uuid, []).copy()
+
+    def update_status(self, task_uuid: str, status: str, **kwargs):
+        """更新任务状态"""
+        if task_uuid not in _task_status:
+            _task_status[task_uuid] = {}
+
+        _task_status[task_uuid]["status"] = status
+        _task_status[task_uuid].update(kwargs)
+
+    def get_status(self, task_uuid: str) -> Optional[dict]:
+        """获取任务状态"""
+        return _task_status.get(task_uuid)
+
+    def cleanup_task(self, task_uuid: str):
+        """清理任务数据"""
+        # 保留日志队列一段时间，以便后续查询
+        # 只清理取消标志
+        if task_uuid in _task_cancelled:
+            del _task_cancelled[task_uuid]
+
+    # ============== 批量任务管理 ==============
+
+    def init_batch(self, batch_id: str, total: int):
+        """初始化批量任务"""
+        _batch_status[batch_id] = {
+            "status": "running",
+            "total": total,
+            "completed": 0,
+            "success": 0,
+            "failed": 0,
+            "skipped": 0,
+            "current_index": 0,
+            "finished": False
+        }
+        logger.info(f"批量任务 {batch_id} 已初始化，总数: {total}")
+
+    def add_batch_log(self, batch_id: str, log_message: str):
+        """添加批量任务日志并推送"""
+        # 先广播到 WebSocket，确保实时推送
+        if self._loop and self._loop.is_running():
+            try:
+                asyncio.run_coroutine_threadsafe(
+                    self._broadcast_batch_log(batch_id, log_message),
+                    self._loop
+                )
+            except Exception as e:
+                logger.warning(f"推送批量日志到 WebSocket 失败: {e}")
+
+        # 广播后再添加到队列
+        with _batch_locks[batch_id]:
+            _batch_logs[batch_id].append(log_message)
+
+    async def _broadcast_batch_log(self, batch_id: str, log_message: str):
+        """广播批量任务日志"""
+        key = f"batch_{batch_id}"
+        with _ws_lock:
+            connections = _ws_connections.get(key, []).copy()
+            # 注意：不在这里更新 sent_index，避免竞态条件
+
+        for ws in connections:
+            try:
+                await ws.send_json({
+                    "type": "log",
+                    "batch_id": batch_id,
+                    "message": log_message,
+                    "timestamp": datetime.utcnow().isoformat()
+                })
+                # 发送成功后更新 sent_index
+                with _ws_lock:
+                    ws_id = id(ws)
+                    if key in _ws_sent_index and ws_id in _ws_sent_index[key]:
+                        _ws_sent_index[key][ws_id] += 1
+            except Exception as e:
+                logger.warning(f"WebSocket 发送批量日志失败: {e}")
+
+    def update_batch_status(self, batch_id: str, **kwargs):
+        """更新批量任务状态"""
+        if batch_id not in _batch_status:
+            logger.warning(f"批量任务 {batch_id} 不存在")
+            return
+
+        _batch_status[batch_id].update(kwargs)
+
+        # 异步广播状态更新
+        if self._loop and self._loop.is_running():
+            try:
+                asyncio.run_coroutine_threadsafe(
+                    self._broadcast_batch_status(batch_id),
+                    self._loop
+                )
+            except Exception as e:
+                logger.warning(f"广播批量状态失败: {e}")
+
+    async def _broadcast_batch_status(self, batch_id: str):
+        """广播批量任务状态"""
+        with _ws_lock:
+            connections = _ws_connections.get(f"batch_{batch_id}", []).copy()
+
+        status = _batch_status.get(batch_id, {})
+
+        for ws in connections:
+            try:
+                await ws.send_json({
+                    "type": "status",
+                    "batch_id": batch_id,
+                    "timestamp": datetime.utcnow().isoformat(),
+                    **status
+                })
+            except Exception as e:
+                logger.warning(f"WebSocket 发送批量状态失败: {e}")
+
+    def get_batch_status(self, batch_id: str) -> Optional[dict]:
+        """获取批量任务状态"""
+        return _batch_status.get(batch_id)
+
+    def get_batch_logs(self, batch_id: str) -> List[str]:
+        """获取批量任务日志"""
+        with _batch_locks[batch_id]:
+            return _batch_logs.get(batch_id, []).copy()
+
+    def is_batch_cancelled(self, batch_id: str) -> bool:
+        """检查批量任务是否已取消"""
+        status = _batch_status.get(batch_id, {})
+        return status.get("cancelled", False)
+
+    def cancel_batch(self, batch_id: str):
+        """取消批量任务"""
+        if batch_id in _batch_status:
+            _batch_status[batch_id]["cancelled"] = True
+            _batch_status[batch_id]["status"] = "cancelling"
+            logger.info(f"批量任务 {batch_id} 已标记为取消")
+
+    def register_batch_websocket(self, batch_id: str, websocket):
+        """注册批量任务 WebSocket 连接"""
+        key = f"batch_{batch_id}"
+        with _ws_lock:
+            if key not in _ws_connections:
+                _ws_connections[key] = []
+            # 避免重复注册同一个连接
+            if websocket not in _ws_connections[key]:
+                _ws_connections[key].append(websocket)
+                # 记录已发送的日志数量，用于发送历史日志时避免重复
+                with _batch_locks[batch_id]:
+                    _ws_sent_index[key][id(websocket)] = len(_batch_logs.get(batch_id, []))
+                logger.info(f"批量任务 WebSocket 连接已注册: {batch_id}")
+            else:
+                logger.warning(f"批量任务 WebSocket 连接已存在，跳过重复注册: {batch_id}")
+
+    def get_unsent_batch_logs(self, batch_id: str, websocket) -> List[str]:
+        """获取未发送给该 WebSocket 的批量任务日志"""
+        key = f"batch_{batch_id}"
+        with _ws_lock:
+            ws_id = id(websocket)
+            sent_count = _ws_sent_index.get(key, {}).get(ws_id, 0)
+
+        with _batch_locks[batch_id]:
+            all_logs = _batch_logs.get(batch_id, [])
+            unsent_logs = all_logs[sent_count:]
+            # 更新已发送索引
+            _ws_sent_index[key][ws_id] = len(all_logs)
+            return unsent_logs
+
+    def unregister_batch_websocket(self, batch_id: str, websocket):
+        """注销批量任务 WebSocket 连接"""
+        key = f"batch_{batch_id}"
+        with _ws_lock:
+            if key in _ws_connections:
+                try:
+                    _ws_connections[key].remove(websocket)
+                except ValueError:
+                    pass
+            # 清理已发送索引
+            if key in _ws_sent_index:
+                _ws_sent_index[key].pop(id(websocket), None)
+        logger.info(f"批量任务 WebSocket 连接已注销: {batch_id}")
+
+    def create_log_callback(self, task_uuid: str) -> Callable[[str], None]:
+        """创建日志回调函数"""
+        def callback(msg: str):
+            self.add_log(task_uuid, msg)
+        return callback
+
+    def create_check_cancelled_callback(self, task_uuid: str) -> Callable[[], bool]:
+        """创建检查取消的回调函数"""
+        def callback() -> bool:
+            return self.is_cancelled(task_uuid)
+        return callback
+
+
+# 全局实例
+task_manager = TaskManager()