Simon
bea288e414
== 已部署上线 (9项) == - 代办事项真实数据源集成 (企微审批API 8bug修复链) - H5/坐席端 Logo样式统一+绿色背景 - 视频引导页修复 (localStorage key v2) - 坐席端 v9 Vue版本修复 (ElMessage._context) - 截图按钮 v10 修复 (getDisplayMedia user gesture) - 扫码样式恢复+H5扫码登录跳转修复 - H5截图快捷键提示 == 代码完成待部署 (3项) == - 知识迭代3Bug修复 (#8 POST端点/#7 MERGE幂等/#6 过期检查) - 会议室预定-小鱼易联终端 (40文件, 40/40测试通过) - IT资产升级审批推送 (asset_service.py) == 需求文档 (2项) == - 坐席端AI辅助消息框-PRD (4项新功能确认) - 坐席端布局优化建议 v2.0 (7天计划) == 新增文档 == - 日报-2026-07-11.md - 知识迭代Bug修复报告-20260711.md - 会议室预定-部署指南.md - CHANGELOG.md 更新 == 测试 == - test_todo_integration.py: 40/40 - test_meetingroom.py: 40/40 - test_bugfix_ki_suggestions.py: 21/21
130 lines
4.9 KiB
Python
130 lines
4.9 KiB
Python
# =============================================================================
|
||
# 企微IT智能服务台 — 复杂场景重构 暂停超时清理定时任务
|
||
# =============================================================================
|
||
# 说明：后台定时任务，扫描 paused 状态且超过 24 小时未恢复的会话，
|
||
# 自动标记为 closed（closed_by = "system(timeout)"），
|
||
# 并清理 Redis 恢复点与暂停会话集合，推送超时关闭 WS 事件。
|
||
#
|
||
# 调用方式：
|
||
# 1. FastAPI lifespan 中 asyncio.create_task(TimeoutCleaner(...).run_scheduled())
|
||
# 2. 或外部调度器定期调用 run_once()
|
||
# =============================================================================
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
from datetime import datetime, timedelta, timezone
|
||
from typing import Any
|
||
|
||
from sqlalchemy import select
|
||
|
||
from app.constants import (
|
||
PAUSE_TIMEOUT_HOURS,
|
||
REDIS_KEY_PAUSED_SESSIONS,
|
||
REDIS_KEY_RESUME_POINT,
|
||
)
|
||
from app.models.automation import AutoSession
|
||
from app.services.automation.progress_publisher import publish_timeout_closed
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class TimeoutCleaner:
|
||
"""暂停超时清理器。
|
||
|
||
扫描 paused 状态的会话，超过 PAUSE_TIMEOUT_HOURS（默认 24 小时）
|
||
未恢复的会话自动关闭，并推送 WS 通知。
|
||
"""
|
||
|
||
def __init__(self, db_factory: Any, redis: Any = None):
|
||
"""初始化。
|
||
|
||
Args:
|
||
db_factory: 异步 DB 会话工厂（如 app.database._get_session_factory()）
|
||
redis: Redis 客户端（可选，用于清理恢复点）
|
||
"""
|
||
self.db_factory = db_factory
|
||
self.redis = redis
|
||
|
||
async def run_once(self) -> int:
|
||
"""执行一次扫描，返回关闭的会话数。
|
||
|
||
Returns:
|
||
int: 本次扫描关闭的会话数量
|
||
"""
|
||
closed_count = 0
|
||
async with self.db_factory() as db:
|
||
# 查询超时的 paused 会话（paused_at < cutoff 隐含 NOT NULL）
|
||
cutoff = datetime.now(timezone.utc) - timedelta(hours=PAUSE_TIMEOUT_HOURS)
|
||
stmt = select(AutoSession).where(
|
||
AutoSession.status == "paused",
|
||
AutoSession.paused_at < cutoff,
|
||
)
|
||
sessions = list((await db.execute(stmt)).scalars().all())
|
||
|
||
if not sessions:
|
||
return 0
|
||
|
||
for session in sessions:
|
||
try:
|
||
# 标记关闭
|
||
session.status = "closed"
|
||
session.closed_by = "system(timeout)"
|
||
closed_at = datetime.now(timezone.utc)
|
||
|
||
# 清理 Redis 恢复点
|
||
if self.redis:
|
||
await self._cleanup_redis(session.id, session.employee_id)
|
||
|
||
await db.flush()
|
||
closed_count += 1
|
||
|
||
# 推送 WS 超时关闭事件
|
||
await publish_timeout_closed(
|
||
session_id=session.id,
|
||
closed_at=closed_at.isoformat(),
|
||
reason=f"暂停超过 {PAUSE_TIMEOUT_HOURS} 小时未恢复，已自动关闭",
|
||
)
|
||
|
||
logger.info(
|
||
f"超时关闭会话: session={session.id} "
|
||
f"paused_at={session.paused_at} closed_at={closed_at}"
|
||
)
|
||
except Exception as e: # noqa: BLE001
|
||
logger.error(f"超时关闭会话失败 session={session.id}: {e}")
|
||
|
||
await db.commit()
|
||
|
||
if closed_count > 0:
|
||
logger.info(f"超时清理完成: 共关闭 {closed_count} 个暂停会话")
|
||
return closed_count
|
||
|
||
async def run_scheduled(self, interval: int = 3600) -> None:
|
||
"""定时扫描（每小时一次）。
|
||
|
||
Args:
|
||
interval: 扫描间隔（秒），默认 3600 = 1 小时
|
||
"""
|
||
logger.info(f"启动暂停超时清理定时任务，间隔 {interval} 秒")
|
||
while True:
|
||
try:
|
||
await self.run_once()
|
||
except Exception as e: # noqa: BLE001
|
||
logger.error(f"超时清理定时任务异常: {e}")
|
||
await asyncio.sleep(interval)
|
||
|
||
async def _cleanup_redis(self, session_id: str, employee_id: str) -> None:
|
||
"""清理 Redis 中的恢复点和暂停会话集合。"""
|
||
if not self.redis:
|
||
return
|
||
try:
|
||
# 删除恢复点
|
||
resume_key = REDIS_KEY_RESUME_POINT.format(session_id=session_id)
|
||
await self.redis.delete(resume_key)
|
||
# 从暂停会话集合中移除
|
||
paused_key = REDIS_KEY_PAUSED_SESSIONS.format(employee_id=employee_id)
|
||
await self.redis.srem(paused_key, session_id)
|
||
except Exception as e: # noqa: BLE001
|
||
logger.warning(f"Redis 清理失败 session={session_id}: {e}")
|