add code merge from test-clean

talkingq-url/banban/dao/__init__.py

@@ -7,7 +7,8 @@ class BaseDAO:
         self.db = db
 
     async def execute(self, query, params: dict = None):
-        return await self.db.execute(text(query), params or {})
+        statement = query if hasattr(query, "_execute_on_connection") else text(query)
+        return await self.db.execute(statement, params or {})
 
     async def commit(self):
         await self.db.commit()

talkingq-url/banban/dao/binding.py

@@ -4,13 +4,21 @@ from collections.abc import Mapping
 from datetime import datetime, timedelta
 from typing import Optional
 
-from sqlalchemy import text
-from sqlalchemy.exc import IntegrityError
-
 from banban.dao import BaseDAO
 
 logger = logging.getLogger("banban.dao.binding")
 
+SESSION_STATUS_PENDING = 1
+SESSION_STATUS_COMPLETED = 2
+SESSION_STATUS_EXPIRED = 3
+SESSION_STATUS_FAILED = 4
+SESSION_STATUS_CANCELLED = 5
+
+BIND_SOURCE_SESSION_CONFIRM = 1
+BIND_SOURCE_DIRECT = 2
+BIND_SOURCE_SET_CHILD = 3
+BIND_SOURCE_NFC = 4
+
 
 class BindingDAO(BaseDAO):
     async def get_device_auth(self, device_id: str) -> Optional[Mapping]:
@@ -55,36 +63,13 @@ class BindingDAO(BaseDAO):
         )
 
     async def _upsert_parent_child_relation(self, user_id: int, child_id: int) -> None:
-        updated = await self.execute(
-            """
-                UPDATE parent_child_relations
-                SET status = 1,
-                    updated_at = CURRENT_TIMESTAMP
-                WHERE user_id = :user_id
-                  AND child_id = :child_id
-                """,
-            {"user_id": user_id, "child_id": child_id},
-        )
-        if updated.rowcount and updated.rowcount > 0:
-            return
-
-        try:
         await self.execute(
             """
                 INSERT INTO parent_child_relations (user_id, child_id, relation_type, is_primary, status)
                 VALUES (:user_id, :child_id, 9, 0, 1)
-                    """,
+                ON DUPLICATE KEY UPDATE
-                {"user_id": user_id, "child_id": child_id},
+                    status = VALUES(status),
-            )
-        except IntegrityError:
-            await self.db.rollback()
-            await self.execute(
-                """
-                    UPDATE parent_child_relations
-                    SET status = 1,
                     updated_at = CURRENT_TIMESTAMP
-                    WHERE user_id = :user_id
-                      AND child_id = :child_id
             """,
             {"user_id": user_id, "child_id": child_id},
         )
@@ -172,14 +157,30 @@ class BindingDAO(BaseDAO):
             {"device_id": device_id, "owner_user_id": user_id, "child_id": child_id},
         )
 
-    async def start_bind(self, user_id: int, device_id: str, child_id: Optional[int]) -> str:
+    async def start_bind(self, user_id: int, device_id: str, child_id: Optional[int]) -> tuple[str, datetime]:
         bind_token = str(uuid.uuid4())
         expires_at = datetime.utcnow() + timedelta(minutes=10)
 
+        await self.execute(
+            """
+                UPDATE device_bind_sessions
+                SET status = :cancelled_status,
+                    consumed_at = COALESCE(consumed_at, CURRENT_TIMESTAMP),
+                    updated_at = CURRENT_TIMESTAMP
+                WHERE device_id = :device_id
+                  AND status = :pending_status
+            """,
+            {
+                "device_id": device_id,
+                "pending_status": SESSION_STATUS_PENDING,
+                "cancelled_status": SESSION_STATUS_CANCELLED,
+            },
+        )
+
         await self.execute(
             """
                 INSERT INTO device_bind_sessions (bind_token, device_id, initiator_user_id, target_child_id, expires_at, status)
-                VALUES (:bind_token, :device_id, :initiator_user_id, :target_child_id, :expires_at, 1)
+                VALUES (:bind_token, :device_id, :initiator_user_id, :target_child_id, :expires_at, :status)
             """,
             {
                 "bind_token": bind_token,
@@ -187,38 +188,125 @@ class BindingDAO(BaseDAO):
                 "initiator_user_id": user_id,
                 "target_child_id": child_id,
                 "expires_at": expires_at,
+                "status": SESSION_STATUS_PENDING,
             },
         )
-        await self.commit()
+        return bind_token, expires_at
-        return bind_token
 
     async def get_session(self, bind_token: str, user_id: int) -> Optional[Mapping]:
         return (
             await self.execute(
-                "SELECT * FROM device_bind_sessions WHERE bind_token = :bind_token AND initiator_user_id = :user_id",
+                """
-                {"bind_token": bind_token, "user_id": user_id},
+                    SELECT
+                        s.*,
+                        CASE
+                            WHEN s.status = :completed_status
+                             AND s.confirmed_at IS NOT NULL
+                             AND c.updated_at >= s.confirmed_at
+                            THEN c.card_uuid
+                            ELSE NULL
+                        END AS card_uuid
+                    FROM device_bind_sessions AS s
+                    LEFT JOIN cards AS c
+                      ON c.device_id = s.device_id
+                    WHERE s.bind_token = :bind_token
+                      AND s.initiator_user_id = :user_id
+                    LIMIT 1
+                """,
+                {
+                    "bind_token": bind_token,
+                    "user_id": user_id,
+                    "completed_status": SESSION_STATUS_COMPLETED,
+                },
             )
         ).mappings().first()
 
+    async def get_latest_pending_session_by_device(self, device_id: str) -> Optional[Mapping]:
+        return (
+            await self.execute(
+                """
+                    SELECT *
+                    FROM device_bind_sessions
+                    WHERE device_id = :device_id
+                      AND status = :status
+                    ORDER BY id DESC
+                    LIMIT 1
+                """,
+                {"device_id": device_id, "status": SESSION_STATUS_PENDING},
+            )
+        ).mappings().first()
+
+    async def mark_session_status(self, session_id: int, status: int) -> None:
+        await self.execute(
+            """
+                UPDATE device_bind_sessions
+                SET status = :status,
+                    consumed_at = COALESCE(consumed_at, CURRENT_TIMESTAMP),
+                    updated_at = CURRENT_TIMESTAMP
+                WHERE id = :id
+            """,
+            {"id": session_id, "status": status},
+        )
+
     async def confirm_bind(self, session_id: int, device_id: str, child_id: Optional[int], user_id: int) -> None:
         if child_id is not None:
             await self._upsert_parent_child_relation(user_id=user_id, child_id=child_id)
 
         await self.execute(
-            "UPDATE device_bind_sessions SET status = 2, confirmed_at = CURRENT_TIMESTAMP WHERE id = :id",
+            """
-            {"id": session_id},
+                UPDATE device_bind_sessions
+                SET status = :status,
+                    confirmed_at = CURRENT_TIMESTAMP,
+                    consumed_at = CURRENT_TIMESTAMP,
+                    updated_at = CURRENT_TIMESTAMP
+                WHERE id = :id
+            """,
+            {"id": session_id, "status": SESSION_STATUS_COMPLETED},
         )
 
         await self._bind_device(device_id=device_id, user_id=user_id, child_id=child_id)
-        await self._insert_bind_history(device_id=device_id, child_id=child_id, user_id=user_id, bind_source=1)
+        await self._insert_bind_history(
-        await self.commit()
+            device_id=device_id,
+            child_id=child_id,
+            user_id=user_id,
+            bind_source=BIND_SOURCE_SESSION_CONFIRM,
+        )
+
+    async def complete_nfc_bind(self, session_id: int, device_id: str, child_id: Optional[int], user_id: int) -> None:
+        if child_id is not None:
+            await self._upsert_parent_child_relation(user_id=user_id, child_id=child_id)
+
+        await self.execute(
+            """
+                UPDATE device_bind_sessions
+                SET status = :status,
+                    confirmed_at = CURRENT_TIMESTAMP,
+                    consumed_at = CURRENT_TIMESTAMP,
+                    updated_at = CURRENT_TIMESTAMP
+                WHERE id = :id
+            """,
+            {"id": session_id, "status": SESSION_STATUS_COMPLETED},
+        )
+
+        await self._bind_device(device_id=device_id, user_id=user_id, child_id=child_id)
+        await self._insert_bind_history(
+            device_id=device_id,
+            child_id=child_id,
+            user_id=user_id,
+            bind_source=BIND_SOURCE_NFC,
+        )
 
     async def direct_bind(self, device_id: str, child_id: Optional[int], user_id: int) -> None:
         if child_id is not None:
             await self._upsert_parent_child_relation(user_id=user_id, child_id=child_id)
 
         await self._bind_device(device_id=device_id, user_id=user_id, child_id=child_id)
-        await self._insert_bind_history(device_id=device_id, child_id=child_id, user_id=user_id, bind_source=2)
+        await self._insert_bind_history(
+            device_id=device_id,
+            child_id=child_id,
+            user_id=user_id,
+            bind_source=BIND_SOURCE_DIRECT,
+        )
         await self.commit()
 
     async def get_current_by_user(self, user_id: int) -> Optional[Mapping]:
@@ -287,7 +375,12 @@ class BindingDAO(BaseDAO):
 
         await self._upsert_parent_child_relation(user_id=user_id, child_id=child_id)
         await self._bind_device(device_id=device_id, user_id=user_id, child_id=child_id)
-        await self._insert_bind_history(device_id=device_id, child_id=child_id, user_id=user_id, bind_source=3)
+        await self._insert_bind_history(
+            device_id=device_id,
+            child_id=child_id,
+            user_id=user_id,
+            bind_source=BIND_SOURCE_SET_CHILD,
+        )
         await self.commit()
         return True
 
@@ -302,7 +395,30 @@ class BindingDAO(BaseDAO):
         )
 
         await self.execute(
-            "INSERT INTO device_bind_history (device_id, child_id, bound_by_user_id, unbound_by_user_id, bind_source, bound_at, unbound_at, unbind_reason) SELECT device_id, child_id, bound_by_user_id, :user_id, bind_source, bound_at, CURRENT_TIMESTAMP, 'user_unbind' FROM device_bind_history WHERE device_id = :device_id AND unbound_at IS NULL",
+            """
+                INSERT INTO device_bind_history (
+                    device_id,
+                    child_id,
+                    bound_by_user_id,
+                    unbound_by_user_id,
+                    bind_source,
+                    bound_at,
+                    unbound_at,
+                    unbind_reason
+                )
+                SELECT
+                    device_id,
+                    child_id,
+                    bound_by_user_id,
+                    :user_id,
+                    bind_source,
+                    bound_at,
+                    CURRENT_TIMESTAMP,
+                    'user_unbind'
+                FROM device_bind_history
+                WHERE device_id = :device_id
+                  AND unbound_at IS NULL
+            """,
             {"device_id": device_id, "user_id": user_id},
         )
         await self.commit()
@@ -318,7 +434,8 @@ class BindingDAO(BaseDAO):
         rows = (
             await self.execute(
                 f"""
-                    SELECT * FROM device_bind_history
+                    SELECT *
+                    FROM device_bind_history
                     WHERE {where}
                     ORDER BY bound_at DESC
                     LIMIT :limit

talkingq-url/banban/dao/child.py

@@ -24,6 +24,7 @@ class ChildDAO(BaseDAO):
         child_name: str,
         child_gender: int = 2,
         child_birthday: Optional[date] = None,
+        auto_commit: bool = True,
     ) -> int:
         result = await self.execute(
             """
@@ -34,6 +35,7 @@ class ChildDAO(BaseDAO):
         )
         child_id = inserted_primary_key(result)
         await self._create_relation(user_id, child_id)
+        if auto_commit:
             await self.commit()
         return child_id
 

talkingq-url/banban/dao/im.py

@@ -25,6 +25,14 @@ class ConversationMessageCreateResult:
     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]:
@@ -179,6 +187,22 @@ class ImDAO(BaseDAO):
             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,
         *, 
@@ -428,8 +452,8 @@ class ImDAO(BaseDAO):
                 FROM im_conversations
                 WHERE conversation_type = :conversation_type
                   AND pair_key = :pair_key
-                {lock_clause}
                 LIMIT 1
+                {lock_clause}
                 """
             ),
             {"conversation_type": conversation_type, "pair_key": pair_key},
@@ -449,8 +473,8 @@ class ImDAO(BaseDAO):
                 SELECT id, conversation_type, last_seq, status
                 FROM im_conversations
                 WHERE id = :conversation_id
-                {lock_clause}
                 LIMIT 1
+                {lock_clause}
                 """
             ),
             {"conversation_id": conversation_id},

talkingq-url/banban/routers/bindings.py

@@ -4,12 +4,15 @@ from datetime import datetime
 from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
 from pydantic import BaseModel
 
-try:
+from banban.dao.binding import (
-    from banban.security import get_current_user_id
+    SESSION_STATUS_CANCELLED,
-    from banban.service.binding import BindingError, BindingService
+    SESSION_STATUS_COMPLETED,
-except ModuleNotFoundError:
+    SESSION_STATUS_EXPIRED,
-    from banban.security import get_current_user_id
+    SESSION_STATUS_FAILED,
-    from banban.service.binding import BindingError, BindingService
+    SESSION_STATUS_PENDING,
+)
+from banban.security import get_current_user_id
+from banban.service.binding import BindingError, BindingService
 
 
 router = APIRouter(prefix="/bindings", tags=["bindings"])
@@ -25,6 +28,16 @@ class BindStartRequest(BaseModel):
 class BindStartResponse(BaseModel):
     bind_token: str
     expires_at: str
+    status: int
+
+
+class BindSessionResponse(BaseModel):
+    bind_token: str
+    device_id: str
+    child_id: int | None = None
+    status: int
+    expires_at: str
+    card_uuid: str | None = None
 
 
 class BindConfirmRequest(BaseModel):
@@ -77,6 +90,7 @@ async def start_bind(
     request: Request,
     current_user_id: int = Depends(get_current_user_id),
 ) -> BindStartResponse:
+    del request
     service = BindingService()
     try:
         bind_token, expires_at = await service.start_bind(
@@ -85,9 +99,35 @@ async def start_bind(
             payload.serial_number,
             payload.child_id,
         )
-    except BindingError as e:
+    except BindingError as exc:
-        raise HTTPException(status_code=e.status_code, detail=str(e))
+        raise HTTPException(status_code=exc.status_code, detail=str(exc))
-    return BindStartResponse(bind_token=bind_token, expires_at=expires_at.isoformat())
+    return BindStartResponse(
+        bind_token=bind_token,
+        expires_at=expires_at.isoformat(),
+        status=SESSION_STATUS_PENDING,
+    )
+
+
+@router.get("/sessions/{bind_token}", response_model=BindSessionResponse)
+async def get_bind_session(
+    bind_token: str,
+    request: Request,
+    current_user_id: int = Depends(get_current_user_id),
+) -> BindSessionResponse:
+    del request
+    service = BindingService()
+    session = await service.get_bind_session(bind_token, current_user_id)
+    if not session:
+        raise HTTPException(status_code=404, detail="bind session not found")
+
+    return BindSessionResponse(
+        bind_token=session["bind_token"],
+        device_id=session["device_id"],
+        child_id=session["target_child_id"],
+        status=int(session["status"]),
+        expires_at=session["expires_at"].isoformat(),
+        card_uuid=session.get("card_uuid"),
+    )
 
 
 @router.post("/confirm", response_model=BindConfirmResponse)
@@ -96,13 +136,14 @@ async def confirm_bind(
     request: Request,
     current_user_id: int = Depends(get_current_user_id),
 ) -> BindConfirmResponse:
+    del request
     service = BindingService()
     try:
         result = await service.confirm_bind(payload.bind_token, current_user_id)
-    except BindingError as e:
+    except BindingError as exc:
-        raise HTTPException(status_code=e.status_code, detail=str(e))
+        raise HTTPException(status_code=exc.status_code, detail=str(exc))
-    except ValueError as e:
+    except ValueError as exc:
-        raise HTTPException(status_code=400, detail=str(e))
+        raise HTTPException(status_code=400, detail=str(exc))
     return BindConfirmResponse(**result)
 
 
@@ -127,6 +168,7 @@ async def direct_bind(
     request: Request,
     current_user_id: int = Depends(get_current_user_id),
 ) -> DirectBindResponse:
+    del request
     service = BindingService()
     try:
         result = await service.direct_bind(
@@ -135,8 +177,8 @@ async def direct_bind(
             payload.child_id,
             current_user_id,
         )
-    except BindingError as e:
+    except BindingError as exc:
-        raise HTTPException(status_code=e.status_code, detail=str(e))
+        raise HTTPException(status_code=exc.status_code, detail=str(exc))
     return DirectBindResponse(**result)
 
 
@@ -147,13 +189,14 @@ async def set_binding_child(
     request: Request,
     current_user_id: int = Depends(get_current_user_id),
 ) -> DirectBindResponse:
+    del request
     service = BindingService()
     try:
         result = await service.set_binding_child(device_id=device_id, child_id=payload.child_id, user_id=current_user_id)
-    except BindingError as e:
+    except BindingError as exc:
-        raise HTTPException(status_code=e.status_code, detail=str(e))
+        raise HTTPException(status_code=exc.status_code, detail=str(exc))
-    except ValueError as e:
+    except ValueError as exc:
-        raise HTTPException(status_code=404, detail=str(e))
+        raise HTTPException(status_code=404, detail=str(exc))
     return DirectBindResponse(**result)
 
 
@@ -162,6 +205,7 @@ async def get_current_binding(
     request: Request,
     current_user_id: int = Depends(get_current_user_id),
 ):
+    del request
     service = BindingService()
     binding = await service.get_current_binding(current_user_id)
     if not binding:
@@ -176,6 +220,7 @@ async def list_bindings(
     limit: int = Query(default=20, ge=1, le=100),
     current_user_id: int = Depends(get_current_user_id),
 ) -> BindingListResponse:
+    del request
     service = BindingService()
     rows, has_more = await service.list_bindings(current_user_id, limit, cursor)
     next_cursor = int(rows[-1]["id"]) if has_more and rows else None
@@ -198,6 +243,7 @@ async def get_binding(
     request: Request,
     current_user_id: int = Depends(get_current_user_id),
 ) -> BindingGetResponse:
+    del request
     service = BindingService()
     binding = await service.get_binding(device_id, current_user_id)
     if not binding:
@@ -211,6 +257,7 @@ async def unbind_device(
     request: Request,
     current_user_id: int = Depends(get_current_user_id),
 ) -> None:
+    del request
     service = BindingService()
     if not await service.unbind(device_id, current_user_id):
         raise HTTPException(status_code=404, detail="binding not found")
@@ -224,6 +271,7 @@ async def get_bind_history(
     limit: int = 20,
     current_user_id: int = Depends(get_current_user_id),
 ) -> BindHistoryResponse:
+    del request, current_user_id
     cursor_dt = datetime.fromisoformat(cursor) if cursor else None
     service = BindingService()
     rows, has_more = await service.list_history(device_id, limit, cursor_dt)

talkingq-url/banban/routers/im.py

@@ -16,7 +16,7 @@ try:
         ConversationMessageCreateResponse,
         ParentChildMessageCreateRequest,
     )
-    from banban.service.im import ImService, im_service
+    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 (
@@ -27,7 +27,7 @@ except ModuleNotFoundError:
         ConversationMessageCreateResponse,
         ParentChildMessageCreateRequest,
     )
-    from banban.service.im import ImService, im_service
+    from banban.service.im import ImService, im_service, present_message_item
 
 
 router = APIRouter(prefix="/children", tags=["im"])
@@ -506,7 +506,7 @@ async def list_child_conversation_messages(
         rows = rows[:limit]
         rows = list(rows)
         rows.reverse()
-        items = [_row_to_message_item(row) for row in rows]
+        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(

talkingq-url/banban/service/binding.py

@@ -2,7 +2,15 @@ from collections.abc import Mapping
 from datetime import datetime
 from typing import Optional
 
-from banban.dao.binding import BindingDAO
+from banban.dao.binding import (
+    SESSION_STATUS_CANCELLED,
+    SESSION_STATUS_COMPLETED,
+    SESSION_STATUS_EXPIRED,
+    SESSION_STATUS_FAILED,
+    SESSION_STATUS_PENDING,
+    BindingDAO,
+)
+from services.card_service import card_service
 from services.database_service_base import DatabaseServiceBase
 
 
@@ -26,6 +34,12 @@ class BindingService(DatabaseServiceBase):
         if int(row["is_active"]) != 1:
             raise BindingError("device is inactive", status_code=400)
 
+    def _normalize_session_status(self, session: Mapping) -> int:
+        status = int(session["status"])
+        if status == SESSION_STATUS_PENDING and datetime.utcnow() > session["expires_at"]:
+            return SESSION_STATUS_EXPIRED
+        return status
+
     async def start_bind(
         self,
         user_id: int,
@@ -37,9 +51,15 @@ class BindingService(DatabaseServiceBase):
         try:
             await self._ensure_bindable_device(db_session, device_id, serial_number)
             dao = BindingDAO(db_session)
-            bind_token = await dao.start_bind(user_id, device_id, child_id)
+            bind_token, expires_at = await dao.start_bind(user_id, device_id, child_id)
             await db_session.commit()
-            return bind_token, datetime.utcnow()
+            from handlers.mqtt_handler import TalkingQMQTTService
+
+            service = await TalkingQMQTTService.get_instance()
+            if service is None:
+                raise BindingError("MQTT service is unavailable", status_code=503)
+            await service.send_bind_nfc_command(device_id)
+            return bind_token, expires_at
         finally:
             await db_session.close()
 
@@ -52,7 +72,7 @@ class BindingService(DatabaseServiceBase):
                 raise ValueError("Bind session not found")
             if datetime.utcnow() > session["expires_at"]:
                 raise ValueError("Bind session expired")
-            if session["status"] != 1:
+            if int(session["status"]) != SESSION_STATUS_PENDING:
                 raise ValueError("Bind session already processed")
 
             await dao.confirm_bind(session["id"], session["device_id"], session["target_child_id"], user_id)
@@ -61,6 +81,76 @@ class BindingService(DatabaseServiceBase):
         finally:
             await db_session.close()
 
+    async def get_bind_session(self, bind_token: str, user_id: int) -> Optional[Mapping]:
+        db_session = await self.get_session()
+        try:
+            dao = BindingDAO(db_session)
+            session = await dao.get_session(bind_token, user_id)
+            if session is None:
+                return None
+
+            normalized_status = self._normalize_session_status(session)
+            if normalized_status == SESSION_STATUS_EXPIRED and int(session["status"]) != SESSION_STATUS_EXPIRED:
+                await dao.mark_session_status(int(session["id"]), SESSION_STATUS_EXPIRED)
+                await db_session.commit()
+                session = await dao.get_session(bind_token, user_id)
+                if session is None:
+                    return None
+                normalized_status = SESSION_STATUS_EXPIRED
+
+            payload = dict(session)
+            payload["status"] = normalized_status
+            payload["card_uuid"] = payload.get("card_uuid")
+            return payload
+        finally:
+            await db_session.close()
+
+    async def finalize_nfc_bind(self, device_id: str, card_uuid: str) -> Optional[Mapping]:
+        db_session = await self.get_session()
+        try:
+            dao = BindingDAO(db_session)
+            session = await dao.get_latest_pending_session_by_device(device_id)
+            if not session:
+                await db_session.rollback()
+                return None
+
+            if datetime.utcnow() > session["expires_at"]:
+                await dao.mark_session_status(int(session["id"]), SESSION_STATUS_EXPIRED)
+                await db_session.commit()
+                return {
+                    "device_id": device_id,
+                    "bind_token": session["bind_token"],
+                    "status": SESSION_STATUS_EXPIRED,
+                }
+
+            try:
+                await card_service.activate_card(
+                    card_uuid=card_uuid,
+                    device_id=device_id,
+                    db_session=db_session,
+                )
+                await dao.complete_nfc_bind(
+                    session_id=int(session["id"]),
+                    device_id=device_id,
+                    child_id=session["target_child_id"],
+                    user_id=int(session["initiator_user_id"]),
+                )
+                await db_session.commit()
+            except Exception:
+                await dao.mark_session_status(int(session["id"]), SESSION_STATUS_FAILED)
+                await db_session.commit()
+                raise
+
+            return {
+                "device_id": device_id,
+                "bind_token": session["bind_token"],
+                "status": SESSION_STATUS_COMPLETED,
+                "child_id": session["target_child_id"],
+                "card_uuid": card_uuid,
+            }
+        finally:
+            await db_session.close()
+
     async def get_binding(self, device_id: str, user_id: int) -> Optional[Mapping]:
         db_session = await self.get_session()
         try:

talkingq-url/banban/service/child.py

@@ -3,6 +3,7 @@ from datetime import date
 from typing import Optional
 
 from banban.dao.child import ChildDAO
+from banban.dao.im import ImDAO
 from services.database_service_base import DatabaseServiceBase
 
 
@@ -19,10 +20,24 @@ class ChildService(DatabaseServiceBase):
     ) -> Mapping:
         db_session = await self.get_session()
         try:
-            dao = ChildDAO(db_session)
+            child_dao = ChildDAO(db_session)
-            child_id = await dao.create(user_id, child_name, child_gender, child_birthday)
+            im_dao = ImDAO(db_session)
+            child_id = await child_dao.create(
+                user_id,
+                child_name,
+                child_gender,
+                child_birthday,
+                auto_commit=False,
+            )
+            await im_dao.ensure_parent_child_conversation(
+                parent_user_id=user_id,
+                child_id=child_id,
+            )
             await db_session.commit()
-            return await self.get(child_id)
+            return await child_dao.get_by_id(child_id)
+        except Exception:
+            await db_session.rollback()
+            raise
         finally:
             await db_session.close()
 

talkingq-url/banban/service/im.py

@@ -1,10 +1,12 @@
 from dataclasses import dataclass
+import hashlib
 import json
 from collections.abc import Mapping
 from typing import Any
 
-from fastapi import HTTPException, status
+from fastapi import HTTPException
 from services.database_service_base import DatabaseServiceBase
+from banban.service.message_audio_storage import MessageAudioStorageService, MessageAudioStorageError
 
 try:
     from banban.dao.im import ImDAO, DeviceIdentity, ConversationMessageCreateResult
@@ -43,6 +45,12 @@ def participant_type_name(participant_type: int) -> str:
     return PARTICIPANT_TYPE_NAMES.get(participant_type, f"unknown_{participant_type}")
 
 
+def build_device_audio_client_msg_id(*, device_id: str, target_device_id: str, audio_url: str) -> str:
+    raw = f"{device_id}|{target_device_id}|{audio_url}"
+    digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()
+    return f"device-audio-{digest[:32]}"
+
+
 def normalize_content_json(value: Any) -> dict[str, Any] | None:
     if value is None:
         return None
@@ -85,9 +93,24 @@ def row_to_message_item(row: Mapping[str, Any]) -> ChildConversationMessageItem:
     )
 
 
+async def present_message_item(
+    row: Mapping[str, Any],
+    *,
+    audio_storage: MessageAudioStorageService,
+) -> ChildConversationMessageItem:
+    item = row_to_message_item(row)
+    if item.content_type == 2 and item.media_file_key:
+        try:
+            item.media_file_key = await audio_storage.get_audio_url(item.media_file_key)
+        except MessageAudioStorageError:
+            pass
+    return item
+
+
 class ImService(DatabaseServiceBase):
     def __init__(self):
         super().__init__(service_name="im_service")
+        self.audio_storage = MessageAudioStorageService()
 
     async def assert_parent_child_access(self, *, user_id: int, child_id: int) -> Mapping[str, Any]:
         db_session = await self.get_session()
@@ -105,6 +128,34 @@ class ImService(DatabaseServiceBase):
         finally:
             await db_session.close()
 
+    async def _build_message_create_result(
+        self,
+        *,
+        dao: ImDAO,
+        idempotent: bool,
+        conversation_id: int,
+        conversation_type: int,
+        client_msg_id: str,
+    ) -> ConversationMessageCreateResult:
+        message_row = await dao._get_message_by_conversation_client_id(
+            conversation_id=conversation_id,
+            client_msg_id=client_msg_id,
+        )
+        if not message_row:
+            raise RuntimeError("message was not found after insert")
+
+        presented_message = await present_message_item(
+            message_row,
+            audio_storage=self.audio_storage,
+        )
+
+        return ConversationMessageCreateResult(
+            idempotent=idempotent,
+            conversation_id=conversation_id,
+            conversation_type=conversation_type,
+            message=presented_message,
+        )
+
     async def create_parent_child_message(
         self,
         *,
@@ -137,18 +188,12 @@ class ImService(DatabaseServiceBase):
                 receiver_avatar_snapshot=None,
                 payload=payload,
             )
-            message_row = await dao._get_message_by_conversation_client_id(
+            return await self._build_message_create_result(
-                conversation_id=conversation_id,
+                dao=dao,
-                client_msg_id=payload.client_msg_id,
-            )
-            if not message_row:
-                raise RuntimeError("message was not found after insert")
-
-            return ConversationMessageCreateResult(
                 idempotent=idempotent,
                 conversation_id=conversation_id,
                 conversation_type=PARENT_CHILD_CONVERSATION_TYPE,
-                message=row_to_message_item(message_row),
+                client_msg_id=payload.client_msg_id,
             )
         except Exception:
             await db_session.rollback()
@@ -156,21 +201,13 @@ class ImService(DatabaseServiceBase):
         finally:
             await db_session.close()
 
-    async def create_device_message(
+    async def _create_device_message_with_payload(
         self,
         *,
-        device_id: str,
+        dao: ImDAO,
-        serial_number: str,
+        device_identity: DeviceIdentity,
         payload: DeviceMessageCreateRequest,
-    ) -> tuple[DeviceIdentity, ConversationMessageCreateResult]:
+    ) -> ConversationMessageCreateResult:
-        db_session = await self.get_session()
-        try:
-            dao = ImDAO(db_session)
-            device_identity = await dao.authenticate_device_identity(
-                device_id=device_id,
-                serial_number=serial_number,
-            )
-
         if payload.conversation_type == CHILD_PEER_CONVERSATION_TYPE:
             if payload.peer_child_id == device_identity.child_id:
                 raise HTTPException(status_code=400, detail="peer_child_id must be different from current child")
@@ -223,18 +260,61 @@ class ImService(DatabaseServiceBase):
                 payload=payload,
             )
 
-            message_row = await dao._get_message_by_conversation_client_id(
+        return await self._build_message_create_result(
-                conversation_id=conversation_id,
+            dao=dao,
-                client_msg_id=payload.client_msg_id,
-            )
-            if not message_row:
-                raise RuntimeError("message was not found after insert")
-
-            result = ConversationMessageCreateResult(
             idempotent=idempotent,
             conversation_id=conversation_id,
             conversation_type=payload.conversation_type,
-                message=row_to_message_item(message_row),
+            client_msg_id=payload.client_msg_id,
+        )
+
+    async def create_device_message(
+        self,
+        *,
+        device_id: str,
+        serial_number: str,
+        payload: DeviceMessageCreateRequest | None = None,
+        target_device_id: str | None = None,
+        audio_url: str | None = None,
+    ) -> tuple[DeviceIdentity, ConversationMessageCreateResult]:
+        if payload is None and (not target_device_id or not audio_url):
+            raise ValueError("payload or target_device_id/audio_url is required")
+        if payload is not None and (target_device_id is not None or audio_url is not None):
+            raise ValueError("payload and target_device_id/audio_url cannot be used together")
+
+        db_session = await self.get_session()
+        try:
+            dao = ImDAO(db_session)
+            device_identity = await dao.authenticate_device_identity(
+                device_id=device_id,
+                serial_number=serial_number,
+            )
+
+            resolved_payload = payload
+            if resolved_payload is None:
+                target_device_identity = await dao.get_device_by_id(device_id=target_device_id)
+                resolved_payload = DeviceMessageCreateRequest(
+                    conversation_type=CHILD_PEER_CONVERSATION_TYPE,
+                    peer_child_id=target_device_identity.child_id,
+                    content_type=2,
+                    media_file_key=audio_url,
+                    media_mime_type="audio/mpeg",
+                    client_msg_id=build_device_audio_client_msg_id(
+                        device_id=device_id,
+                        target_device_id=target_device_id,
+                        audio_url=audio_url,
+                    ),
+                    ext_json={
+                        "source": "device_audio_message",
+                        "source_device_id": device_id,
+                        "target_device_id": target_device_id,
+                    },
+                )
+
+            result = await self._create_device_message_with_payload(
+                dao=dao,
+                device_identity=device_identity,
+                payload=resolved_payload,
             )
             return device_identity, result
         except Exception:
@@ -251,55 +331,5 @@ class ImService(DatabaseServiceBase):
         finally:
             await db_session.close()
 
-    '''
-    Todo 创建设备消息, 还不完善
-    '''
-    async def create_device_message(
-        self,
-        *,
-        device_id: str,
-        serial_number: str,
-        target_device_id: str,
-        audio_url: str,
-    ):
-        db_session = await self.get_session()
-        try:
-            dao = ImDAO(db_session)
-            device_identity = await dao.authenticate_device_identity(
-                device_id=device_id,
-                serial_number=serial_number,
-            )
-            target_device_identity = await dao.get_device_by_id(device_id=target_device_id)
 
-            sender_child_row = await dao.assert_child_exists(child_id=device_identity.child_id)
-            receiver_child_row = await dao.assert_child_exists(child_id=target_device_identity.child_id)
-
-            conversation_id, idempotent = await dao.create_message(
-                conversation_type=PARENT_CHILD_CONVERSATION_TYPE,
-                participant_a_type=CHILD_PARTICIPANT_TYPE,
-                participant_a_id=str(device_identity.child_id),
-                participant_b_type=PARENT_PARTICIPANT_TYPE,
-                participant_b_id=str(receiver_child_row.child_id),
-                pair_key=f"{device_identity.child_id}:{target_device_identity.child_id}",
-                sender_type=CHILD_PARTICIPANT_TYPE,
-                sender_id=str(device_identity.child_id),
-                receiver_type=PARENT_PARTICIPANT_TYPE,
-                receiver_id=str(receiver_child_row.child_id),
-                sender_name_snapshot=sender_child_row["child_name"],
-                sender_avatar_snapshot=None,
-                receiver_name_snapshot=receiver_child_row["child_name"],
-                receiver_avatar_snapshot=None,
-                payload=None,
-            )
-
-            
-            return device_identity
-        except Exception:
-            await db_session.rollback()
-            raise
-        finally:
-            await db_session.close()
-
-
-# 创建全局 ImService 实例
 im_service = ImService()

talkingq-url/banban/service/message_audio_storage.py

@@ -0,0 +1,129 @@
+import asyncio
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from urllib.parse import urlparse
+from uuid import uuid4
+
+try:
+    from qcloud_cos import CosConfig, CosS3Client
+except ModuleNotFoundError:  # pragma: no cover - exercised in runtime env
+    CosConfig = None
+    CosS3Client = None
+
+from config import settings
+
+
+class MessageAudioStorageError(Exception):
+    pass
+
+
+@dataclass(frozen=True)
+class StoredMessageAudio:
+    file_key: str
+    public_url: str
+
+
+class MessageAudioStorageService:
+    def __init__(self) -> None:
+        self._client = None
+
+    def _assert_ready(self) -> None:
+        if CosConfig is None or CosS3Client is None:
+            raise MessageAudioStorageError("COS SDK is not installed")
+
+        required_pairs = {
+            "COS_SECRET_ID": settings.cos_secret_id,
+            "COS_SECRET_KEY": settings.cos_secret_key,
+            "COS_REGION": settings.cos_region,
+            "COS_BUCKET_MESSAGE": settings.cos_bucket_message,
+        }
+        missing = [key for key, value in required_pairs.items() if not value]
+        if missing:
+            raise MessageAudioStorageError(f"missing COS message config: {', '.join(missing)}")
+
+    def _get_client(self):
+        if self._client is None:
+            config = CosConfig(
+                Region=settings.cos_region,
+                SecretId=settings.cos_secret_id,
+                SecretKey=settings.cos_secret_key,
+                Scheme="https",
+            )
+            self._client = CosS3Client(config)
+        return self._client
+
+    def _build_key(self, *, device_id: str, extension: str) -> str:
+        prefix = settings.cos_message_prefix.strip("/") or "messages/audio"
+        now = datetime.now(UTC)
+        return (
+            f"{prefix}/{device_id}/{now.strftime('%Y/%m/%d')}/"
+            f"{uuid4().hex}.{extension}"
+        )
+
+    def _build_public_url(self, *, file_key: str) -> str:
+        base_url = settings.cos_public_base_url.strip().rstrip("/")
+        if not base_url:
+            base_url = f"https://{settings.cos_bucket_message}.cos.{settings.cos_region}.myqcloud.com"
+        return f"{base_url}/{file_key.lstrip('/')}"
+
+    def _normalize_file_key(self, file_key_or_url: str) -> str:
+        value = (file_key_or_url or "").strip()
+        if not value:
+            raise MessageAudioStorageError("audio file key is required")
+        if value.startswith("http://") or value.startswith("https://"):
+            parsed = urlparse(value)
+            path = parsed.path.lstrip("/")
+            if not path:
+                raise MessageAudioStorageError("audio file key is invalid")
+            return path
+        return value.lstrip("/")
+
+    async def upload_audio(
+        self,
+        *,
+        device_id: str,
+        content: bytes,
+        content_type: str = "audio/mpeg",
+        extension: str = "mp3",
+    ) -> StoredMessageAudio:
+        self._assert_ready()
+        if not content:
+            raise MessageAudioStorageError("audio content is empty")
+
+        file_key = self._build_key(device_id=device_id, extension=extension)
+        await asyncio.to_thread(
+            self._upload_audio_sync,
+            file_key=file_key,
+            content=content,
+            content_type=content_type,
+        )
+        return StoredMessageAudio(
+            file_key=file_key,
+            public_url=self._build_public_url(file_key=file_key),
+        )
+
+    async def get_audio_url(self, file_key_or_url: str) -> str:
+        self._assert_ready()
+        normalized_key = self._normalize_file_key(file_key_or_url)
+        return await asyncio.to_thread(
+            self._get_client().get_presigned_url,
+            Bucket=settings.cos_bucket_message,
+            Key=normalized_key,
+            Method="GET",
+            Expired=settings.cos_avatar_url_expire_seconds,
+        )
+
+    def _upload_audio_sync(
+        self,
+        *,
+        file_key: str,
+        content: bytes,
+        content_type: str,
+    ) -> None:
+        self._get_client().put_object(
+            Bucket=settings.cos_bucket_message,
+            Body=content,
+            Key=file_key,
+            ContentType=content_type,
+            EnableMD5=False,
+        )

talkingq-url/config.py

@@ -89,6 +89,7 @@ class Settings(BaseSettings):
     cos_bucket_message: str = Field(default="", validation_alias="COS_BUCKET_MESSAGE")
     cos_bucket_ava: str = Field(default="", validation_alias="COS_BUCKET_AVA")
     cos_public_base_url: str = Field(default="", validation_alias="COS_PUBLIC_BASE_URL")
+    cos_message_prefix: str = Field(default="messages/audio/", validation_alias="COS_MESSAGE_PREFIX")
     cos_avatar_prefix: str = Field(default="avatars/", validation_alias="COS_AVATAR_PREFIX")
     cos_avatar_url_expire_seconds: int = Field(
         default=86400,

talkingq-url/handlers/audio_file_handler.py

@@ -1,36 +1,19 @@
-import os
+from banban.service.message_audio_storage import MessageAudioStorageService
-import uuid
-from config import settings
 from utils.logger import session_logger
-# from utils.audio_denoiser import reduce_background_noise
+
+
+message_audio_storage_service = MessageAudioStorageService()
 
 
 async def save_audio_file(audio_data: bytes, device_id: str) -> str:
-    """
+    """Upload device audio to COS and return its object key."""
-    保存音频数据到 assets/audio 目录
-    
-    Args:
-        audio_data: 音频二进制数据
-        device_id: 设备ID
-        
-    Returns:
-        音频文件的相对路径
-    """
     try:
-        audio_dir = os.path.join(settings.assets_dir, "audio")
+        stored = await message_audio_storage_service.upload_audio(
-        os.makedirs(audio_dir, exist_ok=True)
+            device_id=device_id,
-        
+            content=audio_data,
-        filename = f"{device_id}_{uuid.uuid4().hex[:8]}.mp3"
+        )
-        filepath = os.path.join(audio_dir, filename)
+        session_logger.info(device_id, "audio", f"audio uploaded to COS: {stored.file_key}")
-        
+        return stored.file_key
-        with open(filepath, 'wb') as f:
-            f.write(audio_data)
-        
-        # relative_path = f"assets/audio/{filename}"
-        session_logger.info(device_id, "audio", f"音频文件已保存: {filepath}")
-        
-        # reduce_background_noise(filepath, relative_path,noise_path='assets/audio/noise_sample.wav',normalize_volume=True)
-        return filepath
     except Exception as e:
-        session_logger.error(device_id, "audio", f"保存音频文件时出错: {e}", exc_info=True)
+        session_logger.error(device_id, "audio", f"failed to store audio: {e}", exc_info=True)
         raise

talkingq-url/handlers/mqtt_handler.py

@@ -3,6 +3,7 @@ import time
 import asyncio
 from typing import Optional, Dict, Callable, Awaitable
 from banban.service.device_setting import device_setting_service
+from banban.service.binding import BindingService
 from config import settings
 from services.card_service import card_service
 from services.offline_audio_cache import offline_audio_cache
@@ -166,8 +167,13 @@ class TalkingQMQTTService:
 
         nfc_uuid = params.get("uuid")
         # 卡片与设备绑定
-        await card_service.activate_card(device_id=device_id, card_uuid=nfc_uuid)
+        # await card_service.activate_card(device_id=device_id, card_uuid=nfc_uuid)
-        logger.info(device_id, "", f"[NFC绑定] 设备 {device_id} 请求绑定卡片, UUID={nfc_uuid}")
+        service = BindingService()
+        result = await service.finalize_nfc_bind(device_id=device_id, card_uuid=nfc_uuid)
+        if result is None:
+            logger.warning(device_id, "", "[NFC bind] no pending bind session found")
+            return
+        logger.info(device_id, "", f"[NFC绑定] 设备 {device_id} 请求绑定卡片, UUID={nfc_uuid}, result={result}")
 
     async def _handle_open_response(self, device_id: str, payload: dict):
         params = payload.get("params", {})

talkingq-url/services/card_service.py

@@ -1,12 +1,14 @@
 import asyncio
-from typing import Optional, Dict
+from typing import Dict, Optional
-import time
+
-from sqlalchemy import select, update, insert
-from sqlalchemy.ext.asyncio import AsyncSession
-from utils.logger import session_logger
 from pydantic import BaseModel
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
 from database.models import Card as DBCard
 from services.database_service_base import DatabaseServiceBase
+from utils.logger import session_logger
+
 
 class Card(BaseModel):
     card_id: Optional[int] = None
@@ -20,7 +22,6 @@ class Card(BaseModel):
 
     @classmethod
     def from_db_model(cls, db_model: DBCard):
-        """从数据库模型创建卡片对象"""
         return cls(
             card_id=db_model.card_id,
             card_uuid=db_model.card_uuid,
@@ -29,87 +30,123 @@ class Card(BaseModel):
             status=db_model.status,
             total_swaps=db_model.total_swaps,
             created_at=db_model.created_at.timestamp() if db_model.created_at else None,
-            updated_at=db_model.updated_at.timestamp() if db_model.updated_at else None
+            updated_at=db_model.updated_at.timestamp() if db_model.updated_at else None,
         )
 
+
 class CardService(DatabaseServiceBase):
     def __init__(self):
         super().__init__(service_name="card")
         self.cards: Dict[str, Card] = {}
         self.lock = asyncio.Lock()
 
+    async def _cache_card(self, card: Card) -> None:
+        async with self.lock:
+            self.cards[card.card_uuid] = card
+
     async def _load_card_from_db(self, card_uuid: str, async_session: AsyncSession) -> Optional[Card]:
-        """从数据库加载卡片信息"""
         try:
             query = select(DBCard).where(DBCard.card_uuid == card_uuid)
             result = await async_session.execute(query)
             db_card = result.scalar_one_or_none()
 
-            if db_card:
+            if not db_card:
-                card = Card.from_db_model(db_card)
-                async with self.lock:
-                    self.cards[card_uuid] = card
-                return card
-            return None
-        except Exception as e:
-            session_logger.error("card", "service", f"从数据库加载卡片失败: {str(e)}")
                 return None
 
-    async def _save_card_to_db(self, card: Card, async_session: AsyncSession):
+            card = Card.from_db_model(db_card)
-        """保存卡片信息到数据库"""
+            await self._cache_card(card)
+            return card
+        except Exception as exc:
+            session_logger.error("card", "service", f"load card failed: {exc}")
+            return None
+
+    async def _clear_existing_device_card(
+        self,
+        device_id: str,
+        card_uuid: str,
+        async_session: AsyncSession,
+    ) -> None:
+        query = select(DBCard).where(
+            DBCard.device_id == device_id,
+            DBCard.card_uuid != card_uuid,
+        )
+        result = await async_session.execute(query)
+        existing_cards = result.scalars().all()
+
+        for db_card in existing_cards:
+            db_card.device_id = None
+            db_card.status = 0
+
+            async with self.lock:
+                cached = self.cards.get(db_card.card_uuid)
+                if cached is not None:
+                    cached.device_id = None
+                    cached.status = 0
+
+        if existing_cards:
+            await async_session.flush()
+
+    async def _save_card_to_db(self, card: Card, async_session: AsyncSession, commit: bool = True) -> None:
         try:
             query = select(DBCard).where(DBCard.card_uuid == card.card_uuid)
             result = await async_session.execute(query)
             existing_card = result.scalar_one_or_none()
 
             if existing_card:
-                stmt = update(DBCard).where(
+                existing_card.device_id = card.device_id
-                    DBCard.card_uuid == card.card_uuid
+                existing_card.card_name = card.card_name
-                ).values(
+                existing_card.status = card.status
-                    device_id=card.device_id,
+                existing_card.total_swaps = card.total_swaps
-                    card_name=card.card_name,
+                db_card = existing_card
-                    status=card.status,
-                    total_swaps=card.total_swaps
-                )
             else:
-                stmt = insert(DBCard).values(
+                db_card = DBCard(
                     card_uuid=card.card_uuid,
                     device_id=card.device_id,
                     card_name=card.card_name,
                     status=card.status,
-                    total_swaps=card.total_swaps
+                    total_swaps=card.total_swaps,
                 )
+                async_session.add(db_card)
 
-            await async_session.execute(stmt)
+            await async_session.flush()
+            if commit:
                 await async_session.commit()
 
-            session_logger.info("card", "service", f"卡片已保存到数据库: {card.card_uuid}")
+            card.card_id = db_card.card_id
-        except Exception as e:
+            await self._cache_card(card)
+            session_logger.info("card", "service", f"card saved: {card.card_uuid}")
+        except Exception as exc:
+            if commit:
                 await async_session.rollback()
-            session_logger.error("card", "service", f"保存卡片到数据库失败: {str(e)}")
+            session_logger.error("card", "service", f"save card failed: {exc}")
             raise
 
-    async def get_card_by_uuid(self, card_uuid: str, force_refresh: bool = False) -> Optional[Card]:
+    async def get_card_by_uuid(
-        """根据UUID获取卡片信息"""
+        self,
+        card_uuid: str,
+        force_refresh: bool = False,
+        db_session: Optional[AsyncSession] = None,
+    ) -> Optional[Card]:
         await self._init_database()
 
         card = None
-        
         if not force_refresh:
             async with self.lock:
                 card = self.cards.get(card_uuid)
 
-        if not card:
+        if card:
-            db_session = await self.db_manager.get_session()
-            try:
-                card = await self._load_card_from_db(card_uuid, db_session)
-            finally:
-                await db_session.close()
-        
             return card
 
+        if db_session is not None:
+            return await self._load_card_from_db(card_uuid, db_session)
+
+        session = await self.db_manager.get_session()
+        try:
+            return await self._load_card_from_db(card_uuid, session)
+        finally:
+            await session.close()
+
     async def get_card_by_device_id(self, device_id: str) -> list[Card]:
-        """根据设备ID获取卡片列表"""
         await self._init_database()
 
         db_session = await self.db_manager.get_session()
@@ -121,70 +158,77 @@ class CardService(DatabaseServiceBase):
             cards = []
             for db_card in db_cards:
                 card = Card.from_db_model(db_card)
-                async with self.lock:
+                await self._cache_card(card)
-                    self.cards[card.card_uuid] = card
                 cards.append(card)
-            
             return cards
-        except Exception as e:
+        except Exception as exc:
-            session_logger.error(device_id, "card", f"根据设备ID获取卡片失败: {str(e)}")
+            session_logger.error(device_id, "card", f"load device cards failed: {exc}")
             return []
         finally:
             await db_session.close()
 
-    async def activate_card(self, card_uuid: str, device_id: str, card_name: Optional[str] = None) -> Card:
+    async def activate_card(
-        """激活卡片并绑定到设备"""
+        self,
+        card_uuid: str,
+        device_id: str,
+        card_name: Optional[str] = None,
+        db_session: Optional[AsyncSession] = None,
+    ) -> Card:
         await self._init_database()
 
+        owns_session = db_session is None
+        if db_session is None:
             db_session = await self.db_manager.get_session()
+
         try:
-            # 检查卡片是否已存在
+            existing_card = await self.get_card_by_uuid(card_uuid, db_session=db_session)
-            existing_card = await self.get_card_by_uuid(card_uuid)
+            await self._clear_existing_device_card(device_id=device_id, card_uuid=card_uuid, async_session=db_session)
 
             if existing_card:
-                # 更新现有卡片
                 existing_card.device_id = device_id
                 existing_card.card_name = card_name
-                existing_card.status = 1  # 激活状态
+                existing_card.status = 1
-                await self._save_card_to_db(existing_card, db_session)
+                await self._save_card_to_db(existing_card, db_session, commit=owns_session)
-                session_logger.info(device_id, "card", f"卡片已激活并绑定到设备: {card_uuid}")
+                session_logger.info(device_id, "card", f"card activated: {card_uuid}")
                 return existing_card
-            else:
+
-                # 创建新卡片
             new_card = Card(
                 card_uuid=card_uuid,
                 device_id=device_id,
                 card_name=card_name,
-                    status=1,  # 激活状态
+                status=1,
-                    total_swaps=0
+                total_swaps=0,
             )
-                await self._save_card_to_db(new_card, db_session)
+            await self._save_card_to_db(new_card, db_session, commit=owns_session)
-                session_logger.info(device_id, "card", f"新卡片已创建并激活: {card_uuid}")
+            session_logger.info(device_id, "card", f"new card activated: {card_uuid}")
             return new_card
+        except Exception:
+            if owns_session:
+                await db_session.rollback()
+            raise
         finally:
+            if owns_session:
                 await db_session.close()
 
     async def increment_swap_count(self, card_uuid: str) -> Optional[Card]:
-        """增加卡片交换次数"""
         await self._init_database()
 
         card = await self.get_card_by_uuid(card_uuid)
-        if card:
+        if not card:
+            return None
+
         card.total_swaps += 1
         db_session = await self.db_manager.get_session()
         try:
             await self._save_card_to_db(card, db_session)
-                session_logger.info("card", "service", f"卡片交换次数已增加: {card_uuid}, 总次数: {card.total_swaps}")
+            session_logger.info("card", "service", f"swap count incremented: {card_uuid}, total={card.total_swaps}")
             return card
         finally:
             await db_session.close()
-        return None
 
     async def check_card_ownership(self, card_uuid: str, device_id: str) -> bool:
-        """检查卡片是否属于指定设备"""
         card = await self.get_card_by_uuid(card_uuid)
-        if card and card.device_id == device_id:
+        return bool(card and card.device_id == device_id)
-            return True
+
-        return False
 
 card_service = CardService()