From 7371848a6bf78ef1a87c0aa4dbfe7c03f40698e1 Mon Sep 17 00:00:00 2001 From: ChengCan <783785929@qq.com> Date: Thu, 26 Mar 2026 10:34:18 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E6=B6=88=E6=81=AF=E7=9A=84?= =?UTF-8?q?=E5=82=A8=E5=AD=98=E5=92=8C=E6=9F=A5=E7=9C=8B=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- mini-program/app/main.py | 3 + mini-program/app/routers/messages.py | 357 +++++++++++++++++++++++++++ mini-program/app/schemas/__init__.py | 1 + mini-program/app/schemas/message.py | 61 +++++ 4 files changed, 422 insertions(+) create mode 100644 mini-program/app/routers/messages.py create mode 100644 mini-program/app/schemas/__init__.py create mode 100644 mini-program/app/schemas/message.py diff --git a/mini-program/app/main.py b/mini-program/app/main.py index aa3a223..eeb064d 100644 --- a/mini-program/app/main.py +++ b/mini-program/app/main.py @@ -6,11 +6,13 @@ try: # For module mode: `uvicorn app.main:app` from app.db import check_db_connection, close_db_engine from app.routers.health import router as health_router + from app.routers.messages import router as messages_router from app.settings import settings except ModuleNotFoundError: # For script mode: `python app/main.py` or VS Code "Run Python File" from db import check_db_connection, close_db_engine from routers.health import router as health_router + from routers.messages import router as messages_router from settings import settings @@ -30,6 +32,7 @@ def create_app() -> FastAPI: close_db_engine() app.include_router(health_router) + app.include_router(messages_router) return app diff --git a/mini-program/app/routers/messages.py b/mini-program/app/routers/messages.py new file mode 100644 index 0000000..89a1ae8 --- /dev/null +++ b/mini-program/app/routers/messages.py @@ -0,0 +1,357 @@ +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, + ) diff --git a/mini-program/app/schemas/__init__.py b/mini-program/app/schemas/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/mini-program/app/schemas/__init__.py @@ -0,0 +1 @@ + diff --git a/mini-program/app/schemas/message.py b/mini-program/app/schemas/message.py new file mode 100644 index 0000000..8d7799c --- /dev/null +++ b/mini-program/app/schemas/message.py @@ -0,0 +1,61 @@ +from datetime import datetime +from typing import Any + +from pydantic import BaseModel, Field, model_validator + + +class MessageCreateRequest(BaseModel): + sender_user_id: int = Field(gt=0) + peer_user_id: int = Field(gt=0) + client_msg_id: str = Field(min_length=1, max_length=64) + content_type: int = Field(description="1-text 2-audio 3-image 4-json") + content_text: str | None = Field(default=None) + content_json: dict[str, Any] | None = None + media_file_key: str | None = Field(default=None, max_length=255) + media_duration_ms: int | None = Field(default=None, ge=0) + + @model_validator(mode="after") + def validate_message_fields(self) -> "MessageCreateRequest": + if self.sender_user_id == self.peer_user_id: + raise ValueError("sender_user_id and peer_user_id cannot be the same") + + if self.content_type not in {1, 2, 3, 4}: + raise ValueError("content_type must be one of 1, 2, 3, 4") + + if self.content_type == 1 and not (self.content_text and self.content_text.strip()): + raise ValueError("content_text is required when content_type is 1") + + if self.content_type in {2, 3} and not self.media_file_key: + raise ValueError("media_file_key is required when content_type is 2 or 3") + + if self.content_type == 4 and self.content_json is None: + raise ValueError("content_json is required when content_type is 4") + + return self + + +class MessageItem(BaseModel): + id: int + conversation_id: int + seq: int + sender_user_id: int | None + role: int + content_type: int + content_text: str | None + content_json: dict[str, Any] | None + media_file_key: str | None + media_duration_ms: int | None + client_msg_id: str | None + created_at: datetime + + +class MessageCreateResponse(BaseModel): + idempotent: bool + message: MessageItem + + +class MessageListResponse(BaseModel): + conversation_id: int + has_more: bool + next_cursor_seq: int | None + items: list[MessageItem]