"""Tuya Cloud Gateway routes. URL structure: GET /tuya/ – gateway overview GET /tuya//auth – auth settings page POST /tuya//auth/qr – (AJAX) generate QR code GET /tuya//auth/poll – (AJAX) poll login result POST /tuya//auth/save – (AJAX) persist tokens POST /tuya//sync – sync devices from cloud GET /tuya//device/ – device detail POST /tuya//device//dp//toggle – toggle a DP POST /tuya//device//rename – rename a device """ from flask import ( Blueprint, abort, flash, jsonify, redirect, render_template, request, url_for, ) from flask_login import current_user, login_required from app import db, socketio from app.models.board import Board from app.models.tuya_device import TuyaDevice from app.drivers.registry import registry from app.drivers.tuya_cloud.driver import ( TUYA_CLIENT_ID, TUYA_SCHEMA, category_kind, KIND_ICON, ) tuya_bp = Blueprint("tuya", __name__, url_prefix="/tuya") # ── Helpers ─────────────────────────────────────────────────────────────────── def _get_gateway(board_id: int) -> Board: board = db.get_or_404(Board, board_id) if board.board_type != "tuya_cloud": abort(404) return board def _get_driver(board: Board): drv = registry.get("tuya_cloud") if drv is None: abort(500) return drv # ── Gateway overview ────────────────────────────────────────────────────────── @tuya_bp.route("/") @login_required def gateway(board_id: int): board = _get_gateway(board_id) devices = ( TuyaDevice.query.filter_by(board_id=board_id) .order_by(TuyaDevice.name) .all() ) has_token = bool(board.config.get("tuya_token_info")) return render_template( "tuya/gateway.html", board=board, devices=devices, has_token=has_token, ) # ── Auth settings ───────────────────────────────────────────────────────────── @tuya_bp.route("//auth") @login_required def auth_settings(board_id: int): if not current_user.is_admin(): abort(403) board = _get_gateway(board_id) return render_template("tuya/auth_settings.html", board=board) @tuya_bp.route("//auth/qr", methods=["POST"]) @login_required def generate_qr(board_id: int): """AJAX: generate a new QR-code token for this board. The caller must supply a ``user_code`` in the JSON body. This is a meaningful identifier chosen by the user (e.g. a short string they recognise) that Tuya uses to link the scan to their account. It must **not** be auto-generated – Tuya validates it server-side. """ if not current_user.is_admin(): abort(403) _get_gateway(board_id) data = request.get_json(silent=True) or {} user_code = data.get("user_code", "").strip() if not user_code: return jsonify({"ok": False, "error": "A user code is required to generate the QR code."}), 400 try: from tuya_sharing import LoginControl lc = LoginControl() response = lc.qr_code(TUYA_CLIENT_ID, TUYA_SCHEMA, user_code) if not response.get("success"): return jsonify({"ok": False, "error": response.get("msg", "QR generation failed")}), 400 qr_token = response["result"]["qrcode"] # The URI that Smart Life / Tuya Smart app decodes from the QR: qr_data = f"tuyaSmart--qrLogin?token={qr_token}" return jsonify({ "ok": True, "qr_data": qr_data, "qr_token": qr_token, "user_code": user_code, }) except Exception as exc: return jsonify({"ok": False, "error": str(exc)}), 500 @tuya_bp.route("//auth/poll") @login_required def poll_login(board_id: int): """AJAX: check whether the user has scanned the QR code yet.""" qr_token = request.args.get("token", "") user_code = request.args.get("user_code", "") if not qr_token or not user_code: return jsonify({"ok": False, "error": "Missing token or user_code"}), 400 try: from tuya_sharing import LoginControl lc = LoginControl() ret, info = lc.login_result(qr_token, TUYA_CLIENT_ID, user_code) if ret: return jsonify({"ok": True, "info": info}) return jsonify({"ok": False, "pending": True}) except Exception as exc: return jsonify({"ok": False, "error": str(exc)}), 500 @tuya_bp.route("//auth/save", methods=["POST"]) @login_required def save_auth(board_id: int): """AJAX: persist tokens after successful QR scan.""" if not current_user.is_admin(): abort(403) board = _get_gateway(board_id) data = request.get_json(silent=True) or {} info = data.get("info", {}) user_code = data.get("user_code", "") if not info or not user_code: return jsonify({"ok": False, "error": "Missing data"}), 400 cfg = board.config # Store the token blob (t, uid, expire_time, access_token, refresh_token) cfg["tuya_token_info"] = { "t": info.get("t", 0), "uid": info.get("uid", ""), "expire_time": info.get("expire_time", 7776000), "access_token": info.get("access_token", ""), "refresh_token": info.get("refresh_token", ""), } cfg["tuya_user_code"] = user_code # terminal_id and endpoint are returned by Tuya in the login_result info dict cfg["tuya_terminal_id"] = info.get("terminal_id", "") cfg["tuya_endpoint"] = info.get("endpoint", "https://apigw.iotbing.com") board.config = cfg db.session.commit() return jsonify({"ok": True, "redirect": url_for("tuya.gateway", board_id=board_id)}) # ── Sync devices ────────────────────────────────────────────────────────────── @tuya_bp.route("//sync", methods=["POST"]) @login_required def sync_devices(board_id: int): if not current_user.is_admin(): abort(403) board = _get_gateway(board_id) drv = _get_driver(board) try: n = drv.sync_devices(board) db.session.commit() flash(f"✓ Synced {n} Tuya device(s).", "success") except Exception as exc: flash(f"Sync failed: {exc}", "danger") return redirect(url_for("tuya.gateway", board_id=board_id)) # ── Device detail ───────────────────────────────────────────────────────────── @tuya_bp.route("//device/") @login_required def device_detail(board_id: int, device_id: str): board = _get_gateway(board_id) device = TuyaDevice.query.filter_by( board_id=board_id, device_id=device_id ).first_or_404() return render_template("tuya/device.html", board=board, device=device) # ── Toggle ──────────────────────────────────────────────────────────────────── @tuya_bp.route("//device//dp//toggle", methods=["POST"]) @login_required def toggle_dp(board_id: int, device_id: str, dp_code: str): board = _get_gateway(board_id) device = TuyaDevice.query.filter_by( board_id=board_id, device_id=device_id ).first_or_404() drv = _get_driver(board) ok = drv.toggle_dp(board, device_id, dp_code) new_state = device.status.get(dp_code, False) # Broadcast to all connected clients socketio.emit("tuya_update", { "board_id": board_id, "device_id": device_id, "dp_code": dp_code, "state": new_state, }) if request.headers.get("Accept") == "application/json" or \ request.headers.get("X-Requested-With") == "XMLHttpRequest": return jsonify({"ok": ok, "state": new_state}) return redirect(url_for("tuya.device_detail", board_id=board_id, device_id=device_id)) # ── Rename ──────────────────────────────────────────────────────────────────── @tuya_bp.route("//device//rename", methods=["POST"]) @login_required def rename_device(board_id: int, device_id: str): if not current_user.is_admin(): abort(403) board = _get_gateway(board_id) device = TuyaDevice.query.filter_by( board_id=board_id, device_id=device_id ).first_or_404() new_name = request.form.get("name", "").strip() if new_name: device.name = new_name db.session.commit() flash(f"Device renamed to '{new_name}'.", "success") return redirect(url_for("tuya.device_detail", board_id=board_id, device_id=device_id))