319 lines
11 KiB
Python
319 lines
11 KiB
Python
import logging
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, Response, status
|
|
from sqlalchemy import text
|
|
|
|
try:
|
|
from banban.service import get_db_session
|
|
from banban.routers.im import (
|
|
CHILD_PARTICIPANT_TYPE,
|
|
PARENT_PARTICIPANT_TYPE,
|
|
SUPPORTED_CONVERSATION_TYPES,
|
|
_fetch_child_names,
|
|
_fetch_parent_names,
|
|
_get_conversation_for_child,
|
|
_row_to_conversation_item,
|
|
_row_to_message_item,
|
|
)
|
|
from banban.schemas.im import (
|
|
ChildConversationListResponse,
|
|
ChildConversationMessageListResponse,
|
|
ConversationMessageCreateResponse,
|
|
DeviceMessageCreateRequest,
|
|
)
|
|
from banban.service.im import im_service, present_device_message_item
|
|
except ModuleNotFoundError:
|
|
from banban.service import get_db_session
|
|
from banban.routers.im import (
|
|
CHILD_PARTICIPANT_TYPE,
|
|
PARENT_PARTICIPANT_TYPE,
|
|
SUPPORTED_CONVERSATION_TYPES,
|
|
_fetch_child_names,
|
|
_fetch_parent_names,
|
|
_get_conversation_for_child,
|
|
_row_to_conversation_item,
|
|
_row_to_message_item,
|
|
)
|
|
from banban.schemas.im import (
|
|
ChildConversationListResponse,
|
|
ChildConversationMessageListResponse,
|
|
ConversationMessageCreateResponse,
|
|
DeviceMessageCreateRequest,
|
|
)
|
|
from banban.service.im import im_service, present_device_message_item
|
|
|
|
|
|
router = APIRouter(prefix="/device-im", tags=["device-im"])
|
|
logger = logging.getLogger("app.device_im")
|
|
|
|
|
|
@router.get("/{device_id}/conversations", response_model=ChildConversationListResponse)
|
|
async def list_device_conversations(
|
|
device_id: str,
|
|
request: Request,
|
|
device_serial: str = Header(alias="X-Device-Serial", min_length=1),
|
|
conversation_type: int | None = Query(default=None),
|
|
cursor: int | None = Query(default=None, ge=1),
|
|
limit: int = Query(default=20, ge=1, le=100),
|
|
) -> ChildConversationListResponse:
|
|
if conversation_type is not None and conversation_type not in SUPPORTED_CONVERSATION_TYPES:
|
|
raise HTTPException(status_code=422, detail="unsupported conversation_type")
|
|
|
|
device_identity = await im_service.authenticate_device_identity(
|
|
device_id=device_id,
|
|
serial_number=device_serial,
|
|
)
|
|
child_id = device_identity.child_id
|
|
|
|
from services.database_service_base import DatabaseServiceBase
|
|
db_service = DatabaseServiceBase(service_name="device_im_router")
|
|
db_session = await db_service.get_session()
|
|
try:
|
|
params: dict[str, Any] = {
|
|
"child_id_str": str(child_id),
|
|
"child_participant_type": CHILD_PARTICIPANT_TYPE,
|
|
"fetch_limit": limit + 1,
|
|
}
|
|
where = """
|
|
status = 1
|
|
AND conversation_type IN (1, 2)
|
|
AND (
|
|
(participant_a_type = :child_participant_type AND participant_a_id = :child_id_str)
|
|
OR (participant_b_type = :child_participant_type AND participant_b_id = :child_id_str)
|
|
)
|
|
"""
|
|
if conversation_type is not None:
|
|
where += " AND conversation_type = :conversation_type"
|
|
params["conversation_type"] = conversation_type
|
|
if cursor is not None:
|
|
where += " AND id < :cursor"
|
|
params["cursor"] = cursor
|
|
|
|
result = await db_session.execute(
|
|
text(
|
|
f"""
|
|
SELECT
|
|
id,
|
|
conversation_type,
|
|
participant_a_type,
|
|
participant_a_id,
|
|
participant_b_type,
|
|
participant_b_id,
|
|
last_message_preview,
|
|
last_message_at,
|
|
message_count,
|
|
created_at
|
|
FROM im_conversations
|
|
WHERE {where}
|
|
ORDER BY COALESCE(last_message_at, created_at) DESC, id DESC
|
|
LIMIT :fetch_limit
|
|
"""
|
|
),
|
|
params,
|
|
)
|
|
rows = result.mappings().all()
|
|
|
|
has_more = len(rows) > limit
|
|
rows = rows[:limit]
|
|
next_cursor = int(rows[-1]["id"]) if has_more and rows else None
|
|
|
|
peer_parent_ids: set[int] = set()
|
|
peer_child_ids: set[int] = set()
|
|
child_id_str = str(child_id)
|
|
for row in rows:
|
|
participant_a_type = int(row["participant_a_type"])
|
|
participant_b_type = int(row["participant_b_type"])
|
|
participant_a_id = str(row["participant_a_id"])
|
|
participant_b_id = str(row["participant_b_id"])
|
|
if participant_a_type == CHILD_PARTICIPANT_TYPE and participant_a_id == child_id_str:
|
|
peer_type = participant_b_type
|
|
peer_id = participant_b_id
|
|
else:
|
|
peer_type = participant_a_type
|
|
peer_id = participant_a_id
|
|
|
|
if peer_id.isdigit():
|
|
if peer_type == PARENT_PARTICIPANT_TYPE:
|
|
peer_parent_ids.add(int(peer_id))
|
|
elif peer_type == CHILD_PARTICIPANT_TYPE:
|
|
peer_child_ids.add(int(peer_id))
|
|
|
|
parent_names = await _fetch_parent_names(db_session, peer_parent_ids)
|
|
child_names = await _fetch_child_names(db_session, peer_child_ids)
|
|
items = [
|
|
_row_to_conversation_item(
|
|
row,
|
|
child_id=child_id,
|
|
parent_names=parent_names,
|
|
child_names=child_names,
|
|
)
|
|
for row in rows
|
|
]
|
|
|
|
logger.info(
|
|
"device conversations listed",
|
|
extra={
|
|
"event": "device_conversation_list",
|
|
"request_id": getattr(request.state, "request_id", None),
|
|
"device_id": device_id,
|
|
"child_id": child_id,
|
|
"conversation_type": conversation_type,
|
|
"count": len(items),
|
|
},
|
|
)
|
|
|
|
return ChildConversationListResponse(
|
|
items=items,
|
|
total=len(items),
|
|
next_cursor=next_cursor,
|
|
)
|
|
finally:
|
|
await db_session.close()
|
|
|
|
|
|
@router.get(
|
|
"/{device_id}/conversations/{conversation_id}/messages",
|
|
response_model=ChildConversationMessageListResponse,
|
|
)
|
|
async def list_device_conversation_messages(
|
|
device_id: str,
|
|
conversation_id: int,
|
|
request: Request,
|
|
device_serial: str = Header(alias="X-Device-Serial", min_length=1),
|
|
cursor_seq: int | None = Query(default=None, ge=1),
|
|
limit: int = Query(default=20, ge=1, le=100),
|
|
) -> ChildConversationMessageListResponse:
|
|
device_identity = await im_service.authenticate_device_identity(
|
|
device_id=device_id,
|
|
serial_number=device_serial,
|
|
)
|
|
from services.database_service_base import DatabaseServiceBase
|
|
db_service = DatabaseServiceBase(service_name="device_im_router")
|
|
db_session = await db_service.get_session()
|
|
try:
|
|
await _get_conversation_for_child(
|
|
db=db_session,
|
|
conversation_id=conversation_id,
|
|
child_id=device_identity.child_id,
|
|
)
|
|
finally:
|
|
await db_session.close()
|
|
|
|
from services.database_service_base import DatabaseServiceBase
|
|
db_service = DatabaseServiceBase(service_name="device_im_router")
|
|
db_session = await db_service.get_session()
|
|
try:
|
|
sql = """
|
|
SELECT
|
|
id,
|
|
conversation_id,
|
|
seq,
|
|
sender_type,
|
|
sender_id,
|
|
receiver_type,
|
|
receiver_id,
|
|
content_type,
|
|
content_text,
|
|
content_json,
|
|
media_file_key,
|
|
media_duration_ms,
|
|
media_mime_type,
|
|
media_size_bytes,
|
|
media_transcript_text,
|
|
client_msg_id,
|
|
sender_name_snapshot,
|
|
sender_avatar_snapshot,
|
|
receiver_name_snapshot,
|
|
receiver_avatar_snapshot,
|
|
ext_json,
|
|
created_at
|
|
FROM im_messages
|
|
WHERE conversation_id = :conversation_id
|
|
AND deleted_at IS NULL
|
|
"""
|
|
params: dict[str, Any] = {
|
|
"conversation_id": conversation_id,
|
|
"fetch_limit": limit + 1,
|
|
}
|
|
if cursor_seq is not None:
|
|
sql += " AND seq < :cursor_seq"
|
|
params["cursor_seq"] = cursor_seq
|
|
sql += " ORDER BY seq DESC LIMIT :fetch_limit"
|
|
|
|
result = await db_session.execute(text(sql), params)
|
|
rows = result.mappings().all()
|
|
has_more = len(rows) > limit
|
|
rows = rows[:limit]
|
|
rows = list(rows)
|
|
rows.reverse()
|
|
items = [
|
|
await present_device_message_item(row, device_id=device_id)
|
|
for row in rows
|
|
]
|
|
next_cursor_seq = items[0].seq if has_more and items else None
|
|
|
|
logger.info(
|
|
"device conversation messages listed",
|
|
extra={
|
|
"event": "device_conversation_message_list",
|
|
"request_id": getattr(request.state, "request_id", None),
|
|
"device_id": device_id,
|
|
"child_id": device_identity.child_id,
|
|
"conversation_id": conversation_id,
|
|
"count": len(items),
|
|
"has_more": has_more,
|
|
},
|
|
)
|
|
|
|
return ChildConversationMessageListResponse(
|
|
conversation_id=conversation_id,
|
|
has_more=has_more,
|
|
next_cursor_seq=next_cursor_seq,
|
|
items=items,
|
|
)
|
|
finally:
|
|
await db_session.close()
|
|
|
|
|
|
@router.post(
|
|
"/{device_id}/messages",
|
|
response_model=ConversationMessageCreateResponse,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
async def create_message_from_device(
|
|
device_id: str,
|
|
payload: DeviceMessageCreateRequest,
|
|
request: Request,
|
|
response: Response,
|
|
device_serial: str = Header(alias="X-Device-Serial", min_length=1),
|
|
) -> ConversationMessageCreateResponse:
|
|
device_identity, result = await im_service.create_device_message(
|
|
device_id=device_id,
|
|
serial_number=device_serial,
|
|
payload=payload,
|
|
)
|
|
if result.idempotent:
|
|
response.status_code = status.HTTP_200_OK
|
|
|
|
logger.info(
|
|
"device message created",
|
|
extra={
|
|
"event": "device_message_create",
|
|
"request_id": getattr(request.state, "request_id", None),
|
|
"device_id": device_id,
|
|
"child_id": device_identity.child_id,
|
|
"conversation_id": result.conversation_id,
|
|
"conversation_type": result.conversation_type,
|
|
"idempotent": result.idempotent,
|
|
},
|
|
)
|
|
|
|
return ConversationMessageCreateResponse(
|
|
idempotent=result.idempotent,
|
|
conversation_id=result.conversation_id,
|
|
conversation_type=result.conversation_type,
|
|
conversation_type_name=result.conversation_type_name,
|
|
message=result.message,
|
|
)
|