feat(park): 大屏设备绑定流 — 永久设备ID+连接码+园区端绑定+MQTT下发设备token

- 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/<dev> 下发 token)、
  /park/api/devices/{device_id}/unbind。_resolve_tenant 支持设备 token。
- 验证(TestClient)：register→code→建园区→绑码(bound/命名)→设备token取数据返回绑定园区。

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)

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="")

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"]}

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

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: