commit 1e9a77b90dff57fb9eabba17a35d494e8bd96596 Author: every_holiday Date: Thu Jun 4 00:46:37 2026 +0900 Initial VRC OSC commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..20a9c46 --- /dev/null +++ b/.gitignore @@ -0,0 +1,13 @@ +__pycache__/ +*.py[cod] + +config/config.toml +config/secrets.toml +config/guests.txt +runtime/ +.vrchat_cookie.txt +src/.vrchat_cookie.txt + +*.bak +*.bak_* +backup_*/ diff --git a/bin/vrc_osc.ps1 b/bin/vrc_osc.ps1 new file mode 100644 index 0000000..02fc732 --- /dev/null +++ b/bin/vrc_osc.ps1 @@ -0,0 +1,7 @@ +$ErrorActionPreference = "Stop" + +$ScriptDir = Split-Path -Parent $MyInvocation.MyCommand.Path +$RootDir = Resolve-Path (Join-Path $ScriptDir "..") +$EntryPoint = Join-Path $RootDir "src\app.py" + +python $EntryPoint diff --git a/config/config.example.toml b/config/config.example.toml new file mode 100644 index 0000000..e154d9e --- /dev/null +++ b/config/config.example.toml @@ -0,0 +1,22 @@ +[self] +name = "" + +[staff] +names = [] + +[guest] +file = "config/guests.txt" + +[notice] +missing_count = 0 + +[ocr.crop] +left = 0.05 +top = 0.35 +right = 0.95 +bottom = 0.95 + +[ocr] +preprocess = true +scale = 1.5 +use_angle_cls = false diff --git a/config/guests.example.txt b/config/guests.example.txt new file mode 100644 index 0000000..3124eb2 --- /dev/null +++ b/config/guests.example.txt @@ -0,0 +1,3 @@ +GuestUserA +GuestUserB +GuestUserC diff --git a/config/secrets.example.toml b/config/secrets.example.toml new file mode 100644 index 0000000..8cc68c2 --- /dev/null +++ b/config/secrets.example.toml @@ -0,0 +1,6 @@ +[discord] +webhook_url = "" + +[vrchat] +username = "" +password = "" diff --git a/readme.txt b/readme.txt new file mode 100644 index 0000000..d509807 --- /dev/null +++ b/readme.txt @@ -0,0 +1,237 @@ +# VRC_OSC + +VRChat の OSC パラメータを受けて、Discord ミュート、VRChat ログ監視、OCR、翻訳を実行するローカルツール。 + +## 起動 + +依存関係を入れる。 + +```powershell +python -m pip install -r requirements.txt +``` + +PowerShell から起動する。 + +```powershell +bin\vrc_osc.ps1 +``` + +Python から直接起動する場合。 + +```powershell +python src\app.py +``` + +OSC は `127.0.0.1:9001` で待ち受ける。 + +## 構成 + +```text +bin/ PowerShell 起動スクリプト +config/ example とローカル設定置き場 +src/app.py アプリ本体のエントリ +src/common/ パス、設定ロード、runtime ログなどの共通基盤 +src/discord_control/ Discord ウィンドウ操作 +src/ocr/ OCR 実行 +src/osc/ OSC gateway +src/screen/ VRChat ウィンドウ検出とキャプチャ +src/tools/ 単体実行用の補助ツール +src/translate/ 翻訳 +src/vrc_log/ VRChat ログ解析と名簿照合 +``` + +## 設定ファイル + +実設定は `config/config.toml` に置く。このファイルは Git に入れない。 + +雛形は `config/config.example.toml`。 + +```toml +[self] +name = "自分のVRChat表示名" + +[staff] +names = [ + "StaffUserA" +] + +[guest] +file = "config/guests.txt" + +[notice] +missing_count = 3 + +[ocr.crop] +left = 0.05 +top = 0.35 +right = 0.95 +bottom = 0.95 + +[ocr] +preprocess = true +scale = 1.5 +use_angle_cls = false +``` + +`self.name` は Discord 自動ミュート判定に使う。 + +`staff.names` は表示対象だが guest 人数には含めない。 + +`guest.file` は guest 名簿ファイル。guest 人数の母数と在室判定に使う。 + +`notice.missing_count` は「あと何人以内で全員集合か」の通知範囲。例えば guest が 23 人で `missing_count = 3` の場合、20〜22 人のときに未入室 guest を出す。 + +OCR 設定は速度と精度の調整用。 + +`ocr.crop` は VRChat ウィンドウ内の相対座標。`0.0` から `1.0` で指定する。デフォルトは画面下側の中央寄りだけを読む。 + +フル画面 OCR に戻す場合。 + +```toml +[ocr.crop] +left = 0 +top = 0 +right = 1 +bottom = 1 +``` + +`ocr.preprocess = true` はグレースケール化、コントラスト補正、シャープ化、拡大を行う。 + +`ocr.scale` は前処理時の拡大率。大きいほど小さい文字に強くなるが遅くなる。 + +`ocr.use_angle_cls = false` は角度分類を切って高速化する。縦書きや回転文字を読む必要がある場合だけ `true` にする。 + +## guest 名簿 + +`config/guests.txt` に 1 行 1 名で書く。このファイルは Git に入れない。 + +雛形は `config/guests.example.txt`。 + +```text +GuestUserA +GuestUserB +GuestUserC +``` + +空行と `#` で始まる行は無視する。 + +## 秘密情報 + +秘密情報は `config/secrets.toml` に置く。このファイルは Git に入れない。 + +雛形は `config/secrets.example.toml`。 + +```toml +[discord] +webhook_url = "" + +[vrchat] +username = "" +password = "" +``` + +`discord.webhook_url` は VRC ログ結果の Discord Webhook 通知に使う。 + +`vrchat.username` / `vrchat.password` は `src/tools/resolve_vrchat_user.py` を使う場合だけ必要。 + +## OSC パラメータ + +Avatar Parameters 側の名前と一致させる。 + +| Parameter | 動作 | +| --- | --- | +| `DiscordSend` | 押した瞬間に Discord の mute hotkey を送る | +| `vrc_log` | VRC ログ監視の on/off | +| `ocrEnabled` | VRChat 画面を撮って OCR | +| `translation` | OCR して日本語翻訳 | + +## VRC ログ監視 + +`vrc_log` を押すと、現在の VRChat ログを一度解析して Discord Webhook に送る。その後はログファイル末尾を追い、新しい join / leave / world 系ログが追記されたタイミングで即再解析する。 + +以前のような `output_log = 10` の秒数ポーリング設定は使わない。 + +解析対象は昼 12:00 以降。実行時刻が昼 12:00 前なら前日 12:00 以降を見る。 + +インスタンスが変わったら在室状態はリセットする。 + +出力例。 + +```text +ワールド入室: Midnight Rooftop +現在Guest: 0/3 +[01:20] [join] GuestUserA 1/3 +[01:21] [join] GuestUserB 2/3 +[01:25] [leave] GuestUserA 1/3 +[01:26] [join] [staff] StaffUserA 1/3 +``` + +notice 条件に入った場合。 + +```text +現在Guest: 2/3 +未入室Guest: +- GuestUserC +``` + +## Discord 自動ミュート + +常時 self monitor が VRChat ログの新規追記だけを見る。 + +新しく追記されたログの中で、自分の最後のイベントが `OnPlayerJoined` なら Discord を mute、`OnPlayerLeft` なら unmute する。 + +過去ログ全体の最後が self leave / join だっただけでは反応しない。 + +`setDiscordMuted()` にはロックがあり、同時実行で mute hotkey が連続送信されないようにしている。 + +## OCR / 翻訳 + +`ocrEnabled` は VRChat ウィンドウを撮影して PaddleOCR にかける。結果はクリップボードへコピーする。 + +`translation` は OCR 結果を翻訳し、翻訳結果をクリップボードへコピーする。 + +OCR はデフォルトで画面全体ではなく、画面下側の中央寄りを切り出して読む。会話ログや字幕を読む用途では速く、余計な UI 文字を拾いにくい。 + +## runtime + +新規の実行ログは基本的に以下へまとめる。 + +```text +runtime/runtime.log +runtime/latest_ocr.png +runtime/latest_screenshot.png +``` + +`runtime/` は Git に入れない。 + +OCR text、翻訳 text、VRC ログ結果は `runtime/runtime.log` に追記する。 + +画像は最新ファイルに上書きする。日時別ファイルを増やさない。 + +## ユーザー名照合 + +`src/vrc_log/member_name_matcher.py` は TOML 登録名と VRChat 表示名を安全な順番で紐づける補助実装。 + +照合順。 + +1. 完全一致 +2. 正規化一致 +3. 高信頼 fuzzy +4. guest のみ低信頼 fuzzy +5. pending / ambiguous / unmatched + +完全一致・正規化一致で確定した名前は fuzzy 候補から除外する。 + +## Git に入れないもの + +以下はローカル専用。 + +- `config/config.toml` +- `config/secrets.toml` +- `config/guests.txt` +- `runtime/` +- `__pycache__/` +- `*.pyc` +- `.vrchat_cookie.txt` + +秘密情報を含む可能性があるものはコミットしない。 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..ccef526 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,10 @@ +deep-translator +paddleocr +pillow +pyautogui +pygetwindow +pykakasi +pyperclip +python-osc +rapidfuzz +vrchatapi diff --git a/src/app.py b/src/app.py new file mode 100644 index 0000000..6b25eba --- /dev/null +++ b/src/app.py @@ -0,0 +1,114 @@ +import os +from pathlib import Path +import sys + +os.environ["FLAGS_enable_pir_api"] = "0" +os.environ["FLAGS_use_onednn"] = "0" +os.environ["FLAGS_use_mkldnn"] = "0" +os.environ["FLAGS_use_onednn_bfloat16"] = "0" +os.environ["ONEDNN_VERBOSE"] = "0" + +SRC_DIR = Path(__file__).resolve().parents[1] +if str(SRC_DIR) not in sys.path: + sys.path.insert(0, str(SRC_DIR)) + +from osc.gateway import VrcOscGateway +from discord_control.actions import setDiscordMute +from ocr.ocr_actions import runOcrFromScreen +from translate.translate_actions import saveTranslationText +from translate.translate_actions import translateTextToJapanese +from vrc_log.log_actions import collectVrchatLog +from vrc_log.log_actions import startSelfMonitor +from vrc_log.log_actions import startVrcLogMonitor +from vrc_log.log_actions import stopSelfMonitor +from vrc_log.log_actions import stopVrcLogMonitor + +PARAM_DISCORD_MUTE = "DiscordSend" +PARAM_LOG_GREP = "vrc_log" +PARAM_OCR = "ocrEnabled" +PARAM_TRANSLATE_OCR = "translation" + + +def create_gateway(): + gateway = VrcOscGateway(host="127.0.0.1", port=9001) + + def on_discord_mute(address, *args): + if not args: + gateway.log("ERROR", "DiscordSend args empty") + return + + if gateway.is_rising_edge(address, args[0]): + gateway.log("INFO", f"received {address} args={args}") + setDiscordMute() + + vrc_log_enabled = False + def on_log_grep(address, *args): + nonlocal vrc_log_enabled + + if not args: + gateway.log("ERROR", "vrc_log args empty") + return + + if gateway.is_rising_edge(address, args[0]): + vrc_log_enabled = not vrc_log_enabled + gateway.log("INFO", f"received {address} args={args} enabled={vrc_log_enabled}") + + if vrc_log_enabled: + collectVrchatLog(notify_changed=True) + startVrcLogMonitor() + else: + stopVrcLogMonitor() + + def on_ocr(address, *args): + if not args: + gateway.log("ERROR", "ocrEnabled args empty") + return + + if gateway.is_rising_edge(address, args[0]): + gateway.log("INFO", f"received {address} args={args}") + runOcrFromScreen() + + def on_translate_ocr(address, *args): + if not args: + gateway.log("ERROR", "translateOcr args empty") + return + + if gateway.is_rising_edge(address, args[0]): + gateway.log("INFO", f"received {address} args={args}") + + result = runOcrFromScreen() + if not result: + gateway.log("ERROR", "OCR result is None") + return + + text = result.get("text", "") + if not text.strip(): + gateway.log("INFO", "OCR text is empty. translation skipped") + return + + translated = translateTextToJapanese(text) + if not translated.strip(): + gateway.log("INFO", "translation result is empty") + return + + image_path = result.get("image_path") + stem = image_path.stem if image_path else "translation" + saved = saveTranslationText(stem, translated) + gateway.log("INFO", f"translation saved={saved}") + + gateway.map(PARAM_DISCORD_MUTE, on_discord_mute) + gateway.map(PARAM_LOG_GREP, on_log_grep) + gateway.map(PARAM_OCR, on_ocr) + gateway.map(PARAM_TRANSLATE_OCR, on_translate_ocr) + gateway.set_default_debug(enabled=False) + return gateway + + +def main(): + gateway = create_gateway() + startSelfMonitor() + gateway.serve_forever() + + +if __name__ == "__main__": + main() diff --git a/src/common/config_loader.py b/src/common/config_loader.py new file mode 100644 index 0000000..364ea60 --- /dev/null +++ b/src/common/config_loader.py @@ -0,0 +1,130 @@ +from pathlib import Path +import tomllib + +from common.project_paths import ROOT_DIR + +CONFIG_DIR = ROOT_DIR / "config" +CONFIG_FILE = CONFIG_DIR / "config.toml" +SECRETS_FILE = CONFIG_DIR / "secrets.toml" +DEFAULT_EVENT_FILE = CONFIG_DIR / "event.toml" +DEFAULT_GUEST_FILE = CONFIG_DIR / "guests.txt" + + +def loadTomlFile(path): + if not path.exists(): + return {} + + return tomllib.loads(path.read_text(encoding="utf-8")) + + +def resolveConfigPath(value, default_path): + if not value: + return default_path + + candidate = Path(value) + if candidate.is_absolute(): + return candidate + + return (ROOT_DIR / candidate).resolve() + + +def loadNameListFile(path): + if not path.exists(): + return [] + + names = [] + + for line in path.read_text(encoding="utf-8-sig", errors="replace").splitlines(): + name = line.strip() + if not name or name.startswith("#"): + continue + names.append(name) + + return names + + +def loadSecretConfig(): + return loadTomlFile(SECRETS_FILE) + + +def getSecretValue(section_name, key, default=""): + secrets = loadSecretConfig() + section = secrets.get(section_name, {}) + return str(section.get(key, default)).strip() + + +def loadOcrConfig(): + config = loadTomlFile(CONFIG_FILE) + ocr = config.get("ocr", {}) + crop = ocr.get("crop", {}) + + return { + "crop_left": float(crop.get("left", 0.05)), + "crop_top": float(crop.get("top", 0.35)), + "crop_right": float(crop.get("right", 0.95)), + "crop_bottom": float(crop.get("bottom", 0.95)), + "preprocess": bool(ocr.get("preprocess", True)), + "scale": float(ocr.get("scale", 1.5)), + "use_angle_cls": bool(ocr.get("use_angle_cls", False)), + } + + +def loadVrcLogConfig(): + config = loadTomlFile(CONFIG_FILE) + vrc_log = config.get("vrc_log", {}) + self_section = config.get("self", {}) + event_file_value = ( + vrc_log.get("event_toml") + or vrc_log.get("event_file") + or config.get("event_toml") + or config.get("event_file") + ) + + event_file = resolveConfigPath(event_file_value, DEFAULT_EVENT_FILE) + event_config = loadTomlFile(event_file) + + if not event_config: + event_config = vrc_log or config + + guest_section = event_config.get("guest", {}) + guest_names = guest_section.get("names", []) + guest_file_value = ( + guest_section.get("file") + or event_config.get("guest_file") + or config.get("guest_file") + ) + guest_file = resolveConfigPath(guest_file_value, DEFAULT_GUEST_FILE) + + if not guest_names: + guest_names = loadNameListFile(guest_file) + + staff_names = event_config.get("staff", {}).get("names", []) + + if not guest_names and isinstance(event_config.get("guest_names"), list): + guest_names = event_config.get("guest_names", []) + if not staff_names and isinstance(event_config.get("staff_names"), list): + staff_names = event_config.get("staff_names", []) + + notice_section = event_config.get("notice", {}) + missing_count = notice_section.get("missing_count", 0) + + if not missing_count: + missing_count = event_config.get("missing_count", 0) + + self_name = ( + self_section.get("name") + or config.get("self_name") + or vrc_log.get("self_name") + or event_config.get("self", {}).get("name") + or "" + ) + + return { + "config_file": CONFIG_FILE, + "event_file": event_file, + "guest_file": guest_file, + "guest_names": [str(name).strip() for name in guest_names if str(name).strip()], + "staff_names": [str(name).strip() for name in staff_names if str(name).strip()], + "missing_count": int(missing_count), + "self_name": str(self_name).strip(), + } diff --git a/src/common/project_paths.py b/src/common/project_paths.py new file mode 100644 index 0000000..fa38746 --- /dev/null +++ b/src/common/project_paths.py @@ -0,0 +1,6 @@ +from pathlib import Path + +SRC_DIR = Path(__file__).resolve().parents[1] +ROOT_DIR = SRC_DIR.parent +RUNTIME_DIR = ROOT_DIR / "runtime" +RUNTIME_LOG_FILE = RUNTIME_DIR / "runtime.log" diff --git a/src/common/runtime_log.py b/src/common/runtime_log.py new file mode 100644 index 0000000..862cec0 --- /dev/null +++ b/src/common/runtime_log.py @@ -0,0 +1,19 @@ +from datetime import datetime + +from common.project_paths import RUNTIME_LOG_FILE + + +def appendRuntimeLog(title, text): + RUNTIME_LOG_FILE.parent.mkdir(parents=True, exist_ok=True) + + now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + body = str(text).strip() + if not body: + body = "(empty)" + + with RUNTIME_LOG_FILE.open("a", encoding="utf-8") as file: + file.write(f"\n[{now}] {title}\n") + file.write(body) + file.write("\n") + + return RUNTIME_LOG_FILE diff --git a/src/discord_control/actions.py b/src/discord_control/actions.py new file mode 100644 index 0000000..9d1b68f --- /dev/null +++ b/src/discord_control/actions.py @@ -0,0 +1,203 @@ +from datetime import datetime +import ctypes +import threading +import time + +import pyautogui +import pygetwindow as gw + +BROWSER_KEYWORDS = [ + "google chrome", + "microsoft edge", + "mozilla firefox", +] + +_discord_muted = True +_discord_mute_lock = threading.Lock() + + +def log(level, message): + now = datetime.now().strftime("%H:%M:%S") + print(f"[{level}] {now} {message}", flush=True) + + +def findDiscordWindow(): + candidates = [] + + for window in gw.getAllWindows(): + title = (window.title or "").strip() + if not title: + continue + if "discord" not in title.lower(): + continue + candidates.append(window) + + if not candidates: + raise RuntimeError("Discordウィンドウが見つかりません") + + candidates.sort(key=lambda window: -(window.width * window.height)) + return candidates[0] + + +def findBrowserWindow(): + candidates = [] + + for window in gw.getAllWindows(): + title = (window.title or "").strip() + if not title: + continue + lowered = title.lower() + if not any(keyword in lowered for keyword in BROWSER_KEYWORDS): + continue + candidates.append(window) + + if not candidates: + raise RuntimeError("ブラウザウィンドウが見つかりません") + + candidates.sort(key=lambda window: -(window.width * window.height)) + return candidates[0] + + +def findDiscordBrowserWindow(): + candidates = [] + + for window in gw.getAllWindows(): + title = (window.title or "").strip() + if not title: + continue + lowered = title.lower() + if "discord" not in lowered: + continue + if not any(keyword in lowered for keyword in BROWSER_KEYWORDS): + continue + candidates.append(window) + + if not candidates: + return None + + candidates.sort(key=lambda window: -(window.width * window.height)) + return candidates[0] + + +def activateDiscordWindow(): + window = findDiscordWindow() + hwnd = getattr(window, "_hWnd", None) + + try: + if window.isMinimized: + window.restore() + except Exception: + pass + + if hwnd: + try: + ctypes.windll.user32.ShowWindow(hwnd, 9) + except Exception: + pass + + try: + window.activate() + except Exception as e: + log("ERROR", f"Discord activate() failed detail={e}") + + if hwnd: + try: + ctypes.windll.user32.BringWindowToTop(hwnd) + ctypes.windll.user32.SetForegroundWindow(hwnd) + except Exception as e: + log("ERROR", f"Discord SetForegroundWindow failed detail={e}") + + time.sleep(0.15) + + log("INFO", f"Discord window title={window.title}") + return window + + +def findVrchatWindow(): + candidates = [] + + for window in gw.getAllWindows(): + title = (window.title or "").strip() + if not title: + continue + lowered = title.lower() + if lowered == "vrchat": + candidates.append((0, window)) + continue + if lowered.startswith("vrchat "): + candidates.append((1, window)) + continue + if lowered.startswith("vrchat"): + candidates.append((2, window)) + + if not candidates: + raise RuntimeError("VRChatウィンドウが見つかりません") + + candidates.sort(key=lambda item: (item[0], -(item[1].width * item[1].height))) + return candidates[0][1] + + +def activateWindow(window, label): + if not window: + return + + hwnd = getattr(window, "_hWnd", None) + + try: + if window.isMinimized: + window.restore() + except Exception: + pass + + if hwnd: + try: + ctypes.windll.user32.ShowWindow(hwnd, 9) + except Exception: + pass + + try: + window.activate() + except Exception as e: + log("ERROR", f"{label} activate() failed detail={e}") + + if hwnd: + try: + ctypes.windll.user32.BringWindowToTop(hwnd) + ctypes.windll.user32.SetForegroundWindow(hwnd) + except Exception as e: + log("ERROR", f"{label} SetForegroundWindow failed detail={e}") + + time.sleep(0.15) + try: + pyautogui.click(window.left + window.width // 2, window.top + window.height // 2) + except Exception: + pass + time.sleep(0.1) + log("INFO", f"{label} window title={window.title}") + + +def setDiscordMute(): + log("ACTION", "setDiscordMute called") + try: + activateDiscordWindow() + except Exception as e: + log("ERROR", f"Discord window unavailable detail={e}") + browser = findDiscordBrowserWindow() or findBrowserWindow() + activateWindow(browser, "Browser") + log("INFO", "Discord hotkey Ctrl+Shift+M") + pyautogui.hotkey("ctrl", "shift", "m") + activateWindow(findVrchatWindow(), "VRChat") + + +def setDiscordMuted(muted): + global _discord_muted + + muted = bool(muted) + + with _discord_mute_lock: + if _discord_muted == muted: + log("INFO", f"Discord mute already {'on' if muted else 'off'}") + return + + setDiscordMute() + _discord_muted = muted diff --git a/src/ocr/ocr_actions.py b/src/ocr/ocr_actions.py new file mode 100644 index 0000000..56f2a4a --- /dev/null +++ b/src/ocr/ocr_actions.py @@ -0,0 +1,186 @@ +import os +from datetime import datetime + +os.environ["FLAGS_enable_pir_api"] = "0" +os.environ["FLAGS_use_onednn"] = "0" +os.environ["FLAGS_use_mkldnn"] = "0" +os.environ["FLAGS_use_onednn_bfloat16"] = "0" +os.environ["MKLDNN_DISABLE_WORKSPACE"] = "1" +os.environ["ONEDNN_VERBOSE"] = "0" + +import pyperclip +from paddleocr import PaddleOCR +from PIL import Image +from PIL import ImageFilter +from PIL import ImageOps + +from common.config_loader import loadOcrConfig +from common.runtime_log import appendRuntimeLog +from screen.screen_actions import captureOcrRegion + +_ocr_engine = None + + +def log(level, message): + now = datetime.now().strftime("%H:%M:%S") + print(f"[{level}] {now} {message}", flush=True) + + +def getOcrEngine(): + global _ocr_engine + + if _ocr_engine is None: + log("INFO", "PaddleOCR initializing") + config = loadOcrConfig() + options = { + "lang": "japan", + "use_angle_cls": config["use_angle_cls"], + "show_log": False, + } + + try: + _ocr_engine = PaddleOCR(**options) + except (TypeError, ValueError): + options.pop("show_log", None) + _ocr_engine = PaddleOCR(**options) + + return _ocr_engine + + +def saveOcrText(image_path, text): + return appendRuntimeLog( + "OCR TEXT", + f"image={image_path}\n{text}", + ) + + +def extractTextFromAny(obj): + lines = [] + + def walk(x): + if x is None: + return + + if isinstance(x, dict): + for key in ("rec_texts", "texts"): + value = x.get(key) + if isinstance(value, list): + for item in value: + text = str(item).strip() + if text: + lines.append(text) + + for key in ("text", "rec_text"): + value = x.get(key) + if isinstance(value, str) and value.strip(): + lines.append(value.strip()) + + for value in x.values(): + walk(value) + return + + if isinstance(x, (list, tuple)): + if len(x) >= 2 and isinstance(x[1], (list, tuple)) and len(x[1]) >= 1: + candidate = x[1][0] + if isinstance(candidate, str) and candidate.strip(): + lines.append(candidate.strip()) + + for item in x: + walk(item) + + walk(obj) + + unique_lines = [] + seen = set() + for line in lines: + if line not in seen: + unique_lines.append(line) + seen.add(line) + + return "\n".join(unique_lines) + + +def printOcrResult(text): + cleaned = text.strip() + if not cleaned: + log("INFO", "OCR結果は空です") + return + + log("CHECK", "OCR結果 full") + for line in cleaned.splitlines(): + print(line, flush=True) + + +def preprocessOcrImage(image_path): + config = loadOcrConfig() + enabled = config["preprocess"] + if not enabled: + return image_path + + try: + image = Image.open(image_path) + image = image.convert("L") + image = ImageOps.autocontrast(image) + image = image.filter(ImageFilter.SHARPEN) + + scale = max(1.0, min(2.0, config["scale"])) + if scale > 1.0: + width, height = image.size + resampling = getattr( + getattr(Image, "Resampling", Image), + "LANCZOS", + ) + image = image.resize( + (int(width * scale), int(height * scale)), + resampling, + ) + + image.save(image_path) + log("INFO", f"OCR image preprocessed={image_path}") + except Exception as e: + log("ERROR", f"OCR preprocess failed detail={e}") + + return image_path + + +def runPaddleOcr(engine, image_path): + if hasattr(engine, "predict"): + return engine.predict(str(image_path)) + return engine.ocr(str(image_path)) + + +def runOcrFromImage(image_path, show_result=True): + log("ACTION", f"PaddleOCR start image={image_path}") + + try: + image_path = preprocessOcrImage(image_path) + engine = getOcrEngine() + result = runPaddleOcr(engine, image_path) + except Exception as e: + log("ERROR", f"PaddleOCR failed detail={e}") + return None + + text = extractTextFromAny(result) + text_path = saveOcrText(image_path, text) + + try: + pyperclip.copy(text) + log("INFO", "OCR結果をクリップボードへコピーしました") + except Exception as e: + log("ERROR", f"clipboard copy failed detail={e}") + + log("INFO", f"OCR text appended={text_path}") + if show_result: + printOcrResult(text) + + return { + "image_path": image_path, + "text_path": text_path, + "text": text, + } + + +def runOcrFromScreen(): + log("ACTION", "runOcrFromScreen called") + image_path = captureOcrRegion() + return runOcrFromImage(image_path, show_result=True) diff --git a/src/osc/gateway.py b/src/osc/gateway.py new file mode 100644 index 0000000..419d88b --- /dev/null +++ b/src/osc/gateway.py @@ -0,0 +1,85 @@ +from pythonosc.dispatcher import Dispatcher +from pythonosc.osc_server import BlockingOSCUDPServer +from datetime import datetime +import sys + +class VrcOscGateway: + def __init__(self, host="127.0.0.1", port=9001): + self.host = host + self.port = port + self.dispatcher = Dispatcher() + self.last_values = {} + + def log(self, level, message): + now = datetime.now().strftime("%H:%M:%S") + print(f"[{level}] {now} {message}", flush=True) + + def to_bool(self, value): + if isinstance(value, bool): + return value + + if isinstance(value, int) or isinstance(value, float): + return value != 0 + + if isinstance(value, str): + lowered = value.strip().lower() + if lowered in ("true", "1", "on"): + return True + if lowered in ("false", "0", "off"): + return False + + raise ValueError(f"unsupported OSC value: {value}") + + def is_rising_edge(self, address, value): + current = self.to_bool(value) + previous = self.last_values.get(address, False) + self.last_values[address] = current + return current and not previous + + def is_changed(self, address, value): + current = self.to_bool(value) + previous = self.last_values.get(address, None) + + if previous == current: + return False + + self.last_values[address] = current + return True + + def map(self, parameter_name, handler): + address = f"/avatar/parameters/{parameter_name}" + self.dispatcher.map(address, handler) + self.log("INFO", f"mapped {address}") + + def set_default_debug(self, enabled=False): + if not enabled: + return + + def on_any(address, *args): + if address.startswith("/avatar/parameters/"): + self.log("INFO", f"other address={address} args={args}") + + self.dispatcher.set_default_handler(on_any) + + def serve_forever(self): + self.log("CHECK", "STEP=1 状態確認") + self.log("INFO", f"listen={self.host}:{self.port}") + + self.log("CHECK", "STEP=2 想定状態判定") + self.log("INFO", "VRChat OSC Enabled") + self.log("INFO", "Python は Windows PowerShell 側で実行") + self.log("INFO", "Unity側 MA Parameters と OSC名が一致していること") + + + try: + server = BlockingOSCUDPServer((self.host, self.port), self.dispatcher) + server.serve_forever() + except KeyboardInterrupt: + self.log("INFO", "Ctrl+C detected") + except Exception as e: + self.log("FATAL", f"OSC server error STEP=3 detail={e}") + input("Press Enter to exit...") + sys.exit(1) + + self.log("CHECK", "STEP=4 結果確認") + self.log("INFO", "終了しました") diff --git a/src/screen/screen_actions.py b/src/screen/screen_actions.py new file mode 100644 index 0000000..da0cf0a --- /dev/null +++ b/src/screen/screen_actions.py @@ -0,0 +1,167 @@ +import pyautogui +import pygetwindow as gw + +from common.config_loader import loadOcrConfig +from common.project_paths import RUNTIME_DIR + +SCREENSHOT_PATH = RUNTIME_DIR / "latest_screenshot.png" +OCR_SCREENSHOT_PATH = RUNTIME_DIR / "latest_ocr.png" + +EXCLUDED_WINDOW_KEYWORDS = [ + "Google Chrome", + "Microsoft Edge", + "Mozilla Firefox", + "Visual Studio Code", + "PowerShell", + "Windows Terminal", + "ChatGPT", +] + + +def log(level, message): + from datetime import datetime + + now = datetime.now().strftime("%H:%M:%S") + print(f"[{level}] {now} {message}", flush=True) + + +def findVrchatWindow(): + windows = gw.getAllWindows() + candidates = [] + + for window in windows: + title = window.title or "" + normalized = title.strip().lower() + if not normalized: + continue + if any(keyword.lower() in normalized for keyword in EXCLUDED_WINDOW_KEYWORDS): + continue + + if normalized == "vrchat": + candidates.append((0, window)) + continue + if normalized.startswith("vrchat "): + candidates.append((1, window)) + continue + if normalized.startswith("vrchat"): + candidates.append((2, window)) + + if not candidates: + visible_titles = [] + for window in windows: + title = (window.title or "").strip() + if title: + visible_titles.append(title) + + log("ERROR", "VRChatウィンドウが見つかりません") + log("CHECK", "現在検出できるウィンドウ") + for title in visible_titles[:30]: + print(title, flush=True) + + raise RuntimeError("VRChatウィンドウが見つかりません") + + candidates = sorted( + candidates, + key=lambda item: (item[0], -(item[1].width * item[1].height)), + ) + window = candidates[0][1] + + if window.width <= 0 or window.height <= 0: + raise RuntimeError("VRChatウィンドウサイズが不正です") + + return window + + +def getVrchatRegion(): + window = findVrchatWindow() + + try: + if window.isMinimized: + window.restore() + except Exception: + pass + + left = int(window.left) + top = int(window.top) + width = int(window.width) + height = int(window.height) + + if width <= 0 or height <= 0: + raise RuntimeError("VRChatウィンドウ範囲が不正です") + + log("INFO", f"VRChat window title={window.title}") + log("INFO", f"VRChat region left={left} top={top} width={width} height={height}") + return left, top, width, height + + +def clampRatio(value): + return max(0.0, min(1.0, value)) + + +def getOcrRegion(): + left, top, width, height = getVrchatRegion() + config = loadOcrConfig() + + # Default to the lower central area. It is much faster than OCRing the + # whole VRChat window and usually contains chat/subtitle text. + crop_left = clampRatio(config["crop_left"]) + crop_top = clampRatio(config["crop_top"]) + crop_right = clampRatio(config["crop_right"]) + crop_bottom = clampRatio(config["crop_bottom"]) + + if crop_right <= crop_left: + crop_left = 0.0 + crop_right = 1.0 + + if crop_bottom <= crop_top: + crop_top = 0.0 + crop_bottom = 1.0 + + region_left = left + int(width * crop_left) + region_top = top + int(height * crop_top) + region_width = max(1, int(width * (crop_right - crop_left))) + region_height = max(1, int(height * (crop_bottom - crop_top))) + + log( + "INFO", + f"OCR crop left={region_left} top={region_top} width={region_width} height={region_height}", + ) + return region_left, region_top, region_width, region_height + + +def captureRegionTo(image_path, region): + image_path.parent.mkdir(parents=True, exist_ok=True) + + left, top, width, height = region + + image = pyautogui.screenshot(region=(left, top, width, height)) + image.save(image_path) + + log("ACTION", f"captureVrchatWindow saved={image_path}") + return image_path + + +def captureVrchatWindowTo(image_path): + return captureRegionTo( + image_path, + getVrchatRegion(), + ) + + +def captureVrchatWindow(): + return captureVrchatWindowTo(SCREENSHOT_PATH) + + +def captureOcrRegion(): + return captureRegionTo( + OCR_SCREENSHOT_PATH, + getOcrRegion(), + ) + + +def captureScreen(): + return captureVrchatWindow() + + +def takeScreenshot(): + return captureVrchatWindow() diff --git a/src/tools/discord_hold_mute_listener.py b/src/tools/discord_hold_mute_listener.py new file mode 100644 index 0000000..03514d2 --- /dev/null +++ b/src/tools/discord_hold_mute_listener.py @@ -0,0 +1,124 @@ +from pythonosc.dispatcher import Dispatcher +from pythonosc.osc_server import BlockingOSCUDPServer +from datetime import datetime +import pyautogui +import time +import sys + +HOST = "127.0.0.1" +PORT = 9001 +PARAMETER = "/avatar/parameters/DiscordSend" + +# Discordの初期状態を「ミュート中」と仮定 +discord_muted = True +button_pressed = False +last_action_time = 0.0 +DEBOUNCE_SECONDS = 0.3 + +def log(level, message): + now = datetime.now().strftime("%H:%M:%S") + print(f"[{level}] {now} {message}", flush=True) + +def wait_exit(): + input("Press Enter to exit...") + +def press_discord_mute_hotkey(): + pyautogui.hotkey("ctrl", "shift", "m") + +def handle_discord_send(address, *args): + global discord_muted + global button_pressed + global last_action_time + + if not args: + log("ERROR", "OSC args is empty") + return + + value = args[0] + now_ts = time.time() + + if isinstance(value, bool): + is_pressed = value + elif isinstance(value, int) or isinstance(value, float): + is_pressed = value != 0 + else: + log("ERROR", f"unsupported value type value={value}") + return + + if now_ts - last_action_time < DEBOUNCE_SECONDS: + return + + # 押した瞬間: ミュート解除 + if is_pressed and not button_pressed: + button_pressed = True + + if discord_muted: + log("ACTION", "VRChat button pressed -> Discord unmute hotkey") + press_discord_mute_hotkey() + discord_muted = False + last_action_time = now_ts + else: + log("INFO", "already unmuted") + + return + + # 離した瞬間: ミュート + if not is_pressed and button_pressed: + button_pressed = False + + if not discord_muted: + log("ACTION", "VRChat button released -> Discord mute hotkey") + press_discord_mute_hotkey() + discord_muted = True + last_action_time = now_ts + else: + log("INFO", "already muted") + + return + +def main(): + log("CHECK", "STEP=1 OSC受信状態確認") + log("INFO", f"host={HOST}") + log("INFO", f"port={PORT}") + log("INFO", f"parameter={PARAMETER}") + + log("CHECK", "STEP=2 想定状態判定") + log("INFO", "Windows PowerShellで実行していること") + log("INFO", "VRChat OSC Enabledであること") + log("INFO", "Discordのミュート切替キーが Ctrl+Shift+M であること") + log("INFO", "Discordの現在状態はミュート中として扱います") + + answer = input("[ACTION] 続行しますか？ yes/no: ").strip().lower() + + if answer != "yes": + if answer == "no": + log("FATAL", "ユーザー操作により停止 STEP=2") + wait_exit() + sys.exit(1) + + log("ERROR", "不正入力です STEP=2") + wait_exit() + sys.exit(1) + + log("ACTION", "STEP=3 OSC受信開始") + log("INFO", "VRChatのメニューを押している間だけDiscordミュート解除します") + log("INFO", "停止する場合は Ctrl+C") + + dispatcher = Dispatcher() + dispatcher.map(PARAMETER, handle_discord_send) + + try: + server = BlockingOSCUDPServer((HOST, PORT), dispatcher) + server.serve_forever() + except KeyboardInterrupt: + log("INFO", "Ctrl+C detected") + except Exception as e: + log("FATAL", f"OSC server error STEP=3 detail={e}") + wait_exit() + sys.exit(1) + + log("CHECK", "STEP=4 結果確認") + log("INFO", "終了しました") + +if __name__ == "__main__": + main() diff --git a/src/tools/osc_debug_listener.py b/src/tools/osc_debug_listener.py new file mode 100644 index 0000000..c21b324 --- /dev/null +++ b/src/tools/osc_debug_listener.py @@ -0,0 +1,21 @@ +from pythonosc.dispatcher import Dispatcher +from pythonosc.osc_server import BlockingOSCUDPServer +from datetime import datetime + +HOST = "127.0.0.1" +PORT = 9001 + +def on_any(address, *args): + now = datetime.now().strftime("%H:%M:%S") + print(f"[INFO] {now} address={address} args={args}", flush=True) + +dispatcher = Dispatcher() +dispatcher.set_default_handler(on_any) + +print("[CHECK] OSC受信開始") +print(f"[INFO] host={HOST}") +print(f"[INFO] port={PORT}") +print("[ACTION] VRChatでメニューを押してください") + +server = BlockingOSCUDPServer((HOST, PORT), dispatcher) +server.serve_forever() diff --git a/src/tools/osc_test_print_listener.py b/src/tools/osc_test_print_listener.py new file mode 100644 index 0000000..5b67fb4 --- /dev/null +++ b/src/tools/osc_test_print_listener.py @@ -0,0 +1,54 @@ +from pythonosc.dispatcher import Dispatcher +from pythonosc.osc_server import BlockingOSCUDPServer +from datetime import datetime + +HOST = "127.0.0.1" +PORT = 9001 + +def log(msg): + print(msg, flush=True) + +def on_test(address, *args): + now = datetime.now().strftime("%H:%M:%S") + + log("") + log(f"[INFO] time={now}") + log(f"[ACTION] OSC parameter received") + log(f"[INFO] address={address}") + log(f"[INFO] args={args}") + +dispatcher = Dispatcher() + +dispatcher.map( + "/avatar/parameters/TestPrint", + on_test +) + +log("[CHECK] STEP=1 OSC待受開始") +log(f"[INFO] host={HOST}") +log(f"[INFO] port={PORT}") + +try: + server = BlockingOSCUDPServer( + (HOST, PORT), + dispatcher + ) + + log("[INFO] OSC server started") + log("[ACTION] VRChatでTestPrintを押してください") + + server.serve_forever() + +except KeyboardInterrupt: + log("") + log("[INFO] Ctrl+C detected") + log("[INFO] shutdown") + +except Exception as e: + log("") + log("[FATAL] OSC server error") + log(f"[ERROR] detail={e}") + + print("") + input("Press Enter to exit...") + raise SystemExit(1) diff --git a/src/tools/resolve_vrchat_user.py b/src/tools/resolve_vrchat_user.py new file mode 100644 index 0000000..9841885 --- /dev/null +++ b/src/tools/resolve_vrchat_user.py @@ -0,0 +1,134 @@ +import re +from http.cookiejar import MozillaCookieJar + +import vrchatapi +from vrchatapi.api import authentication_api +from vrchatapi.api import users_api +from vrchatapi.exceptions import UnauthorizedException +from vrchatapi.models.two_factor_auth_code import TwoFactorAuthCode +from vrchatapi.models.two_factor_email_code import TwoFactorEmailCode + +from common.config_loader import getSecretValue + +COOKIE_FILE = ".vrchat_cookie.txt" + +USER_AGENT = ( + "VRC_OSC/1.0 " + "(contact: test@gmail.com)" +) + + +def get_required_secret(section, key): + value = getSecretValue(section, key) + if not value: + raise RuntimeError(f"{section}.{key} is not set in config/secrets.toml") + return value + + +def setup_cookie_jar(api_client): + cookie_jar = MozillaCookieJar(COOKIE_FILE) + + if os.path.exists(COOKIE_FILE): + try: + cookie_jar.load( + ignore_discard=True, + ignore_expires=True, + ) + except Exception: + pass + + api_client.rest_client.cookie_jar = cookie_jar + return cookie_jar + + +def save_cookie_jar(cookie_jar): + try: + cookie_jar.save( + ignore_discard=True, + ignore_expires=True, + ) + except Exception: + pass + + +def login(api_client, cookie_jar): + auth_api = authentication_api.AuthenticationApi(api_client) + + try: + current_user = auth_api.get_current_user() + save_cookie_jar(cookie_jar) + return current_user + + except UnauthorizedException as e: + reason = str(e.reason) + + if "Email 2 Factor Authentication" in reason: + code = input("Email 2FA Code: ").strip() + + auth_api.verify2_fa_email_code( + TwoFactorEmailCode(code) + ) + + elif "2 Factor Authentication" in reason: + code = input("TOTP Code: ").strip() + + auth_api.verify2_fa( + TwoFactorAuthCode(code) + ) + + else: + raise + + current_user = auth_api.get_current_user() + + save_cookie_jar(cookie_jar) + + return current_user + + +def extract_user_id(text): + match = re.search( + r"(usr_[0-9a-fA-F\-]+)", + text, + ) + + if not match: + return None + + return match.group(1) + + +def main(): + text = input("参加者名: ").strip() + + user_id = extract_user_id(text) + + if not user_id: + print(text) + return + + configuration = vrchatapi.Configuration( + username=get_required_secret("vrchat", "username"), + password=get_required_secret("vrchat", "password"), + ) + + with vrchatapi.ApiClient(configuration) as api_client: + api_client.user_agent = USER_AGENT + + api_client.default_headers["User-Agent"] = USER_AGENT + + cookie_jar = setup_cookie_jar(api_client) + + login(api_client, cookie_jar) + + users = users_api.UsersApi(api_client) + + user = users.get_user(user_id) + + print(user.display_name) + + save_cookie_jar(cookie_jar) + + +if __name__ == "__main__": + main() diff --git a/src/translate/translate_actions.py b/src/translate/translate_actions.py new file mode 100644 index 0000000..6df4117 --- /dev/null +++ b/src/translate/translate_actions.py @@ -0,0 +1,102 @@ +from datetime import datetime +import re + +import pyperclip +from deep_translator import GoogleTranslator + +from common.runtime_log import appendRuntimeLog + +IGNORE_EXACT = { + "戻る", + "翻訳", + "OCR", + "VRCLOG", + "Discord_Mute", + "Screenshot", + "OCR_Toggle", + "DCR_Toggle", + "HI", +} + + +def log(level, message): + now = datetime.now().strftime("%H:%M:%S") + print(f"[{level}] {now} {message}", flush=True) + + +def saveTranslationText(stem, text): + return appendRuntimeLog( + "TRANSLATION TEXT", + f"source={stem}\n{text}", + ) + + +def looksLikeNoise(line): + value = line.strip() + if not value: + return True + if value in IGNORE_EXACT: + return True + if len(value) <= 2: + return True + if "_" in value: + return True + if re.fullmatch(r"[A-Za-z0-9\-_/]+", value): + return True + return False + + +def normalizeSourceText(text): + lines = [] + + for raw_line in text.splitlines(): + line = raw_line.strip() + if looksLikeNoise(line): + continue + lines.append(line) + + unique_lines = [] + seen = set() + for line in lines: + if line not in seen: + unique_lines.append(line) + seen.add(line) + + return "\n".join(unique_lines).strip() + + +def translateTextToJapanese(text): + source = normalizeSourceText(text) + if not source: + log("INFO", "翻訳対象が空です") + return "" + + log("CHECK", "Translation source") + for line in source.splitlines(): + print(line, flush=True) + + try: + translated = GoogleTranslator( + source="auto", + target="ja", + ).translate(source) + except Exception as e: + log("ERROR", f"translation failed detail={e}") + return "" + + translated = translated.strip() + if not translated: + log("INFO", "翻訳結果は空です") + return "" + + try: + pyperclip.copy(translated) + log("INFO", "翻訳結果をクリップボードへコピーしました") + except Exception as e: + log("ERROR", f"clipboard copy failed detail={e}") + + log("CHECK", "Translation result") + for line in translated.splitlines(): + print(line, flush=True) + + return translated diff --git a/src/vrc_log/log_actions.py b/src/vrc_log/log_actions.py new file mode 100644 index 0000000..c70a385 --- /dev/null +++ b/src/vrc_log/log_actions.py @@ -0,0 +1,565 @@ +import os +import re +import subprocess +import tempfile +import threading +import time +from datetime import datetime +from datetime import timedelta +from pathlib import Path + +from common.config_loader import loadVrcLogConfig +from common.config_loader import getSecretValue +from common.runtime_log import appendRuntimeLog +from discord_control.actions import setDiscordMuted + +_monitor_thread = None +_monitor_stop_event = threading.Event() +_last_sent_output = None +_self_monitor_thread = None +_self_monitor_stop_event = threading.Event() +_last_self_state = None +REALTIME_IDLE_SECONDS = 0.2 + +JOIN_PATTERN = re.compile( + r"OnPlayerJoined\s+(.+?)\s+\(usr_[0-9a-fA-F-]+\)" +) +LEFT_PATTERN = re.compile( + r"OnPlayerLeft\s+(.+?)\s+\(usr_[0-9a-fA-F-]+\)" +) +WORLD_PATTERN = re.compile( + r"(wrld_[0-9a-fA-F-]+(?::[^\s\]\)\"']+)?)" +) +ENTERING_ROOM_PATTERN = re.compile( + r"\[Behaviour\]\s+(?:Entering Room|Joining or Creating Room):\s+(.+)$" +) +WORLD_NAME_PATTERN = re.compile( + r"worldId=(wrld_[0-9a-fA-F-]+)(?::[^,}]*)?,\s*worldName=([^,}]+)" +) +TIME_PATTERN = re.compile( + r"(\d{4}\.\d{2}\.\d{2}\s+\d{2}:\d{2}:\d{2})" +) +RELEVANT_LOG_PATTERN = re.compile( + r"OnPlayerJoined|OnPlayerLeft|Entering Room|Joining or Creating Room|worldId=|wrld_" +) + + +def log(level, message): + now = datetime.now().strftime("%H:%M:%S") + print(f"[{level}] {now} {message}", flush=True) + + +def getWebhookUrl(): + return getSecretValue("discord", "webhook_url") + + +def sendWebhook(text): + webhook_url = getWebhookUrl().strip() + if not webhook_url: + log("ERROR", "Discord webhook URL is not set config=config/secrets.toml") + return False + + message = str(text).strip() + if not message: + log("INFO", "Webhook送信内容が空です") + return False + + if len(message) > 1800: + message = message[-1800:] + + ps_script = r''' +param( + [string]$WebhookUrl, + [string]$Message +) + +[Console]::OutputEncoding = [System.Text.Encoding]::UTF8 +$Payload = @{ content = $Message } | ConvertTo-Json -Compress +$Body = [System.Text.Encoding]::UTF8.GetBytes($Payload) +Invoke-WebRequest -Uri $WebhookUrl -Method Post -ContentType 'application/json; charset=utf-8' -Body $Body -UseBasicParsing | Out-Null +''' + + ps_path = None + + try: + with tempfile.NamedTemporaryFile( + mode="w", + suffix=".ps1", + delete=False, + encoding="utf-8-sig", + ) as tmp: + tmp.write(ps_script) + ps_path = tmp.name + + result = subprocess.run( + [ + "powershell", + "-NoProfile", + "-ExecutionPolicy", + "Bypass", + "-File", + ps_path, + webhook_url, + message, + ], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=20, + ) + except Exception as e: + log("ERROR", f"Discord webhook PowerShell failed detail={e}") + return False + finally: + if ps_path: + try: + Path(ps_path).unlink(missing_ok=True) + except Exception: + pass + + if result.returncode != 0: + stderr = (result.stderr or "").strip() + stdout = (result.stdout or "").strip() + log("ERROR", f"Discord webhook PowerShell failed code={result.returncode} stderr={stderr} stdout={stdout}") + return False + + log("INFO", "Discord webhook sent via PowerShell") + return True + + +def getVrchatLogDir(): + userprofile = os.environ.get("USERPROFILE") + if not userprofile: + raise RuntimeError("USERPROFILE が取得できません") + return Path(userprofile) / "AppData" / "LocalLow" / "VRChat" / "VRChat" + + +def findLatestVrchatLog(): + log_dir = getVrchatLogDir() + if not log_dir.exists(): + raise RuntimeError(f"VRChatログディレクトリが存在しません: {log_dir}") + + candidates = [] + for pattern in ("output_log_*.txt", "Player.log", "*.log", "*.txt"): + candidates.extend(log_dir.glob(pattern)) + + candidates = [path for path in candidates if path.is_file()] + if not candidates: + raise RuntimeError(f"VRChatログファイルが見つかりません: {log_dir}") + + candidates.sort(key=lambda path: path.stat().st_mtime, reverse=True) + return candidates[0] + + +def readTextSafe(path): + for encoding in ("utf-8", "utf-8-sig", "cp932", "shift_jis"): + try: + return path.read_text(encoding=encoding, errors="replace") + except Exception: + continue + + raise RuntimeError(f"ログファイルを読めません: {path}") + + +def parseLineTime(line): + matched = TIME_PATTERN.search(line) + if not matched: + return None + + return datetime.strptime(matched.group(1), "%Y.%m.%d %H:%M:%S") + + +def formatClock(dt): + return dt.strftime("%H:%M") + + +def getLogStartTime(now): + noon_today = now.replace(hour=12, minute=0, second=0, microsecond=0) + if now >= noon_today: + return noon_today + return noon_today - timedelta(days=1) + + +def splitWorldLocation(location): + if ":" not in location: + return location, "" + + world_id, instance_id = location.split(":", 1) + return world_id, instance_id + + +def extractWorldNameMap(text): + world_name_map = {} + pending_world_name = "" + + for raw_line in text.splitlines(): + room_match = ENTERING_ROOM_PATTERN.search(raw_line) + if room_match: + pending_world_name = room_match.group(1).strip() + + match = WORLD_NAME_PATTERN.search(raw_line) + if match: + world_id = match.group(1).strip() + world_name = match.group(2).strip() + if world_id and world_name: + world_name_map[world_id] = world_name + continue + + world_match = WORLD_PATTERN.search(raw_line) + if world_match and pending_world_name: + location = world_match.group(1).strip() + world_id, _ = splitWorldLocation(location) + if world_id: + world_name_map[world_id] = pending_world_name + pending_world_name = "" + + return world_name_map + + +def formatWorldLabel(world_id, world_name_map): + world_name = world_name_map.get(world_id, "").strip() + if not world_name: + return "不明" + return world_name + + +def classifyUser(name, guest_names, staff_names): + if name in staff_names: + return "staff" + if name in guest_names: + return "guest" + return "other" + + +def isSelf(name, self_name): + return bool(self_name) and name.strip() == self_name.strip() + + +def buildNoticeLines(guest_names, guest_present_set, missing_count): + total_guest = len(guest_names) + current_guest = len(guest_present_set) + threshold = total_guest - missing_count + + if missing_count <= 0: + return [] + if not (threshold <= current_guest < total_guest): + return [] + + missing_guests = [name for name in guest_names if name not in guest_present_set] + if not missing_guests: + return [] + + lines = [f"現在Guest: {current_guest}/{total_guest}", "未入室Guest:"] + lines.extend(f"- {name}" for name in missing_guests) + return lines + + +def appendNoticeIfNeeded(output_lines, state): + notice_lines = buildNoticeLines( + state["guest_names"], + state["guest_present_set"], + state["missing_count"], + ) + notice_key = tuple(notice_lines) + + if not notice_lines: + state["last_notice_key"] = None + return + + if state["last_notice_key"] == notice_key: + return + + output_lines.extend(notice_lines) + state["last_notice_key"] = notice_key + + +def parseVrchatEvents(text, config): + start_time = getLogStartTime(datetime.now()) + guest_names = config["guest_names"] + staff_names = set(config["staff_names"]) + self_name = config.get("self_name", "") + world_name_map = extractWorldNameMap(text) + + state = { + "location": "", + "world_id": "", + "instance_id": "", + "guest_names": guest_names, + "guest_name_set": set(guest_names), + "staff_names": staff_names, + "guest_present_set": set(), + "staff_present_set": set(), + "member_present_set": set(), + "missing_count": config["missing_count"], + "last_notice_key": None, + } + output_lines = [] + self_state = None + + for raw_line in text.splitlines(): + line_time = parseLineTime(raw_line) + if not line_time or line_time < start_time: + continue + + world_match = WORLD_PATTERN.search(raw_line) + if world_match: + location = world_match.group(1).strip() + world_id, instance_id = splitWorldLocation(location) + + if location != state["location"]: + state["location"] = location + state["world_id"] = world_id + state["instance_id"] = instance_id + state["guest_present_set"].clear() + state["staff_present_set"].clear() + state["member_present_set"].clear() + state["last_notice_key"] = None + output_lines = [] + output_lines.append(f"ワールド入室: {formatWorldLabel(world_id, world_name_map)}") + output_lines.append(f"現在Guest: 0/{len(guest_names)}") + + join_match = JOIN_PATTERN.search(raw_line) + if join_match and state["world_id"]: + name = join_match.group(1).strip() + if isSelf(name, self_name): + self_state = "joined" + state["member_present_set"].add(name) + output_lines.append( + f'[{formatClock(line_time)}] [join] [self] {name} {len(state["member_present_set"])}/{len(guest_names)}' + ) + continue + + role = classifyUser(name, state["guest_name_set"], state["staff_names"]) + + if role == "guest": + state["guest_present_set"].add(name) + state["member_present_set"].add(name) + output_lines.append( + f'[{formatClock(line_time)}] [join] {name} {len(state["member_present_set"])}/{len(guest_names)}' + ) + appendNoticeIfNeeded(output_lines, state) + elif role == "staff": + state["staff_present_set"].add(name) + output_lines.append( + f'[{formatClock(line_time)}] [join] [staff] {name} {len(state["member_present_set"])}/{len(guest_names)}' + ) + else: + state["member_present_set"].add(name) + output_lines.append( + f'[{formatClock(line_time)}] [join] {name} {len(state["member_present_set"])}/{len(guest_names)}' + ) + continue + + left_match = LEFT_PATTERN.search(raw_line) + if left_match and state["world_id"]: + name = left_match.group(1).strip() + if isSelf(name, self_name): + self_state = "left" + state["member_present_set"].discard(name) + output_lines.append( + f'[{formatClock(line_time)}] [leave] [self] {name} {len(state["member_present_set"])}/{len(guest_names)}' + ) + continue + + role = classifyUser(name, state["guest_name_set"], state["staff_names"]) + + if role == "guest": + state["guest_present_set"].discard(name) + state["member_present_set"].discard(name) + output_lines.append( + f'[{formatClock(line_time)}] [leave] {name} {len(state["member_present_set"])}/{len(guest_names)}' + ) + appendNoticeIfNeeded(output_lines, state) + elif role == "staff": + state["staff_present_set"].discard(name) + output_lines.append( + f'[{formatClock(line_time)}] [leave] [staff] {name} {len(state["member_present_set"])}/{len(guest_names)}' + ) + else: + state["member_present_set"].discard(name) + output_lines.append( + f'[{formatClock(line_time)}] [leave] {name} {len(state["member_present_set"])}/{len(guest_names)}' + ) + + return output_lines, self_state + + +def extractLatestSelfLogState(text, self_name): + if not self_name: + return None + + latest_state = None + + for raw_line in text.splitlines(): + join_match = JOIN_PATTERN.search(raw_line) + if join_match and isSelf(join_match.group(1).strip(), self_name): + latest_state = "joined" + continue + + left_match = LEFT_PATTERN.search(raw_line) + if left_match and isSelf(left_match.group(1).strip(), self_name): + latest_state = "left" + + return latest_state + + +def grepVrchatLog(pattern=None): + return collectVrchatLog(pattern=pattern, notify_changed=False) + + +def collectVrchatLog(pattern=None, notify_changed=True): + config = loadVrcLogConfig() + latest_log = findLatestVrchatLog() + log("ACTION", f"parseVrchatLog file={latest_log}") + log("INFO", f'event_file={config["event_file"]}') + + text = readTextSafe(latest_log) + lines, self_state = parseVrchatEvents(text, config) + + output_text = "\n".join(lines) if lines else "[INFO] 対象イベントなし" + output_path = appendRuntimeLog("VRC LOG", output_text) + + if notify_changed: + global _last_sent_output + if output_text != _last_sent_output: + try: + ok = sendWebhook(output_text) + if ok: + log("INFO", "Discord webhook sent") + _last_sent_output = output_text + else: + log("ERROR", "Discord webhook send failed") + except Exception as e: + log("ERROR", f"Discord webhook failed detail={e}") + + log("INFO", f"result appended={output_path}") + for line in lines or ["[INFO] 対象イベントなし"]: + print(line, flush=True) + + return output_path + + +def _readNewText(path, offset): + for encoding in ("utf-8", "utf-8-sig", "cp932", "shift_jis"): + try: + with path.open("r", encoding=encoding, errors="replace") as file: + file.seek(offset) + return file.read(), file.tell() + except Exception: + continue + + raise RuntimeError(f"ログファイルを読めません: {path}") + + +def _watchVrchatLogChanges(stop_event, label, on_relevant_change): + log("ACTION", f"{label} realtime watcher started") + + current_log = None + offset = 0 + + while not stop_event.is_set(): + try: + latest_log = findLatestVrchatLog() + latest_size = latest_log.stat().st_size + + if current_log is None: + current_log = latest_log + offset = latest_size + log("INFO", f"{label} watching file={current_log}") + continue + + if latest_log != current_log: + current_log = latest_log + offset = 0 + log("INFO", f"{label} watching new file={current_log}") + + if latest_size < offset: + offset = 0 + + if latest_size > offset: + text, offset = _readNewText(latest_log, offset) + + if RELEVANT_LOG_PATTERN.search(text): + on_relevant_change(text) + except Exception as e: + log("ERROR", f"{label} realtime watcher failed detail={e}") + + stop_event.wait(REALTIME_IDLE_SECONDS) + + log("INFO", f"{label} realtime watcher stopped") + + +def _monitor_loop(): + def on_relevant_change(_text): + collectVrchatLog(notify_changed=True) + + _watchVrchatLogChanges( + _monitor_stop_event, + "VRC log monitor", + on_relevant_change, + ) + + +def _self_monitor_loop(): + def on_relevant_change(text): + try: + config = loadVrcLogConfig() + self_state = extractLatestSelfLogState( + text, + config.get("self_name", ""), + ) + + global _last_self_state + if self_state and self_state != _last_self_state: + if self_state == "joined": + setDiscordMuted(True) + elif self_state == "left": + setDiscordMuted(False) + _last_self_state = self_state + except Exception as e: + log("ERROR", f"VRC self monitor failed detail={e}") + + _watchVrchatLogChanges( + _self_monitor_stop_event, + "VRC self monitor", + on_relevant_change, + ) + + +def startVrcLogMonitor(_interval_seconds=None): + global _monitor_thread + + if _monitor_thread and _monitor_thread.is_alive(): + log("INFO", "VRC log monitor already running") + return + + _monitor_stop_event.clear() + _monitor_thread = threading.Thread( + target=_monitor_loop, + daemon=True, + ) + _monitor_thread.start() + + +def startSelfMonitor(_interval_seconds=None): + global _self_monitor_thread + + if _self_monitor_thread and _self_monitor_thread.is_alive(): + log("INFO", "VRC self monitor already running") + return + + _self_monitor_stop_event.clear() + _self_monitor_thread = threading.Thread( + target=_self_monitor_loop, + daemon=True, + ) + _self_monitor_thread.start() + + +def stopVrcLogMonitor(): + _monitor_stop_event.set() + + +def stopSelfMonitor(): + _self_monitor_stop_event.set() diff --git a/src/vrc_log/member_name_matcher.py b/src/vrc_log/member_name_matcher.py new file mode 100644 index 0000000..efe2a34 --- /dev/null +++ b/src/vrc_log/member_name_matcher.py @@ -0,0 +1,623 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +""" +目的: +VRChat表示名とTOML登録名を安全な順番で紐づける + +特徴: +- 完全一致を先に確定して候補から除外 +- 正規化一致を確定して候補から除外 +- 残った候補だけを high confidence fuzzy 判定 +- guest のみ low confidence fuzzy で再選別 +- pending / ambiguous / unmatched 判定 +- 記号除去 +- 全角半角統一 +- 小文字化 +- 日本語 -> ローマ字変換 +- rapidfuzz による類似度 +- テストケース付き + +前提: +pip install rapidfuzz pykakasi + +期待結果: +- 100%一致した名前を fuzzy 候補に残さない +- 似た名前同士の誤リンクを減らす +""" + +import re +import sys +import unicodedata +from dataclasses import dataclass, field +from typing import Dict, Iterable, List, Optional, Tuple + +try: + from rapidfuzz import fuzz +except Exception as e: + if __name__ == "__main__": + print(f"[FATAL] rapidfuzz import failed: {e}") + input("Press Enter to exit...") + sys.exit(1) + raise + +try: + from pykakasi import kakasi +except Exception as e: + if __name__ == "__main__": + print(f"[FATAL] pykakasi import failed: {e}") + input("Press Enter to exit...") + sys.exit(1) + raise + + +# ========================= +# 設定 +# ========================= + +AUTO_LINK_SCORE = 95 +LOW_CONFIDENCE_SCORE = 80 +AMBIGUOUS_DIFF = 10 +LOW_CONFIDENCE_MAX_REMAINING = 3 +WEAK_MATCH_SCORE = 50 + +STATUS_MATCHED = "matched" +STATUS_MATCHED_LOW_CONFIDENCE = "matched_low_confidence" +STATUS_PENDING = "pending" +STATUS_AMBIGUOUS = "ambiguous" +STATUS_UNMATCHED = "unmatched" +STATUS_UNKNOWN_MEMBER = "unknown_member" + +ROLE_GUEST = "guest" +ROLE_SELF = "self" +ROLE_STAFF = "staff" + +TEST_GUESTS = [ + "okada", + "tanaka", + "yamada", + "messykenty", +] + +TEST_CURRENT_USERS = [ + "岡田", + "ikada", + "okada_", + "ＯＫＡＤＡ", + "tanaka!!!", + "田中", + "やまだ", + "messy_kenty", + "messy.kenty", + "MESSYKENTY", + "山田", +] + + +@dataclass(frozen=True) +class TomlMember: + name: str + role: str = ROLE_GUEST + + +@dataclass +class MatchResult: + toml_name: str + role: str + status: str + vrchat_name: Optional[str] = None + score: Optional[float] = None + second_score: Optional[float] = None + reason: str = "" + candidates: List[Tuple[str, float]] = field(default_factory=list) + + +# ========================= +# ローマ字変換 +# ========================= + +kks = kakasi() + + +def jp_to_romaji(text: str) -> str: + try: + converted = kks.convert(text) + + result = [] + for item in converted: + result.append(item["hepburn"]) + + return "".join(result) + + except Exception as e: + print(f"[ERROR] romaji convert failed: {text} : {e}") + return text + + +# ========================= +# 正規化 +# ========================= + +def normalize_name(name: str) -> str: + """ + 正規化内容: + - 全角半角統一 + - 小文字化 + - 日本語→ローマ字 + - 記号除去 + """ + + # Unicode normalize + name = unicodedata.normalize("NFKC", name) + + # 小文字化 + name = name.lower() + + # 日本語→ローマ字 + name = jp_to_romaji(name) + + # 記号除去 + name = re.sub(r"[^a-z0-9]", "", name) + + return name.strip() + + +# ========================= +# 類似候補検索 +# ========================= + +def find_candidates( + target_guest: str, + current_users: List[str], +) -> List[Tuple[str, float]]: + + normalized_target = normalize_name(target_guest) + + normalized_map = {} + + for user in current_users: + normalized_map[user] = normalize_name(user) + + results = [] + + for original_name, normalized_name in normalized_map.items(): + + score = fuzz.ratio( + normalized_target, + normalized_name, + ) + + results.append((original_name, score)) + + results.sort(key=lambda x: x[1], reverse=True) + + return results + + +def _to_toml_member(member) -> TomlMember: + if isinstance(member, TomlMember): + return member + + if isinstance(member, str): + return TomlMember(name=member) + + if isinstance(member, dict): + return TomlMember( + name=str(member.get("name", "")), + role=str(member.get("role", ROLE_GUEST)), + ) + + name = getattr(member, "name", "") + role = getattr(member, "role", ROLE_GUEST) + + return TomlMember( + name=str(name), + role=str(role), + ) + + +def _make_candidates( + toml_name: str, + current_users: Iterable[str], +) -> List[Tuple[str, float]]: + normalized_toml = normalize_name(toml_name) + results = [] + + for user in current_users: + score = fuzz.ratio( + normalized_toml, + normalize_name(user), + ) + results.append((user, score)) + + results.sort(key=lambda x: x[1], reverse=True) + + return results + + +def _classify_remaining( + member: TomlMember, + current_users: Iterable[str], +) -> MatchResult: + candidates = _make_candidates( + member.name, + current_users, + ) + + if not candidates: + return MatchResult( + toml_name=member.name, + role=member.role, + status=STATUS_UNMATCHED, + reason="no_candidate", + ) + + top_name, top_score = candidates[0] + second_score = candidates[1][1] if len(candidates) >= 2 else 0 + diff = top_score - second_score + + if top_score < WEAK_MATCH_SCORE: + status = STATUS_UNMATCHED + reason = "weak_candidate" + elif diff < AMBIGUOUS_DIFF: + status = STATUS_AMBIGUOUS + reason = "close_candidates" + else: + status = STATUS_PENDING + reason = "below_auto_link_threshold" + + return MatchResult( + toml_name=member.name, + role=member.role, + status=status, + vrchat_name=top_name, + score=top_score, + second_score=second_score, + reason=reason, + candidates=candidates, + ) + + +def _best_fuzzy_match( + member: TomlMember, + current_users: Iterable[str], + min_score: float, +) -> Optional[MatchResult]: + candidates = _make_candidates( + member.name, + current_users, + ) + + if not candidates: + return None + + top_name, top_score = candidates[0] + second_score = candidates[1][1] if len(candidates) >= 2 else 0 + + if ( + top_score >= min_score + and top_score - second_score >= AMBIGUOUS_DIFF + ): + return MatchResult( + toml_name=member.name, + role=member.role, + status=STATUS_MATCHED, + vrchat_name=top_name, + score=top_score, + second_score=second_score, + reason="fuzzy", + candidates=candidates, + ) + + return None + + +def _apply_fuzzy_round( + remaining_members: Dict[str, TomlMember], + remaining_users: set, + min_score: float, + status: str, + reason: str, + eligible_roles: Optional[set] = None, +) -> List[MatchResult]: + results = [] + + while True: + possible_matches = [] + + for key, member in remaining_members.items(): + if ( + eligible_roles is not None + and member.role not in eligible_roles + ): + continue + + match = _best_fuzzy_match( + member, + remaining_users, + min_score, + ) + + if match is None: + continue + + possible_matches.append((key, match)) + + if not possible_matches: + break + + possible_matches.sort( + key=lambda x: ( + x[1].score if x[1].score is not None else 0, + (x[1].score or 0) - (x[1].second_score or 0), + ), + reverse=True, + ) + + key, match = possible_matches[0] + match.status = status + match.reason = reason + + results.append(match) + remaining_members.pop(key) + remaining_users.remove(match.vrchat_name) + + return results + + +def match_username_sets( + toml_members, + current_users: List[str], +) -> Tuple[List[MatchResult], List[str]]: + """ + TOML登録名とVRChat現在ユーザーを段階的に1対1照合する。 + + 戻り値: + - TOML側の照合結果 + - TOMLに紐づかなかったVRChat名 + """ + + members = [_to_toml_member(member) for member in toml_members] + remaining_members = { + f"{idx}:{member.name}": member + for idx, member in enumerate(members) + } + remaining_users = set(current_users) + results = [] + + # 1. 完全一致: fuzzy判定せず即確定して候補から除外 + for key, member in list(remaining_members.items()): + if member.name in remaining_users: + results.append( + MatchResult( + toml_name=member.name, + role=member.role, + status=STATUS_MATCHED, + vrchat_name=member.name, + score=100, + second_score=None, + reason="exact", + ) + ) + remaining_members.pop(key) + remaining_users.remove(member.name) + + # 2. 正規化一致: 一意に対応できるものだけを確定して除外 + normalized_members: Dict[str, List[str]] = {} + normalized_users: Dict[str, List[str]] = {} + + for key, member in remaining_members.items(): + normalized_members.setdefault( + normalize_name(member.name), + [], + ).append(key) + + for user in remaining_users: + normalized_users.setdefault( + normalize_name(user), + [], + ).append(user) + + for normalized_name, member_keys in normalized_members.items(): + users = normalized_users.get(normalized_name, []) + + if len(member_keys) != 1 or len(users) != 1: + continue + + key = member_keys[0] + user = users[0] + member = remaining_members[key] + + results.append( + MatchResult( + toml_name=member.name, + role=member.role, + status=STATUS_MATCHED, + vrchat_name=user, + score=100, + second_score=None, + reason="normalized", + ) + ) + remaining_members.pop(key) + remaining_users.remove(user) + + # 3. 高信頼fuzzy: 残った候補だけで、確定ごとに候補集合を縮める + results.extend( + _apply_fuzzy_round( + remaining_members, + remaining_users, + AUTO_LINK_SCORE, + STATUS_MATCHED, + "high_confidence_fuzzy", + ) + ) + + # 4. 低信頼guest再選別: guestのみ、候補集合が少ない場合だけ + if len(remaining_members) <= LOW_CONFIDENCE_MAX_REMAINING: + results.extend( + _apply_fuzzy_round( + remaining_members, + remaining_users, + LOW_CONFIDENCE_SCORE, + STATUS_MATCHED_LOW_CONFIDENCE, + "guest_low_confidence_fuzzy", + eligible_roles={ROLE_GUEST}, + ) + ) + + # 5. 残りは無理に確定しない + for member in remaining_members.values(): + results.append( + _classify_remaining( + member, + remaining_users, + ) + ) + + unknown_users = sorted(remaining_users) + + return results, unknown_users + + +# ========================= +# 判定 +# ========================= + +def evaluate_guest_match( + guest_name: str, + current_users: List[str], +): + + print("") + print("=" * 60) + print(f"[CHECK] guest={guest_name}") + + normalized_guest = normalize_name(guest_name) + + print(f"[INFO] normalized_guest={normalized_guest}") + + candidates = find_candidates( + guest_name, + current_users, + ) + + if not candidates: + print("[ERROR] no candidates") + return + + print("") + print("[CHECK] candidate list") + + for idx, (name, score) in enumerate(candidates[:5], start=1): + print(f" {idx}. {name:<20} score={score:.2f}") + + top_name, top_score = candidates[0] + + second_score = 0 + + if len(candidates) >= 2: + second_score = candidates[1][1] + + diff = top_score - second_score + + print("") + print(f"[INFO] top_score={top_score:.2f}") + print(f"[INFO] second_score={second_score:.2f}") + print(f"[INFO] diff={diff:.2f}") + + # 自動確定 + if ( + top_score >= AUTO_LINK_SCORE + and diff >= AMBIGUOUS_DIFF + ): + print("") + print("[INFO] AUTO LINK SAFE") + print(f"[INFO] linked => {top_name}") + return + + # 曖昧 + print("") + print("[WARN] ambiguous match") + print("[WARN] auto link disabled") + + print("") + print("[ACTION] manual review required") + + +# ========================= +# main +# ========================= + +def main(): + + print("[CHECK] normalize preview") + + for user in TEST_CURRENT_USERS: + print( + f"{user:<20}" + f" => " + f"{normalize_name(user)}" + ) + + print("") + print("[CHECK] username sync ordered matching test") + + results, unknown_users = match_username_sets( + TEST_GUESTS, + TEST_CURRENT_USERS, + ) + + for result in results: + if result.vrchat_name is None: + print( + f"{result.toml_name:<20}" + f" => " + f"{result.status:<22}" + f" reason={result.reason}" + ) + continue + + score_text = "" + if result.score is not None: + score_text = f" score={result.score:.2f}" + + second_text = "" + if result.second_score is not None: + second_text = f" second={result.second_score:.2f}" + + print( + f"{result.toml_name:<20}" + f" => " + f"{result.vrchat_name:<20}" + f" {result.status:<22}" + f" reason={result.reason}" + f"{score_text}" + f"{second_text}" + ) + + for user in unknown_users: + print( + f"{user:<20}" + f" => " + f"{STATUS_UNKNOWN_MEMBER}" + ) + + print("") + print("[INFO] completed") + + +if __name__ == "__main__": + + print("[CHECK] start") + + answer = input( + "[ACTION] run fuzzy test? (yes/no): " + ).strip().lower() + + if answer != "yes": + print("[FATAL] cancelled") + input("Press Enter to exit...") + sys.exit(1) + + main()