358 lines
11 KiB
Python
358 lines
11 KiB
Python
import json
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
|
|
from sqlalchemy import text
|
|
from sqlalchemy.orm import Session
|
|
|
|
try:
|
|
# For module mode: `uvicorn app.main:app`
|
|
from app.db import get_db
|
|
from app.schemas.message import (
|
|
MessageCreateRequest,
|
|
MessageCreateResponse,
|
|
MessageItem,
|
|
MessageListResponse,
|
|
)
|
|
except ModuleNotFoundError:
|
|
# For script mode: `python app/main.py` or VS Code "Run Python File"
|
|
from db import get_db
|
|
from schemas.message import (
|
|
MessageCreateRequest,
|
|
MessageCreateResponse,
|
|
MessageItem,
|
|
MessageListResponse,
|
|
)
|
|
|
|
|
|
router = APIRouter(prefix="/messages", tags=["messages"])
|
|
|
|
|
|
def _build_preview(content_type: int, content_text: str | None) -> str:
|
|
if content_type == 1:
|
|
preview = f"user: {(content_text or '').strip()}"
|
|
elif content_type == 2:
|
|
preview = "user: [audio]"
|
|
elif content_type == 3:
|
|
preview = "user: [image]"
|
|
else:
|
|
preview = "user: [json]"
|
|
return preview[:255]
|
|
|
|
|
|
def _normalize_content_json(value: Any) -> dict[str, Any] | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, dict):
|
|
return value
|
|
if isinstance(value, str):
|
|
try:
|
|
parsed = json.loads(value)
|
|
if isinstance(parsed, dict):
|
|
return parsed
|
|
except json.JSONDecodeError:
|
|
return None
|
|
return None
|
|
|
|
|
|
def _row_to_message_item(row: Mapping[str, Any]) -> MessageItem:
|
|
return MessageItem(
|
|
id=int(row["id"]),
|
|
conversation_id=int(row["conversation_id"]),
|
|
seq=int(row["seq"]),
|
|
sender_user_id=row["sender_user_id"],
|
|
role=int(row["role"]),
|
|
content_type=int(row["content_type"]),
|
|
content_text=row["content_text"],
|
|
content_json=_normalize_content_json(row["content_json"]),
|
|
media_file_key=row["media_file_key"],
|
|
media_duration_ms=row["media_duration_ms"],
|
|
client_msg_id=row["client_msg_id"],
|
|
created_at=row["created_at"],
|
|
)
|
|
|
|
|
|
def _get_existing_message(
|
|
db: Session, conversation_id: int, client_msg_id: str
|
|
) -> Mapping[str, Any] | None:
|
|
return (
|
|
db.execute(
|
|
text(
|
|
"""
|
|
SELECT
|
|
id,
|
|
conversation_id,
|
|
seq,
|
|
sender_user_id,
|
|
role,
|
|
content_type,
|
|
content_text,
|
|
content_json,
|
|
media_file_key,
|
|
media_duration_ms,
|
|
client_msg_id,
|
|
created_at
|
|
FROM chat_message
|
|
WHERE conversation_id = :conversation_id
|
|
AND client_msg_id = :client_msg_id
|
|
LIMIT 1
|
|
"""
|
|
),
|
|
{"conversation_id": conversation_id, "client_msg_id": client_msg_id},
|
|
)
|
|
.mappings()
|
|
.first()
|
|
)
|
|
|
|
|
|
@router.post("", response_model=MessageCreateResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_message(
|
|
payload: MessageCreateRequest,
|
|
response: Response,
|
|
db: Session = Depends(get_db),
|
|
) -> MessageCreateResponse:
|
|
sender_user_id = payload.sender_user_id
|
|
peer_user_id = payload.peer_user_id
|
|
user_low_id = min(sender_user_id, peer_user_id)
|
|
user_high_id = max(sender_user_id, peer_user_id)
|
|
|
|
with db.begin():
|
|
user_count = db.execute(
|
|
text(
|
|
"""
|
|
SELECT COUNT(*) AS cnt
|
|
FROM chat_user
|
|
WHERE id IN (:sender_user_id, :peer_user_id)
|
|
AND status = 1
|
|
"""
|
|
),
|
|
{"sender_user_id": sender_user_id, "peer_user_id": peer_user_id},
|
|
).scalar_one()
|
|
|
|
if int(user_count) != 2:
|
|
raise HTTPException(status_code=404, detail="sender or peer user not found")
|
|
|
|
db.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO chat_conversation (
|
|
user_low_id,
|
|
user_high_id,
|
|
status,
|
|
last_seq,
|
|
message_count,
|
|
created_at,
|
|
updated_at
|
|
)
|
|
VALUES (
|
|
:user_low_id,
|
|
:user_high_id,
|
|
1,
|
|
0,
|
|
0,
|
|
CURRENT_TIMESTAMP(3),
|
|
CURRENT_TIMESTAMP(3)
|
|
)
|
|
ON DUPLICATE KEY UPDATE
|
|
id = LAST_INSERT_ID(id),
|
|
updated_at = CURRENT_TIMESTAMP(3)
|
|
"""
|
|
),
|
|
{"user_low_id": user_low_id, "user_high_id": user_high_id},
|
|
)
|
|
|
|
conversation_id = int(db.execute(text("SELECT LAST_INSERT_ID()")).scalar_one())
|
|
|
|
conversation = (
|
|
db.execute(
|
|
text(
|
|
"""
|
|
SELECT id, last_seq, status
|
|
FROM chat_conversation
|
|
WHERE id = :conversation_id
|
|
FOR UPDATE
|
|
"""
|
|
),
|
|
{"conversation_id": conversation_id},
|
|
)
|
|
.mappings()
|
|
.first()
|
|
)
|
|
if not conversation:
|
|
raise HTTPException(status_code=500, detail="failed to load conversation")
|
|
|
|
if int(conversation["status"]) != 1:
|
|
raise HTTPException(status_code=409, detail="conversation is not active")
|
|
|
|
existing = _get_existing_message(db, conversation_id, payload.client_msg_id)
|
|
if existing:
|
|
response.status_code = status.HTTP_200_OK
|
|
return MessageCreateResponse(
|
|
idempotent=True,
|
|
message=_row_to_message_item(existing),
|
|
)
|
|
|
|
next_seq = int(conversation["last_seq"]) + 1
|
|
preview = _build_preview(payload.content_type, payload.content_text)
|
|
|
|
insert_result = db.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO chat_message (
|
|
conversation_id,
|
|
seq,
|
|
sender_user_id,
|
|
role,
|
|
content_type,
|
|
content_text,
|
|
content_json,
|
|
media_file_key,
|
|
media_duration_ms,
|
|
client_msg_id,
|
|
created_at
|
|
)
|
|
VALUES (
|
|
:conversation_id,
|
|
:seq,
|
|
:sender_user_id,
|
|
1,
|
|
:content_type,
|
|
:content_text,
|
|
:content_json,
|
|
:media_file_key,
|
|
:media_duration_ms,
|
|
:client_msg_id,
|
|
CURRENT_TIMESTAMP(3)
|
|
)
|
|
"""
|
|
),
|
|
{
|
|
"conversation_id": conversation_id,
|
|
"seq": next_seq,
|
|
"sender_user_id": sender_user_id,
|
|
"content_type": payload.content_type,
|
|
"content_text": payload.content_text,
|
|
"content_json": json.dumps(payload.content_json, ensure_ascii=False)
|
|
if payload.content_json is not None
|
|
else None,
|
|
"media_file_key": payload.media_file_key,
|
|
"media_duration_ms": payload.media_duration_ms,
|
|
"client_msg_id": payload.client_msg_id,
|
|
},
|
|
)
|
|
|
|
db.execute(
|
|
text(
|
|
"""
|
|
UPDATE chat_conversation
|
|
SET
|
|
last_seq = :last_seq,
|
|
message_count = message_count + 1,
|
|
last_message_preview = :last_message_preview,
|
|
last_message_at = CURRENT_TIMESTAMP(3),
|
|
updated_at = CURRENT_TIMESTAMP(3)
|
|
WHERE id = :conversation_id
|
|
"""
|
|
),
|
|
{
|
|
"conversation_id": conversation_id,
|
|
"last_seq": next_seq,
|
|
"last_message_preview": preview,
|
|
},
|
|
)
|
|
|
|
created = (
|
|
db.execute(
|
|
text(
|
|
"""
|
|
SELECT
|
|
id,
|
|
conversation_id,
|
|
seq,
|
|
sender_user_id,
|
|
role,
|
|
content_type,
|
|
content_text,
|
|
content_json,
|
|
media_file_key,
|
|
media_duration_ms,
|
|
client_msg_id,
|
|
created_at
|
|
FROM chat_message
|
|
WHERE id = :message_id
|
|
LIMIT 1
|
|
"""
|
|
),
|
|
{"message_id": insert_result.lastrowid},
|
|
)
|
|
.mappings()
|
|
.first()
|
|
)
|
|
if not created:
|
|
raise HTTPException(status_code=500, detail="failed to load created message")
|
|
|
|
return MessageCreateResponse(
|
|
idempotent=False,
|
|
message=_row_to_message_item(created),
|
|
)
|
|
|
|
|
|
@router.get("", response_model=MessageListResponse)
|
|
def list_messages(
|
|
conversation_id: int = Query(gt=0),
|
|
cursor_seq: int | None = Query(default=None, ge=1),
|
|
limit: int = Query(default=20, ge=1, le=100),
|
|
db: Session = Depends(get_db),
|
|
) -> MessageListResponse:
|
|
conversation_exists = (
|
|
db.execute(
|
|
text("SELECT 1 FROM chat_conversation WHERE id = :conversation_id LIMIT 1"),
|
|
{"conversation_id": conversation_id},
|
|
).first()
|
|
is not None
|
|
)
|
|
if not conversation_exists:
|
|
raise HTTPException(status_code=404, detail="conversation not found")
|
|
|
|
sql = """
|
|
SELECT
|
|
id,
|
|
conversation_id,
|
|
seq,
|
|
sender_user_id,
|
|
role,
|
|
content_type,
|
|
content_text,
|
|
content_json,
|
|
media_file_key,
|
|
media_duration_ms,
|
|
client_msg_id,
|
|
created_at
|
|
FROM chat_message
|
|
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"
|
|
|
|
rows = db.execute(text(sql), params).mappings().all()
|
|
has_more = len(rows) > limit
|
|
rows = rows[:limit]
|
|
rows.reverse()
|
|
items = [_row_to_message_item(row) for row in rows]
|
|
|
|
next_cursor_seq = items[0].seq if has_more and items else None
|
|
return MessageListResponse(
|
|
conversation_id=conversation_id,
|
|
has_more=has_more,
|
|
next_cursor_seq=next_cursor_seq,
|
|
items=items,
|
|
)
|