banban/talkingq-url/banban/routers/im.py

import json
import logging
from collections.abc import Mapping
from typing import Any

from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status
from sqlalchemy import text

try:
    from banban.security import get_current_user_id
    from banban.schemas.im import (
        ChildConversationItem,
        ChildConversationListResponse,
        ChildConversationMessageItem,
        ChildConversationMessageListResponse,
        ConversationMessageCreateResponse,
        ParentChildMessageCreateRequest,
    )
    from banban.service.im import ImService, im_service, present_message_item
except ModuleNotFoundError:
    from banban.security import get_current_user_id
    from banban.schemas.im import (
        ChildConversationItem,
        ChildConversationListResponse,
        ChildConversationMessageItem,
        ChildConversationMessageListResponse,
        ConversationMessageCreateResponse,
        ParentChildMessageCreateRequest,
    )
    from banban.service.im import ImService, im_service, present_message_item


router = APIRouter(prefix="/children", tags=["im"])
logger = logging.getLogger("app.im")

PARENT_PARTICIPANT_TYPE = 1
CHILD_PARTICIPANT_TYPE = 2

CHILD_PEER_CONVERSATION_TYPE = 1
PARENT_CHILD_CONVERSATION_TYPE = 2
SUPPORTED_CONVERSATION_TYPES = {
    CHILD_PEER_CONVERSATION_TYPE,
    PARENT_CHILD_CONVERSATION_TYPE,
}

CONVERSATION_TYPE_NAMES = {
    CHILD_PEER_CONVERSATION_TYPE: "child_peer",
    PARENT_CHILD_CONVERSATION_TYPE: "parent_child",
}

PARTICIPANT_TYPE_NAMES = {
    PARENT_PARTICIPANT_TYPE: "parent",
    CHILD_PARTICIPANT_TYPE: "child",
}


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 _participant_type_name(participant_type: int) -> str:
    return PARTICIPANT_TYPE_NAMES.get(participant_type, f"unknown_{participant_type}")


def _conversation_type_name(conversation_type: int) -> str:
    return CONVERSATION_TYPE_NAMES.get(conversation_type, f"unknown_{conversation_type}")


def _build_in_params(prefix: str, values: list[int]) -> tuple[str, dict[str, int]]:
    placeholders: list[str] = []
    params: dict[str, int] = {}
    for index, value in enumerate(values):
        key = f"{prefix}_{index}"
        placeholders.append(f":{key}")
        params[key] = value
    return ", ".join(placeholders), params


async def _fetch_parent_names(db, user_ids: set[int]) -> dict[int, str | None]:
    if not user_ids:
        return {}

    values = sorted(user_ids)
    placeholders, params = _build_in_params("user_id", values)
    result = await db.execute(
        text(
            f"""
            SELECT user_id, nickname
            FROM parents
            WHERE status = 1
              AND user_id IN ({placeholders})
            """
        ),
        params,
    )
    return {int(row["user_id"]): row["nickname"] for row in result.mappings().all()}


async def _fetch_child_names(db, child_ids: set[int]) -> dict[int, str | None]:
    if not child_ids:
        return {}

    values = sorted(child_ids)
    placeholders, params = _build_in_params("child_id", values)
    result = await db.execute(
        text(
            f"""
            SELECT child_id, child_name
            FROM children
            WHERE status = 1
              AND child_id IN ({placeholders})
            """
        ),
        params,
    )
    return {int(row["child_id"]): row["child_name"] for row in result.mappings().all()}


async def _assert_child_access(db, child_id: int, user_id: int) -> None:
    result = await db.execute(
        text(
            """
            SELECT child_id
            FROM children
            WHERE child_id = :child_id
              AND status = 1
            LIMIT 1
            """
        ),
        {"child_id": child_id},
    )
    child_row = result.mappings().first()
    if not child_row:
        raise HTTPException(status_code=404, detail="child not found")

    result = await db.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},
    )
    has_access = result.scalar_one_or_none() is not None
    if not has_access:
        raise HTTPException(status_code=403, detail="no permission to access this child")


async def _get_conversation_for_child(
    db,
    *,
    conversation_id: int,
    child_id: int,
) -> Mapping[str, Any]:
    child_id_str = str(child_id)
    result = await db.execute(
        text(
            """
            SELECT
              id,
              conversation_type,
              participant_a_type,
              participant_a_id,
              participant_b_type,
              participant_b_id,
              status
            FROM im_conversations
            WHERE id = :conversation_id
            LIMIT 1
            """
        ),
        {"conversation_id": conversation_id},
    )
    row = result.mappings().first()
    if not row:
        raise HTTPException(status_code=404, detail="conversation not found")

    conversation_type = int(row["conversation_type"])
    if conversation_type not in SUPPORTED_CONVERSATION_TYPES or int(row["status"]) != 1:
        raise HTTPException(status_code=404, detail="conversation not found")

    is_child_participant = (
        (
            int(row["participant_a_type"]) == CHILD_PARTICIPANT_TYPE
            and row["participant_a_id"] == child_id_str
        )
        or (
            int(row["participant_b_type"]) == CHILD_PARTICIPANT_TYPE
            and row["participant_b_id"] == child_id_str
        )
    )
    if not is_child_participant:
        raise HTTPException(status_code=404, detail="conversation not found")

    return row


def _row_to_conversation_item(
    row: Mapping[str, Any],
    *,
    child_id: int,
    parent_names: Mapping[int, str | None],
    child_names: Mapping[int, str | None],
) -> ChildConversationItem:
    child_id_str = str(child_id)
    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

    peer_name: str | None = None
    if peer_type == PARENT_PARTICIPANT_TYPE and peer_id.isdigit():
        peer_name = parent_names.get(int(peer_id))
    elif peer_type == CHILD_PARTICIPANT_TYPE and peer_id.isdigit():
        peer_name = child_names.get(int(peer_id))

    return ChildConversationItem(
        conversation_id=int(row["id"]),
        conversation_type=int(row["conversation_type"]),
        conversation_type_name=_conversation_type_name(int(row["conversation_type"])),
        peer_type=_participant_type_name(peer_type),
        peer_id=peer_id,
        peer_name=peer_name,
        last_message_preview=row["last_message_preview"],
        last_message_at=row["last_message_at"],
        message_count=int(row["message_count"]),
    )


def _row_to_message_item(row: Mapping[str, Any]) -> ChildConversationMessageItem:
    return ChildConversationMessageItem(
        id=int(row["id"]),
        conversation_id=int(row["conversation_id"]),
        seq=int(row["seq"]),
        sender_type=_participant_type_name(int(row["sender_type"])),
        sender_id=str(row["sender_id"]),
        receiver_type=_participant_type_name(int(row["receiver_type"])),
        receiver_id=str(row["receiver_id"]),
        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"],
        media_mime_type=row.get("media_mime_type"),
        media_size_bytes=row.get("media_size_bytes"),
        media_transcript_text=row.get("media_transcript_text"),
        client_msg_id=row["client_msg_id"],
        sender_name_snapshot=row["sender_name_snapshot"],
        sender_avatar_snapshot=row.get("sender_avatar_snapshot"),
        receiver_name_snapshot=row["receiver_name_snapshot"],
        receiver_avatar_snapshot=row.get("receiver_avatar_snapshot"),
        ext_json=_normalize_content_json(row.get("ext_json")),
        created_at=row["created_at"],
    )


@router.get("/{child_id}/conversations", response_model=ChildConversationListResponse)
async def list_child_conversations(
    child_id: int,
    request: Request,
    conversation_type: int | None = Query(default=None),
    cursor: int | None = Query(default=None, ge=1),
    limit: int = Query(default=20, ge=1, le=100),
    current_user_id: int = Depends(get_current_user_id),
) -> ChildConversationListResponse:
    if conversation_type is not None and conversation_type not in SUPPORTED_CONVERSATION_TYPES:
        raise HTTPException(status_code=422, detail="unsupported conversation_type")

    await im_service.assert_parent_child_access(user_id=current_user_id, child_id=child_id)

    # 从 DatabaseServiceBase 获取数据库会话
    from services.database_service_base import DatabaseServiceBase
    db_service = DatabaseServiceBase(service_name="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,
            "child_peer_conversation_type": CHILD_PEER_CONVERSATION_TYPE,
            "parent_child_conversation_type": PARENT_CHILD_CONVERSATION_TYPE,
            "fetch_limit": limit + 1,
        }
        where = """
            status = 1
            AND conversation_type IN (:child_peer_conversation_type, :parent_child_conversation_type)
            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(
            "child conversations listed",
            extra={
                "event": "child_conversation_list",
                "request_id": getattr(request.state, "request_id", None),
                "user_id": current_user_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.post(
    "/{child_id}/messages",
    response_model=ConversationMessageCreateResponse,
    status_code=status.HTTP_201_CREATED,
)
async def create_child_message_for_parent(
    child_id: int,
    payload: ParentChildMessageCreateRequest,
    request: Request,
    response: Response,
    current_user_id: int = Depends(get_current_user_id),
) -> ConversationMessageCreateResponse:
    result = await im_service.create_parent_child_message(
        parent_user_id=current_user_id,
        child_id=child_id,
        payload=payload,
    )
    if result.idempotent:
        response.status_code = status.HTTP_200_OK

    logger.info(
        "parent child message created",
        extra={
            "event": "parent_child_message_create",
            "request_id": getattr(request.state, "request_id", None),
            "user_id": current_user_id,
            "child_id": child_id,
            "conversation_id": result.conversation_id,
            "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,
    )


@router.get(
    "/{child_id}/conversations/{conversation_id}/messages",
    response_model=ChildConversationMessageListResponse,
)
async def list_child_conversation_messages(
    child_id: int,
    conversation_id: int,
    request: Request,
    cursor_seq: int | None = Query(default=None, ge=1),
    limit: int = Query(default=20, ge=1, le=100),
    current_user_id: int = Depends(get_current_user_id),
) -> ChildConversationMessageListResponse:
    await im_service.assert_parent_child_access(user_id=current_user_id, child_id=child_id)
    
    # 从 DatabaseServiceBase 获取数据库会话
    from services.database_service_base import DatabaseServiceBase
    db_service = DatabaseServiceBase(service_name="im_router")
    db_session = await db_service.get_session()
    try:
        await _get_conversation_for_child(db=db_session, conversation_id=conversation_id, child_id=child_id)

        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_message_item(row, audio_storage=im_service.audio_storage) for row in rows]
        next_cursor_seq = items[0].seq if has_more and items else None

        logger.info(
            "child conversation messages listed",
            extra={
                "event": "child_conversation_message_list",
                "request_id": getattr(request.state, "request_id", None),
                "user_id": current_user_id,
                "child_id": 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()