feat: add takeover service core logic
Implement TakeoverService for avatar takeover chat feature: - check_takeover_enabled: query Authorization for enabled takeover - generate_reply: call local chat API via httpx to generate responses - execute_takeover: get owner IM credentials, generate reply, send via BoxIM - enqueue_delayed_message: Redis delayed queue with graceful degradation - process_delayed_queue: placeholder for scheduled worker Service degrades gracefully when Redis is unavailable.
This commit is contained in:
stefanfeng
@@ -0,0 +1,111 @@
|
||||
"""Takeover service — message listening, decision, reply execution."""
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from typing import Optional
|
||||
|
||||
import httpx
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from models import Authorization
|
||||
from services.boxim_client import BoxIMClient
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TakeoverService:
|
||||
"""Service for handling avatar takeover — generating replies and sending them via IM."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
db: Session,
|
||||
boxim_client: BoxIMClient,
|
||||
redis_client=None,
|
||||
):
|
||||
self.db = db
|
||||
self.boxim = boxim_client
|
||||
self.redis = redis_client
|
||||
self._chat_api_base = os.getenv(
|
||||
"TAKEOVER_CHAT_API_BASE", "http://localhost:8000/api"
|
||||
)
|
||||
|
||||
def check_takeover_enabled(
|
||||
self, owner_huihui_id: str, from_user_id: str
|
||||
) -> Optional[Authorization]:
|
||||
"""Check whether takeover is enabled for the given target user."""
|
||||
auth = (
|
||||
self.db.query(Authorization)
|
||||
.filter(Authorization.target_id == from_user_id)
|
||||
.filter(Authorization.takeover_enabled == True)
|
||||
.first()
|
||||
)
|
||||
if auth and auth.takeover_enabled:
|
||||
return auth
|
||||
return None
|
||||
|
||||
async def generate_reply(self, avatar_id: str, message: str) -> str:
|
||||
"""Call the avatar chat endpoint to generate a reply."""
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=30) as client:
|
||||
r = await client.post(
|
||||
f"{self._chat_api_base}/avatar/{avatar_id}/chat",
|
||||
json={"message": message, "history": []},
|
||||
)
|
||||
data = r.json()
|
||||
if data.get("code") in (0, 200):
|
||||
return data.get("data", {}).get("answer", "")
|
||||
logger.warning(f"Avatar chat API returned error code: {data}")
|
||||
return ""
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to call avatar chat API: {e}")
|
||||
return ""
|
||||
|
||||
async def execute_takeover(self, auth: Authorization, message: dict) -> bool:
|
||||
"""Execute takeover: generate a reply and send it as the owner via IM."""
|
||||
try:
|
||||
owner_huihui_id = auth.owner_id if hasattr(auth, "owner_id") else ""
|
||||
credentials = await self.boxim.get_credentials(owner_huihui_id)
|
||||
if not credentials:
|
||||
logger.warning(f"Cannot obtain IM credentials for owner: {owner_huihui_id}")
|
||||
return False
|
||||
|
||||
reply = await self.generate_reply(auth.avatar_id, message.get("content", ""))
|
||||
if not reply:
|
||||
logger.warning("Avatar did not generate a reply")
|
||||
return False
|
||||
|
||||
success = await self.boxim.send_p2p_message(
|
||||
from_accid=credentials["accid"],
|
||||
to_accid=message.get("from_accid", ""),
|
||||
content=reply,
|
||||
)
|
||||
if success:
|
||||
logger.info(f"Takeover reply sent successfully: {reply[:50]}...")
|
||||
return success
|
||||
except Exception as e:
|
||||
logger.error(f"Takeover execution failed: {e}")
|
||||
return False
|
||||
|
||||
def enqueue_delayed_message(self, auth: Authorization, message: dict):
|
||||
"""Write a message into the Redis delayed queue (TTL = delay + 10s buffer)."""
|
||||
if not self.redis:
|
||||
logger.warning("Redis not configured, degrading to immediate takeover")
|
||||
return
|
||||
key = f"takeover:delayed:{auth.target_id}:{message.get('msg_id', '')}"
|
||||
value = json.dumps({
|
||||
"avatar_id": auth.avatar_id,
|
||||
"from_accid": message.get("from_accid", ""),
|
||||
"content": message.get("content", ""),
|
||||
"owner_huihui_id": auth.owner_id if hasattr(auth, "owner_id") else "",
|
||||
})
|
||||
self.redis.setex(key, auth.takeover_delay_seconds + 10, value)
|
||||
logger.info(f"Message enqueued to delayed queue: {key}")
|
||||
|
||||
async def process_delayed_queue(self):
|
||||
"""Process expired messages from the delayed queue.
|
||||
|
||||
Redis handles expiration automatically; this is a placeholder
|
||||
for a future scheduled worker that scans and dispatches.
|
||||
"""
|
||||
if not self.redis:
|
||||
return
|
||||
@@ -0,0 +1,136 @@
|
||||
"""Tests for the TakeoverService — message listening, decision, reply execution."""
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, patch, MagicMock
|
||||
from services.takeover_service import TakeoverService
|
||||
from models import Authorization
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_db():
|
||||
db = MagicMock()
|
||||
return db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_boxim():
|
||||
client = AsyncMock()
|
||||
client.get_credentials.return_value = {"accid": "owner_acc", "token": "tok"}
|
||||
client.send_p2p_message.return_value = True
|
||||
return client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_auth():
|
||||
auth = MagicMock(spec=Authorization)
|
||||
auth.takeover_enabled = True
|
||||
auth.takeover_mode = "immediate"
|
||||
auth.takeover_delay_seconds = 30
|
||||
auth.avatar_id = "avatar_123"
|
||||
auth.target_id = "target_user_123"
|
||||
auth.owner_id = "owner_huihui_123"
|
||||
return auth
|
||||
|
||||
|
||||
def test_check_takeover_enabled_returns_auth_when_enabled(mock_db, mock_auth, mock_boxim):
|
||||
mock_db.query.return_value.filter.return_value.filter.return_value.first.return_value = mock_auth
|
||||
service = TakeoverService(mock_db, mock_boxim)
|
||||
result = service.check_takeover_enabled("owner_123", "target_123")
|
||||
assert result == mock_auth
|
||||
|
||||
|
||||
def test_check_takeover_enabled_returns_none_when_disabled(mock_db, mock_boxim):
|
||||
disabled_auth = MagicMock(spec=Authorization)
|
||||
disabled_auth.takeover_enabled = False
|
||||
mock_db.query.return_value.filter.return_value.filter.return_value.first.return_value = disabled_auth
|
||||
service = TakeoverService(mock_db, mock_boxim)
|
||||
result = service.check_takeover_enabled("owner_123", "target_123")
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generate_reply_returns_answer(mock_boxim):
|
||||
mock_db = MagicMock()
|
||||
with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class:
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {"code": 200, "data": {"answer": "Hello back"}}
|
||||
mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response
|
||||
|
||||
service = TakeoverService(mock_db, mock_boxim)
|
||||
result = await service.generate_reply("avatar_123", "Hello")
|
||||
assert result == "Hello back"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_takeover_success(mock_db, mock_boxim, mock_auth):
|
||||
with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class:
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {"code": 200, "data": {"answer": "Hello back"}}
|
||||
mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response
|
||||
|
||||
service = TakeoverService(mock_db, mock_boxim)
|
||||
message = {"from_accid": "user_acc", "content": "Hello"}
|
||||
|
||||
result = await service.execute_takeover(mock_auth, message)
|
||||
|
||||
assert result is True
|
||||
mock_boxim.get_credentials.assert_called_once()
|
||||
mock_boxim.send_p2p_message.assert_called_once()
|
||||
|
||||
|
||||
def test_enqueue_delayed_message_with_redis(mock_db, mock_boxim, mock_auth):
|
||||
mock_redis = MagicMock()
|
||||
service = TakeoverService(mock_db, mock_boxim, mock_redis)
|
||||
message = {"msg_id": "msg_1", "from_accid": "user_acc", "content": "Hello"}
|
||||
|
||||
service.enqueue_delayed_message(mock_auth, message)
|
||||
|
||||
mock_redis.setex.assert_called_once()
|
||||
|
||||
|
||||
def test_enqueue_delayed_message_without_redis_logs_warning(mock_db, mock_boxim, mock_auth):
|
||||
"""When Redis is not configured, enqueue_delayed_message should log a warning and not crash."""
|
||||
service = TakeoverService(mock_db, mock_boxim)
|
||||
message = {"msg_id": "msg_1", "from_accid": "user_acc", "content": "Hello"}
|
||||
|
||||
# Should not raise
|
||||
service.enqueue_delayed_message(mock_auth, message)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_takeover_fails_when_no_credentials(mock_db, mock_boxim, mock_auth):
|
||||
"""execute_takeover should return False when boxim.get_credentials returns None."""
|
||||
mock_boxim.get_credentials.return_value = None
|
||||
service = TakeoverService(mock_db, mock_boxim)
|
||||
message = {"from_accid": "user_acc", "content": "Hello"}
|
||||
|
||||
result = await service.execute_takeover(mock_auth, message)
|
||||
|
||||
assert result is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generate_reply_handles_empty_answer(mock_boxim):
|
||||
"""generate_reply should return empty string when answer is missing."""
|
||||
mock_db = MagicMock()
|
||||
with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class:
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {"code": 200, "data": {}}
|
||||
mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response
|
||||
|
||||
service = TakeoverService(mock_db, mock_boxim)
|
||||
result = await service.generate_reply("avatar_123", "Hello")
|
||||
assert result == ""
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generate_reply_handles_error_code(mock_boxim):
|
||||
"""generate_reply should return empty string when API returns error code."""
|
||||
mock_db = MagicMock()
|
||||
with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class:
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {"code": 500, "message": "Internal error"}
|
||||
mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response
|
||||
|
||||
service = TakeoverService(mock_db, mock_boxim)
|
||||
result = await service.generate_reply("avatar_123", "Hello")
|
||||
assert result == ""
|
||||
Reference in New Issue
Block a user