Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 1 addition & 14 deletions astrbot/core/db/migration/migra_webchat_session.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
from sqlalchemy import func, select
from sqlmodel import col

from astrbot.api import logger, sp
from astrbot.api import logger
from astrbot.core.db import BaseDatabase
from astrbot.core.db.po import ConversationV2, PlatformMessageHistory, PlatformSession

Expand All @@ -23,13 +23,6 @@ async def migrate_webchat_session(db_helper: BaseDatabase) -> None:
This migration extracts all unique user_ids from platform_message_history
where platform_id='webchat' and creates corresponding PlatformSession records.
"""
# 检查是否已经完成迁移
migration_done = await db_helper.get_preference(
"global", "global", "migration_done_webchat_session_1"
)
if migration_done:
return

logger.info("开始执行数据库迁移(WebChat 会话迁移)...")

try:
Expand All @@ -52,9 +45,6 @@ async def migrate_webchat_session(db_helper: BaseDatabase) -> None:

if not webchat_users:
logger.info("没有找到需要迁移的 WebChat 数据")
await sp.put_async(
"global", "global", "migration_done_webchat_session_1", True
)
return

logger.info(f"找到 {len(webchat_users)} 个 WebChat 会话需要迁移")
Expand Down Expand Up @@ -123,9 +113,6 @@ async def migrate_webchat_session(db_helper: BaseDatabase) -> None:
else:
logger.info("没有新会话需要迁移")

# 标记迁移完成
await sp.put_async("global", "global", "migration_done_webchat_session_1", True)

except Exception as e:
logger.error(f"迁移过程中发生错误: {e}", exc_info=True)
raise
19 changes: 19 additions & 0 deletions astrbot/dashboard/services/chat_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -1115,6 +1115,25 @@ async def build_chat_stream(
"Message content is empty (reply only is not allowed)"
)

if platform_history_id == "webchat":
try:
platform_session = await self.db.get_platform_session_by_id(
webchat_conv_id
)
if platform_session is None:
await self.db.create_platform_session(
creator=username,
platform_id="webchat",
session_id=webchat_conv_id,
is_group=0,
)
except Exception as exc:
logger.warning(
"Failed to ensure WebChat platform session %s: %s",
webchat_conv_id,
exc,
)

message_id = str(uuid.uuid4())
llm_checkpoint_id = post_data.get("_llm_checkpoint_id") or str(uuid.uuid4())
skip_user_history = bool(post_data.get("_skip_user_history"))
Expand Down
37 changes: 36 additions & 1 deletion tests/test_chat_route.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,12 @@ def chat_service_instance(monkeypatch, tmp_path):
platform_message_history_manager=platform_history_mgr,
umop_config_router=Mock(),
)
service = ChatService(Mock(), core_lifecycle)
db = Mock()
db.get_platform_session_by_id = AsyncMock(
return_value=SimpleNamespace(session_id="existing-session")
)
db.create_platform_session = AsyncMock()
service = ChatService(db, core_lifecycle)
service.build_user_message_parts = AsyncMock(
return_value=[{"type": "plain", "text": "hello"}]
)
Expand All @@ -40,6 +45,36 @@ def chat_service_instance(monkeypatch, tmp_path):
return service


@pytest.mark.asyncio
async def test_chat_stream_creates_missing_webchat_platform_session(
chat_service_instance,
):
service = chat_service_instance
session_id = "missing-platform-session"
service.db.get_platform_session_by_id.return_value = None

stream = await service.build_chat_stream(
"alice",
{"message": "hello", "session_id": session_id},
)
run = next(iter(service.chat_runs.values()))

try:
service.db.get_platform_session_by_id.assert_awaited_once_with(session_id)
service.db.create_platform_session.assert_awaited_once_with(
creator="alice",
platform_id="webchat",
session_id=session_id,
is_group=0,
)
finally:
await stream.aclose()
if run.task and not run.task.done():
run.task.cancel()
await asyncio.gather(run.task, return_exceptions=True)
chat_service.webchat_queue_mgr.remove_queues(session_id)


def _decode_sse_event(event: str) -> dict:
"""Decode one JSON SSE event emitted by ChatService.

Expand Down