523 lines
18 KiB
Python
523 lines
18 KiB
Python
import json
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass
|
|
from typing import Any, Optional, Tuple
|
|
|
|
from sqlalchemy import text
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from banban.dao import BaseDAO
|
|
from banban.db_compat import inserted_primary_key
|
|
from banban.schemas.im import ChildConversationMessageItem
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DeviceIdentity:
|
|
device_id: str
|
|
child_id: int
|
|
child_name: str | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ConversationMessageCreateResult:
|
|
idempotent: bool
|
|
conversation_id: int
|
|
conversation_type: int
|
|
message: dict
|
|
|
|
@property
|
|
def conversation_type_name(self) -> str:
|
|
if self.conversation_type == 1:
|
|
return "child_peer"
|
|
if self.conversation_type == 2:
|
|
return "parent_child"
|
|
return f"unknown_{self.conversation_type}"
|
|
|
|
|
|
class ImDAO(BaseDAO):
|
|
async def assert_parent_child_access(self, *, user_id: int, child_id: int) -> Mapping[str, Any]:
|
|
child_row = await self._get_child_row(child_id)
|
|
if not child_row:
|
|
from fastapi import HTTPException, status
|
|
raise HTTPException(status_code=404, detail="child not found")
|
|
|
|
has_access = (
|
|
await self.execute(
|
|
text(
|
|
"""
|
|
SELECT 1
|
|
FROM parent_child_relations
|
|
WHERE user_id = :user_id
|
|
AND child_id = :child_id
|
|
AND status = 1
|
|
LIMIT 1
|
|
"""
|
|
),
|
|
{"user_id": user_id, "child_id": child_id},
|
|
)
|
|
).scalar_one_or_none() is not None
|
|
if not has_access:
|
|
from fastapi import HTTPException, status
|
|
raise HTTPException(status_code=403, detail="no permission to access this child")
|
|
|
|
return child_row
|
|
|
|
async def authenticate_device_identity(self, *, device_id: str, serial_number: str) -> DeviceIdentity:
|
|
row = (
|
|
await self.execute(
|
|
text(
|
|
"""
|
|
SELECT
|
|
da.device_id,
|
|
db.child_id,
|
|
c.child_name
|
|
FROM device_auth AS da
|
|
LEFT JOIN device_bindings AS db
|
|
ON db.device_id = da.device_id
|
|
AND db.status = 1
|
|
LEFT JOIN children AS c
|
|
ON c.child_id = db.child_id
|
|
AND c.status = 1
|
|
WHERE da.device_id = :device_id
|
|
AND da.serial_number = :serial_number
|
|
AND da.is_active = 1
|
|
LIMIT 1
|
|
"""
|
|
),
|
|
{"device_id": device_id, "serial_number": serial_number},
|
|
)
|
|
).mappings().first()
|
|
if not row:
|
|
from fastapi import HTTPException, status
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid device credentials")
|
|
if row["child_id"] is None:
|
|
from fastapi import HTTPException, status
|
|
raise HTTPException(status_code=404, detail="device not bound to a child")
|
|
return DeviceIdentity(
|
|
device_id=str(row["device_id"]),
|
|
child_id=int(row["child_id"]),
|
|
child_name=row["child_name"],
|
|
)
|
|
|
|
async def get_device_by_id(self, *, device_id: str) -> DeviceIdentity:
|
|
row = (
|
|
await self.execute(
|
|
text(
|
|
"""
|
|
SELECT
|
|
da.device_id,
|
|
db.child_id,
|
|
c.child_name
|
|
FROM device_auth AS da
|
|
LEFT JOIN device_bindings AS db
|
|
ON db.device_id = da.device_id
|
|
AND db.status = 1
|
|
LEFT JOIN children AS c
|
|
ON c.child_id = db.child_id
|
|
AND c.status = 1
|
|
WHERE da.device_id = :device_id
|
|
AND da.is_active = 1
|
|
LIMIT 1
|
|
"""
|
|
),
|
|
{"device_id": device_id},
|
|
)
|
|
).mappings().first()
|
|
if not row:
|
|
from fastapi import HTTPException, status
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid device credentials")
|
|
if row["child_id"] is None:
|
|
from fastapi import HTTPException, status
|
|
raise HTTPException(status_code=404, detail="device not bound to a child")
|
|
return DeviceIdentity(
|
|
device_id=str(row["device_id"]),
|
|
child_id=int(row["child_id"]),
|
|
child_name=row["child_name"],
|
|
)
|
|
|
|
async def assert_child_exists(self, *, child_id: int) -> Mapping[str, Any]:
|
|
child_row = await self._get_child_row(child_id)
|
|
if not child_row:
|
|
from fastapi import HTTPException, status
|
|
raise HTTPException(status_code=404, detail="child not found")
|
|
return child_row
|
|
|
|
async def _get_child_row(self, child_id: int) -> Mapping[str, Any] | None:
|
|
result = await self.execute(
|
|
text(
|
|
"""
|
|
SELECT child_id, child_name, status
|
|
FROM children
|
|
WHERE child_id = :child_id
|
|
AND status = 1
|
|
LIMIT 1
|
|
"""
|
|
),
|
|
{"child_id": child_id},
|
|
)
|
|
return result.mappings().first()
|
|
|
|
async def _get_parent_row(self, user_id: int) -> Mapping[str, Any] | None:
|
|
result = await self.execute(
|
|
text(
|
|
"""
|
|
SELECT user_id, nickname, avatar_url, status
|
|
FROM parents
|
|
WHERE user_id = :user_id
|
|
AND status = 1
|
|
LIMIT 1
|
|
"""
|
|
),
|
|
{"user_id": user_id},
|
|
)
|
|
return result.mappings().first()
|
|
|
|
def _build_child_peer_pair(self, child_a_id: int, child_b_id: int) -> Tuple[str, str, str]:
|
|
low_id, high_id = sorted((child_a_id, child_b_id))
|
|
participant_a_id = str(low_id)
|
|
participant_b_id = str(high_id)
|
|
return participant_a_id, participant_b_id, f"{participant_a_id}:{participant_b_id}"
|
|
|
|
async def _build_preview(
|
|
self,
|
|
content_type: int,
|
|
content_text: str | None,
|
|
ext_json: dict[str, Any] | None = None,
|
|
) -> str:
|
|
if content_type == 1:
|
|
return (content_text or "").strip()[:255]
|
|
if content_type == 2:
|
|
if (ext_json or {}).get("message_kind") == "leave_message":
|
|
return "[留言]"
|
|
return "[语音]"
|
|
if content_type == 3:
|
|
return "[image]"
|
|
return "[json]"
|
|
|
|
async def ensure_parent_child_conversation(self, *, parent_user_id: int, child_id: int) -> int:
|
|
child_row = await self.assert_parent_child_access(user_id=parent_user_id, child_id=child_id)
|
|
parent_row = await self._get_parent_row(parent_user_id)
|
|
if not parent_row:
|
|
from fastapi import HTTPException
|
|
raise HTTPException(status_code=404, detail="parent not found")
|
|
|
|
return await self._get_or_create_conversation(
|
|
conversation_type=2,
|
|
participant_a_type=2,
|
|
participant_a_id=str(child_id),
|
|
participant_b_type=1,
|
|
participant_b_id=str(parent_user_id),
|
|
pair_key=f"{child_id}:{parent_user_id}",
|
|
)
|
|
|
|
async def create_message(
|
|
self,
|
|
*,
|
|
conversation_type: int,
|
|
participant_a_type: int,
|
|
participant_a_id: str,
|
|
participant_b_type: int,
|
|
participant_b_id: str,
|
|
pair_key: str,
|
|
sender_type: int,
|
|
sender_id: str,
|
|
receiver_type: int,
|
|
receiver_id: str,
|
|
sender_name_snapshot: str | None,
|
|
sender_avatar_snapshot: str | None,
|
|
receiver_name_snapshot: str | None,
|
|
receiver_avatar_snapshot: str | None,
|
|
payload: Any,
|
|
) -> tuple[int, bool]:
|
|
conversation_id = await self._get_or_create_conversation(
|
|
conversation_type=conversation_type,
|
|
participant_a_type=participant_a_type,
|
|
participant_a_id=participant_a_id,
|
|
participant_b_type=participant_b_type,
|
|
participant_b_id=participant_b_id,
|
|
pair_key=pair_key,
|
|
)
|
|
|
|
existing = await self._get_message_by_conversation_client_id(
|
|
conversation_id=conversation_id,
|
|
client_msg_id=payload.client_msg_id,
|
|
)
|
|
if existing:
|
|
return conversation_id, True
|
|
|
|
now_sql = "CURRENT_TIMESTAMP(3)"
|
|
preview = await self._build_preview(
|
|
payload.content_type,
|
|
payload.content_text,
|
|
payload.ext_json,
|
|
)
|
|
|
|
try:
|
|
conversation_row = await self._get_conversation_by_id(conversation_id=conversation_id, lock=True)
|
|
if not conversation_row:
|
|
from fastapi import HTTPException, status
|
|
raise HTTPException(status_code=404, detail="conversation not found")
|
|
next_seq = int(conversation_row["last_seq"]) + 1
|
|
|
|
message_id = await self._next_primary_key("im_messages")
|
|
insert_sql = f"""
|
|
INSERT INTO im_messages (
|
|
{'id,' if message_id is not None else ''}
|
|
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
|
|
) VALUES (
|
|
{':id,' if message_id is not None else ''}
|
|
: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,
|
|
{now_sql}
|
|
)
|
|
"""
|
|
params: dict[str, Any] = {
|
|
"conversation_id": conversation_id,
|
|
"seq": next_seq,
|
|
"sender_type": sender_type,
|
|
"sender_id": sender_id,
|
|
"receiver_type": receiver_type,
|
|
"receiver_id": receiver_id,
|
|
"content_type": payload.content_type,
|
|
"content_text": payload.content_text,
|
|
"content_json": json.dumps(payload.content_json) if payload.content_json is not None else None,
|
|
"media_file_key": payload.media_file_key,
|
|
"media_duration_ms": payload.media_duration_ms,
|
|
"media_mime_type": payload.media_mime_type,
|
|
"media_size_bytes": payload.media_size_bytes,
|
|
"media_transcript_text": payload.media_transcript_text,
|
|
"client_msg_id": payload.client_msg_id,
|
|
"sender_name_snapshot": sender_name_snapshot,
|
|
"sender_avatar_snapshot": sender_avatar_snapshot,
|
|
"receiver_name_snapshot": receiver_name_snapshot,
|
|
"receiver_avatar_snapshot": receiver_avatar_snapshot,
|
|
"ext_json": json.dumps(payload.ext_json) if payload.ext_json is not None else None,
|
|
}
|
|
if message_id is not None:
|
|
params["id"] = message_id
|
|
|
|
result = await self.execute(text(insert_sql), params)
|
|
if message_id is None:
|
|
message_id = inserted_primary_key(result)
|
|
|
|
await self.execute(
|
|
text(
|
|
f"""
|
|
UPDATE im_conversations
|
|
SET status = 1,
|
|
last_seq = :last_seq,
|
|
message_count = message_count + 1,
|
|
last_message_preview = :last_message_preview,
|
|
last_message_at = {now_sql},
|
|
updated_at = {now_sql}
|
|
WHERE id = :conversation_id
|
|
"""
|
|
),
|
|
{
|
|
"conversation_id": conversation_id,
|
|
"last_seq": next_seq,
|
|
"last_message_preview": preview,
|
|
},
|
|
)
|
|
await self.commit()
|
|
return conversation_id, False
|
|
except IntegrityError:
|
|
await self.db.rollback()
|
|
existing = await self._get_message_by_conversation_client_id(
|
|
conversation_id=conversation_id,
|
|
client_msg_id=payload.client_msg_id,
|
|
)
|
|
if existing:
|
|
return conversation_id, True
|
|
raise
|
|
except Exception:
|
|
await self.db.rollback()
|
|
raise
|
|
|
|
async def _get_or_create_conversation(
|
|
self,
|
|
*,
|
|
conversation_type: int,
|
|
participant_a_type: int,
|
|
participant_a_id: str,
|
|
participant_b_type: int,
|
|
participant_b_id: str,
|
|
pair_key: str,
|
|
) -> int:
|
|
row = await self._get_conversation_by_pair(
|
|
conversation_type=conversation_type,
|
|
pair_key=pair_key,
|
|
lock=False,
|
|
)
|
|
if row:
|
|
return int(row["id"])
|
|
|
|
now_sql = "CURRENT_TIMESTAMP(3)"
|
|
conversation_id = await self._next_primary_key("im_conversations")
|
|
insert_sql = f"""
|
|
INSERT INTO im_conversations (
|
|
{'id,' if conversation_id is not None else ''}
|
|
conversation_type,
|
|
participant_a_type,
|
|
participant_a_id,
|
|
participant_b_type,
|
|
participant_b_id,
|
|
pair_key,
|
|
status,
|
|
last_seq,
|
|
message_count,
|
|
created_at,
|
|
updated_at
|
|
) VALUES (
|
|
{':id,' if conversation_id is not None else ''}
|
|
:conversation_type,
|
|
:participant_a_type,
|
|
:participant_a_id,
|
|
:participant_b_type,
|
|
:participant_b_id,
|
|
:pair_key,
|
|
1,
|
|
0,
|
|
0,
|
|
{now_sql},
|
|
{now_sql}
|
|
)
|
|
"""
|
|
params = {
|
|
"conversation_type": conversation_type,
|
|
"participant_a_type": participant_a_type,
|
|
"participant_a_id": participant_a_id,
|
|
"participant_b_type": participant_b_type,
|
|
"participant_b_id": participant_b_id,
|
|
"pair_key": pair_key,
|
|
}
|
|
if conversation_id is not None:
|
|
params["id"] = conversation_id
|
|
|
|
try:
|
|
result = await self.execute(text(insert_sql), params)
|
|
if conversation_id is not None:
|
|
return conversation_id
|
|
return inserted_primary_key(result)
|
|
except IntegrityError:
|
|
await self.db.rollback()
|
|
row = await self._get_conversation_by_pair(
|
|
conversation_type=conversation_type,
|
|
pair_key=pair_key,
|
|
lock=False,
|
|
)
|
|
if row:
|
|
return int(row["id"])
|
|
raise
|
|
|
|
async def _get_conversation_by_pair(
|
|
self,
|
|
*,
|
|
conversation_type: int,
|
|
pair_key: str,
|
|
lock: bool,
|
|
) -> Mapping[str, Any] | None:
|
|
lock_clause = " FOR UPDATE" if lock else ""
|
|
result = await self.execute(
|
|
text(
|
|
f"""
|
|
SELECT id, conversation_type, last_seq, status
|
|
FROM im_conversations
|
|
WHERE conversation_type = :conversation_type
|
|
AND pair_key = :pair_key
|
|
LIMIT 1
|
|
{lock_clause}
|
|
"""
|
|
),
|
|
{"conversation_type": conversation_type, "pair_key": pair_key},
|
|
)
|
|
return result.mappings().first()
|
|
|
|
async def _get_conversation_by_id(
|
|
self,
|
|
*,
|
|
conversation_id: int,
|
|
lock: bool,
|
|
) -> Mapping[str, Any] | None:
|
|
lock_clause = " FOR UPDATE" if lock else ""
|
|
result = await self.execute(
|
|
text(
|
|
f"""
|
|
SELECT id, conversation_type, last_seq, status
|
|
FROM im_conversations
|
|
WHERE id = :conversation_id
|
|
LIMIT 1
|
|
{lock_clause}
|
|
"""
|
|
),
|
|
{"conversation_id": conversation_id},
|
|
)
|
|
return result.mappings().first()
|
|
|
|
async def _get_message_by_conversation_client_id(
|
|
self,
|
|
*,
|
|
conversation_id: int,
|
|
client_msg_id: str,
|
|
) -> Mapping[str, Any] | None:
|
|
result = await self.execute(
|
|
text(
|
|
"""
|
|
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 client_msg_id = :client_msg_id
|
|
LIMIT 1
|
|
"""
|
|
),
|
|
{"conversation_id": conversation_id, "client_msg_id": client_msg_id},
|
|
)
|
|
return result.mappings().first()
|
|
|
|
async def _next_primary_key(self, table_name: str) -> int | None:
|
|
result = await self.execute(text(f"SELECT 1"))
|
|
return None
|