# -*- coding: utf-8 -*- from __future__ import print_function from flask import Flask, request, jsonify import sqlite3, os, time # ====== (참고용) 환경 변수 ====== # 자동 갱신을 제거했으므로 아래 값들은 사용하지 않습니다. AQARA_DOMAIN = os.getenv("AQARA_DOMAIN", "https://open-kr.aqara.com/v3.0/open/api") AQARA_APP_ID = os.getenv("AQARA_APP_ID", "13947123405048053765eeb5") AQARA_KEY_ID = os.getenv("AQARA_KEY_ID", "K.1394712340626440194") AQARA_APP_KEY = os.getenv("AQARA_APP_KEY", "fk8o1bjhq60s1kgy9kicqmkgivuus4yi") DB_PATH = os.path.join(os.path.dirname(__file__), "tokens.db") app = Flask(__name__) # ===================== 유틸 ===================== def now_ts(): return int(time.time()) # ===================== DB 초기화/액세스 ===================== def init_db(): """ 기본 테이블 생성 + '마지막 변경 기기' 메타 컬럼 보강 """ with sqlite3.connect(DB_PATH) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS user_tokens ( user TEXT PRIMARY KEY, authtoken TEXT, refreshtoken TEXT, updated_at INTEGER ) """) # 컬럼 존재 확인 후 없다면 추가 (구버전과 호환) def ensure_col(col, type_): cur = conn.execute("PRAGMA table_info(user_tokens)") cols = [r[1] for r in cur.fetchall()] if col not in cols: sql = "ALTER TABLE user_tokens ADD COLUMN {} {}".format(col, type_) conn.execute(sql) ensure_col("last_updater_device_id", "TEXT") ensure_col("last_updater_model", "TEXT") ensure_col("last_updater_platform", "TEXT") ensure_col("last_updater_os_version", "TEXT") ensure_col("last_updater_app_version", "TEXT") ensure_col("last_updater_server_host", "TEXT") ensure_col("last_updater_ip", "TEXT") init_db() def upsert_tokens(user, authtoken, refreshtoken): """단일 슬롯 upsert""" with sqlite3.connect(DB_PATH) as conn: cur = conn.execute( "UPDATE user_tokens SET authtoken=?, refreshtoken=?, updated_at=? WHERE user=?", (authtoken, refreshtoken, now_ts(), user) ) if cur.rowcount == 0: conn.execute( "INSERT INTO user_tokens(user, authtoken, refreshtoken, updated_at) VALUES (?,?,?,?)", (user, authtoken, refreshtoken, now_ts()) ) def get_tokens_row(user): with sqlite3.connect(DB_PATH) as conn: cur = conn.execute( "SELECT authtoken, refreshtoken, updated_at FROM user_tokens WHERE user=?", (user,) ) row = cur.fetchone() if not row: return None return {"authtoken": row[0], "refreshtoken": row[1], "updated_at": int(row[2] or 0)} def update_last_updater_meta(user, device_id, device_model, device_platform, os_version, app_version, server_host, ip_addr): with sqlite3.connect(DB_PATH) as conn: conn.execute(""" UPDATE user_tokens SET last_updater_device_id=?, last_updater_model=?, last_updater_platform=?, last_updater_os_version=?, last_updater_app_version=?, last_updater_server_host=?, last_updater_ip=? WHERE user=? """, (device_id or None, device_model, device_platform, os_version, app_version, server_host, ip_addr, user)) def read_last_updater_meta(user): with sqlite3.connect(DB_PATH) as conn: try: cur = conn.execute(""" SELECT last_updater_device_id, last_updater_model, last_updater_platform, last_updater_os_version, last_updater_app_version, last_updater_server_host, last_updater_ip FROM user_tokens WHERE user=? """, (user,)) row = cur.fetchone() except sqlite3.OperationalError: row = None if not row: return { "device_id": None, "model": None, "platform": None, "os_version": None, "app_version": None, "server_host": None, "ip": None } return { "device_id": row[0], "model": row[1], "platform": row[2], "os_version": row[3], "app_version": row[4], "server_host": row[5], "ip": row[6] } # ===================== 라우트 ===================== # 헬스체크 @app.route("/", methods=["GET"]) def index(): return "tokens service alive" # --- 저장 엔드포인트 (오타 포함 + /tokens 접두 포함 모두 지원) --- @app.route("/setToekn", methods=["GET"]) @app.route("/setToken", methods=["GET"]) @app.route("/tokens/setToekn", methods=["GET"]) @app.route("/tokens/setToken", methods=["GET"]) def set_tokens(): """ 필수: user, authtoken, refreshtoken 선택(권장): device_id, device_model, device_platform, os_version, app_version, server_host 동작: 토큰 upsert + '마지막 변경 기기' 메타 갱신 """ user = (request.args.get("user") or "").strip() authtoken = request.args.get("authtoken") refreshtoken = request.args.get("refreshtoken") # 선택 메타 (마지막 변경 기기 기록용) device_id = (request.args.get("device_id") or "").strip() device_model = request.args.get("device_model") device_platform = request.args.get("device_platform") os_version = request.args.get("os_version") app_version = request.args.get("app_version") server_host = request.args.get("server_host") last_ip = request.headers.get("X-Forwarded-For", request.remote_addr) if not user: return jsonify({"ok": False, "error": "user required"}), 400 if authtoken is None or refreshtoken is None: return jsonify({"ok": False, "error": "authtoken and refreshtoken required"}), 400 # 1) 토큰 저장/갱신 upsert_tokens(user, authtoken, refreshtoken) # 2) 마지막 변경 기기 정보 기록 update_last_updater_meta(user, device_id, device_model, device_platform, os_version, app_version, server_host, last_ip) return jsonify({ "ok": True, "user": user, "authtoken": authtoken, "refreshtoken": refreshtoken, "last_updater": { "device_id": device_id or None, "model": device_model, "platform": device_platform, "os_version": os_version, "app_version": app_version, "server_host": server_host, "ip": last_ip } }) # --- 조회 엔드포인트 (/tokens 접두 포함/미포함 모두 지원) --- @app.route("/getToken", methods=["GET"]) @app.route("/tokens/getToken", methods=["GET"]) def get_token(): """ 입력: user (필수) 출력: ok, authtoken, refreshtoken, last_updater{...} 주의: 자동 갱신 없음 — DB에 저장된 값 그대로 반환 """ user = (request.args.get("user") or "").strip() if not user: return jsonify({"ok": False, "error": "user required"}), 400 row = get_tokens_row(user) if not row: return jsonify({"ok": False, "error": "not found"}), 404 at, rt = row["authtoken"], row["refreshtoken"] last_updater = read_last_updater_meta(user) return jsonify({ "ok": True, "user": user, "authtoken": at, "refreshtoken": rt, "last_updater": last_updater }) # ========= 로컬 실행용 진입점 ========= if __name__ == "__main__": # 개발 편의를 위해 모든 인터페이스 바인딩 (운영은 리버스 프록시 권장) app.run(host="0.0.0.0", port=int(os.getenv("PORT", "9022")))