From b6315f69b0db82f19994025d6feda08af3343a47 Mon Sep 17 00:00:00 2001 From: Pine Date: Mon, 24 Aug 2026 20:10:46 +0800 Subject: [PATCH] =?UTF-8?q?feat(park):=20=E5=A4=A7=E5=B1=8F=E8=AE=BE?= =?UTF-8?q?=E5=A4=87=E7=BB=91=E5=AE=9A=E6=B5=81=20=E2=80=94=20=E6=B0=B8?= =?UTF-8?q?=E4=B9=85=E8=AE=BE=E5=A4=87ID+=E8=BF=9E=E6=8E=A5=E7=A0=81+?= =?UTF-8?q?=E5=9B=AD=E5=8C=BA=E7=AB=AF=E7=BB=91=E5=AE=9A+MQTT=E4=B8=8B?= =?UTF-8?q?=E5=8F=91=E8=AE=BE=E5=A4=87token?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - park_screens 加 status/code/code_expires/bound_at、tenant_id 可空(alembic 0002)。 - auth 加 create/parse_device_token {tenant_id,device_id}。 - tenants 加 ensures_device/set_device_code/find_device_by_code/bind_device/unbind_device。 - 端点：/park/api/devices/register、/park/api/devices/{device_id}/code(8位码TTL10min)、 /park/tenants/{tid}/screens/bind-by-code(校验码→绑定→MQTT opc/display/bind/ 下发 token)、 /park/api/devices/{device_id}/unbind。_resolve_tenant 支持设备 token。 - 验证(TestClient)：register→code→建园区→绑码(bound/命名)→设备token取数据返回绑定园区。 --- alembic/versions/0002_park_screen_bind.py | 41 ++++++++++++ app/infrastructure/models.py | 8 ++- app/park/auth.py | 20 ++++++ app/park/routers.py | 79 +++++++++++++++++++++-- app/park/tenants.py | 74 ++++++++++++++++++++- 5 files changed, 214 insertions(+), 8 deletions(-) create mode 100644 alembic/versions/0002_park_screen_bind.py diff --git a/alembic/versions/0002_park_screen_bind.py b/alembic/versions/0002_park_screen_bind.py new file mode 100644 index 0000000..5866597 --- /dev/null +++ b/alembic/versions/0002_park_screen_bind.py @@ -0,0 +1,41 @@ +"""park_screens 绑定字段（status/code/code_expires/bound_at），tenant_id 可空（幂等） + +Revision ID: 0002_park_screen_bind +Revises: 0001_initial +Create Date: 2026-08-24 +""" +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +revision = "0002_park_screen_bind" +down_revision = "0001_initial" +branch_labels = None +depends_on = None + + +def _existing_cols(table: str) -> set[str]: + conn = op.get_bind() + rows = conn.exec_driver_sql(f"PRAGMA table_info({table})").fetchall() + return {r[1] for r in rows} + + +def upgrade() -> None: + cols = _existing_cols("park_screens") + spec = { + "status": sa.Column("status", sa.String(), server_default="unbound"), + "code": sa.Column("code", sa.String(), server_default=""), + "code_expires": sa.Column("code_expires", sa.String(), server_default=""), + "bound_at": sa.Column("bound_at", sa.String(), server_default=""), + } + for name, col in spec.items(): + if name not in cols: + op.add_column("park_screens", col) + + +def downgrade() -> None: + cols = _existing_cols("park_screens") + for name in ("status", "code", "code_expires", "bound_at"): + if name in cols: + op.drop_column("park_screens", name) diff --git a/app/infrastructure/models.py b/app/infrastructure/models.py index ddc1771..7faa894 100644 --- a/app/infrastructure/models.py +++ b/app/infrastructure/models.py @@ -758,9 +758,13 @@ class ParkKbDoc(Base): class ParkScreen(Base): __tablename__ = "park_screens" id: Mapped[str] = mapped_column(String, primary_key=True) - tenant_id: Mapped[str] = mapped_column(ForeignKey("park_tenants.id", ondelete="CASCADE"), nullable=False, index=True) - device_id: Mapped[str] = mapped_column(String, default="") + tenant_id: Mapped[str | None] = mapped_column(ForeignKey("park_tenants.id", ondelete="CASCADE"), nullable=True, index=True) + device_id: Mapped[str] = mapped_column(String, default="", index=True) name: Mapped[str] = mapped_column(String, default="") role: Mapped[str] = mapped_column(String, default="main") location: Mapped[str] = mapped_column(String, default="") + status: Mapped[str] = mapped_column(String, default="unbound") # unbound | bound + code: Mapped[str] = mapped_column(String, default="") # 待绑定连接码 + code_expires: Mapped[str] = mapped_column(String, default="") # 码过期时间(ISO) + bound_at: Mapped[str] = mapped_column(String, default="") created_at: Mapped[str] = mapped_column(String, default="") diff --git a/app/park/auth.py b/app/park/auth.py index b527ed9..715a013 100644 --- a/app/park/auth.py +++ b/app/park/auth.py @@ -41,3 +41,23 @@ def require_tenant(authorization: str | None = Header(default=None, description= if not authorization or not authorization.startswith("Bearer "): raise HTTPException(status_code=401, detail="缺少 park token") return parse_token(authorization.split(" ", 1)[1]) + + +# ---------------- 设备 token（大屏绑定后用，含 tenant_id + device_id） ---------------- + +def create_device_token(tenant_id: str, device_id: str) -> str: + now = datetime.datetime.now(datetime.timezone.utc) + payload = {"typ": "device", "tid": tenant_id, "dev": device_id, + "iat": now, "exp": now + datetime.timedelta(days=_TTL_DAYS)} + return jwt.encode(payload, _SECRET, algorithm=_ALG) + + +def parse_device_token(token: str) -> dict: + """返回 {tenant_id, device_id}；非法抛 401。""" + try: + payload = jwt.decode(token, _SECRET, algorithms=[_ALG]) + except jwt.PyJWTError as e: # noqa: BLE001 + raise HTTPException(status_code=401, detail=f"设备 token 无效: {e}") from e + if payload.get("typ") != "device" or not payload.get("tid") or not payload.get("dev"): + raise HTTPException(status_code=401, detail="设备 token 无效") + return {"tenant_id": payload["tid"], "device_id": payload["dev"]} diff --git a/app/park/routers.py b/app/park/routers.py index 18d62fd..2d1cf94 100644 --- a/app/park/routers.py +++ b/app/park/routers.py @@ -18,7 +18,7 @@ from fastapi.responses import FileResponse, JSONResponse, StreamingResponse from pydantic import BaseModel, Field from . import park_config, tenants -from .auth import create_token, parse_token, require_tenant +from .auth import create_device_token, create_token, parse_device_token, parse_token, require_tenant from .config import settings from .event_bus import bus from .mqtt import hub @@ -29,15 +29,19 @@ logger = logging.getLogger("dpm.api") async def _resolve_tenant(authorization: str | None, tenant_id: str | None) -> str: - """解析目标园区：?tenant_id=（admin 亮传）> 大屏 park-token > 默认园区。 - admin 携带平台 token，不应被当作 park-token 解析失败；故 tenant_id 优先。""" + """解析目标园区：?tenant_id=（admin 亮传）> 大屏 tenant-token > 设备 token > 默认园区。""" if tenant_id: return tenant_id if authorization and authorization.startswith("Bearer "): + tok = authorization.split(" ", 1)[1] try: - return parse_token(authorization.split(" ", 1)[1]) + return parse_token(tok) # 大屏 tenant token except HTTPException: - pass # 平台 token → 回退默认 + pass + try: + return parse_device_token(tok)["tenant_id"] # 绑定后的设备 token + except HTTPException: + pass return await tenants.ensure_default_tenant() log = logging.getLogger("dpm.api") @@ -453,6 +457,71 @@ async def screens_delete(sid: str, authorization: str | None = Header(None), ten return {"ok": ok, "id": sid} +# ==================== 大屏设备注册 / 绑定（连接码 + MQTT 通知） ==================== + +class DeviceBody(BaseModel): + device_id: str = "" + + +class BindByCodeBody(BaseModel): + code: str + name: str = "" + role: str = "main" + location: str = "" + + +def _gen_code() -> str: + import random + return str(random.randint(10000000, 99999999)) + + +def _publish_bind(device_id: str, payload: dict) -> None: + """未绑定大屏仅订阅 bind 频道；绑定成功后经此频道通知并下发 token。""" + hub.publish(f"opc/display/bind/{device_id}", payload, qos=1) + + +@router.post("/api/devices/register") +async def device_register(body: DeviceBody): + device_id = (body.device_id or "").strip() + if not device_id: + return JSONResponse({"ok": False, "error": "device_id 必填"}, status_code=400) + d = await tenants.ensure_device(device_id) + return {"ok": True, "device_id": device_id, "bound": d.get("status") == "bound", "tenant_id": d.get("tenant_id")} + + +@router.post("/api/devices/{device_id}/code") +async def device_code(device_id: str): + d = await tenants.ensure_device(device_id) + code = _gen_code() + expires = time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(time.time() + 600)) + await tenants.set_device_code(device_id, code, expires) + return {"ok": True, "code": code, "expires": expires} + + +@router.post("/tenants/{tid}/screens/bind-by-code") +async def bind_screen_by_code(tid: str, body: BindByCodeBody): + """园区端输入/扫码绑定：校验码 → 绑定到该园区 → 签发设备 token + MQTT 通知大屏。""" + d = await tenants.find_device_by_code(body.code.strip()) + if d is None: + return JSONResponse({"ok": False, "error": "连接码无效或已过期"}, status_code=404) + dev_id = d["device_id"] + if d.get("tenant_id") and d["tenant_id"] != tid: + return JSONResponse({"ok": False, "error": "该大屏已绑定其它园区"}, status_code=409) + screen = await tenants.bind_device(tid, dev_id, body.name or "", body.role or "main", body.location or "") + token = create_device_token(tid, dev_id) + tenant = await tenants.get_tenant(tid) + _publish_bind(dev_id, {"event": "bound", "token": token, "tenant_id": tid, + "name": (tenant or {}).get("name", ""), "intro": (tenant or {}).get("intro", [])}) + return {"ok": True, "screen": screen, "token": token} + + +@router.post("/api/devices/{device_id}/unbind") +async def device_unbind(device_id: str, authorization: str | None = Header(None), tenant_id: str | None = None): + await _resolve_tenant(authorization, tenant_id) + ok = await tenants.unbind_device(device_id) + return {"ok": ok, "device_id": device_id} + + # ==================== 入驻企业（租户 data.companies 主数据源） ==================== @router.get("/api/park/companies") diff --git a/app/park/tenants.py b/app/park/tenants.py index 2e6821c..2627a70 100644 --- a/app/park/tenants.py +++ b/app/park/tenants.py @@ -146,7 +146,8 @@ def _kb(r) -> dict: def _sc(r) -> dict: - return {"id": r.id, "device_id": r.device_id, "name": r.name, "role": r.role, "location": r.location, "created_at": r.created_at} + return {"id": r.id, "tenant_id": r.tenant_id, "device_id": r.device_id, "name": r.name, "role": r.role, + "location": r.location, "status": r.status, "code": r.code, "created_at": r.created_at} def _summary(tid: str, name: str, intro: list, username: str, admin_username: str = "", status: str = "active") -> dict: @@ -407,6 +408,77 @@ async def delete_screen(tenant_id: str, sid: str) -> bool: return r.rowcount > 0 +# ---------------- 设备绑定（大屏 → 园区） ---------------- + +async def get_device_by_id(device_id: str) -> dict | None: + async with _get_session() as s: + row = (await s.execute(select(ParkScreen).where(ParkScreen.device_id == device_id).limit(1))).scalars().first() + return _sc(row) if row else None + + +async def ensure_device(device_id: str) -> dict: + """确保存在该设备记录（未绑定）。""" + existing = await get_device_by_id(device_id) + if existing: + return existing + async with _get_session() as s: + s.add(ParkScreen(id=_new_id("SCR"), device_id=device_id, status="unbound", tenant_id=None, + name="", role="main", location="", created_at=_now())) + await s.commit() + return await get_device_by_id(device_id) + + +async def set_device_code(device_id: str, code: str, expires: str) -> dict: + async with _get_session() as s: + row = (await s.execute(select(ParkScreen).where(ParkScreen.device_id == device_id).limit(1))).scalars().first() + if row is None: + return None + row.code = code + row.code_expires = expires + row.status = "unbound" + row.tenant_id = None + await s.commit() + await s.refresh(row) + return _sc(row) + + +async def find_device_by_code(code: str) -> dict | None: + async with _get_session() as s: + row = (await s.execute(select(ParkScreen).where(ParkScreen.code == code, ParkScreen.status == "unbound").limit(1))).scalars().first() + return _sc(row) if row else None + + +async def bind_device(tenant_id: str, device_id: str, name: str = "", role: str = "main", location: str = "") -> dict | None: + async with _get_session() as s: + row = (await s.execute(select(ParkScreen).where(ParkScreen.device_id == device_id).limit(1))).scalars().first() + if row is None: + return None + row.tenant_id = tenant_id + row.status = "bound" + row.name = name or row.name or "未命名屏" + row.role = role + row.location = location + row.bound_at = _now() + row.code = "" + row.code_expires = "" + await s.commit() + await s.refresh(row) + return _sc(row) + + +async def unbind_device(device_id: str) -> bool: + async with _get_session() as s: + row = (await s.execute(select(ParkScreen).where(ParkScreen.device_id == device_id).limit(1))).scalars().first() + if row is None: + return False + row.tenant_id = None + row.status = "unbound" + row.code = "" + row.bound_at = "" + await s.commit() + return True + + # ---------------- 大屏数据 ---------------- async def screen_view(tenant_id: str) -> dict: