diff --git a/talkingq-url/banban/dao/__init__.py b/talkingq-url/banban/dao/__init__.py index fd6faa8..793454d 100644 --- a/talkingq-url/banban/dao/__init__.py +++ b/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() diff --git a/talkingq-url/banban/dao/binding.py b/talkingq-url/banban/dao/binding.py index 9ef916a..73d5383 100644 --- a/talkingq-url/banban/dao/binding.py +++ b/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]: @@ -21,7 +29,7 @@ class BindingDAO(BaseDAO): FROM device_auth WHERE device_id = :device_id LIMIT 1 - """, + """, {"device_id": device_id}, ) ).mappings().first() @@ -39,7 +47,7 @@ class BindingDAO(BaseDAO): unbound_at = NULL, updated_at = CURRENT_TIMESTAMP WHERE id = :id - """, + """, {"id": row_id}, ) return @@ -50,51 +58,28 @@ class BindingDAO(BaseDAO): SET child_id = NULL, updated_at = CURRENT_TIMESTAMP WHERE id = :id - """, + """, {"id": row_id}, ) async def _upsert_parent_child_relation(self, user_id: int, child_id: int) -> None: - updated = await self.execute( + await self.execute( """ - UPDATE parent_child_relations - SET status = 1, + 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 + status = VALUES(status), 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) - """, - {"user_id": user_id, "child_id": child_id}, - ) - 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}, - ) async def _insert_bind_history(self, device_id: str, child_id: Optional[int], user_id: int, bind_source: int) -> None: await self.execute( """ INSERT INTO device_bind_history (device_id, child_id, bound_by_user_id, bind_source, bound_at) VALUES (:device_id, :child_id, :user_id, :bind_source, CURRENT_TIMESTAMP) - """, + """, { "device_id": device_id, "child_id": child_id, @@ -122,7 +107,7 @@ class BindingDAO(BaseDAO): unbound_at = NULL, updated_at = CURRENT_TIMESTAMP WHERE device_id = :device_id - """, + """, {"owner_user_id": user_id, "device_id": device_id}, ) else: @@ -130,7 +115,7 @@ class BindingDAO(BaseDAO): """ INSERT INTO device_bindings (device_id, owner_user_id, child_id, status, bound_at) VALUES (:device_id, :owner_user_id, NULL, 1, CURRENT_TIMESTAMP) - """, + """, {"device_id": device_id, "owner_user_id": user_id}, ) return @@ -156,8 +141,8 @@ class BindingDAO(BaseDAO): unbound_at = NULL, updated_at = CURRENT_TIMESTAMP WHERE device_id = :device_id - """, - {"owner_user_id": user_id, "child_id": child_id, "device_id": device_id}, + """, + {"owner_user_id": user_id, "child_id": child_id, "device_id": device_id}, ) return @@ -168,57 +153,160 @@ class BindingDAO(BaseDAO): """ INSERT INTO device_bindings (device_id, owner_user_id, child_id, status, bound_at) VALUES (:device_id, :owner_user_id, :child_id, 1, CURRENT_TIMESTAMP) - """, + """, {"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, "device_id": device_id, "initiator_user_id": user_id, "target_child_id": child_id, "expires_at": expires_at, + "status": SESSION_STATUS_PENDING, }, ) - await self.commit() - return bind_token + return bind_token, expires_at 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.commit() + await self._insert_bind_history( + 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]: @@ -231,7 +319,7 @@ class BindingDAO(BaseDAO): AND status = 1 ORDER BY bound_at DESC LIMIT 1 - """, + """, {"user_id": user_id}, ) ).mappings().first() @@ -260,7 +348,7 @@ class BindingDAO(BaseDAO): WHERE {where} ORDER BY db.id DESC LIMIT :limit - """, + """, params, ) ).mappings().all() @@ -275,7 +363,7 @@ class BindingDAO(BaseDAO): WHERE device_id = :device_id AND owner_user_id = :user_id AND status = 1 - """, + """, {"device_id": device_id, "user_id": user_id}, ) ).mappings().first() @@ -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,11 +434,12 @@ 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 - """, + """, params, ) ).mappings().all() diff --git a/talkingq-url/banban/dao/child.py b/talkingq-url/banban/dao/child.py index 65e072e..adf3c7c 100644 --- a/talkingq-url/banban/dao/child.py +++ b/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,7 +35,8 @@ class ChildDAO(BaseDAO): ) child_id = inserted_primary_key(result) await self._create_relation(user_id, child_id) - await self.commit() + if auto_commit: + await self.commit() return child_id async def get_by_id(self, child_id: int) -> Optional[Mapping]: diff --git a/talkingq-url/banban/dao/im.py b/talkingq-url/banban/dao/im.py index b4ab869..dcc0447 100644 --- a/talkingq-url/banban/dao/im.py +++ b/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}, @@ -484,4 +508,4 @@ class ImDAO(BaseDAO): async def _next_primary_key(self, table_name: str) -> int | None: result = await self.execute(text(f"SELECT 1")) - return None \ No newline at end of file + return None diff --git a/talkingq-url/banban/routers/bindings.py b/talkingq-url/banban/routers/bindings.py index 8db2d62..3104c04 100644 --- a/talkingq-url/banban/routers/bindings.py +++ b/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.security import get_current_user_id - from banban.service.binding import BindingError, BindingService -except ModuleNotFoundError: - from banban.security import get_current_user_id - from banban.service.binding import BindingError, BindingService +from banban.dao.binding import ( + SESSION_STATUS_CANCELLED, + SESSION_STATUS_COMPLETED, + SESSION_STATUS_EXPIRED, + SESSION_STATUS_FAILED, + 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: - raise HTTPException(status_code=e.status_code, detail=str(e)) - return BindStartResponse(bind_token=bind_token, expires_at=expires_at.isoformat()) + except BindingError as exc: + raise HTTPException(status_code=exc.status_code, detail=str(exc)) + 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: - raise HTTPException(status_code=e.status_code, detail=str(e)) - except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) + except BindingError as exc: + raise HTTPException(status_code=exc.status_code, detail=str(exc)) + except ValueError as exc: + 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: - raise HTTPException(status_code=e.status_code, detail=str(e)) + except BindingError as exc: + 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: - raise HTTPException(status_code=e.status_code, detail=str(e)) - except ValueError as e: - raise HTTPException(status_code=404, detail=str(e)) + except BindingError as exc: + raise HTTPException(status_code=exc.status_code, detail=str(exc)) + except ValueError as exc: + 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) diff --git a/talkingq-url/banban/routers/im.py b/talkingq-url/banban/routers/im.py index 757d1a2..958973d 100644 --- a/talkingq-url/banban/routers/im.py +++ b/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( diff --git a/talkingq-url/banban/service/binding.py b/talkingq-url/banban/service/binding.py index 34bc87a..cc7b364 100644 --- a/talkingq-url/banban/service/binding.py +++ b/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: diff --git a/talkingq-url/banban/service/child.py b/talkingq-url/banban/service/child.py index 269d6d2..d304cdf 100644 --- a/talkingq-url/banban/service/child.py +++ b/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_id = await dao.create(user_id, child_name, child_gender, child_birthday) + child_dao = ChildDAO(db_session) + 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() diff --git a/talkingq-url/banban/service/im.py b/talkingq-url/banban/service/im.py index 9c5574d..74d3b86 100644 --- a/talkingq-url/banban/service/im.py +++ b/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( - conversation_id=conversation_id, - client_msg_id=payload.client_msg_id, - ) - if not message_row: - raise RuntimeError("message was not found after insert") - - return ConversationMessageCreateResult( + return await self._build_message_create_result( + dao=dao, 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,13 +201,87 @@ class ImService(DatabaseServiceBase): finally: await db_session.close() + async def _create_device_message_with_payload( + self, + *, + dao: ImDAO, + device_identity: DeviceIdentity, + payload: DeviceMessageCreateRequest, + ) -> ConversationMessageCreateResult: + 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") + sender_child_row = await dao.assert_child_exists(child_id=device_identity.child_id) + receiver_child_row = await dao.assert_child_exists(child_id=payload.peer_child_id) + participant_a_id, participant_b_id, pair_key = dao._build_child_peer_pair( + device_identity.child_id, + payload.peer_child_id, + ) + + conversation_id, idempotent = await dao.create_message( + conversation_type=CHILD_PEER_CONVERSATION_TYPE, + participant_a_type=CHILD_PARTICIPANT_TYPE, + participant_a_id=participant_a_id, + participant_b_type=CHILD_PARTICIPANT_TYPE, + participant_b_id=participant_b_id, + pair_key=pair_key, + sender_type=CHILD_PARTICIPANT_TYPE, + sender_id=str(device_identity.child_id), + receiver_type=CHILD_PARTICIPANT_TYPE, + receiver_id=str(payload.peer_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=payload, + ) + else: + sender_child_row = await dao.assert_child_exists(child_id=device_identity.child_id) + parent_row = await dao._get_parent_row(payload.parent_user_id) + if not parent_row: + raise HTTPException(status_code=404, detail="parent not found") + await dao.assert_parent_child_access(user_id=payload.parent_user_id, child_id=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(payload.parent_user_id), + pair_key=f"{device_identity.child_id}:{payload.parent_user_id}", + sender_type=CHILD_PARTICIPANT_TYPE, + sender_id=str(device_identity.child_id), + receiver_type=PARENT_PARTICIPANT_TYPE, + receiver_id=str(payload.parent_user_id), + sender_name_snapshot=sender_child_row["child_name"], + sender_avatar_snapshot=None, + receiver_name_snapshot=parent_row["nickname"], + receiver_avatar_snapshot=parent_row["avatar_url"], + payload=payload, + ) + + return await self._build_message_create_result( + dao=dao, + idempotent=idempotent, + conversation_id=conversation_id, + conversation_type=payload.conversation_type, + client_msg_id=payload.client_msg_id, + ) + async def create_device_message( self, *, device_id: str, serial_number: str, - payload: DeviceMessageCreateRequest, + 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) @@ -171,70 +290,31 @@ class ImService(DatabaseServiceBase): 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") - sender_child_row = await dao.assert_child_exists(child_id=device_identity.child_id) - receiver_child_row = await dao.assert_child_exists(child_id=payload.peer_child_id) - participant_a_id, participant_b_id, pair_key = dao._build_child_peer_pair( - device_identity.child_id, - payload.peer_child_id, - ) - - conversation_id, idempotent = await dao.create_message( + 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, - participant_a_type=CHILD_PARTICIPANT_TYPE, - participant_a_id=participant_a_id, - participant_b_type=CHILD_PARTICIPANT_TYPE, - participant_b_id=participant_b_id, - pair_key=pair_key, - sender_type=CHILD_PARTICIPANT_TYPE, - sender_id=str(device_identity.child_id), - receiver_type=CHILD_PARTICIPANT_TYPE, - receiver_id=str(payload.peer_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=payload, - ) - else: - sender_child_row = await dao.assert_child_exists(child_id=device_identity.child_id) - parent_row = await dao._get_parent_row(payload.parent_user_id) - if not parent_row: - raise HTTPException(status_code=404, detail="parent not found") - await dao.assert_parent_child_access(user_id=payload.parent_user_id, child_id=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(payload.parent_user_id), - pair_key=f"{device_identity.child_id}:{payload.parent_user_id}", - sender_type=CHILD_PARTICIPANT_TYPE, - sender_id=str(device_identity.child_id), - receiver_type=PARENT_PARTICIPANT_TYPE, - receiver_id=str(payload.parent_user_id), - sender_name_snapshot=sender_child_row["child_name"], - sender_avatar_snapshot=None, - receiver_name_snapshot=parent_row["nickname"], - receiver_avatar_snapshot=parent_row["avatar_url"], - payload=payload, + 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, + }, ) - message_row = await dao._get_message_by_conversation_client_id( - conversation_id=conversation_id, - 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), + 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() diff --git a/talkingq-url/banban/service/message_audio_storage.py b/talkingq-url/banban/service/message_audio_storage.py new file mode 100644 index 0000000..eb749d3 --- /dev/null +++ b/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, + ) diff --git a/talkingq-url/config.py b/talkingq-url/config.py index 28ed176..7b175e9 100644 --- a/talkingq-url/config.py +++ b/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, diff --git a/talkingq-url/handlers/audio_file_handler.py b/talkingq-url/handlers/audio_file_handler.py index efd34ab..e695baf 100644 --- a/talkingq-url/handlers/audio_file_handler.py +++ b/talkingq-url/handlers/audio_file_handler.py @@ -1,36 +1,19 @@ -import os -import uuid -from config import settings +from banban.service.message_audio_storage import MessageAudioStorageService 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: - """ - 保存音频数据到 assets/audio 目录 - - Args: - audio_data: 音频二进制数据 - device_id: 设备ID - - Returns: - 音频文件的相对路径 - """ + """Upload device audio to COS and return its object key.""" try: - audio_dir = os.path.join(settings.assets_dir, "audio") - os.makedirs(audio_dir, exist_ok=True) - - filename = f"{device_id}_{uuid.uuid4().hex[:8]}.mp3" - filepath = os.path.join(audio_dir, filename) - - 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 + stored = await message_audio_storage_service.upload_audio( + device_id=device_id, + content=audio_data, + ) + session_logger.info(device_id, "audio", f"audio uploaded to COS: {stored.file_key}") + return stored.file_key except Exception as e: - session_logger.error(device_id, "audio", f"保存音频文件时出错: {e}", exc_info=True) - raise \ No newline at end of file + session_logger.error(device_id, "audio", f"failed to store audio: {e}", exc_info=True) + raise diff --git a/talkingq-url/handlers/mqtt_handler.py b/talkingq-url/handlers/mqtt_handler.py index d70b2fb..d4ca8c4 100644 --- a/talkingq-url/handlers/mqtt_handler.py +++ b/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) - logger.info(device_id, "", f"[NFC绑定] 设备 {device_id} 请求绑定卡片, UUID={nfc_uuid}") + # await card_service.activate_card(device_id=device_id, card_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", {}) diff --git a/talkingq-url/services/card_service.py b/talkingq-url/services/card_service.py index 737f32b..41957cf 100644 --- a/talkingq-url/services/card_service.py +++ b/talkingq-url/services/card_service.py @@ -1,12 +1,14 @@ import asyncio -from typing import Optional, Dict -import time -from sqlalchemy import select, update, insert -from sqlalchemy.ext.asyncio import AsyncSession -from utils.logger import session_logger +from typing import Dict, Optional + 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 @@ -17,10 +19,9 @@ class Card(BaseModel): total_swaps: int = 0 created_at: Optional[float] = None updated_at: Optional[float] = None - + @classmethod def from_db_model(cls, db_model: DBCard): - """从数据库模型创建卡片对象""" return cls( card_id=db_model.card_id, card_uuid=db_model.card_uuid, @@ -29,162 +30,205 @@ 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: - card = Card.from_db_model(db_card) - async with self.lock: - self.cards[card_uuid] = card - return card + + if not db_card: + return None + + 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 - 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): - """保存卡片信息到数据库""" + + 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( - DBCard.card_uuid == card.card_uuid - ).values( - device_id=card.device_id, - card_name=card.card_name, - status=card.status, - total_swaps=card.total_swaps - ) + existing_card.device_id = card.device_id + existing_card.card_name = card.card_name + existing_card.status = card.status + existing_card.total_swaps = card.total_swaps + db_card = existing_card 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, ) - - await async_session.execute(stmt) - await async_session.commit() - - session_logger.info("card", "service", f"卡片已保存到数据库: {card.card_uuid}") - except Exception as e: - await async_session.rollback() - session_logger.error("card", "service", f"保存卡片到数据库失败: {str(e)}") + async_session.add(db_card) + + await async_session.flush() + if commit: + await async_session.commit() + + card.card_id = db_card.card_id + 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"save card failed: {exc}") raise - - async def get_card_by_uuid(self, card_uuid: str, force_refresh: bool = False) -> Optional[Card]: - """根据UUID获取卡片信息""" + + async def get_card_by_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: - 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 card: + 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() try: query = select(DBCard).where(DBCard.device_id == device_id) result = await db_session.execute(query) db_cards = result.scalars().all() - + cards = [] for db_card in db_cards: card = Card.from_db_model(db_card) - async with self.lock: - self.cards[card.card_uuid] = card + await self._cache_card(card) cards.append(card) - return cards - except Exception as e: - session_logger.error(device_id, "card", f"根据设备ID获取卡片失败: {str(e)}") + except Exception as exc: + 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() - - db_session = await self.db_manager.get_session() + + 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) - + existing_card = await self.get_card_by_uuid(card_uuid, db_session=db_session) + 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 # 激活状态 - await self._save_card_to_db(existing_card, db_session) - session_logger.info(device_id, "card", f"卡片已激活并绑定到设备: {card_uuid}") + existing_card.status = 1 + await self._save_card_to_db(existing_card, db_session, commit=owns_session) + 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, # 激活状态 - total_swaps=0 - ) - await self._save_card_to_db(new_card, db_session) - session_logger.info(device_id, "card", f"新卡片已创建并激活: {card_uuid}") - return new_card + + new_card = Card( + card_uuid=card_uuid, + device_id=device_id, + card_name=card_name, + status=1, + total_swaps=0, + ) + await self._save_card_to_db(new_card, db_session, commit=owns_session) + 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 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"swap count incremented: {card_uuid}, total={card.total_swaps}") + return card finally: 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: - 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}") - 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 True - return False + return bool(card and card.device_id == device_id) + card_service = CardService()