# 파일 /mnt/d/_St/AIWK_PC/aiwk_mypc_agent.py # 버전 v126-mypc-watch-ws-request-summary-v001 import argparse import datetime import json import os import hashlib import secrets import platform import re import socket import subprocess import sys import time import uuid import urllib.request import base64 import struct import threading import traceback VERSION = "v126-mypc-watch-ws-request-summary-v001" DEFAULT_CONFIG_NAME = "aiwk_mypc_config.json" ZERO_ID_RE = re.compile(r"(?:^|_)0{8,}(?:$|_)") def now_iso(): return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") def script_dir(): return os.path.dirname(os.path.abspath(__file__)) def die(code, message, extra=None): out = { "ok": 0, "version": VERSION, "error_code": code, "message": message, "extra": extra or {}, "created_at": now_iso() } print(json.dumps(out, ensure_ascii=False, indent=2)) sys.exit(1) def is_state_file(path): name = os.path.basename(str(path or "")).lower() return "state" in name and name.endswith(".json") def backup_text_file(path, suffix): try: ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") backup_path = path + "." + suffix + "." + ts with open(path, "r", encoding="utf-8", errors="replace") as src: raw = src.read() with open(backup_path, "w", encoding="utf-8") as dst: dst.write(raw) return backup_path except Exception: return "" def recover_json_object_from_extra_data(path): # v123: state 파일 끝에 }5" 같은 찌꺼기가 붙은 경우 첫 번째 JSON object만 살려서 복구한다. with open(path, "r", encoding="utf-8", errors="replace") as f: raw = f.read() start = raw.find("{") if start < 0: raise ValueError("JSON_OBJECT_START_NOT_FOUND") decoder = json.JSONDecoder() obj, end = decoder.raw_decode(raw[start:]) if not isinstance(obj, dict): raise ValueError("RECOVERED_JSON_NOT_OBJECT") trailing = raw[start + end:].strip() backup_path = backup_text_file(path, "corrupt") obj["_recovered_from_extra_data"] = { "ok": 1, "at": now_iso(), "backup_path": backup_path, "trailing_preview": trailing[:80] } write_json_file(path, obj) return obj def read_json_file(path, required=True): if not os.path.isfile(path): if required: die("AIWK_MYPC_CONFIG_MISSING", "mypc 설정 파일이 없습니다.", { "required_path": path, "how_to_fix": "aiwk_mypc_config.example.json을 aiwk_mypc_config.json으로 복사한 뒤 device_label 등을 확인하십시오." }) return None try: with open(path, "r", encoding="utf-8") as f: return json.load(f) except json.JSONDecodeError as e: if is_state_file(path) and "Extra data" in str(e): try: return recover_json_object_from_extra_data(path) except Exception as re_err: die("AIWK_MYPC_STATE_RECOVER_FAILED", "state JSON 자동 복구에 실패했습니다.", { "path": path, "error": str(e), "recover_error": str(re_err) }) code = "AIWK_MYPC_STATE_READ_FAILED" if is_state_file(path) else "AIWK_MYPC_CONFIG_READ_FAILED" msg = "state 파일을 읽지 못했습니다." if is_state_file(path) else "mypc 설정 파일을 읽지 못했습니다." die(code, msg, {"path": path, "error": str(e)}) except Exception as e: code = "AIWK_MYPC_STATE_READ_FAILED" if is_state_file(path) else "AIWK_MYPC_CONFIG_READ_FAILED" msg = "state 파일을 읽지 못했습니다." if is_state_file(path) else "mypc 설정 파일을 읽지 못했습니다." die(code, msg, {"path": path, "error": str(e)}) def write_json_file(path, data): parent = os.path.dirname(os.path.abspath(path)) if parent and not os.path.isdir(parent): die("AIWK_MYPC_STATE_DIR_MISSING", "state 저장 폴더가 없습니다.", { "path": path, "dir": parent }) # v123: 여러 local_ws 요청/감시 루프가 동시에 state를 쓰더라도 같은 .tmp를 공유하지 않게 한다. tmp_path = path + "." + str(os.getpid()) + "." + str(threading.get_ident()) + ".tmp" try: with open(tmp_path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) f.write("\n") try: f.flush() os.fsync(f.fileno()) except Exception: pass os.replace(tmp_path, path) except Exception as e: try: if os.path.exists(tmp_path): os.remove(tmp_path) except Exception: pass die("AIWK_MYPC_STATE_WRITE_FAILED", "state 파일 저장에 실패했습니다.", { "path": path, "error": str(e) }) def resolve_path(base_dir, path): if not isinstance(path, str) or not path.strip(): die("AIWK_MYPC_PATH_EMPTY", "필수 경로 설정값이 비어 있습니다.", {"value": path}) if os.path.isabs(path): return path return os.path.abspath(os.path.join(base_dir, path)) def require_config(cfg, key): if key not in cfg: die("AIWK_MYPC_CONFIG_KEY_MISSING", "mypc 필수 설정값이 없습니다.", {"key": key}) value = cfg.get(key) if value is None: die("AIWK_MYPC_CONFIG_KEY_EMPTY", "mypc 필수 설정값이 비어 있습니다.", {"key": key}) if isinstance(value, str) and value.strip() == "": die("AIWK_MYPC_CONFIG_KEY_EMPTY", "mypc 필수 설정값이 비어 있습니다.", {"key": key}) return value def get_system_hostname(): name = platform.node() or socket.gethostname() name = str(name).strip() if not name: die("AIWK_MYPC_HOSTNAME_EMPTY", "운영체제 컴퓨터명을 읽지 못했습니다.", { "how_to_fix": "aiwk_mypc_config.json의 device_label에 표시용 이름을 직접 넣어도, system_hostname은 읽혀야 합니다." }) return name def get_device_label(cfg, system_hostname): raw = cfg.get("device_label") if raw is None: raw = cfg.get("device_name", "__HOSTNAME__") raw = str(raw).strip() if raw in ("", "__HOSTNAME__", "__AUTO__"): return system_hostname return raw def safe_id_part(value): safe = re.sub(r"[^0-9A-Za-z가-힣_.-]+", "_", str(value)).strip("_") if not safe: die("AIWK_MYPC_DEVICE_LABEL_INVALID", "device_label로 device_instance_id를 만들 수 없습니다.", {"device_label": value}) return safe def make_device_instance_id(device_label): return "dev_" + safe_id_part(device_label) + "_" + uuid.uuid4().hex[:12] def make_device_secret(): return secrets.token_hex(32) def mask_secret(value): s = str(value or "") if not s: return "" if len(s) <= 12: return "***" return s[:6] + "...." + s[-6:] def make_device_auth_hash(device_instance_id, device_secret): raw = str(device_instance_id) + ":" + str(device_secret) return hashlib.sha256(raw.encode("utf-8")).hexdigest() def build_aiwk_address(room_id, group, project, client_id, endpoint, url5=""): route = str(client_id) if str(url5 or "").strip(): route += ":" + str(url5).strip() return str(room_id) + "@" + str(group) + "." + str(project) + "." + route + "#" + str(endpoint) def build_legacy_address(group, project, client_id, endpoint, url5=""): route = str(client_id) if str(url5 or "").strip(): route += ":" + str(url5).strip() return str(group) + "." + str(project) + "." + route + "#" + str(endpoint) def is_placeholder_device_id(device_instance_id): s = str(device_instance_id or "") if not s: return True if "000000000000" in s: return True if ZERO_ID_RE.search(s): return True return False def load_or_create_state(state_path, device_label, system_hostname): created_new = False migrated_placeholder = False if os.path.isfile(state_path): state = read_json_file(state_path, required=True) if not isinstance(state, dict): die("AIWK_MYPC_STATE_INVALID", "state 파일 JSON 구조가 object가 아닙니다.", {"path": state_path}) else: state = { "version": VERSION, "device_label": device_label, "device_name": device_label, "system_hostname": system_hostname, "device_instance_id": make_device_instance_id(device_label), "device_secret": make_device_secret(), "created_at": now_iso(), "updated_at": now_iso(), "recent_logs": [] } created_new = True if is_placeholder_device_id(state.get("device_instance_id")): state["device_instance_id_prev"] = state.get("device_instance_id", "") state["device_instance_id"] = make_device_instance_id(device_label) state["device_instance_id_migrated_at"] = now_iso() migrated_placeholder = True if not state.get("device_secret"): state["device_secret"] = make_device_secret() state["device_secret_created_at"] = now_iso() state["version"] = VERSION state["device_label"] = device_label state["device_name"] = device_label state["system_hostname"] = system_hostname state["updated_at"] = now_iso() if created_new or migrated_placeholder: write_json_file(state_path, state) return state, created_new, migrated_placeholder def append_log(state, line, keep_lines): logs = state.get("recent_logs") if not isinstance(logs, list): logs = [] logs.append(now_iso() + " " + line) state["recent_logs"] = logs[-keep_lines:] state["updated_at"] = now_iso() def fetch_remote_ip(url, timeout_sec): result = { "ok": 0, "remote_ip": "", "url": url, "error": "" } try: req = urllib.request.Request(url, headers={"User-Agent": "AIWK-MYPC/" + VERSION}) with urllib.request.urlopen(req, timeout=timeout_sec) as res: body = res.read(4096).decode("utf-8", errors="replace").strip() m = re.search(r"\b(?:\d{1,3}\.){3}\d{1,3}\b", body) if m: result["ok"] = 1 result["remote_ip"] = m.group(0) return result m6 = re.search(r"\b[0-9A-Fa-f:]{3,}\b", body) if m6: result["ok"] = 1 result["remote_ip"] = m6.group(0) return result result["error"] = "IP_PATTERN_NOT_FOUND" result["body_preview"] = body[:200] return result except Exception as e: result["error"] = str(e) return result def check_tcp(host, port, timeout_sec): started = time.time() try: with socket.create_connection((host, int(port)), timeout=timeout_sec): return { "ok": 1, "host": host, "port": int(port), "latency_ms": int((time.time() - started) * 1000), "error": "" } except Exception as e: return { "ok": 0, "host": host, "port": int(port), "latency_ms": int((time.time() - started) * 1000), "error": str(e) } def run_command(args, timeout_sec): try: p = subprocess.run(args, capture_output=True, text=True, timeout=timeout_sec, errors="replace") return p.returncode, (p.stdout or "") + "\n" + (p.stderr or "") except Exception as e: return 999, str(e) def check_chrome_process(process_names): names = [str(x).lower() for x in process_names if str(x).strip()] if not names: die("AIWK_MYPC_CHROME_PROCESS_NAMES_EMPTY", "chrome_process_names가 비어 있습니다.", {}) system = platform.system().lower() if "windows" in system: rc, out = run_command(["tasklist"], 5) else: rc, out = run_command(["ps", "-eo", "pid,comm,args"], 5) out_lower = out.lower() found = [] for name in names: if name in out_lower: found.append(name) return { "ok": 1 if found else 0, "process_names": process_names, "found": sorted(list(set(found))), "check_method": "tasklist" if "windows" in system else "ps", "command_rc": rc } def build_status(cfg, state, cfg_path, state_path): timeout_sec = float(cfg.get("timeout_sec", 3)) keep_lines = int(cfg.get("log_keep_lines", 20)) group = str(require_config(cfg, "group")).strip() project = str(require_config(cfg, "project")).strip() room_id = str(require_config(cfg, "room_id")).strip() client_id = str(require_config(cfg, "client_id")).strip() endpoint = str(cfg.get("endpoint") or "aiwk_pc").strip() url5 = str(cfg.get("url5") or cfg.get("route_key") or "pc001").strip() tab_kind = "pc-agent" if not group or not project or not room_id or not client_id: die("AIWK_MYPC_ROOT_CONTEXT_EMPTY", "mypc 루트 컨텍스트(group/project/room_id/client_id)가 비어 있습니다.", { "group": group, "project": project, "room_id": room_id, "client_id": client_id }) ip = fetch_remote_ip(require_config(cfg, "remote_ip_url"), timeout_sec) node_red = check_tcp(require_config(cfg, "node_red_host"), int(require_config(cfg, "node_red_port")), timeout_sec) chrome = check_chrome_process(require_config(cfg, "chrome_process_names")) if ip.get("ok"): state["remote_ip"] = ip.get("remote_ip", "") state["remote_ip_checked_at"] = now_iso() append_log(state, "remote_ip ok " + state["remote_ip"], keep_lines) else: append_log(state, "remote_ip fail " + str(ip.get("error", "")), keep_lines) append_log(state, "node_red " + ("ok" if node_red.get("ok") else "fail") + " " + str(node_red.get("host")) + ":" + str(node_red.get("port")), keep_lines) append_log(state, "chrome " + ("ok" if chrome.get("ok") else "not_found"), keep_lines) write_json_file(state_path, state) return { "ok": 1, "version": VERSION, "created_at": now_iso(), "config_path": cfg_path, "state_path": state_path, "identity": { "identity_kind": "AIWK_ROOT_DEVICE", "root_identity": 1, "device_label": state.get("device_label"), "device_name": state.get("device_name"), "system_hostname": state.get("system_hostname"), "device_instance_id": state.get("device_instance_id"), "device_auth_hash": make_device_auth_hash(state.get("device_instance_id", ""), state.get("device_secret", "")), "device_secret_masked": mask_secret(state.get("device_secret", "")), "remote_ip": state.get("remote_ip", ""), "group": group, "project": project, "room_id": room_id, "default_group": group, "default_project": project, "default_room_id": room_id, "client_id": client_id, "url5": url5, "route_key": url5, "endpoint": endpoint, "tab_kind": tab_kind, "tab_id": "vtab_aiwk_pc_" + url5, "profile_instance_id": "pf_mypc_" + state.get("device_instance_id", ""), "address": build_aiwk_address(room_id, group, project, client_id, endpoint, url5), "legacy_address": build_legacy_address(group, project, client_id, endpoint, url5) }, "checks": { "identity_guard": { "ok": 1, "role": "root-device", "device_secret_present": 1 if state.get("device_secret") else 0, "token_present": 1 if str(cfg.get("token") or cfg.get("auth_token") or "").strip() else 0, "root_key": build_aiwk_address(room_id, group, project, client_id, endpoint, url5), "unique_key": str(state.get("device_instance_id")) + "/" + client_id + ":" + url5 }, "remote_ip": ip, "node_red": node_red, "chrome": chrome }, "recent_logs": state.get("recent_logs", [])[-keep_lines:] } def chrome_user_data_dir(): local = os.environ.get("LOCALAPPDATA", "") if local: return os.path.join(local, "Google", "Chrome", "User Data") home = os.path.expanduser("~") return os.path.join(home, "AppData", "Local", "Google", "Chrome", "User Data") def list_chrome_profiles(cfg=None): cfg = cfg or {} ext4 = get_ext4_cfg(cfg) if isinstance(cfg, dict) else {} # 목록은 실제 실행에 쓰는 user_data_dir 기준이어야 혼란이 없다. # 우선순위: chrome_user_data_dir 명시값 > ext4.user_data_dir > OS 기본 Chrome User Data base = str(cfg.get("chrome_user_data_dir") or ext4.get("user_data_dir") or chrome_user_data_dir()).strip() out = {"ok": 0, "chrome_user_data": base, "profiles": [], "error": ""} if not base or not os.path.isdir(base): out["error"] = "CHROME_USER_DATA_NOT_FOUND" return out info_cache = {} local_state_path = os.path.join(base, "Local State") if os.path.isfile(local_state_path): try: with open(local_state_path, "r", encoding="utf-8") as f: local_state = json.load(f) info_cache = ((local_state.get("profile") or {}).get("info_cache") or {}) except Exception as e: out["local_state_error"] = str(e) candidates = [] for name in os.listdir(base): p = os.path.join(base, name) if not os.path.isdir(p): continue if name == "Default" or name.startswith("Profile "): candidates.append(name) def sort_key(x): if x == "Default": return (0, 0) m = re.search(r"(\d+)$", x) return (1, int(m.group(1)) if m else 9999, x) for name in sorted(candidates, key=sort_key): meta = info_cache.get(name) or {} ext_settings = os.path.join(base, name, "Local Extension Settings") out["profiles"].append({ "profile_dir": name, "profile_name": meta.get("name") or name, "gaia_name": meta.get("gaia_name") or "", "user_name": meta.get("user_name") or "", "path": os.path.join(base, name), "local_extension_settings": ext_settings, "local_extension_settings_exists": 1 if os.path.isdir(ext_settings) else 0 }) out["ok"] = 1 out["count"] = len(out["profiles"]) return out def get_ext4_cfg(cfg): ext4 = cfg.get("ext4") if isinstance(cfg.get("ext4"), dict) else {} return ext4 def resolve_runtime_path(base_dir, path): s = str(path or "").strip() if not s: return "" # Windows absolute paths are not treated as absolute when this file is checked on Linux/WSL. if re.match(r"^[A-Za-z]:[\\/]", s) or s.startswith("\\\\"): return s if os.path.isabs(s): return s return os.path.abspath(os.path.join(base_dir, s)) def default_chrome_candidates(): return [ r"C:\Program Files\Google\Chrome\Application\chrome.exe", r"C:\Program Files (x86)\Google\Chrome\Application\chrome.exe", "chrome.exe", "chrome" ] def resolve_chrome_path(cfg): ext4 = get_ext4_cfg(cfg) configured = str(ext4.get("chrome_path") or cfg.get("chrome_path") or "").strip() if configured: return configured for p in default_chrome_candidates(): if re.match(r"^[A-Za-z]:[\\/]", p): if os.path.isfile(p): return p else: return p return "chrome.exe" def check_ext4_dir(cfg, base_dir): ext4 = get_ext4_cfg(cfg) extension_dir = resolve_runtime_path(base_dir, ext4.get("extension_dir") or "") out = { "ok": 0, "extension_dir": extension_dir, "manifest_path": "", "exists": 0, "manifest_exists": 0, "manifest": {}, "error": "" } if not extension_dir: out["error"] = "EXT4_EXTENSION_DIR_EMPTY" return out out["exists"] = 1 if os.path.isdir(extension_dir) else 0 manifest_path = os.path.join(extension_dir, "manifest.json") out["manifest_path"] = manifest_path out["manifest_exists"] = 1 if os.path.isfile(manifest_path) else 0 if not out["exists"]: out["error"] = "EXT4_EXTENSION_DIR_NOT_FOUND" return out if not out["manifest_exists"]: out["error"] = "EXT4_MANIFEST_NOT_FOUND" return out try: with open(manifest_path, "r", encoding="utf-8") as f: manifest = json.load(f) out["manifest"] = { "name": manifest.get("name", ""), "version": manifest.get("version", ""), "manifest_version": manifest.get("manifest_version", ""), "description": manifest.get("description", "") } out["ok"] = 1 return out except Exception as e: out["error"] = "EXT4_MANIFEST_READ_FAILED: " + str(e) return out def build_chrome_load_ext4_args(cfg, base_dir, opts=None): opts = opts or {} ext4 = get_ext4_cfg(cfg) chrome_path = resolve_chrome_path(cfg) extension_dir = resolve_runtime_path(base_dir, ext4.get("extension_dir") or "") user_data_dir_raw = opts.get("user_data_dir") if opts.get("user_data_dir") is not None else ext4.get("user_data_dir") profile_directory_raw = opts.get("profile_directory") if opts.get("profile_directory") is not None else ext4.get("profile_directory") remote_debugging_port_raw = opts.get("remote_debugging_port") if opts.get("remote_debugging_port") is not None else ext4.get("remote_debugging_port") start_url_raw = opts.get("start_url") if opts.get("start_url") is not None else ext4.get("start_url") user_data_dir = resolve_runtime_path(base_dir, user_data_dir_raw or "") profile_directory = str(profile_directory_raw or "").strip() remote_debugging_port = str(remote_debugging_port_raw or "").strip() start_url = str(start_url_raw or "https://chatgpt.com/").strip() args = [chrome_path] if user_data_dir: args.append("--user-data-dir=" + user_data_dir) if profile_directory: args.append("--profile-directory=" + profile_directory) if remote_debugging_port: args.append("--remote-debugging-port=" + remote_debugging_port) args.append("--load-extension=" + extension_dir) if start_url: args.append(start_url) meta = { "user_data_dir": user_data_dir, "profile_directory": profile_directory, "remote_debugging_port": remote_debugging_port, "start_url": start_url, "extension_dir": extension_dir } return args, meta def chrome_load_ext4(cfg, base_dir, opts=None): opts = opts or {} check = check_ext4_dir(cfg, base_dir) args, meta = build_chrome_load_ext4_args(cfg, base_dir, opts) out = { "ok": 0, "act": "pc_chrome_load_ext4_result", "ext4": check, "chrome_path": args[0] if args else "", "chrome_launch": meta, "args_preview": args, "pid": 0, "error": "" } if not meta.get("profile_directory"): out["error"] = "CHROME_PROFILE_DIRECTORY_REQUIRED" out["message"] = "여러 Chrome 프로필 환경에서는 --profile-directory 를 지정해야 합니다. 예: --profile-directory \"Profile 3\" 또는 Default" out["profiles_hint"] = list_chrome_profiles(cfg) return out if not check.get("ok"): out["error"] = check.get("error") or "EXT4_CHECK_FAILED" return out chrome_path = args[0] if re.match(r"^[A-Za-z]:[\\/]", chrome_path) and not os.path.isfile(chrome_path): out["error"] = "CHROME_PATH_NOT_FOUND" return out try: p = subprocess.Popen(args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, close_fds=True) out["ok"] = 1 out["pid"] = int(p.pid) return out except Exception as e: out["error"] = str(e) return out def save_profile_bind(state_path, req, addr): data = req.get("data") if isinstance(req.get("data"), dict) else {} profile_id = str(data.get("profile_instance_id") or data.get("profileInstanceId") or req.get("profile_instance_id") or "").strip() client_id = str(data.get("client_id") or req.get("from_client") or "EXT").strip().upper() tab_id = str(data.get("tab_id") or data.get("chrome_tab_id") or "tab").strip() if not profile_id: profile_id = "pf_unknown" state = read_json_file(state_path, required=True) if os.path.isfile(state_path) else {} profile_map = state.get("profile_map") if isinstance(state.get("profile_map"), dict) else {} profile_map.setdefault(profile_id, {"profile_instance_id": profile_id, "tabs": {}, "created_at": now_iso()}) row = profile_map[profile_id] row["updated_at"] = now_iso() row["last_remote"] = str(addr) row["last_client_id"] = client_id row["tabs"][client_id + "/" + tab_id] = {"updated_at": now_iso(), "client_id": client_id, "tab_id": tab_id, "data": data} state["profile_map"] = profile_map state["updated_at"] = now_iso() write_json_file(state_path, state) return {"ok": 1, "profile_instance_id": profile_id, "client_id": client_id, "tab_id": tab_id, "saved": 1, "map_count": len(profile_map)} def cfg_bool(cfg, key, default=False): v = cfg.get(key, default) if isinstance(v, bool): return v if isinstance(v, (int, float)): return bool(v) return str(v).strip().lower() in ("1", "y", "yes", "true", "on") def get_wss_client_module(): try: import websocket # type: ignore return websocket except Exception as e: die("AIWK_MYPC_WSS_PACKAGE_MISSING", "WSS 등록에 필요한 websocket-client 패키지가 없습니다.", { "error": str(e), "how_to_fix": "\\opt\\venv\\Scripts\\python.exe -m pip install websocket-client" }) def build_register_packet(cfg, status): ident = status.get("identity", {}) client_id = str(ident.get("client_id") or cfg.get("client_id") or "PC1").strip() group = str(require_config(cfg, "group")).strip() project = str(require_config(cfg, "project")).strip() room_id = str(require_config(cfg, "room_id")).strip() token = str(cfg.get("token") or cfg.get("auth_token") or "").strip() endpoint = str(cfg.get("endpoint") or ident.get("endpoint") or "aiwk_pc").strip() url5 = str(ident.get("url5") or cfg.get("url5") or cfg.get("route_key") or "pc001").strip() address = build_aiwk_address(room_id, group, project, client_id, endpoint, url5) legacy_address = build_legacy_address(group, project, client_id, endpoint, url5) packet = { "type": "register", "role": str(cfg.get("role") or "PC").upper(), "identity_kind": "AIWK_ROOT_DEVICE", "root_identity": 1, "client_id": client_id, "from_client": client_id, "url5": url5, "from_url5": url5, "route_key": url5, "group": group, "project": project, "room_id": room_id, "default_group": group, "default_project": project, "default_room_id": room_id, "endpoint": endpoint, "from_endpoint": endpoint, "tab_kind": "pc-agent", "tab_id": ident.get("tab_id"), "from_tab_id": ident.get("tab_id"), "profile_instance_id": ident.get("profile_instance_id"), "device_label": ident.get("device_label"), "device_name": ident.get("device_name"), "system_hostname": ident.get("system_hostname"), "device_instance_id": ident.get("device_instance_id"), "device_auth_hash": ident.get("device_auth_hash"), "remote_ip": ident.get("remote_ip"), "address": address, "legacy_address": legacy_address, "endpoints": [address, legacy_address], "debug_title": "AIWK PC " + str(ident.get("device_label") or ""), "debug_url": endpoint + "://" + str(ident.get("device_instance_id") or ""), "tab": { "title": "AIWK PC " + str(ident.get("device_label") or ""), "url": endpoint + "://" + str(ident.get("device_instance_id") or ""), "tab_id": ident.get("tab_id"), "profile_instance_id": ident.get("profile_instance_id"), "endpoint": endpoint, "tab_kind": "pc-agent", "address": address } } if token: packet["token"] = token return packet def wss_register_once(cfg, status): if not cfg_bool(cfg, "wss_enabled", False): return {"ok": 0, "enabled": 0, "error": "WSS_DISABLED"} if cfg_bool(cfg, "wss_require_token", False) and not str(cfg.get("token") or cfg.get("auth_token") or "").strip(): die("AIWK_MYPC_WSS_TOKEN_MISSING", "wss_require_token=true 인데 token이 비어 있습니다.", { "how_to_fix": "aiwk_mypc_config.json에 token을 넣거나 테스트 목적이면 wss_require_token=false로 바꾸십시오." }) websocket = get_wss_client_module() url = str(require_config(cfg, "wss_url")).strip() timeout_sec = float(cfg.get("wss_timeout_sec", cfg.get("timeout_sec", 3))) packet = build_register_packet(cfg, status) result = { "ok": 0, "enabled": 1, "url": url, "registered": 0, "ack": None, "error": "" } ws = None try: ws = websocket.create_connection(url, timeout=timeout_sec) ws.send(json.dumps(packet, ensure_ascii=False)) raw = ws.recv() ack = json.loads(raw) result["ack"] = ack result["registered"] = 1 if ack.get("ok") else 0 result["ok"] = 1 if ack.get("ok") else 0 if not ack.get("ok"): result["error"] = str(ack.get("reason") or ack.get("error") or "REGISTER_FAILED") return result except Exception as e: result["error"] = str(e) return result finally: try: if ws is not None: ws.close() except Exception: pass def compact_text(value, max_len=120): if value is None: return "" text = str(value).replace("\r", " ").replace("\n", " ").strip() if len(text) > max_len: return text[:max_len - 3] + "..." return text def pick_nested(req, *names): if not isinstance(req, dict): return "" data = req.get("data") if isinstance(req.get("data"), dict) else {} for name in names: val = req.get(name) if val not in (None, ""): return val val = data.get(name) if val not in (None, ""): return val return "" def local_ws_request_summary(req, addr): if not isinstance(req, dict): return "REQ invalid-object from:%s" % (addr,) act = compact_text(req.get("act") or req.get("type") or "?") req_id = compact_text(req.get("id") or "-") client_id = compact_text(pick_nested(req, "client_id", "from_client") or "-") profile_dir = compact_text(pick_nested(req, "profile_directory", "profile_dir", "chrome_profile") or "-") aiwk_from = compact_text(pick_nested(req, "aiwk_from_url", "from_url", "from") or "-") aiwk_to = compact_text(pick_nested(req, "aiwk_to_url", "to_url", "to") or "-") return "REQ act:%s id:%s from:%s client:%s profile:%s aiwk_from:%s aiwk_to:%s" % (act, req_id, addr, client_id, profile_dir, aiwk_from, aiwk_to) def local_ws_result_summary(req, res, addr): if not isinstance(res, dict): return "RES invalid-result from:%s" % (addr,) req_act = "?" if isinstance(req, dict): req_act = compact_text(req.get("act") or req.get("type") or "?") res_act = compact_text(res.get("act") or "-") ok = res.get("ok") if ok is True: ok_text = "1" elif ok is False: ok_text = "0" elif ok in (0, 1): ok_text = str(ok) else: ok_text = "?" err = compact_text(res.get("error_code") or res.get("error") or "") extra = "" if "chrome" in res and isinstance(res.get("chrome"), dict): extra = " chrome_ok:%s" % (1 if res["chrome"].get("ok") else 0) elif "node_red" in res and isinstance(res.get("node_red"), dict): extra = " nr_ok:%s" % (1 if res["node_red"].get("ok") else 0) elif "ext4" in res and isinstance(res.get("ext4"), dict): extra = " ext4_ok:%s" % (1 if res["ext4"].get("ok") else 0) elif "profile_map" in res and isinstance(res.get("profile_map"), dict): extra = " profiles:%s" % len(res.get("profile_map") or {}) return "RES req:%s act:%s ok:%s err:%s%s" % (req_act, res_act, ok_text, err or "-", extra) def print_local_ws_event(line): # watch 모드에서 . 이 같은 줄에 누적 중이어도 WS 이벤트는 새 줄로 분리한다. print("\n[" + now_iso() + "] LOCAL_WS " + line, flush=True) class LocalWsBridgeServer: def __init__(self, cfg, cfg_path, state_path, log_path): self.cfg = cfg self.cfg_path = cfg_path self.state_path = state_path self.log_path = log_path lb = cfg.get("local_bridge") if isinstance(cfg.get("local_bridge"), dict) else {} self.host = str(lb.get("host", "127.0.0.1")).strip() or "127.0.0.1" self.port = int(lb.get("port", 8768)) self.token = str(lb.get("token") or lb.get("auth_token") or "").strip() self.sock = None self.thread = None self.stop_event = threading.Event() self.last_error = "" if not self.token: die("AIWK_LOCAL_BRIDGE_TOKEN_MISSING", "local_bridge.enabled=true 인데 token이 비어 있습니다.", { "how_to_fix": "aiwk_mypc_config.json의 local_bridge.token에 긴 랜덤 문자열을 넣으십시오.", "example": {"local_bridge": {"enabled": True, "host": "127.0.0.1", "port": 8768, "token": "CHANGE_ME_LONG_RANDOM"}} }) if self.host not in ("127.0.0.1", "localhost", "::1"): die("AIWK_LOCAL_BRIDGE_HOST_BLOCKED", "local_bridge.host는 개발 단계에서 localhost만 허용합니다.", { "host": self.host, "how_to_fix": "host를 127.0.0.1로 설정하십시오." }) @staticmethod def enabled_in_config(cfg): lb = cfg.get("local_bridge") if isinstance(cfg.get("local_bridge"), dict) else {} return cfg_bool(lb, "enabled", False) def start(self): if self.thread and self.thread.is_alive(): return {"ok": 1, "enabled": 1, "host": self.host, "port": self.port, "already": 1} self.thread = threading.Thread(target=self._serve, name="AIWKLocalWsBridge", daemon=True) self.thread.start() return {"ok": 1, "enabled": 1, "host": self.host, "port": self.port, "url": "ws://%s:%s" % (self.host, self.port)} def close(self): self.stop_event.set() try: if self.sock is not None: self.sock.close() except Exception: pass self.sock = None def _serve(self): try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind((self.host, self.port)) sock.listen(16) sock.settimeout(1.0) self.sock = sock print("[" + now_iso() + "] LOCAL_WS listening ws://" + self.host + ":" + str(self.port), flush=True) while not self.stop_event.is_set(): try: conn, addr = sock.accept() except socket.timeout: continue except OSError: break threading.Thread(target=self._client_thread, args=(conn, addr), daemon=True).start() except Exception as e: self.last_error = str(e) print(json.dumps({ "ok": 0, "version": VERSION, "error_code": "AIWK_LOCAL_WS_START_FAILED", "message": "local websocket bridge 시작 실패", "extra": {"host": self.host, "port": self.port, "error": str(e), "trace": traceback.format_exc()[:2000]}, "created_at": now_iso() }, ensure_ascii=False, indent=2), flush=True) def _client_thread(self, conn, addr): try: conn.settimeout(30) self._handshake(conn) while not self.stop_event.is_set(): msg = self._recv_text(conn) if msg is None: break req = None try: req = json.loads(msg) print_local_ws_event(local_ws_request_summary(req, addr)) res = self.handle_packet(req, addr) except Exception as e: res = {"ok": 0, "version": VERSION, "act": "error", "error_code": "LOCAL_WS_PACKET_FAILED", "error": str(e)} print_local_ws_event(local_ws_result_summary(req, res, addr)) self._send_text(conn, json.dumps(res, ensure_ascii=False)) except Exception: pass finally: try: conn.close() except Exception: pass def _handshake(self, conn): data = b"" while b"\r\n\r\n" not in data and len(data) < 8192: chunk = conn.recv(1024) if not chunk: raise RuntimeError("HANDSHAKE_EMPTY") data += chunk text = data.decode("utf-8", errors="replace") key = "" for line in text.split("\r\n"): if line.lower().startswith("sec-websocket-key:"): key = line.split(":", 1)[1].strip() break if not key: raise RuntimeError("WEBSOCKET_KEY_MISSING") accept = base64.b64encode(hashlib.sha1((key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode("ascii")).digest()).decode("ascii") resp = ( "HTTP/1.1 101 Switching Protocols\r\n" "Upgrade: websocket\r\n" "Connection: Upgrade\r\n" "Sec-WebSocket-Accept: " + accept + "\r\n" "\r\n" ) conn.sendall(resp.encode("ascii")) def _recv_exact(self, conn, n): buf = b"" while len(buf) < n: chunk = conn.recv(n - len(buf)) if not chunk: return None buf += chunk return buf def _recv_text(self, conn): h = self._recv_exact(conn, 2) if not h: return None b1, b2 = h[0], h[1] opcode = b1 & 0x0F masked = bool(b2 & 0x80) length = b2 & 0x7F if opcode == 0x8: return None if opcode == 0x9: # ping: consume and ignore pass if length == 126: ext = self._recv_exact(conn, 2) if not ext: return None length = struct.unpack("!H", ext)[0] elif length == 127: ext = self._recv_exact(conn, 8) if not ext: return None length = struct.unpack("!Q", ext)[0] if length > 1024 * 1024: raise RuntimeError("LOCAL_WS_FRAME_TOO_LARGE") mask = self._recv_exact(conn, 4) if masked else b"" payload = self._recv_exact(conn, length) if length else b"" if payload is None: return None if masked: payload = bytes(b ^ mask[i % 4] for i, b in enumerate(payload)) if opcode == 0x1: return payload.decode("utf-8", errors="replace") if opcode == 0x9: self._send_frame(conn, 0xA, payload) return "{}" return None def _send_frame(self, conn, opcode, payload): if isinstance(payload, str): payload = payload.encode("utf-8") length = len(payload) head = bytearray([0x80 | opcode]) if length < 126: head.append(length) elif length <= 65535: head.append(126) head.extend(struct.pack("!H", length)) else: head.append(127) head.extend(struct.pack("!Q", length)) conn.sendall(bytes(head) + payload) def _send_text(self, conn, text): self._send_frame(conn, 0x1, text) def _auth_ok(self, req): token = str((req or {}).get("token") or (req or {}).get("auth_token") or "").strip() data = (req or {}).get("data") if not token and isinstance(data, dict): token = str(data.get("token") or data.get("auth_token") or "").strip() return token and secrets.compare_digest(token, self.token) def _fresh_status(self): # 로컬 브리지 응답용 상태 생성. WSS 재등록 남발을 피하기 위해 wss_enabled를 임시로 끈다. cfg2 = dict(self.cfg) cfg2["wss_enabled"] = False return run_once(cfg2, self.cfg_path, self.state_path, self.log_path, print_json=False, wss_live=None) def handle_packet(self, req, addr): if not isinstance(req, dict): return {"ok": 0, "version": VERSION, "error_code": "REQUEST_NOT_OBJECT", "solution": "JSON object를 보내십시오."} req_id = req.get("id") or "" act = str(req.get("act") or req.get("type") or "").strip() if not act: return {"ok": 0, "version": VERSION, "id": req_id, "error_code": "ACT_REQUIRED", "solution": "act 값을 넣으십시오."} if not self._auth_ok(req): return {"ok": 0, "version": VERSION, "id": req_id, "act": act + "_result", "error_code": "AUTH_FAILED", "solution": "local_bridge.token 값을 확인하십시오."} if act in ("hello", "identity_hello"): status = self._fresh_status() return {"ok": 1, "version": VERSION, "id": req_id, "act": "identity_hello_result", "identity": status.get("identity", {}), "checks": {"identity_guard": status.get("checks", {}).get("identity_guard", {})}, "local_identity_service": {"host": self.host, "port": self.port, "scope": "pc/profile/tab identity only"}} if act == "chrome_profiles_list": return {"ok": 1, "version": VERSION, "id": req_id, "act": "chrome_profiles_list_result", "chrome": list_chrome_profiles(self.cfg)} if act in ("profile_bind", "ext_tab_report"): saved = save_profile_bind(self.state_path, req, addr) return dict({"version": VERSION, "id": req_id, "act": "profile_bind_result"}, **saved) if act == "profile_map": state = read_json_file(self.state_path, required=True) if os.path.isfile(self.state_path) else {} return {"ok": 1, "version": VERSION, "id": req_id, "act": "profile_map_result", "profile_map": state.get("profile_map", {})} if act == "pc_status": status = self._fresh_status() return {"ok": 1, "version": VERSION, "id": req_id, "act": "pc_status_result", "status": status, "note": "진단용입니다. Local Identity Service 핵심 기능은 identity_hello/chrome_profiles_list/profile_bind/profile_map 입니다."} if act == "node_red_ping": timeout_sec = float(self.cfg.get("timeout_sec", 3)) node_red = check_tcp(require_config(self.cfg, "node_red_host"), int(require_config(self.cfg, "node_red_port")), timeout_sec) return {"ok": 1 if node_red.get("ok") else 0, "version": VERSION, "id": req_id, "act": "node_red_ping_result", "node_red": node_red, "note": "진단용 보조 기능입니다."} if act == "chrome_check": chrome = check_chrome_process(require_config(self.cfg, "chrome_process_names")) return {"ok": 1 if chrome.get("ok") else 0, "version": VERSION, "id": req_id, "act": "chrome_check_result", "chrome": chrome, "note": "진단용 보조 기능입니다."} if act == "pc_ext4_check": ext4 = check_ext4_dir(self.cfg, os.path.dirname(self.cfg_path)) return {"ok": 1 if ext4.get("ok") else 0, "version": VERSION, "id": req_id, "act": "pc_ext4_check_result", "ext4": ext4} if act == "pc_chrome_load_ext4": res = chrome_load_ext4(self.cfg, os.path.dirname(self.cfg_path)) return dict({"version": VERSION, "id": req_id}, **res) return {"ok": 0, "version": VERSION, "id": req_id, "act": act + "_result", "error_code": "UNKNOWN_ACT", "solution": "허용 act: identity_hello, chrome_profiles_list, profile_bind, profile_map, pc_status(진단), node_red_ping(진단), chrome_check(진단), pc_ext4_check, pc_chrome_load_ext4"} class WssLiveClient: def __init__(self, cfg): self.cfg = cfg self.websocket = None self.ws = None self.last_result = {"ok": 0, "enabled": 1, "registered": 0, "error": "not_connected"} def enabled(self): return cfg_bool(self.cfg, "wss_enabled", False) def close(self): try: if self.ws is not None: self.ws.close() except Exception: pass self.ws = None def ensure_registered(self, status): if not self.enabled(): self.last_result = {"ok": 0, "enabled": 0, "error": "WSS_DISABLED"} return self.last_result if cfg_bool(self.cfg, "wss_require_token", False) and not str(self.cfg.get("token") or self.cfg.get("auth_token") or "").strip(): die("AIWK_MYPC_WSS_TOKEN_MISSING", "wss_require_token=true 인데 token이 비어 있습니다.", { "how_to_fix": "aiwk_mypc_config.json에 token을 넣거나 테스트 목적이면 wss_require_token=false로 바꾸십시오." }) if self.websocket is None: self.websocket = get_wss_client_module() url = str(require_config(self.cfg, "wss_url")).strip() timeout_sec = float(self.cfg.get("wss_timeout_sec", self.cfg.get("timeout_sec", 3))) if self.ws is not None: try: self.ws.ping() self.last_result["ok"] = 1 return self.last_result except Exception: self.close() result = {"ok": 0, "enabled": 1, "url": url, "registered": 0, "ack": None, "error": ""} try: self.ws = self.websocket.create_connection(url, timeout=timeout_sec) packet = build_register_packet(self.cfg, status) self.ws.send(json.dumps(packet, ensure_ascii=False)) raw = self.ws.recv() ack = json.loads(raw) result["ack"] = ack result["registered"] = 1 if ack.get("ok") else 0 result["ok"] = 1 if ack.get("ok") else 0 if not ack.get("ok"): result["error"] = str(ack.get("reason") or ack.get("error") or "REGISTER_FAILED") self.close() self.last_result = result return result except Exception as e: result["error"] = str(e) self.close() self.last_result = result return result def write_status_log(log_path, status): try: with open(log_path, "a", encoding="utf-8") as f: f.write(json.dumps(status, ensure_ascii=False) + "\n") except Exception as e: status["log_write_error"] = str(e) def print_one_line(status): ident = status.get("identity", {}) checks = status.get("checks", {}) node_red_ok = "OK" if checks.get("node_red", {}).get("ok") else "FAIL" chrome_ok = "OK" if checks.get("chrome", {}).get("ok") else "FAIL" if "wss" in checks: wss_ok = "OK" if checks.get("wss", {}).get("ok") else "FAIL" else: wss_ok = "OFF" ip = ident.get("remote_ip", "") or "-" print("[" + now_iso() + "] " + str(ident.get("client_id", "PC1")) + " " + str(ident.get("device_label", "")) + " host:" + str(ident.get("system_hostname", "")) + " IP:" + ip + " WSS:" + wss_ok + " NR:" + node_red_ok + " Chrome:" + chrome_ok, flush=True) def watch_status_signature(status): # v125: watch 모드는 같은 상태를 매 주기 반복 출력하지 않는다. # 상태 비교에는 운영자가 실제로 보고 싶은 주요 값만 사용한다. ident = status.get("identity", {}) if isinstance(status, dict) else {} checks = status.get("checks", {}) if isinstance(status, dict) else {} chrome = checks.get("chrome", {}) if isinstance(checks.get("chrome", {}), dict) else {} node_red = checks.get("node_red", {}) if isinstance(checks.get("node_red", {}), dict) else {} wss = checks.get("wss", {}) if isinstance(checks.get("wss", {}), dict) else {} local_bridge = checks.get("local_bridge", {}) if isinstance(checks.get("local_bridge", {}), dict) else {} return json.dumps({ "client_id": ident.get("client_id", "PC1"), "device_instance_id": ident.get("device_instance_id", ""), "device_label": ident.get("device_label", ""), "system_hostname": ident.get("system_hostname", ""), "remote_ip": ident.get("remote_ip", ""), "node_red_ok": 1 if node_red.get("ok") else 0, "chrome_ok": 1 if chrome.get("ok") else 0, "wss_present": 1 if "wss" in checks else 0, "wss_ok": 1 if wss.get("ok") else 0, "local_bridge_ok": 1 if local_bridge.get("ok") else 0, "local_bridge_error": local_bridge.get("error", ""), }, ensure_ascii=False, sort_keys=True) def print_watch_idle_dot(last_dot_line_at, dot_count): # 같은 상태가 유지되면 한 줄 로그를 반복하지 않고 . 만 누적한다. # 1시간마다 보기 좋게 줄바꿈한다. now = time.time() if last_dot_line_at <= 0: last_dot_line_at = now if now - last_dot_line_at >= 3600: print("", flush=True) last_dot_line_at = now dot_count = 0 print(".", end="", flush=True) return last_dot_line_at, dot_count + 1 def run_once(cfg, cfg_path, state_path, log_path, print_json=True, wss_live=None): system_hostname = get_system_hostname() device_label = get_device_label(cfg, system_hostname) state, created_new, migrated_placeholder = load_or_create_state(state_path, device_label, system_hostname) keep_lines = int(cfg.get("log_keep_lines", 20)) if created_new: append_log(state, "device_instance_id created " + state.get("device_instance_id", ""), keep_lines) if migrated_placeholder: append_log(state, "device_instance_id migrated " + state.get("device_instance_id_prev", "") + " -> " + state.get("device_instance_id", ""), keep_lines) status = build_status(cfg, state, cfg_path, state_path) if cfg_bool(cfg, "wss_enabled", False): if wss_live is not None: status.setdefault("checks", {})["wss"] = wss_live.ensure_registered(status) else: status.setdefault("checks", {})["wss"] = wss_register_once(cfg, status) wss_ok = "ok" if status.get("checks", {}).get("wss", {}).get("ok") else "fail" append_log(state, "wss " + wss_ok + " " + str(status.get("checks", {}).get("wss", {}).get("error", "")), keep_lines) write_json_file(state_path, state) status["recent_logs"] = state.get("recent_logs", [])[-keep_lines:] write_status_log(log_path, status) if print_json: print(json.dumps(status, ensure_ascii=False, indent=2)) return status def main(): parser = argparse.ArgumentParser(description="AIWK MyPC root identity agent v107") parser.add_argument("--config", default=os.path.join(script_dir(), DEFAULT_CONFIG_NAME), help="aiwk_mypc_config.json path") parser.add_argument("--once", action="store_true", help="run one status check and print JSON") parser.add_argument("--watch", action="store_true", help="keep running; print status changes, idle dots, and local_ws request/result summaries") parser.add_argument("--local-ws", action="store_true", help="run local websocket bridge server and keep process alive") parser.add_argument("--ext4-check", action="store_true", help="check configured EXT4 extension directory and manifest.json") parser.add_argument("--chrome-load-ext4", action="store_true", help="start Chrome with configured EXT4 directory by --load-extension") parser.add_argument("--chrome-profiles-list", action="store_true", help="list Chrome profiles detected on this PC") parser.add_argument("--profile-directory", default="", help="Chrome profile directory to launch, e.g. Default or Profile 3") parser.add_argument("--user-data-dir", default="", help="override Chrome user data dir") parser.add_argument("--start-url", default="", help="override startup URL") parser.add_argument("--remote-debugging-port", default="", help="override Chrome remote debugging port") parser.add_argument("--interval", type=float, default=5.0, help="watch interval seconds") args = parser.parse_args() if args.interval <= 0: die("AIWK_MYPC_INTERVAL_INVALID", "--interval 값은 0보다 커야 합니다.", {"interval": args.interval}) cfg_path = os.path.abspath(args.config) cfg = read_json_file(cfg_path, required=True) if not isinstance(cfg, dict): die("AIWK_MYPC_CONFIG_INVALID", "config JSON 구조가 object가 아닙니다.", {"path": cfg_path}) base_dir = os.path.dirname(cfg_path) state_path = resolve_path(base_dir, require_config(cfg, "state_path")) log_path = resolve_path(base_dir, require_config(cfg, "log_path")) if args.chrome_profiles_list: print(json.dumps({ "ok": 1, "version": VERSION, "act": "chrome_profiles_list_result", "chrome": list_chrome_profiles(cfg), "created_at": now_iso() }, ensure_ascii=False, indent=2)) return if args.ext4_check: print(json.dumps({ "ok": 1, "version": VERSION, "act": "pc_ext4_check_result", "ext4": check_ext4_dir(cfg, base_dir), "created_at": now_iso() }, ensure_ascii=False, indent=2)) return if args.chrome_load_ext4: launch_opts = { "profile_directory": args.profile_directory or None, "user_data_dir": args.user_data_dir or None, "start_url": args.start_url or None, "remote_debugging_port": args.remote_debugging_port or None } print(json.dumps(dict({ "version": VERSION, "created_at": now_iso() }, **chrome_load_ext4(cfg, base_dir, launch_opts)), ensure_ascii=False, indent=2)) return local_bridge = None local_bridge_enabled = LocalWsBridgeServer.enabled_in_config(cfg) if args.watch or args.local_ws: if local_bridge_enabled: local_bridge = LocalWsBridgeServer(cfg, cfg_path, state_path, log_path) started = local_bridge.start() print(json.dumps({"ok": 1, "version": VERSION, "local_bridge": started, "created_at": now_iso()}, ensure_ascii=False), flush=True) elif args.local_ws: die("AIWK_LOCAL_BRIDGE_DISABLED", "--local-ws가 지정됐지만 local_bridge.enabled=false 입니다.", { "how_to_fix": "aiwk_mypc_config.json에서 local_bridge.enabled=true, token 설정 후 다시 실행하십시오." }) if args.local_ws and not args.watch: # v123: local_ws는 외부 요청이 들어올 때만 상세 표시한다. # 주기 상태 출력/주기 state 저장은 JSON 경합과 로그 소음을 만들 수 있어 기본 중단한다. try: while True: time.sleep(args.interval) finally: if local_bridge is not None: local_bridge.close() elif args.watch: wss_live = WssLiveClient(cfg) if cfg_bool(cfg, "wss_enabled", False) else None last_signature = None last_dot_line_at = 0.0 dot_count = 0 try: while True: status = run_once(cfg, cfg_path, state_path, log_path, print_json=False, wss_live=wss_live) if local_bridge_enabled: status.setdefault("checks", {})["local_bridge"] = { "ok": 1 if local_bridge and local_bridge.thread and local_bridge.thread.is_alive() else 0, "enabled": 1, "host": local_bridge.host if local_bridge else "", "port": local_bridge.port if local_bridge else 0, "error": local_bridge.last_error if local_bridge else "not_started" } signature = watch_status_signature(status) if signature != last_signature: if last_signature is not None and dot_count > 0: print("", flush=True) print_one_line(status) last_signature = signature last_dot_line_at = time.time() dot_count = 0 else: last_dot_line_at, dot_count = print_watch_idle_dot(last_dot_line_at, dot_count) time.sleep(args.interval) finally: if dot_count > 0: print("", flush=True) if wss_live is not None: wss_live.close() if local_bridge is not None: local_bridge.close() else: status = run_once(cfg, cfg_path, state_path, log_path, print_json=False) lb = cfg.get("local_bridge") if isinstance(cfg.get("local_bridge"), dict) else {} status.setdefault("checks", {})["local_bridge"] = { "ok": 0, "enabled": 1 if cfg_bool(lb, "enabled", False) else 0, "host": str(lb.get("host", "127.0.0.1")), "port": int(lb.get("port", 8768)), "running": 0, "note": "once 모드에서는 서버를 시작하지 않습니다. --watch 또는 --local-ws를 사용하십시오." } print(json.dumps(status, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()