huangyicao/banban
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(binding): support qr scan and nfc card binding flow

Browse Source
This commit is contained in:
ChengCan
2026-04-30 00:58:56 +08:00
parent efb13de8ed
commit 9952fc4365
9 changed files with 984 additions and 489 deletions
Download Patch File Download Diff File Expand all files Collapse all files

237
talkingq-url/banban/dao/binding.py
View File

@@ -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()

86
talkingq-url/banban/routers/bindings.py
View File

@@ -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)

98
talkingq-url/banban/service/binding.py
View File

@@ -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:

327
talkingq-url/handlers/mqtt_handler.py
View File

@@ -1,16 +1,18 @@
import asyncio
import json
import time
import asyncio
from typing import Optional, Dict, Callable, Awaitable
from banban.service.device_setting import device_setting_service
from config import settings
from services.card_service import card_service
from services.offline_audio_cache import offline_audio_cache
import aiomqtt
from banban.service.location import location_service
from database.models import ChildLocationCurrent
from typing import Awaitable, Callable, Dict, Optional
import aiomqtt
from banban.service.binding import BindingService
from banban.service.device_setting import device_setting_service
from banban.service.location import location_service
from config import settings
from database.models import ChildLocationCurrent
from services.card_service import card_service
from services.device_target_cache import device_target_cache
from services.offline_audio_cache import offline_audio_cache
from utils.logger import session_logger as logger
@@ -64,130 +66,117 @@ class TalkingQMQTTService:
try:
topic = str(message.topic)
parts = topic.split("/")
if parts[0] == "device":
device_id = parts[1] if len(parts) >= 2 else "unknown"
if not parts or parts[0] != "device":
continue
if not device_id.startswith(f"{self.device_prefix}_"):
continue
device_id = parts[1] if len(parts) >= 2 else "unknown"
if not device_id.startswith(f"{self.device_prefix}_"):
continue
payload = json.loads(message.payload.decode("utf-8"))
msg_id = payload.get("msg_id")
handler = self._msg_handlers.get(msg_id)
if handler:
await handler(device_id, payload)
else:
logger.warning(device_id, "", f"未知 msg_id: {msg_id}, 设备: {device_id}")
except json.JSONDecodeError as e:
logger.error("", "", f"消息解析失败: {e}")
except Exception as e:
logger.error("", "", f"消息处理异常: {e}")
payload = json.loads(message.payload.decode("utf-8"))
msg_id = payload.get("msg_id")
handler = self._msg_handlers.get(msg_id)
if handler:
await handler(device_id, payload)
else:
logger.warning(device_id, "", f"unknown msg_id={msg_id}")
except json.JSONDecodeError as exc:
logger.error("", "", f"invalid mqtt payload: {exc}")
except Exception as exc:
logger.error("", "", f"mqtt message handling failed: {exc}")
except asyncio.CancelledError:
pass
except Exception as e:
logger.error("", "", f"消息循环异常: {e}")
except Exception as exc:
logger.error("", "", f"mqtt loop failed: {exc}")
self._connected = False
async def _handle_device_info(self, device_id: str, payload: dict):
# status = payload.get("status")
data = payload.get("data", {})
# if status == "success":
d_id = data.get("id")
power = data.get("power")
signal = data.get("signal")
version = data.get("version")
voice = data.get("voice")
logger.info(device_id, "", f"[设备信息] 设备 {d_id} 信息: 电量={power}, 信号强度={signal}, 版本号={version}, 音量={voice}")
# 插入到数据库
await device_setting_service.insert_or_update(device_id=device_id, power=power, signal_strength=signal, version_str=version, volume=voice)
# else:
# logger.warning(device_id, "", f"[设备信息] 设备 {device_id} 查询失败: {payload}")
await device_setting_service.insert_or_update(
device_id=device_id,
power=data.get("power"),
signal_strength=data.get("signal"),
version_str=data.get("version"),
volume=data.get("voice"),
)
await self._publish(f"device/{device_id}/event_resp", {"msg_id": "000", "status": "success"})
async def _handle_gps_response(self, device_id: str, payload: dict):
status = payload.get("status")
if payload.get("status") != "success":
logger.warning(device_id, "", f"[GPS] query failed: {payload}")
return
data = payload.get("data", {})
if status == "success":
lat = data.get("latitude")
lon = data.get("longitude")
logger.info(device_id, "", f"[GPS] 设备 {device_id} 位置: 纬度={lat}, 经度={lon}")
# 插入到数据库
location = ChildLocationCurrent(device_id=device_id, lat=lat, lon=lon)
await location_service.insert_or_update(device_id=device_id, location=location)
else:
logger.warning(device_id, "", f"[GPS] 设备 {device_id} 查询失败: {payload}")
location = ChildLocationCurrent(
device_id=device_id,
lat=data.get("latitude"),
lon=data.get("longitude"),
)
await location_service.insert_or_update(device_id=device_id, location=location)
async def _handle_volume_response(self, device_id: str, payload: dict):
status = payload.get("status")
data = payload.get("data", {})
if status == "success":
level = data.get("current_level")
logger.info(device_id, "", f"[音量] 设备 {device_id} 当前音量: {level}")
# 更新设备音量
await device_setting_service.insert_or_update(device_id=device_id, volume=level)
if payload.get("status") != "success":
logger.warning(device_id, "", f"[volume] command failed: {payload}")
return
else:
logger.warning(device_id, "", f"[音量] 设备 {device_id} 调节失败: {payload}")
data = payload.get("data", {})
await device_setting_service.insert_or_update(device_id=device_id, volume=data.get("current_level"))
async def _handle_ota_response(self, device_id: str, payload: dict):
status = payload.get("status")
data = payload.get("data", {})
if status == "accepted":
current = data.get("current_version")
target = data.get("target_version")
logger.info(device_id, "", f"[OTA] 设备 {device_id} 已接受升级: {current} -> {target}")
elif status == "success":
new_ver = data.get("new_version")
logger.info(device_id, "", f"[OTA] 设备 {device_id} 升级完成: {new_ver}")
# 更新设备版本号
await device_setting_service.insert_or_update(device_id=device_id, version_str=new_ver)
else:
logger.warning(device_id, "", f"[OTA] 设备 {device_id} 升级异常: {payload}")
if status == "success":
await device_setting_service.insert_or_update(device_id=device_id, version_str=data.get("new_version"))
elif status != "accepted":
logger.warning(device_id, "", f"[OTA] command failed: {payload}")
async def _handle_nfc_notice_response(self, device_id: str, payload: dict):
status = payload.get("status")
if status == "success":
logger.info(device_id, "", f"[NFC通知] 设备 {device_id} 已收到留言提示")
else:
logger.warning(device_id, "", f"[NFC通知] 设备 {device_id} 留言提示失败: {payload}")
if status != "success":
logger.warning(device_id, "", f"[NFC notice] command failed: {payload}")
async def _handle_nfc_listen_report(self, device_id: str, payload: dict):
params = payload.get("params", {})
nfc_uuid = params.get("uuid")
logger.info(device_id, "", f"[NFC收听] 设备 {device_id} 请求收听留言, UUID={nfc_uuid}")
await self._send_nfc_listen_response(device_id, nfc_uuid)
async def _handle_bind_response(self, device_id: str, payload: dict):
params = payload.get("params", {})
status = params.get("status")
logger.info(device_id, "", f"[NFC绑定] 设备 {device_id} 请求绑定卡片返回状态: {status}")
data = payload.get("data", {})
status = payload.get("status") or params.get("status")
logger.info(device_id, "", f"[NFC bind] device={device_id} status={status}")
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}")
if status not in (None, "success", "accepted"):
logger.warning(device_id, "", f"[NFC bind] bind failed: {payload}")
return
nfc_uuid = data.get("uuid") or params.get("uuid")
if not nfc_uuid:
logger.warning(device_id, "", f"[NFC bind] missing uuid: {payload}")
return
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 bind] bind completed, uuid={nfc_uuid}, result={result}")
async def _handle_open_response(self, device_id: str, payload: dict):
params = payload.get("params", {})
status = params.get("status")
logger.info(device_id, "", f"[设备状态] 设备 {device_id} 请求返回设备状态: {status}")
data = params.get("data", {})
device_status = data.get("status")
logger.info(device_id, "", f"[设备状态] 设备 {device_id} 请求设备状态: {device_status}")
logger.info(device_id, "", f"[device open] response={params}")
async def _send_nfc_listen_response(self, device_id: str, nfc_uuid: str):
topic = f"device/{device_id}/event_resp"
card = await card_service.get_card_by_uuid(nfc_uuid)
if not card:
logger.warning("", "", f"[NFC收听] 设备 {device_id} 没有找到卡片{nfc_uuid}，无法发送留言提示")
payload = {
"msg_id": "005",
"type": 0,
"params": {
"url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/error_card.mp3"
}
},
}
await self._publish(topic, payload)
return
@@ -201,75 +190,54 @@ class TalkingQMQTTService:
"type": 0,
"params": {
"url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/test_zh.mp3"
}
},
}
await self._publish(topic, payload)
return
return
if has_pending:
audio_urls = await offline_audio_cache.get_audio_urls(device_id)
# 53D92B6DA20001 测试卡片
if nfc_uuid == "53D92B6DA20001":
payload = {
"msg_id": "005",
"type": 0,
"params": {
"url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/test_zh.mp3"
}
}
await self._publish(topic, payload)
return
if len(audio_urls) == 0:
payload = {
"msg_id": "005",
"type": 0,
"params": {
"url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/no_message.mp3"
}
},
}
else:
params = {}
for k, audio_url in enumerate(audio_urls, start=1):
params[f"url_{k}"] = audio_url
payload = {
"msg_id": "005",
"type": 0,
"params": params
}
params = {f"url_{k}": audio_url for k, audio_url in enumerate(audio_urls, start=1)}
payload = {"msg_id": "005", "type": 0, "params": params}
await offline_audio_cache.clear_audio_urls(device_id)
await self._publish(topic, payload)
else:
payload = {
"msg_id": "005",
"type": 0,
"params": {
"url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/no_message.mp3"
}
}
await self._publish(topic, payload)
else:
# 检查卡片是否存在
existing_card = await card_service.get_card_by_uuid(nfc_uuid)
if existing_card:
# 卡片已存在，使用卡片绑定的设备ID作为目标设备ID
target_device_id = existing_card.device_id
logger.info(device_id, "card", f"卡片已存在，绑定的设备ID: {target_device_id}")
else:
# 卡片不存在，创建新卡片并绑定到当前设备
new_card = await card_service.activate_card(nfc_uuid, device_id)
logger.info(device_id, "card", f"新卡片{nfc_uuid}已创建并激活，绑定到设备: {device_id}")
return
# 设置目标设备
await device_target_cache.set_target(device_id, target_device_id)
payload = {
"msg_id": "005",
"type": 1,
"type": 0,
"params": {
"url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/welcome.mp3"
}
"url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/no_message.mp3"
},
}
await self._publish(topic, payload)
return
existing_card = await card_service.get_card_by_uuid(nfc_uuid)
if existing_card:
target_device_id = existing_card.device_id
else:
await card_service.activate_card(nfc_uuid, device_id)
return
await device_target_cache.set_target(device_id, target_device_id)
payload = {
"msg_id": "005",
"type": 1,
"params": {
"url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/welcome.mp3"
},
}
await self._publish(topic, payload)
async def connect(self):
async with self._connect_lock:
@@ -286,19 +254,15 @@ class TalkingQMQTTService:
)
await self._client.__aenter__()
self._connected = True
logger.system_info("", f"TalkingQ MQTT 已连接: {self.broker}:{self.port}")
await self._client.subscribe("device/+/response", qos=self.qos)
logger.system_info("", "已订阅: device/+/response")
await self._client.subscribe("device/+/event", qos=self.qos)
logger.system_info("", "已订阅: device/+/event")
self._message_task = asyncio.create_task(self._message_loop())
logger.info("", "", "TalkingQ MQTT 服务启动中...")
except Exception as e:
logger.info("", "", f"mqtt connected: {self.broker}:{self.port}")
except Exception as exc:
self._connected = False
logger.error("", "", f"TalkingQ MQTT 连接失败: {e}")
logger.error("", "", f"mqtt connect failed: {exc}")
async def disconnect(self):
if self._message_task and not self._message_task.done():
@@ -325,69 +289,48 @@ class TalkingQMQTTService:
await self._ensure_connected()
try:
await self._client.publish(topic, json.dumps(payload, ensure_ascii=False), qos=self.qos)
logger.info("", "", f"下发命令 -> {topic}: {payload}")
except Exception as e:
logger.error("", "", f"命令下发失败: {e}")
logger.info("", "", f"mqtt publish topic={topic} payload={payload}")
except Exception as exc:
logger.error("", "", f"mqtt publish failed: {exc}")
async def send_gps_query(self, device_id: str) -> str:
topic = f"device/{device_id}/command"
payload = {"msg_id": "001"}
await self._publish(topic, payload)
await self._publish(f"device/{device_id}/command", {"msg_id": "001"})
return "001"
async def send_volume_command(self, device_id: str, level: int) -> str:
topic = f"device/{device_id}/command"
payload = {
"msg_id": "002",
"params": {"level": level}
}
await self._publish(topic, payload)
await self._publish(
f"device/{device_id}/command",
{"msg_id": "002", "params": {"level": level}},
)
return "002"
async def send_ota_command(self, device_id: str, url: str, version: str) -> str:
topic = f"device/{device_id}/command"
payload = {
"msg_id": "003",
"params": {
"url": url,
"version": version,
}
}
await self._publish(topic, payload)
await self._publish(
f"device/{device_id}/command",
{"msg_id": "003", "params": {"url": url, "version": version}},
)
return "003"
async def send_nfc_notice(self, device_id: str, url: str) -> str:
topic = f"device/{device_id}/command"
payload = {
"msg_id": "004",
"params": {"url_1": url}
}
await self._publish(topic, payload)
await self._publish(
f"device/{device_id}/command",
{"msg_id": "004", "params": {"url_1": url}},
)
return "004"
async def send_bind_nfc_command(self, device_id: str, uuid: str) -> str:
topic = f"device/{device_id}/command"
payload = {
"msg_id": "006"
}
await self._publish(topic, payload)
async def send_bind_nfc_command(self, device_id: str, uuid: Optional[str] = None) -> str:
del uuid
await self._publish(f"device/{device_id}/command", {"msg_id": "006"})
return "006"
async def send_open_command(self, device_id: str, open_type: int, is_open: int) -> str:
topic = f"device/{device_id}/command"
if open_type not in [0, 1]:
raise ValueError("open_type must be 0 or 1")
if open_type == 0:
payload = {
"msg_id": "007",
"type": open_type
}
elif open_type == 1:
payload = {
"msg_id": "007",
"type": open_type,
"status": is_open
}
await self._publish(topic, payload)
payload = {"msg_id": "007", "type": open_type}
else:
payload = {"msg_id": "007", "type": open_type, "status": is_open}
await self._publish(f"device/{device_id}/command", payload)
return "007"

262
talkingq-url/services/card_service.py
View File

@@ -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()