#!/usr/bin/env python3 # BASE_TARGET_V094: http://149.118.48.126:8080/persistent/scripts/watcher.py (67,160B | 1,477L | md5=cb11d45f639722650e3ebe156897da51 | sha256=b634251dc94632819cd7e1d69c8b90a2fe9f052a56df11b8b40328a066f111e2 | measured_at=2026-10-06T11:36:00Z UTC) # BASE_ROOT_8080 : http://149.118.48.126:8080/watcher.py (62,239B | 1,379L | md5=797357a7755baa381bee249121e05b3d | sha256=1b89afc9655d19a0b73647ebc8dae36fd8cf442a28196694f8e1c968537f7251 | measured_at=2026-10-06T11:36:00Z UTC) # SPEC_V3_3_PIN : watcher-url-rotation-spec-v3.3.txt (29,999B | 328L | md5=086d1debba34ca92b1640ef7d6177326 | sha256=43746d5fd5245f47fddba4b9c9ad8df32cfdeb63b3e9ac06490abfbd00d048b3 | measured_at=2026-10-06T11:36:00Z UTC) """ Arena 持久 DOM Watcher - 陣列式多席位系統（v3 + ca.js 精準版） 每個席位一個獨立長駐 browser，每 30 秒一個 cycle - 見 Stop 掣（生成緊）→ 唔郁 - 見 Keep working → 撳 → 等 5 秒 → reload → 等頁面生成 → 等 5 秒 → 確認消失 → send - 閒置 → 直接 send - Send 後：等 10 秒 → reload → 等生成 → 等 5 秒 → 有 Stop 掣/Keep working = 成功 - Browser crash → 自動重開 - 登入檢查：跟 ca.js（電郵 + Log In 掣），未登入自動行 login flow（見密碼即停手） - 不登出、不清 cookies/cache/profile 用法: watcher.py <席位ID> [--once] --once: 只跑一次（測試用） """ import sys import os import time import json from pathlib import Path # 配置 SYSTEM_DIR = Path.home() / "workspace" / "arena-system" SEATS_CONF = SYSTEM_DIR / "seats.conf" VPN_CONF = Path.home() / "workspace" / "vpn-tunnel" / "vpn-proxies.conf" PROFILES_DIR = Path.home() / "workspace" / "arena-system" / "local-profiles" # 直接用 workspace，唔用 RAM（owner 2026-10-05） # v0.93: CA 兩套 checkbox 完整跟足（owner 2026-10-05 15:38 下令） # 第1套（橙色）：檢測 (527,409) #f6821f → 點擊 (270,420) # 第2套（藍色 Cloudflare）：檢測 (910,420) 200x60 #1A73E8 → 點擊 (276,425) # v0.93 第三套（紫色） PURPLE_CHECK_X = 715 PURPLE_CHECK_Y = 415 PURPLE_CHECK_COLOR = "#f3e9f7" PURPLE_CLICK_X = 580 PURPLE_CLICK_Y = 605 ORANGE_CHECK_X = 527 ORANGE_CHECK_Y = 409 ORANGE_CHECK_COLOR = '#f6821f' ORANGE_CLICK_X = 270 ORANGE_CLICK_Y = 420 BLUE_CHECK_X = 910 BLUE_CHECK_Y = 420 BLUE_CHECK_W = 200 BLUE_CHECK_H = 60 BLUE_CHECK_COLOR = '#1A73E8' BLUE_CLICK_X = 276 BLUE_CLICK_Y = 425 MASTER_PROFILES_DIR = SYSTEM_DIR / "profiles" # 只讀 master LOGS_DIR = SYSTEM_DIR / "logs" RECALL_MSG = "Continue SESSION 2147483647" RECALL_BASE = 2147483647 # 固定起始數字（owner 2026-10-04 14:58 HKT：每次推減1，重啟重置） _recall_counter = RECALL_BASE # 內存計數器，重啟 watcher 即重置 _last_sent_num = None # 上次實際發出嘅數字（畀精準驗證用） def get_recall_msg(): """每次推攞一條新召回句：第一次 2147483647，第二次 2147483646，如此類推。 watcher 重啟 → 計數器重置返 2147483647。""" global _recall_counter, _last_sent_num num = _recall_counter _recall_counter -= 1 _last_sent_num = num return f"Continue SESSION {num}" CHECK_INTERVAL = 30 # 秒 def load_seats(): seats = {} with open(SEATS_CONF) as f: for line in f: line = line.strip() if not line or line.startswith("#"): continue parts = line.split() if len(parts) >= 6: sid, profile, vpn, url, account = parts[0], parts[1], parts[2], parts[3], parts[4] status = parts[5] if len(parts) > 5 else "active" seats[sid] = { "profile": profile, "vpn": vpn, "url": url, "account": account, "status": status } return seats # === URL 輪換 + Zoom 100% 硬鎖 + _seat_window_id / W1-W5 視窗定位 v5.8 (Spec v3.3 Conformance) === # BASE_TARGET_V094: http://149.118.48.126:8080/persistent/scripts/watcher.py (67,160B | 1,477L | md5=cb11d45f639722650e3ebe156897da51 | sha256=b634251dc94632819cd7e1d69c8b90a2fe9f052a56df11b8b40328a066f111e2 | measured_at=2026-10-06T11:36:00Z UTC) # BASE_ROOT_8080 : http://149.118.48.126:8080/watcher.py (62,239B | 1,379L | md5=797357a7755baa381bee249121e05b3d | sha256=1b89afc9655d19a0b73647ebc8dae36fd8cf442a28196694f8e1c968537f7251 | measured_at=2026-10-06T11:36:00Z UTC) # SPEC_V3_3_PIN : watcher-url-rotation-spec-v3.3.txt (29,999B | 328L | md5=086d1debba34ca92b1640ef7d6177326 | sha256=43746d5fd5245f47fddba4b9c9ad8df32cfdeb63b3e9ac06490abfbd00d048b3 | measured_at=2026-10-06T11:36:00Z UTC) import fcntl, json, os, re, shutil, subprocess, time, urllib.request, urllib.error ROTATION_TAILS = {"A2": "2", "A3": "3", "A4": "4"} ROTATION_SERVER_URL_TEMPLATE = "http://149.118.48.126:8080/url-rotation/url-rotation-tail{tail}.json" ROTATION_HTTP_TIMEOUT_S = 3 _DEFAULT_SYSTEM_DIR = SYSTEM_DIR TARGET_VIEWPORT_W = 1400 TARGET_VIEWPORT_H = 900 EXPECTED_WIN_W = 1408 EXPECTED_WIN_H = 900 _LAST_VALID_ENTRIES = {} _LAST_VALID_INDEX = {} _LAST_VALID_LABEL = {} _PREFS_CLEANED_AT_BOOT = set() _SEAT_DPR_BASELINE = {"A2": 1.0, "A3": 1.0, "A4": 1.0} _BASELINE_LOGGED = set() def _seats_fallback_url(sid): try: sconf = SYSTEM_DIR / "seats.conf" if sconf.exists(): for line in sconf.read_text(encoding="utf-8", errors="ignore").splitlines(): line = line.strip() if not line or line.startswith("#"): continue parts = line.split() if len(parts) >= 4 and parts[0] == sid: return parts[3] except Exception: pass return "" def _slot_target_y(sid): """由 SLOT_Y_POS 或 seats.conf 席位順序（三個一組 0/900/1800）決定目標 Y 座標。""" if "SLOT_Y_POS" in globals() and sid in SLOT_Y_POS: return int(SLOT_Y_POS[sid]) try: sconf = SYSTEM_DIR / "seats.conf" if sconf.exists(): seat_ids = [] for line in sconf.read_text(encoding="utf-8", errors="ignore").splitlines(): line = line.strip() if not line or line.startswith("#"): continue parts = line.split() if parts: seat_ids.append(parts[0]) if sid in seat_ids: return (seat_ids.index(sid) % 3) * 900 except Exception: pass m = re.match(r"^A(\d+)$", str(sid or "")) if m: return ((int(m.group(1)) - 2) % 3) * 900 return 0 def _save_corrupt_copy(bad_path): try: ts_utc = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) bak = bad_path.with_name(f"{bad_path.name}.corrupt-{ts_utc}") if not bak.exists(): shutil.copy2(bad_path, bak) return bak.name except Exception: return f"{bad_path.name}.corrupt-err" def _fetch_server_tail_entries(sid, tail): # Tier 1 (主持 47600)：先經 HTTP 讀 http://149.118.48.126:8080/url-rotation/url-rotation-tail{tail}.json（server 唯一本） if SYSTEM_DIR != _DEFAULT_SYSTEM_DIR: return None srv_url = ROTATION_SERVER_URL_TEMPLATE.format(tail=tail) try: req = urllib.request.Request(srv_url, headers={"Cache-Control": "no-cache"}) with urllib.request.urlopen(req, timeout=ROTATION_HTTP_TIMEOUT_S) as resp: if getattr(resp, "status", 200) != 200: log(sid, f"⚠️ [URL-ROT] [HTTP_FALLBACK] Server HTTP {getattr(resp, 'status', None)} ({srv_url})，fallback 本機 JSON") return None raw_text = resp.read().decode("utf-8") data = json.loads(raw_text) urls = data.get("urls") if isinstance(data, dict) else None if not isinstance(urls, list) or len(urls) == 0: log(sid, f"⚠️ [URL-ROT] [HTTP_FALLBACK] {srv_url} urls 為空或格式錯誤，fallback 本機 JSON") return None valid = [ {"label": str(item.get("label", "")).strip(), "url": str(item.get("url", "")).strip()} for item in urls if isinstance(item, dict) and str(item.get("url", "")).strip() ] if not valid: log(sid, f"⚠️ [URL-ROT] [HTTP_FALLBACK] {srv_url} urls 無合法條目，fallback 本機 JSON") return None _LAST_VALID_ENTRIES[sid] = valid log(sid, f"ℹ️ [URL-ROT] [SERVER_JSON] 讀自 server 唯一本 {srv_url} ({len(valid)} 條)") return valid except Exception as e: log(sid, f"⚠️ [URL-ROT] [HTTP_FALLBACK] 讀取 server 唯一本 {srv_url} 失敗 ({type(e).__name__}: {e})，fallback 本機 JSON") return None def _parse_tail_entries(sid, tail): srv_valid = _fetch_server_tail_entries(sid, tail) if srv_valid is not None: return srv_valid, "json" jpath = SYSTEM_DIR / f"url-rotation-tail{tail}.json" if not jpath.exists(): log(sid, f"⚠️ [URL-ROT] [JSON_MISSING] {jpath.name} urls 設定檔唔存在 (missing)") missing_reason = "missing" return None, missing_reason try: data = json.loads(jpath.read_text(encoding="utf-8")) urls = data.get("urls") if not isinstance(urls, list) or len(urls) == 0: log(sid, f"⚠️ [URL-ROT] [JSON_CORRUPT] {jpath.name} urls 為空或格式錯誤") return _LAST_VALID_ENTRIES.get(sid), "corrupt" valid = [ {"label": str(item.get("label", "")).strip(), "url": str(item.get("url", "")).strip()} for item in urls if isinstance(item, dict) and str(item.get("url", "")).strip() ] if not valid: log(sid, f"⚠️ [URL-ROT] [JSON_CORRUPT] {jpath.name} urls 無合法條目") return _LAST_VALID_ENTRIES.get(sid), "corrupt" _LAST_VALID_ENTRIES[sid] = valid return valid, "json" except json.JSONDecodeError as e: log(sid, f"⚠️ [URL-ROT] [JSON_CORRUPT] {jpath.name} JSON 語法錯誤／Torn Read ({e})，Fail-Closed 保留快取") return _LAST_VALID_ENTRIES.get(sid), "corrupt" except Exception as e: log(sid, f"⚠️ [URL-ROT] [JSON_CORRUPT] {jpath.name} 讀取異常 ({type(e).__name__}: {e})") return _LAST_VALID_ENTRIES.get(sid), "corrupt" def _state_read_idx(sid, total): st_file = SYSTEM_DIR / "url-rotation.json" tail = ROTATION_TAILS.get(sid, sid[-1] if sid else "") per_tail = SYSTEM_DIR / f"url-rotation-tail{tail}.state.json" if not st_file.exists() and not per_tail.exists(): _LAST_VALID_INDEX.pop(sid, None) log(sid, "ℹ️ [URL-ROT] [INDEX_INIT] 狀態檔唔存在（首次初始化），由 index=0 起行") return 0 if st_file.exists(): try: obj = json.loads(st_file.read_text(encoding="utf-8")) if isinstance(obj, dict): if sid in obj: v = obj[sid] idx = v.get("index") if isinstance(v, dict) else v else: idx = obj.get("index") if isinstance(idx, int) and not isinstance(idx, bool) and 0 <= idx < total: _LAST_VALID_INDEX[sid] = idx return idx raise ValueError(f"out-of-bounds or invalid index={idx!r} (total={total})") raise ValueError("url-rotation.json root not dict") except Exception as e: bak_name = _save_corrupt_copy(st_file) log(sid, f"⚠️ [URL-ROT] [INDEX_CORRUPT] url-rotation.json 損毀或越界 ({type(e).__name__}: {e})，已存化驗副本 {bak_name}，嚴守 §四 鐵律保持原 index={_LAST_VALID_INDEX.get(sid)}（永不覆寫壞檔、嚴禁歸零）") if sid in _LAST_VALID_INDEX: return _LAST_VALID_INDEX[sid] if per_tail.exists(): try: obj = json.loads(per_tail.read_text(encoding="utf-8")) idx = obj.get("index") if isinstance(obj, dict) else None if isinstance(idx, int) and not isinstance(idx, bool) and 0 <= idx < total: _LAST_VALID_INDEX[sid] = idx return idx raise ValueError(f"out-of-bounds or invalid per_tail index={idx!r} (total={total})") except Exception as e: bak_name = _save_corrupt_copy(per_tail) log(sid, f"⚠️ [URL-ROT] [INDEX_CORRUPT] {per_tail.name} 損毀或越界 ({type(e).__name__}: {e})，已存化驗副本 {bak_name}，嚴守 §四 鐵律保持原 index={_LAST_VALID_INDEX.get(sid)}（永不覆寫壞檔、嚴禁歸零）") if sid in _LAST_VALID_INDEX: return _LAST_VALID_INDEX[sid] return None def _state_write_idx(sid, idx, label="", url=""): idx = int(idx) _LAST_VALID_INDEX[sid] = idx if label: _LAST_VALID_LABEL[sid] = label tail = ROTATION_TAILS.get(sid, sid[-1] if sid else "") per_tail = SYSTEM_DIR / f"url-rotation-tail{tail}.state.json" st_file = SYSTEM_DIR / "url-rotation.json" try: payload = {"seat": sid, "index": idx, "label": label, "url": url, "updated_at": int(time.time())} tmp_s = per_tail.with_suffix(".tmp") tmp_s.write_text(json.dumps(payload, ensure_ascii=False) + "\n", encoding="utf-8") tmp_s.replace(per_tail) except Exception: pass try: state = {"index": idx} if st_file.exists(): try: loaded = json.loads(st_file.read_text(encoding="utf-8")) if isinstance(loaded, dict): state = loaded except Exception: pass state["index"] = idx state[sid] = idx tmp_f = st_file.with_suffix(".json.tmp") tmp_f.write_text(json.dumps(state, ensure_ascii=False) + "\n", encoding="utf-8") tmp_f.replace(st_file) except Exception: pass def load_rotation_entries(sid): # 主持 47600 同名入口：1. 先經 HTTP 讀 server 唯一本 -> 2. fallback 本機 JSON -> 3. fallback seats.conf return load_url_rotation(sid) def load_url_rotation(sid): """Seam 1 (R2/R3/R4/R7)：每次攞 URL 重讀 JSON；回傳 {rotation,label,url,index,count,source}。""" tail = ROTATION_TAILS.get(sid) if not tail: return {"rotation": False, "label": sid, "url": _seats_fallback_url(sid), "index": 0, "count": 0, "source": "ineligible"} entries, _ = _parse_tail_entries(sid, tail) if not entries: log(sid, f"⚠️ [URL-ROT] [FALLBACK-START:tail{tail}] 退回 seats.conf 預設 URL") return {"rotation": True, "label": sid, "url": _seats_fallback_url(sid), "index": -1, "count": 0, "source": "fallback"} idx = _state_read_idx(sid, len(entries)) if idx is None: log(sid, f"⚠️ [URL-ROT] [INDEX_CORRUPT] [FALLBACK-START:tail{tail}] 重啟後狀態檔損毀且無記憶體 index，嚴禁歸零或覆寫壞檔，退回 seats.conf") return {"rotation": True, "label": sid, "url": _seats_fallback_url(sid), "index": -1, "count": 0, "source": "fallback"} chosen = entries[idx] prev_lbl = _LAST_VALID_LABEL.get(sid) if prev_lbl and prev_lbl != chosen["label"]: log(sid, f"⚠️ [URL-ROT] [INDEX_LABEL_MISMATCH] index={idx} 標籤由 {prev_lbl} 變為 {chosen['label']}（只准 append 唔准重排）") _LAST_VALID_LABEL[sid] = chosen["label"] log(sid, f"URL 輪換：用第 {idx + 1}/{len(entries)} 條（{chosen['label']}） {chosen['url']}") return {"rotation": True, "label": chosen["label"], "url": chosen["url"], "index": idx, "count": len(entries), "source": "json"} def rotation_advance(sid): """Seam 2 (R5/R6)：耗盡時嚴格只向前 +1（10->1 wrap），persist 狀態，永不後退。""" tail = ROTATION_TAILS.get(sid) if not tail: return {"rotation": False, "label": sid, "url": _seats_fallback_url(sid), "index": 0, "count": 0, "source": "ineligible"} entries, _ = _parse_tail_entries(sid, tail) if not entries: log(sid, f"⚠️ [URL-ROT] [FALLBACK-START:tail{tail}] 退回 seats.conf 預設 URL") return {"rotation": True, "label": sid, "url": _seats_fallback_url(sid), "index": -1, "count": 0, "source": "fallback"} cur_idx = _state_read_idx(sid, len(entries)) if cur_idx is None: log(sid, f"⚠️ [URL-ROT] [INDEX_CORRUPT] [FALLBACK-START:tail{tail}] 重啟後狀態檔損毀且無記憶體 index，嚴禁歸零或覆寫壞檔，退回 seats.conf") return {"rotation": True, "label": sid, "url": _seats_fallback_url(sid), "index": -1, "count": 0, "source": "fallback"} next_idx = (cur_idx + 1) % len(entries) chosen = entries[next_idx] _state_write_idx(sid, next_idx, chosen["label"], chosen["url"]) log(sid, f"URL 輪換：用第 {next_idx + 1}/{len(entries)} 條（{chosen['label']}） {chosen['url']}") return {"rotation": True, "label": chosen["label"], "url": chosen["url"], "index": next_idx, "count": len(entries), "source": "json"} def _get_descendant_chrome_pids(root_pid=None): """由本席 watcher PID 向下遍歷進程樹，搵出所有子孫 PID（優先排列 chrome/chromium 進程）。""" root_pid = int(root_pid if root_pid is not None else os.getpid()) children_map = {} comm_map = {} proc_root = Path("/proc") if proc_root.exists(): for entry in proc_root.iterdir(): if not entry.name.isdigit(): continue pid_i = int(entry.name) try: stat_txt = (entry / "stat").read_text(encoding="utf-8", errors="ignore") rp = stat_txt.rfind(")") if rp != -1: comm = stat_txt[stat_txt.find("(") + 1:rp].lower() rest = stat_txt[rp + 2:].split() ppid_i = int(rest[1]) children_map.setdefault(ppid_i, []).append(pid_i) comm_map[pid_i] = comm except Exception: pass try: out = subprocess.run(["pgrep", "-P", str(root_pid)], capture_output=True, text=True, timeout=2) for line in out.stdout.splitlines(): if line.strip().isdigit(): cp = int(line.strip()) if cp not in children_map.get(root_pid, []): children_map.setdefault(root_pid, []).append(cp) except Exception: pass visited = [] queue = list(children_map.get(root_pid, [])) seen = {root_pid} while queue: cur = queue.pop(0) if cur in seen: continue seen.add(cur) visited.append(cur) queue.extend(children_map.get(cur, [])) chrome_pids = [p for p in visited if "chrom" in comm_map.get(p, "")] other_pids = [p for p in visited if "chrom" not in comm_map.get(p, "")] return chrome_pids + other_pids + [root_pid] def _parse_window_geom(geom_text): """解析 xdotool getwindowgeometry 輸出，回傳 (x, y, w, h) 或 None。""" if not geom_text: return None x = y = w = h = None m_pos = re.search(r"Position:\s*(-?\d+),(-?\d+)", geom_text) if m_pos: x, y = int(m_pos.group(1)), int(m_pos.group(2)) m_geo = re.search(r"Geometry:\s*(\d+)x(\d+)", geom_text) if m_geo: w, h = int(m_geo.group(1)), int(m_geo.group(2)) if x is None: mx = re.search(r"^X=(-?\d+)", geom_text, re.M) my = re.search(r"^Y=(-?\d+)", geom_text, re.M) if mx and my: x, y = int(mx.group(1)), int(my.group(1)) if w is None: mw = re.search(r"^WIDTH=(\d+)", geom_text, re.M) mh = re.search(r"^HEIGHT=(\d+)", geom_text, re.M) if mw and mh: w, h = int(mw.group(1)), int(mh.group(1)) if x is not None and y is not None and w is not None and h is not None: return (x, y, w, h) return None def _descendant_x11_wids(root_pid=None): """按本席 PID 樹調用 xdotool search --onlyvisible --pid 認返自己個 Chrome window（嚴禁全屏見 Arena 就郁）。""" env = {**os.environ, "DISPLAY": ":99"} cand_pids = _get_descendant_chrome_pids(root_pid) found_wids = [] for cpid in cand_pids: try: res = subprocess.run( ["xdotool", "search", "--onlyvisible", "--pid", str(cpid)], capture_output=True, text=True, timeout=3, env=env ) for w in res.stdout.strip().splitlines(): w = w.strip() if w and w not in found_wids: found_wids.append(w) except Exception: pass return found_wids def _seat_window_id(sid=None, page=None, root_pid=None): """統一本席視窗解析器（主持 47526 + A4 47528【2】）：只准由本席 PID 樹 (--pid) 或本席 page ZOOMLOCK 臨時標題認窗，徹底廢除面積啟發式。""" wids = _descendant_x11_wids(root_pid=root_pid) if wids: return wids[-1] if page is not None and sid: env = {**os.environ, "DISPLAY": ":99"} tag = f"ZOOMLOCK_{sid}_{int(time.time()*1000)%100000}" try: old_t = page.evaluate(f"() => {{ const t = document.title; document.title = '{tag}'; return t; }}") time.sleep(0.08) out = subprocess.run(["xdotool", "search", "--onlyvisible", "--name", tag], env=env, timeout=3, capture_output=True, text=True) twids = [w.strip() for w in out.stdout.splitlines() if w.strip()] page.evaluate("(t) => { document.title = t; }", old_t) if twids: return twids[-1] except Exception: pass return None def _get_seat_dpr_baseline(sid): '''讀取 operator 簽認或 seats.conf 宣告之 DPR 基準（永不自動學習未驗證讀數）；回傳 float 或 None。''' try: sconf = SYSTEM_DIR / "seats.conf" if sconf.exists(): for line in sconf.read_text(encoding="utf-8", errors="ignore").splitlines(): s = line.strip() m = re.match(r"^#\s*zoom_baseline\s*:\s*(A\d+)\s*=\s*([0-9.]+)", s) if m and m.group(1) == sid: val = float(m.group(2)) if val > 0: _SEAT_DPR_BASELINE[sid] = val if sid not in _BASELINE_LOGGED: _BASELINE_LOGGED.add(sid) ts_utc = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) log(sid, f"[ZOOM-BASELINE:{sid}={val}@{ts_utc}]") return val except Exception: pass val = _SEAT_DPR_BASELINE.get(sid, 1.0) try: val = float(val) return val if val > 0 else None except Exception: return None def _sanitize_profile_zoom_prefs(profile_dir, sid): if not profile_dir: return False prefs_path = Path(profile_dir) / "Default" / "Preferences" if not prefs_path.exists(): return False try: data = json.loads(prefs_path.read_text(encoding="utf-8", errors="ignore")) if not isinstance(data, dict): return False changed = False part = data.get("partition") if isinstance(part, dict): for k in ("per_host_zoom_levels", "partition_default_zoom_levels"): if part.get(k): log(sid, f"🔍 [ZOOM-PREFS] 捉到 Preferences partition.{k}={part[k]}，即刻清零") part[k] = {} changed = True if "default_zoom_level" in part and part.get("default_zoom_level") not in (None, 0, 0.0, {}): log(sid, f"🔍 [ZOOM-PREFS] 捉到 Preferences partition.default_zoom_level={part.get('default_zoom_level')}，即刻移除") part.pop("default_zoom_level", None) changed = True for k in ("per_host_zoom_levels", "partition_default_zoom_levels"): if data.get(k): log(sid, f"🔍 [ZOOM-PREFS] 捉到 Preferences {k}={data[k]}，即刻清零") data[k] = {} changed = True if "default_zoom_level" in data and data.get("default_zoom_level") not in (None, 0, 0.0, {}): log(sid, f"🔍 [ZOOM-PREFS] 捉到 Preferences default_zoom_level={data.get('default_zoom_level')}，即刻移除") data.pop("default_zoom_level", None) changed = True if changed: tmp_p = prefs_path.with_suffix(".tmp") tmp_p.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") tmp_p.replace(prefs_path) _PREFS_CLEANED_AT_BOOT.add(sid) return changed except Exception as e: log(sid, f"⚠️ [ZOOM-PREFS] 清理 Preferences 異常 ({type(e).__name__}: {e})") return False def _apply_css_zoom_reset(page, sid): '''冪等 CSS zoom reset（A4 47581【2】）：若 documentElement.style.zoom 已為空/'100%'/'1' 則零動作；否則重設為 '100%'。''' try: cur_z = page.evaluate("document.documentElement && document.documentElement.style ? document.documentElement.style.zoom : ''") if isinstance(cur_z, str) and cur_z.strip() not in ("", "100%", "1"): page.evaluate("if (document.documentElement && document.documentElement.style) { document.documentElement.style.zoom = '100%'; }") return True return False except Exception as e: log(sid, f"⚠️ [ZOOM-CSS] CSS zoom reset 異常 ({type(e).__name__}: {e})") return False def _send_windowactivate_ctrl0(page, sid): try: if hasattr(page, "keyboard") and hasattr(page.keyboard, "press"): page.keyboard.press("Control+0") except Exception as e: log(sid, f"⚠️ [ZOOM-KEY] Playwright Control+0 異常 ({type(e).__name__}: {e})") if os.environ.get("DISPLAY") or Path("/tmp/.X11-unix/X99").exists(): try: env = {**os.environ, "DISPLAY": ":99"} wid = _seat_window_id(sid, page=page) if wid: subprocess.run(["xdotool", "windowraise", str(wid)], env=env, timeout=3, capture_output=True) time.sleep(0.12) subprocess.run(["xdotool", "windowactivate", "--sync", str(wid)], env=env, timeout=3, capture_output=True) time.sleep(0.12) subprocess.run(["xdotool", "key", "ctrl+0"], env=env, timeout=3, capture_output=True) except Exception as e: log(sid, f"⚠️ [ZOOM-X11] xdotool ctrl+0 異常 ({type(e).__name__}: {e})") def ensure_zoom_100(page, sid, stage="CYCLE", profile_dir=None): '''Seam 3 (Z1-Z6)：四源讀回驗證（含 operator 基準比值 dpr/baseline）+ 3步遞進校正 -> 失敗 HOLD 禁座標點擊。''' base_dpr = _get_seat_dpr_baseline(sid) def _probe(): if base_dpr is None or base_dpr <= 0: return {"dpr": 0.0, "dprRatio": 0.0, "innerW": 0, "innerH": 0, "vvScale": 0.0, "unconfirmed": True} try: raw_dpr = page.evaluate("window.devicePixelRatio") raw_w = page.evaluate("window.innerWidth") raw_h = page.evaluate("window.innerHeight") vv_raw = page.evaluate("JSON.stringify({scale: (window.visualViewport && window.visualViewport.scale) || 1.0})") dpr = float(raw_dpr if raw_dpr is not None else 1.0) w = int(raw_w if raw_w is not None else TARGET_VIEWPORT_W) h = int(raw_h if raw_h is not None else TARGET_VIEWPORT_H) if isinstance(vv_raw, str): vv_scale = float(json.loads(vv_raw).get("scale", 1.0)) elif isinstance(vv_raw, (int, float)): vv_scale = float(vv_raw) else: vv_scale = 1.0 ratio = round(dpr / base_dpr, 4) return {"dpr": dpr, "dprRatio": ratio, "innerW": w, "innerH": h, "vvScale": vv_scale, "unconfirmed": False} except Exception as e: log(sid, f"⚠️ [ZOOM-PROBE] 縮放探針讀取失敗 ({type(e).__name__}: {e}) -> UNCONFIRMED") return {"dpr": 0.0, "dprRatio": 0.0, "innerW": 0, "innerH": 0, "vvScale": 0.0, "unconfirmed": True} def _ok(m): return ( not m.get("unconfirmed", False) and abs(float(m.get("dprRatio", m.get("dpr", 0))) - 1.0) < 0.01 and int(m.get("innerW", 0)) == TARGET_VIEWPORT_W and int(m.get("innerH", 0)) == TARGET_VIEWPORT_H and abs(float(m.get("vvScale", 1.0)) - 1.0) < 0.01 ) def _z5(m, att, meth, res): w = int(m.get("innerW", 0)) pz = round(TARGET_VIEWPORT_W / w, 4) if w > 0 else 0.0 return f"zoom dpr={m.get('dpr')} ratio={m.get('dprRatio', m.get('dpr'))} innerW={w} innerH={m.get('innerH')} vvScale={m.get('vvScale', 1.0)} pageZoom={pz} attempt={att} method={meth} result={res}" m0 = _probe() if _ok(m0): _PREFS_CLEANED_AT_BOOT.discard(sid) if stage != "CYCLE": log(sid, _z5(m0, 0, "noop", "OK")) return True, m0 ts_drift_utc = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) log(sid, f"🚨 ERROR [ZOOM-LOCK] [ZOOM-DRIFT-ALERT:{sid}@{ts_drift_utc}] 偵測到 Zoom/Scale 偏離 100% (stage={stage} baseline={base_dpr} dpr={m0.get('dpr')} ratio={m0.get('dprRatio')} innerW={m0.get('innerW')} innerH={m0.get('innerH')} vvScale={m0.get('vvScale')} unconfirmed={m0.get('unconfirmed')})，拒絕靜默修復，立即報錯並啟動三步強制校正") if sid in _PREFS_CLEANED_AT_BOOT or stage == "TARGET-URL": log(sid, f"⚠️ [CAUSAL-NOTE:PREFS-NOT-PERSISTED] {sid} 啟動清 Preferences 後首輪仍測得非 100% ({m0})，進入三步強制校正") _PREFS_CLEANED_AT_BOOT.discard(sid) # Attempt 1: 鎖定本席 PID 視窗 (_seat_window_id) + windowactivate --sync + ctrl+0 _send_windowactivate_ctrl0(page, sid) m1 = _probe() if _ok(m1): log(sid, _z5(m1, 1, "windowactivate+ctrl+0", "OK")) return True, m1 log(sid, _z5(m1, 1, "windowactivate+ctrl+0", "FAIL")) # Attempt 2: 清 Preferences + windowactivate --sync + ctrl+0 _sanitize_profile_zoom_prefs(profile_dir, sid) _send_windowactivate_ctrl0(page, sid) m2 = _probe() if _ok(m2): log(sid, _z5(m2, 2, "clean_prefs+windowactivate+ctrl+0", "OK")) return True, m2 log(sid, _z5(m2, 2, "clean_prefs+windowactivate+ctrl+0", "FAIL")) # Attempt 3: 冪等 CSS zoom reset + windowactivate + ctrl+0, then reread（依主持 47596 真機三席實測，刪去 runtime 視口重設與 CDP 覆寫） _apply_css_zoom_reset(page, sid) _send_windowactivate_ctrl0(page, sid) m3 = _probe() if _ok(m3): log(sid, _z5(m3, 3, "css_zoom_reset+windowactivate+ctrl+0", "OK")) return True, m3 log(sid, f"🚨 [CAUSAL-NOTE:Z2-HOLD:{stage}] {_z5(m3, 3, 'escalate_skip_click_and_reopen', 'FAIL')} -> HOLD（因 Zoom 未達 100%，本輪嚴禁座標點擊並觸發收 Browser 清 Preferences 重開）") return False, m3 def load_vpn_ports(): ports = {} with open(VPN_CONF) as f: for line in f: line = line.strip() if not line or line.startswith("#"): continue parts = line.split() if len(parts) >= 2: ports[parts[0]] = int(parts[1]) return ports def log(sid, msg): ts = time.strftime("%Y-%m-%d %H:%M:%S") line = f"[{ts}] [{sid}] {msg}" print(line, flush=True) try: LOGS_DIR.mkdir(parents=True, exist_ok=True) logfile = LOGS_DIR / f"watcher-{sid}.log" with open(logfile, "a") as f: f.write(line + "\n") # trim to 5000 lines（errors=ignore 防亂碼 crash，2026-10-05 修） with open(logfile, errors="ignore") as f: lines = f.readlines() if len(lines) > 5000: with open(logfile, "w") as f: f.writelines(lines[-5000:]) except Exception: pass # 席位→Y 座標映射（Browser 排位鐵律，2026-10-05 19:44 HKT owner 下令必須） # 三個一組：第一席 y=0，第二席 y=900，第三席 y=1800 SLOT_Y_POS = { 'A2': 0, 'A3': 900, 'A4': 1800, 'A5': 0, 'A6': 900, 'A7': 1800, 'A8': 0, 'A9': 900, 'A10': 1800, } def enforce_window_geometry(sid, stage="CYCLE", root_pid=None): '''強制本席 browser window 位置與尺寸（主持 47526 + A4 47528 W1-W8：按本席 PID 樹認窗 + windowmove + windowsize 雙重驗） 1) W2：只准由本席 PID 樹 (`_descendant_x11_wids`) 認返自己個 window（嚴禁見 Arena 就郁晒）。 2) W1/W3：檢查及校正位置 (0, _slot_target_y(sid)) ＋ 尺寸 1408x900，重讀雙重驗。 3) W4/W5/W8：重開/導航後及每 cycle 執行；CYCLE 已合規時零動作零 spam。 ''' target_x = 0 target_y = _slot_target_y(sid) env = {**os.environ, 'DISPLAY': ':99'} try: wids = _descendant_x11_wids(root_pid=root_pid) if not wids: log(sid, f"⚠️ [GEOM-W2] 按本席 PID 樹搵唔到 Chrome window (pid={root_pid or os.getpid()})，跳過呢個 cycle") return False all_ok = True for wid in wids: geom_out = subprocess.run( ['xdotool', 'getwindowgeometry', str(wid)], capture_output=True, text=True, timeout=5, env=env ).stdout parsed = _parse_window_geom(geom_out) pos_ok = parsed is not None and parsed[0] == target_x and parsed[1] == target_y size_ok = parsed is not None and parsed[2] == EXPECTED_WIN_W and parsed[3] == EXPECTED_WIN_H if pos_ok and size_ok: if stage != "CYCLE": log(sid, f"geom wid={wid} x={parsed[0]} y={parsed[1]} w={parsed[2]} h={parsed[3]} attempt=0 result=OK") continue subprocess.run( ['xdotool', 'windowmove', str(wid), str(target_x), str(target_y)], timeout=5, env=env, capture_output=True ) subprocess.run( ['xdotool', 'windowsize', str(wid), str(EXPECTED_WIN_W), str(EXPECTED_WIN_H)], timeout=5, env=env, capture_output=True ) time.sleep(0.5) geom2_out = subprocess.run( ['xdotool', 'getwindowgeometry', str(wid)], capture_output=True, text=True, timeout=5, env=env ).stdout parsed2 = _parse_window_geom(geom2_out) pos_ok2 = parsed2 is not None and parsed2[0] == target_x and parsed2[1] == target_y size_ok2 = parsed2 is not None and parsed2[2] == EXPECTED_WIN_W and parsed2[3] == EXPECTED_WIN_H if not (pos_ok2 and size_ok2): log(sid, f"🚨 [GEOM-W1] window {wid} 較唔返 ({target_x},{target_y}) {EXPECTED_WIN_W}x{EXPECTED_WIN_H}（實測={parsed2}） attempt=1 result=FAIL，呢個 cycle 唔運行") all_ok = False else: log(sid, f"⚠️ [GEOM-W1] window {wid} 已校正至位置 ({target_x},{target_y}) 及尺寸 {EXPECTED_WIN_W}x{EXPECTED_WIN_H}（實測={parsed2}） attempt=1 result=OK") return all_ok except Exception as e: log(sid, f"enforce_window_geometry 出錯 ({type(e).__name__}: {e})，唔准運行") return False def is_generating(page): """檢查係咪生成緊（跟 ca.js：唔係搵 ■ 字，係搵 Stop 掣） selector: button[aria-label="Stop generating"], button[aria-label="Stop generation"] 用 .last()，check is_visible()。有 = 生成緊。 """ try: stop_btn = page.locator('button[aria-label="Stop generating"], button[aria-label="Stop generation"]').last if stop_btn.count() > 0: try: if stop_btn.is_visible(timeout=300): return True except Exception: pass return False except Exception: return False def get_arena_logged_in_email(page): """跟 ca.js getArenaLoggedInEmail：用 JS evaluate 喺 sidebar DOM 搵電郵。 搵：ul[data-sidebar="menu"] button → ul[data-sidebar="menu"] → button → 成個 body 用 email regex 抽。搵到 = 已登入（ca.js：唯一登入證據＝真實電郵）。 """ try: email = page.evaluate("""() => { const emailRe = /[A-Z0-9._%+-]+@[A-Z0-9.-]+\\.[A-Z]{2,}/i; const roots = Array.from(document.querySelectorAll('ul[data-sidebar="menu"] button, ul[data-sidebar="menu"], button')); for (const el of roots) { const text = (el.innerText || el.textContent || '').trim(); const m = text.match(emailRe); if (m) return m[0]; } const bodyText = (document.body && document.body.innerText || '').trim(); const m = bodyText.match(emailRe); return m ? m[0] : null; }""") return email except Exception: return None def find_exact_text_login_button(page): """跟 ca.js findExactTextLoginButton（已修正：唔分大小寫，因頁面有時係 "Log in"）： 先 get_by_role('button', name='Log In', exact=True)，唔得就逐個 button 睇 innerText（case-insensitive） """ import re try: rb = page.get_by_role("button", name="Log In", exact=True).first if rb.count() > 0: try: if rb.is_visible(timeout=700): return rb except Exception: pass except Exception: pass try: # 唔分大小寫：Log In / Log in 都得 rb2 = page.get_by_role("button", name=re.compile(r"^log in$", re.IGNORECASE)).first if rb2.count() > 0: try: if rb2.is_visible(timeout=700): return rb2 except Exception: pass except Exception: pass try: buttons = page.locator('button[type="button"], button') count = buttons.count() for i in range(count): try: b = buttons.nth(i) if not b.is_visible(timeout=300): continue t = b.inner_text(timeout=1000).replace("\n", " ").strip() t = re.sub(r"\s+", " ", t) if t.lower() == "log in": return b except Exception: continue except Exception: pass return None def wait_for_sidebar_button(page, sid, timeout_ms=120000): """120s 內一直搵左面 slide menu 個制，搵到 = loaded。 用 expand_sidebar 嗰兩組 selector（group/toggle-btn + ca.js aria-label），淨係搵唔撳。 """ import time as _time deadline = _time.time() * 1000 + timeout_ms sels = [ 'button[class*="group/toggle-btn"]', 'button[aria-label*="sidebar" i], button[aria-label*="menu" i], button[aria-label*="open" i]', ] while _time.time() * 1000 < deadline: for sel in sels: try: btn = page.locator(sel).first if btn.count() > 0 and btn.is_visible(timeout=1000): log(sid, f"搵到 slide menu 制（{sel}）= loaded") return True except Exception: pass try: page.wait_for_timeout(1000) except Exception: break log(sid, "120s 都搵唔到 slide menu 制") return False def expand_sidebar(page, sid): """撳 sidebar toggle，等 sidebar 內容 load 出嚟。 跟 ca.js + 用戶俾嘅正確 DOM： 1. agent page：button[class*="group/toggle-btn"]（無 aria-label，用戶俾嘅） 2. Landing/homepage：button[aria-label*="sidebar" i], button[aria-label*="menu" i], button[aria-label*="open" i]（ca.js） 撳之前先 check 開咗未（見到 Log In 掣或電郵 = 已開），開咗就唔撳，避免開閂打架。 撳一下，delay 3s。 """ # 先 check：sidebar 已開（見到 Log In 掣或電郵）就唔使撳 try: if find_exact_text_login_button(page) is not None: return True except Exception: pass try: if get_arena_logged_in_email(page): return True except Exception: pass # 方法 1：agent page 版（用戶俾嘅正確 DOM） try: btn = page.locator('button[class*="group/toggle-btn"]').first if btn.count() > 0 and btn.is_visible(timeout=3000): btn.click(timeout=8000) page.wait_for_timeout(5000) log(sid, "已撳 sidebar toggle（group/toggle-btn）") return True except Exception as e: log(sid, f"撳 group/toggle-btn 出錯: {e}") # 方法 2：ca.js 版（三個 aria-label pattern） try: tg = page.locator('button[aria-label*="sidebar" i], button[aria-label*="menu" i], button[aria-label*="open" i]').first if tg.count() > 0 and tg.is_visible(timeout=3000): tg.click(timeout=8000) page.wait_for_timeout(5000) log(sid, "已撳 sidebar toggle（ca.js aria-label 版）") return True except Exception as e: log(sid, f"撳 ca.js sidebar toggle 出錯/冇掣: {e}") return False def ca_page_logged_in(page, sid): """跟 ca.js caPageLoggedIn：5 次輪詢搵電郵，奇數次開 sidebar。搵到電郵 = 已登入。""" for i in range(5): try: email = get_arena_logged_in_email(page) if email: log(sid, f"搵到電郵：{email}（第 {i+1} 次）→ 已登入") return True if i % 2 == 1: expand_sidebar(page, sid) email2 = get_arena_logged_in_email(page) if email2: log(sid, f"開 sidebar 後搵到電郵：{email2} → 已登入") return True except Exception as e: log(sid, f"登入輪詢第 {i+1} 次出錯: {e}") page.wait_for_timeout(900) return False def find_keep_working(page, sid): """用 CSS :has-text() selector，逐個試，is_visible 先回傳。 順序跟 ca.js fallback，包埋「有进展/有進展/Some progress」變體。 """ candidates = [ 'button:has-text("Keep working")', 'button:has-text("有进展")', 'button:has-text("有進展")', 'button:has-text("Some progress")', 'span:has-text("Keep working")', 'span:has-text("有进展")', 'span:has-text("有進展")', 'button:has-text("继续工作")', 'button:has-text("繼續工作")', 'span:has-text("继续工作")', 'span:has-text("繼續工作")', ] for sel in candidates: try: loc = page.locator(sel).first if loc.count() > 0 and loc.is_visible(timeout=600): log(sid, f"找到 Keep working：{sel}") return loc except Exception: continue return None def wait_for_login_indicators(page, sid, timeout_ms=120000): """等登入指標出現：電郵（已登入）或 Log In 掣（未登入），唔用死時間。 用 ca_page_logged_in 同 find_exact_text_login_button 輪詢。 返 True = 見到其中一個，False = 超時。 """ import time as _time deadline = _time.time() * 1000 + timeout_ms while _time.time() * 1000 < deadline: try: if ca_page_logged_in(page, sid): log(sid, "等到登入指標：已登入（見到電郵）") return True except Exception: pass try: if find_exact_text_login_button(page) is not None: log(sid, "等到登入指標：未登入（見到 Log In 掣）") return True except Exception: pass try: page.wait_for_timeout(1000) except Exception: break log(sid, "等登入指標超時（未見電郵亦未見 Log In 掣）") return False def wait_for_dom_stable(page, sid, timeout_ms=30000, stable_rounds=3, poll_ms=1000): """等 DOM 穩定（唔用死時間）：每 poll_ms 計一次 body hash， 連續 stable_rounds 次無變化即當穩定。超時照返。 """ last_h = None stable = 0 waited = 0 while waited < timeout_ms: try: body = page.content() h = hash(body) & 0xFFFFFFFF except Exception: h = None if h is not None and h == last_h: stable += 1 if stable >= stable_rounds: log(sid, f"DOM 已穩定（hash={h:08x}，{waited}ms）") return True else: stable = 0 last_h = h page.wait_for_timeout(poll_ms) waited += poll_ms log(sid, f"等 DOM 穩定超時（{timeout_ms}ms），繼續") return False def wait_for_page_generated(page, sid, timeout_ms=15000): """等頁面生成（有實際內容）後先返。唔係齋等 reload 完。 用 body innerText 長度 > 100 做指標，代表頁面內容 render 咗。 """ try: page.wait_for_function( "document.body && document.body.innerText && document.body.innerText.length > 100", timeout=timeout_ms ) return True except Exception as e: log(sid, f"等頁面生成超時: {e}") return False def find_input_box(page, sid): """ 搵輸入框，三個方法逐個試（跟 8080：textarea 用 .last，contenteditable 用 .first，textbox 用 .last） 避開 password field（Chrome autofill 會填密碼，唔可以掂）。 返 locator 或 None """ def is_password_field(loc): """檢查係咪 password field""" try: tag = loc.evaluate("el => el.tagName.toLowerCase()").lower() typ = loc.evaluate("el => (el.getAttribute('type') || '').toLowerCase()") if tag == "input" and typ == "password": return True # 檢查附近有冇密碼相關（name/id 含 password/passwd/pwd） name_id = loc.evaluate("el => (el.getAttribute('name') || '') + ' ' + (el.getAttribute('id') || '') + ' ' + (el.getAttribute('autocomplete') || '')").lower() if any(k in name_id for k in ["password", "passwd", "current-password", "new-password"]): return True except Exception: pass return False cands = [ ("textarea-last", lambda: page.locator("textarea").last), ("contenteditable-first", lambda: page.locator('[contenteditable="true"]').first), ("role-textbox-last", lambda: page.get_by_role("textbox").last), ] for name, fn in cands: try: loc = fn() if loc.count() > 0 and loc.is_visible(timeout=3000): if is_password_field(loc): log(sid, f"⚠️ 跳過 password field（{name}），唔掂密碼") continue log(sid, f"找到輸入框: {name}") return loc except Exception: pass return None def get_page_heartbeat(page, sid): """ 心跳：記錄 page 狀態，防假成功。 記錄：hash、Stop 掣、Keep working、輸入框、body 字數 """ try: body = page.inner_text("body") h = hash(body) & 0xFFFFFFFF has_gen = is_generating(page) has_kw = find_keep_working(page, sid) is not None inp = find_input_box(page, sid) inp_state = "無" if inp: try: inp_state = "有(可見)" if inp.is_visible(timeout=1000) else "有(隱藏)" except: inp_state = "有(未知)" hb = f"hash={h:08x} stop={'有' if has_gen else '無'} kw={'有' if has_kw else '無'} 輸入框={inp_state} body={len(body)}字" return hb except Exception as e: return f"心跳出錯: {e}" def verify_email_via_sidebar(page, sid): """明確用 slide menu 開閂去確認有無 email： 開 sidebar → 睇有無 email → 有＝已登入，冇＝要 login。 返 email 字串或 None。 """ try: # 確保 sidebar 開咗 expand_sidebar(page, sid) page.wait_for_timeout(2000) email = get_arena_logged_in_email(page) if email: log(sid, f"sidebar 確認有 email：{email}") return email else: log(sid, "sidebar 開咗但搵唔到 email → 要 login") return None except Exception as e: log(sid, f"verify_email_via_sidebar 出錯: {e}") return None def check_login_state(page, sid, account_email, browser_ctx=None): """ 登入狀態檢查（跟 ca.js）： 1. ca_page_logged_in：5 次輪詢搵電郵 → 搵到＝已登入 2. 搵唔到 → 睇有冇 Log In 掣 → 有＝未登入 → 自動行 login flow 3. 兩個都冇＝異常 返：'logged_in' / 'logged_out' / 'unknown' / 'login_failed' / 'session_cleared_retry' """ try: # 1. 用 ca.js 方法搵電郵 if ca_page_logged_in(page, sid): return 'logged_in' # 2. 冇電郵 → 睇下係咪有 Log In 掣 login_btn = find_exact_text_login_button(page) if login_btn is not None: log(sid, "見到 Log In 掣 → 嘗試自動登入") try: login_btn.click(timeout=8000) page.wait_for_timeout(3000) except Exception as e: log(sid, f"撳 Log In 出錯: {e}") # 行自動登入流程 login_result = handle_arena_login_flow(page, sid, browser_ctx) if login_result == "SESSION_CLEARED": log(sid, "Arena session missing，已清 arena，要重跑 watcher") return 'session_cleared_retry' if login_result: log(sid, "自動登入成功，再驗證一次") page.wait_for_timeout(3000) if ca_page_logged_in(page, sid): return 'logged_in' else: log(sid, "自動登入後都搵唔到電郵") return 'login_failed' else: log(sid, "自動登入失敗（可能要密碼或超時）") return 'login_failed' # 3. 兩個都冇＝異常 log(sid, "⚠️ 電郵同 Log In 都搵唔到＝頁面異常，截圖留證") try: page.screenshot(path=f"/tmp/{sid.lower()}-login-unknown-{int(time.time())}.png") except: pass return 'unknown' except Exception as e: log(sid, f"登入檢查出錯: {e}") return 'unknown' def wait_for_login_signal(page, predicate, timeout_ms=2000, poll_ms=200): """跟 ca.js waitForLoginSignal：poll predicate，成功即返，唔使等足 timeout""" import time as _time deadline = _time.time() * 1000 + timeout_ms while _time.time() * 1000 < deadline: try: if predicate(): return True except Exception: pass page.wait_for_timeout(poll_ms) return False def is_login_flow_stuck_on_session_missing(page, sid): """檢測 login flow 卡住（Google missing Arena session）： 在 Arena 按 Log In 後、Google login 期間，DOM 出現 "session missing" 文字。 返 True = 卡住，要清 Arena 後重行 first-run login flow。 """ try: body_text = page.locator("body").inner_text(timeout=3000).lower() if "session missing" in body_text: log(sid, f"⚠️ 檢測到 login flow 卡住（DOM 有 'session missing'，URL: {page.url[:80]}）") return True except Exception: pass return False def is_conversation_too_long(page, sid): """檢測對話超長：頁面出現紅字 "The conversation is too long for the model. Please start a new chat." 呢個代表 AI 已完結，URL 無效，要交 Muse AI（worker Muses）handle。 返 True = 對話超長。 """ try: body_text = page.locator("body").inner_text(timeout=3000) if "The conversation is too long for the model" in body_text: log(sid, "⚠️ 檢測到對話超長（The conversation is too long）→ AI 已完結，URL 無效") return True except Exception: pass return False def is_arena_session_missing(page, sid): """檢查 Arena（經 Google login）有冇出現 session missing。 Google 本身 login 冇問題，係 Arena 個 session 唔見咗。 成日出現但難捉 DOM，用多種 pattern 鬆匹配。 """ try: body_text = page.locator("body").inner_text(timeout=3000).lower() # session missing 相關字眼（中英文） has_session = "session" in body_text or "工作階段" in body_text or "會話" in body_text has_missing = "missing" in body_text or "expired" in body_text or "遺失" in body_text or "過期" in body_text or "不存在" in body_text if has_session and has_missing: log(sid, f"偵測到 Arena session missing（URL: {page.url[:80]}）") return True return False except Exception: return False def clear_arena_site_data(browser_ctx, page, sid): """清 arena.ai site data（經 CDP），保留 Google cookies。 用 Storage.clearDataForOrigin 只清 arena.ai 個 origin。 """ try: cdp = browser_ctx.new_cdp_session(page) cdp.send("Storage.clearDataForOrigin", { "origin": "https://arena.ai", "storageTypes": "all" }) log(sid, "已清 arena.ai site data（保留 Google cookies）") return True except Exception as e: log(sid, f"清 arena site data 出錯: {e}") return False def handle_arena_login_flow(page, sid, browser_ctx=None): """照搬 ca.js _handleArenaLoginFlow，但改咗：見到要入密碼就即刻停手（唔填密碼）。 步驟：Terms of Use → Continue with Google → 帳號選擇 →（密碼→停手）→ 2FA/challenge → 返 arena.ai 返 True = 登入成功（返到 arena.ai），False = 失敗/要密碼/超時， "SESSION_CLEARED" = Arena session missing，已清 arena data，要重跑 watcher。 """ import time as _time MAX_LOGIN_WAIT = 60000 login_start = _time.time() * 1000 log(sid, "開始自動登入流程（見到密碼框即停手）…") def need_password_stop(): """檢查有冇密碼框（pwd challenge 頁除外，由 3c 自動打）""" try: url = page.url if "/challenge/pwd" in url or "/signin/v3/challenge/pwd" in url: return False # 3c 會自動打，唔停手 pwd = page.locator('input[type="password"]').first if pwd.count() > 0 and pwd.is_visible(timeout=500): log(sid, "見到密碼框（非 pwd 頁）→ 停手") return True except Exception: pass return False # 1. 處理 Terms of Use dialog（15 秒） terms_deadline = _time.time() * 1000 + 15000 while _time.time() * 1000 < terms_deadline: try: dlg = page.locator('[role="dialog"]').filter(has=page.locator('h2:has-text("Terms of Use")')).first if dlg.count() > 0 and dlg.is_visible(timeout=1000): agree = page.locator('[role="dialog"] button[type="submit"]:has-text("Agree")').first if agree.count() > 0 and agree.is_visible(timeout=3000): agree.click(timeout=5000) else: page.keyboard.press("Enter") log(sid, "已處理 Terms of Use") wait_for_login_signal(page, lambda: not (dlg.count() > 0 and dlg.is_visible(timeout=250)), 2000) break except Exception: pass if _time.time() * 1000 - login_start > MAX_LOGIN_WAIT: break page.wait_for_timeout(200) # 2. 撳 Continue with Google（20 秒） google_deadline = _time.time() * 1000 + 20000 while _time.time() * 1000 < google_deadline: try: gbtn = page.locator('button:has-text("Continue with Google")').first if gbtn.count() > 0 and gbtn.is_visible(timeout=1000): url_before = page.url gbtn.click(timeout=5000) log(sid, "已撳 Continue with Google") wait_for_login_signal(page, lambda: page.url != url_before or "accounts.google.com" in page.url or page.locator('input[type="password"]').first.is_visible(timeout=200), 3000) break except Exception: pass if _time.time() * 1000 - login_start > MAX_LOGIN_WAIT * 2: break page.wait_for_timeout(200) # 3. Google 帳號頁處理（25 秒） account_handled = False account_deadline = _time.time() * 1000 + 25000 while _time.time() * 1000 < account_deadline: try: url = page.url # 3a0. Arena session missing（經 Google login）→ 即刻清 arena 再重跑 # 時機限定：Arena 按 Log In 後、Google login 期間先算 if is_login_flow_stuck_on_session_missing(page, sid) or is_arena_session_missing(page, sid): log(sid, "Arena session missing → 清 arena.ai 重跑 watcher") if browser_ctx is not None: clear_arena_site_data(browser_ctx, page, sid) return "SESSION_CLEARED" # 3a. 先檢查係咪要密碼 if need_password_stop(): return False # 3b. Challenge（2FA）頁 if "accounts.google.com" in url and "/challenge/" in url: if "/challenge/pwd" not in url and "/challenge/selection" not in url: log(sid, f"遇到二次驗證，等 10 秒睇下會唔會自動過…") if wait_for_login_signal(page, lambda: "arena.ai" in page.url, 10000): log(sid, "二次驗證後返到 arena.ai，視為通過") account_handled = True break continue if "/challenge/selection" in url: pwd_opt = page.locator('div[data-challengeentrytype="1"]').first if pwd_opt.count() > 0 and pwd_opt.is_visible(timeout=1000): url_before = page.url pwd_opt.click(timeout=5000) wait_for_login_signal(page, lambda: page.url != url_before or page.locator('input[type="password"]').first.is_visible(timeout=200), 2000) continue else: log(sid, "挑戰選擇頁搵唔到密碼選項，終止登入") return False # 3c. 密碼頁 → 自動打密碼（跟 CA arena-headed-login.py，owner 2026-10-05 下令） if "/challenge/pwd" in url or "/signin/v3/challenge/pwd" in url: try: pwd_input = page.locator('input[type="password"]').first pwd_input.wait_for(state="visible", timeout=15000) pwd_input.fill(os.environ.get("ARENA_PW", "")) log(sid, "已自動填密碼") # 撳 Next/繼續（timeout 加長到 15000，owner 2026-10-05 14:34 下令） for btn_sel in ['button:has-text("Next")', 'button:has-text("繼續")', '#passwordNext']: btn = page.locator(btn_sel).first if btn.count() > 0 and btn.is_visible(timeout=5000): btn.click(timeout=15000) log(sid, "已撳密碼 Next") break continue except Exception as e: log(sid, f"自動填密碼失敗: {e}，停手") return False # 3d. 帳號選擇 for sel in ['[data-identifier]', 'div[role="link"][data-authuser]', '.VV3oRb[role="link"]']: acc = page.locator(sel).first if acc.count() > 0 and acc.is_visible(timeout=3000): url_before = page.url acc.click(timeout=5000) log(sid, f"已揀帳號（{sel}）") wait_for_login_signal(page, lambda: page.url != url_before or page.locator('input[type="password"]').first.is_visible(timeout=200), 2000) break # 3e. 返 arena.ai 確認 return_deadline = _time.time() * 1000 + 20000 while _time.time() * 1000 < return_deadline: url = page.url if "arena.ai" in url and "accounts.google.com" not in url: page.wait_for_load_state("domcontentloaded", timeout=10000) page.wait_for_timeout(500) account_handled = True break if "accounts.google.com" in url: if need_password_stop(): return False for sel in ['[jsname="uRHG6"] button', 'button:has-text("繼續")', 'button:has-text("Continue")', 'button:has-text("Allow")', '.tyoyWc button']: cb = page.locator(sel).first if cb.count() > 0 and cb.is_visible(timeout=1000): url_before = page.url cb.click(timeout=5000) log(sid, f"已撳繼續/Allow（{sel}）") wait_for_login_signal(page, lambda: page.url != url_before or ("arena.ai" in page.url and "accounts.google.com" not in page.url), 3000) break page.wait_for_timeout(200) if account_handled: break except Exception as e: log(sid, f"登入流程出錯: {e}") if _time.time() * 1000 - login_start > MAX_LOGIN_WAIT * 3: break page.wait_for_timeout(200) if account_handled: log(sid, "自動登入成功，已返 arena.ai") else: log(sid, "自動登入失敗（超時或中斷）") return account_handled def _do_send_once(page, sid): """搵輸入框 → click → 打字 → Enter。成功返 True""" inp = find_input_box(page, sid) if not inp: log(sid, "ERROR: 搵唔到輸入框 → 截圖＋dump DOM") try: page.screenshot(path=f"/tmp/{sid.lower()}-no-input.png") except: pass try: dom = page.content() with open(f"/tmp/{sid.lower()}-no-input-dom.html", "w") as f: f.write(dom) except: pass return False try: inp.click(timeout=8000) page.wait_for_timeout(800) inp.fill("") msg = get_recall_msg() # 每次推攞新數字：2147483647 → 2147483646 → … page.keyboard.type(msg, delay=25) page.wait_for_timeout(800) page.keyboard.press("Enter") log(sid, "已按 Enter 發送") # v0.93: 發完訊息後過 checkbox 關（跟 CA POST-SEND 模式） # 過唔到就唔繼續，直接返 page.wait_for_timeout(2000) if not ca_checkbox_gate(page, sid, "POST-SEND"): log(sid, "POST-SEND checkbox 未過，停") return False # owner 2026-10-05 15:23 下令：send 完 3 秒後自動 cap 圖（捉藍 checkbox） page.wait_for_timeout(3000) try: page.screenshot(path=f"/tmp/{sid}-postsend-3s.png") log(sid, f"已cap圖 /tmp/{sid}-postsend-3s.png（send後3秒）") except Exception as e: log(sid, f"cap圖失敗: {e}") return True except Exception as e: log(sid, f"打字/發送出錯: {e}") return False def do_send_cycle(page, sid): """v3 運作邏輯（用戶逐點修正）： 1. 見 Stop 掣（生成緊）→ 乜都唔做，SKIP_GENERATING 2. 見 Keep working → 撳 → 等 5 秒 → reload → 等頁面生成 → 等 5 秒 → 確認消失 → send 仲有 → 再撳一次 → 等 5 秒 → reload → 等生成 → 等 5 秒 → 仲有 → HOLD_ERROR 3. 冇 Stop 掣、冇 Keep working → 直接 send 4. Send 後：等 10 秒 → reload → 等頁面生成 → 等 5 秒 有 Stop 掣 → SUCCESS；有 Keep working → SUCCESS_NEXT_CYCLE（即刻下一個循環） 兩個都冇 → 再 send 一次 5. 第二次：等 10 秒 → reload → 等生成 → 等 5 秒 有 Stop 掣 → SUCCESS；有 Keep working → SUCCESS_NEXT_CYCLE；兩個都冇 → ERROR """ # 1. 見 Stop 掣（生成緊）→ 乜都唔做 if is_generating(page): log(sid, "見到 Stop 掣（生成緊），乜都唔做，等下個 cycle") return "SKIP_GENERATING" # 2. 見 Keep working → 撳 kw = find_keep_working(page, sid) if kw: try: kw.click(timeout=8000) log(sid, "撳咗 Keep working，等 5 秒") page.wait_for_timeout(3000) page.reload(wait_until="domcontentloaded", timeout=60000) wait_for_page_generated(page, sid) page.wait_for_timeout(5000) log(sid, "Reload 完（等頁面生成後 5 秒），確認有冇 Keep working") if find_keep_working(page, sid) is not None: log(sid, "仲有 Keep working，再試一次") kw2 = find_keep_working(page, sid) if kw2: kw2.click(timeout=8000) page.wait_for_timeout(5000) page.reload(wait_until="domcontentloaded", timeout=60000) wait_for_page_generated(page, sid) page.wait_for_timeout(5000) if find_keep_working(page, sid) is not None: log(sid, "HOLD: 兩次都仲有 Keep working → 報告錯誤，停手") return "HOLD_ERROR" log(sid, "確認冇 Keep working，可以 send") except Exception as e: log(sid, f"撳 Keep working 出錯: {e}") return "KW_CLICK_ERROR" else: log(sid, "無 Keep working 掣，直接 send") # 3. Send if not _do_send_once(page, sid): return "NO_INPUT" # 4. 驗證（第一次）：等 10 秒 → 10秒內有冇出現 → 無就 skip gate 直接 reload # owner 2026-10-05 16:30 下令：10s 內冇出現就唔使過 checkbox 關 log(sid, "等 10 秒，監測有冇 checkbox 出現…") import time wid = _seat_window_id(sid, page=page) appeared = False for i in range(10): time.sleep(1) if not wid: break # 快 check 三套色（唔點，淨睇有冇） c_orange = _read_window_pixel(wid, ORANGE_CHECK_X, ORANGE_CHECK_Y) c_purple = _read_window_pixel(wid, PURPLE_CHECK_X, PURPLE_CHECK_Y) c_blue = _check_window_area_color(wid, BLUE_CHECK_X, BLUE_CHECK_Y, BLUE_CHECK_W, BLUE_CHECK_H, BLUE_CHECK_COLOR) if (c_orange and _color_match(c_orange, ORANGE_CHECK_COLOR)) or \ (c_purple and _color_match(c_purple, PURPLE_CHECK_COLOR)) or c_blue: log(sid, f"第 {i+1} 秒見到 checkbox，行完整 gate") appeared = True break if appeared: # 有出現先過完整 gate if not ca_checkbox_gate(page, sid, "VERIFY-BEFORE-RELOAD"): log(sid, "VERIFY checkbox 未過，唔 reload") return "CHECKBOX_BLOCKED" else: log(sid, "10 秒內冇 checkbox，直接 reload") page.reload(wait_until="domcontentloaded", timeout=60000) wait_for_page_generated(page, sid) page.wait_for_timeout(5000) has_gen = is_generating(page) has_kw = find_keep_working(page, sid) is not None if has_gen or has_kw: if has_kw: log(sid, "Reload 後有 Keep working → 成功，即刻開始下一個循環") return "SUCCESS_NEXT_CYCLE" log(sid, "Reload 後有 Stop 掣（生成緊）→ 成功") return "SUCCESS" # 5. 兩個都冇 → 再送一次 log(sid, "Reload 後無 Stop 掣又無 Keep working → 再送一次") if not _do_send_once(page, sid): return "NO_INPUT_2" log(sid, "第二次送出，等 10 秒…") page.wait_for_timeout(10000) page.reload(wait_until="domcontentloaded", timeout=60000) wait_for_page_generated(page, sid) page.wait_for_timeout(5000) has_gen2 = is_generating(page) has_kw2 = find_keep_working(page, sid) is not None if has_gen2 or has_kw2: if has_kw2: log(sid, "第二次 reload 有 Keep working → 成功，即刻開始下一個循環") return "SUCCESS_NEXT_CYCLE" log(sid, "第二次 reload 有 Stop 掣（生成緊）→ 成功") return "SUCCESS" else: log(sid, "ERROR: 第二次都無 Stop 掣又無 Keep working → 報告錯誤") return "ERROR" def _get_browser_window_id(sid=None, page=None): """兼容轉發至統一本席 PID 樹視窗解析器 _seat_window_id（已徹底移除面積啟發式）""" return _seat_window_id(sid=sid, page=page) def _read_window_pixel(window_id, x, y): """跟 CA：用 import -window 讀 window 相對座標 pixel""" import subprocess try: out = subprocess.run(['import', '-window', str(window_id), '-crop', f'1x1+{x}+{y}', '-depth', '8', 'txt:-'], capture_output=True, text=True, timeout=5, env={**__import__('os').environ, 'DISPLAY': ':99'}) import re m = re.search(r'#([0-9a-fA-F]{6})', out.stdout) return '#' + m.group(1).lower() if m else None except: return None def _color_match(c1, c2, tolerance=10): """顏色比對（容差，2026-10-05 19:50 HKT owner 指出 exact match 太嚴加嘅） 每個 RGB channel 差唔多過 tolerance 就算 match。 """ try: c1 = c1.lstrip('#') c2 = c2.lstrip('#') r1, g1, b1 = int(c1[0:2], 16), int(c1[2:4], 16), int(c1[4:6], 16) r2, g2, b2 = int(c2[0:2], 16), int(c2[2:4], 16), int(c2[4:6], 16) return abs(r1-r2) <= tolerance and abs(g1-g2) <= tolerance and abs(b1-b2) <= tolerance except: return False def _check_window_area_color(window_id, cx, cy, w, h, hex_color): """跟 CA checkAreaColor：檢查範圍內有冇目標色（有容差）""" import subprocess try: # 抽樣檢查範圍內幾點 for dx, dy in [(0,0), (w//2,h//2), (w-1,0), (0,h-1), (w-1,h-1)]: c = _read_window_pixel(window_id, cx+dx, cy+dy) if c and _color_match(c, hex_color, tolerance=10): return True return False except: return False def _xdotool_click(window_id, x, y): """跟 CA：xdotool 三步點擊""" import subprocess, time env = {**__import__('os').environ, 'DISPLAY': ':99'} try: subprocess.run(['xdotool', 'windowraise', str(window_id)], env=env, timeout=3, capture_output=True) time.sleep(0.2) subprocess.run(['xdotool', 'windowactivate', '--sync', str(window_id)], env=env, timeout=3, capture_output=True) time.sleep(0.3) subprocess.run(['xdotool', 'mousemove', '--window', str(window_id), str(x), str(y), 'click', '1'], env=env, timeout=5, capture_output=True) return True except: return False def ca_checkbox_gate(page, sid, label): _zok, _ = ensure_zoom_100(page, sid, f"GATE-{label}") if not _zok: log(sid, f"🚨 [CAUSAL-NOTE:Z2-HOLD] {label} 因 Zoom 未達 100% 拒絕執行座標點擊") return False """v0.93: CA 兩套 checkbox 關卡（跟足 CA） 第1套（橙色）：import -window 讀 (527,409) #f6821f → xdotool (270,420) 第2套（藍色）：import -window 讀 (910,420,200x60) #1A73E8 → xdotool (276,425) """ try: wid = _seat_window_id(sid, page=page) if not wid: return True # 第1套：紫色（優先） — check (715,415) #f3e9f7 → click (580,605) c3 = _read_window_pixel(wid, PURPLE_CHECK_X, PURPLE_CHECK_Y) if c3 and _color_match(c3, PURPLE_CHECK_COLOR): _xdotool_click(wid, PURPLE_CLICK_X, PURPLE_CLICK_Y) log(sid, f"[CHECKBOX-GATE] {label} 紫色 ({c3}) 已點 ({PURPLE_CLICK_X},{PURPLE_CLICK_Y})") # 點完等 8 秒，再 5 秒內每 1 秒 check log(sid, f"[CHECKBOX-GATE] {label} 等 8 秒後開始 5 秒監測...") time.sleep(8) reappeared = False for i in range(5): time.sleep(1) cc = _read_window_pixel(wid, PURPLE_CHECK_X, PURPLE_CHECK_Y) if cc and _color_match(cc, PURPLE_CHECK_COLOR): log(sid, f"[CHECKBOX-GATE] {label} 點後第 {i+1} 秒紫色再出現") reappeared = True break if not reappeared: log(sid, f"[CHECKBOX-GATE] {label} 5 秒內冇再出現，pass") return True # 再出現：第 2 次 click log(sid, f"[CHECKBOX-GATE] {label} 第 2 次點擊 ({PURPLE_CLICK_X},{PURPLE_CLICK_Y})") _xdotool_click(wid, PURPLE_CLICK_X, PURPLE_CLICK_Y) time.sleep(2) c4 = _read_window_pixel(wid, PURPLE_CHECK_X, PURPLE_CHECK_Y) if c4 and _color_match(c4, PURPLE_CHECK_COLOR): log(sid, f"[CHECKBOX-GATE] {label} 第 2 次點完仲喺度，HOLD 報 Muse") return False # hold住，等 Muse 處理 log(sid, f"[CHECKBOX-GATE] {label} 第 2 次點完消失，pass") # 第2套：藍色（200x60 範圍）— v0.93：藍色＝圖片挑戰（owner 2026-10-05 20:07 HKT 下令） # 邏輯：紫色過咗之後，如果藍色出現又清唔走＝中圖片挑戰，停低乜都唔做等人手 # 第一次：點之前 check if _check_window_area_color(wid, BLUE_CHECK_X, BLUE_CHECK_Y, BLUE_CHECK_W, BLUE_CHECK_H, BLUE_CHECK_COLOR): _xdotool_click(wid, BLUE_CLICK_X, BLUE_CLICK_Y) log(sid, f"[CHECKBOX-GATE] {label} 藍色已點 ({BLUE_CLICK_X},{BLUE_CLICK_Y})，再驗一次") time.sleep(2) if _check_window_area_color(wid, BLUE_CHECK_X, BLUE_CHECK_Y, BLUE_CHECK_W, BLUE_CHECK_H, BLUE_CHECK_COLOR): log(sid, f"[CHECKBOX-GATE] {label} 🚨 藍色點完仲喺度＝圖片挑戰") log(sid, f"[CHECKBOX-GATE] {label} 🚨 停低，乜都唔做，等 MUSE 人手處理") return False return False # 點 checkbox _xdotool_click(wid, BLUE_CLICK_X, BLUE_CLICK_Y) log(sid, f"[CHECKBOX-GATE] {label} 已點 checkbox ({BLUE_CLICK_X},{BLUE_CLICK_Y})") # v0.93：藍色出現＝圖片挑戰，HOLD 等人手（owner 2026-10-05 20:07 HKT） # owner 2026-10-05 20:08 HKT：中圖片挑戰一樣停低搵 MUSE 處理，乜都唔做 for i in range(10): time.sleep(0.5) if _check_window_area_color(wid, BLUE_CHECK_X, BLUE_CHECK_Y, BLUE_CHECK_W, BLUE_CHECK_H, BLUE_CHECK_COLOR): log(sid, f"[CHECKBOX-GATE] {label} 🚨 點後 {(i+1)*0.5} 秒出現藍色＝圖片挑戰") log(sid, f"[CHECKBOX-GATE] {label} 🚨 停低，乜都唔做，等 MUSE 人手處理") return False # 藍色搞掂，繼續第3套 # 第3套：橙色 — 跟足 CA 完整流程 unreadable = 0 for attempt in range(3): # CA: 連續讀唔到先放行 c1 = _read_window_pixel(wid, ORANGE_CHECK_X, ORANGE_CHECK_Y) if c1 is None: unreadable += 1 if unreadable >= 3: log(sid, f"[CHECKBOX-GATE] {label} 連續 3 次讀唔到像素，放行") break time.sleep(0.7) continue unreadable = 0 if c1.lower() != ORANGE_CHECK_COLOR.lower(): break # 唔係橙色，放行 # 係橙色：跟 CA 流程 log(sid, f"[CHECKBOX-GATE] {label} 橙色出現 ({c1})") time.sleep(5) # CA: 等 5 秒先點 _xdotool_click(wid, ORANGE_CLICK_X, ORANGE_CLICK_Y) log(sid, f"[CHECKBOX-GATE] {label} 已點 ({ORANGE_CLICK_X},{ORANGE_CLICK_Y})，等 5 秒再驗") time.sleep(5) # 點完再驗 c2 = _read_window_pixel(wid, ORANGE_CHECK_X, ORANGE_CHECK_Y) if c2 and c2.lower() != ORANGE_CHECK_COLOR.lower(): log(sid, f"[CHECKBOX-GATE] {label} 橙色已消失，放行") break # 仲係橙色：CA 會 reload（呢度簡化，返 False 等上層處理） log(sid, f"[CHECKBOX-GATE] {label} 橙色仲喺度，要處理") return False # 如果上面 loop 係因為橙色而 return False，已經返咗 # 嚟到呢度即係冇橙色，繼續藍色檢查 return True except Exception as e: log(sid, f"[CHECKBOX-GATE] {label} 異常: {e}") return True def ca_checkbox_click(window_id, x, y): """CA 式：用 xdotool 做 OS 層級滑鼠點擊（唔係 DOM click）""" import subprocess env = {'DISPLAY': ':99'} try: subprocess.run(['xdotool', 'windowraise', str(window_id)], env={**os.environ, **env}, timeout=3, capture_output=True) subprocess.run(['xdotool', 'windowactivate', '--sync', str(window_id)], env={**os.environ, **env}, timeout=3, capture_output=True) subprocess.run(['xdotool', 'mousemove', '--window', str(window_id), str(x), str(y), 'click', '1'], env={**os.environ, **env}, timeout=5, capture_output=True) return True except Exception as e: return False def is_page_hung(page, timeout=5000): """ 檢測頁面 JS 是否 hang 咗（用戶 2026-10-02 22:25 提出嘅關鍵問題） Arena 係動態 JS 頁，有時 renderer 死咗但 DOM 停喺舊狀態呃人 用 F12 Inspector 級別嘅檢測：心跳＋event loop """ try: # 心跳 1: Date.now() 應該要郁 t1 = page.evaluate("Date.now()") time.sleep(0.5) t2 = page.evaluate("Date.now()") if t2 <= t1: return True # 時間冇郁，JS 已死 # 心跳 2: requestAnimationFrame 測試 event loop raf_ok = page.evaluate("""new Promise((resolve) => { let done = false; requestAnimationFrame(() => { done = true; resolve(true); }); setTimeout(() => { if (!done) resolve(false); }, 2000); })""") if not raf_ok: return True # event loop 唔郁 return False except Exception: return True # evaluate 都 timeout，肯定 hang 咗 def run_watcher(sid, once=False): try: from playwright.sync_api import sync_playwright except ImportError: log(sid, "ERROR: playwright 未安裝") return 1 seats = load_seats() vpn_ports = load_vpn_ports() if sid not in seats: log(sid, f"ERROR: 席位 {sid} 不在 seats.conf") return 1 cfg = seats[sid] if cfg["status"] != "active": log(sid, f"席位狀態為 {cfg['status']}，跳過") return 0 if cfg["vpn"] not in vpn_ports: log(sid, f"ERROR: VPN {cfg['vpn']} 不在 vpn-proxies.conf") return 1 port = vpn_ports[cfg["vpn"]] # 直接用 workspace storage（唔用 RAM，owner 2026-10-05 下令） profile_dir = PROFILES_DIR / cfg["profile"] profile_dir.mkdir(parents=True, exist_ok=True) log(sid, f"Profile 直接用 workspace: {profile_dir}") _sanitize_profile_zoom_prefs(profile_dir, sid) _fallback_seat_url = cfg["url"] if sid in ROTATION_TAILS: _rot_init = load_url_rotation(sid) if _rot_init.get("url"): cfg["url"] = _rot_init["url"] log(sid, f"啟動 watcher: profile={cfg['profile']}, vpn={cfg['vpn']}:{port}, url={cfg['url']}") with sync_playwright() as p: browser = None try: browser = p.chromium.launch_persistent_context( str(profile_dir), headless=False, device_scale_factor=1, proxy={"server": f"socks5://127.0.0.1:{port}"}, args=[ "--test-type", "--no-sandbox", "--disable-setuid-sandbox", "--disable-gpu", "--disable-dev-shm-usage", "--disable-blink-features=AutomationControlled", "--disable-infobars", "--no-first-run", "--no-default-browser-check", "--disable-crash-reporter", "--disable-breakpad", "--disable-component-update", "--disable-extensions", "--hide-crash-restore-bubble", "--disable-session-crashed-bubble", "--disable-features=TranslateUI,PrivacySandboxSettings4,OptimizationGuideModelExecution,ChromeWhatsNewUI,AutomationControlled,RestoreOnStartup", "--use-gl=swiftshader", "--disable-popup-blocking", "--disable-notifications", "--force-device-scale-factor=1", "--high-dpi-support=1", "--window-size=1408,900", # 外框1408x900 → content 1400x900 ], ignore_default_args=["--enable-automation"], viewport={'width': 1400, 'height': 900}, ) page = browser.new_page() if not browser.pages else browser.pages[0] first_load = True cycle_count = 0 # 計 cycle 數，每 120 個 cycle（約 1 小時）做一次登入檢查 while True: try: # 檢查 browser 仲生唔生 if browser is None: log(sid, "Browser 已死，重開中...") _sanitize_profile_zoom_prefs(profile_dir, sid) browser = p.chromium.launch_persistent_context( str(profile_dir), headless=False, device_scale_factor=1, proxy={"server": f"socks5://127.0.0.1:{port}"}, args=[ "--test-type", "--no-sandbox", "--disable-setuid-sandbox", "--disable-gpu", "--disable-dev-shm-usage", "--disable-blink-features=AutomationControlled", "--disable-infobars", "--no-first-run", "--no-default-browser-check", "--disable-crash-reporter", "--disable-breakpad", "--disable-component-update", "--disable-extensions", "--hide-crash-restore-bubble", "--disable-session-crashed-bubble", "--disable-features=TranslateUI,PrivacySandboxSettings4,OptimizationGuideModelExecution,ChromeWhatsNewUI,AutomationControlled,RestoreOnStartup", "--use-gl=swiftshader", "--disable-popup-blocking", "--disable-notifications", "--force-device-scale-factor=1", "--high-dpi-support=1", "--window-size=1408,900", # 外框1408x900 → content 1400x900 ], ignore_default_args=["--enable-automation"], viewport={'width': 1400, 'height': 900}, ) page = browser.new_page() if not browser.pages else browser.pages[0] enforce_window_geometry(sid, stage="REOPEN") first_load = True # 首次載入或需要時先 goto，之後直接 check DOM（唔好每 30 秒 reload） if first_load: # 先去 https://arena.ai/agent 登入，登入咗先再去指定 URL # 搵到email=登入成功；搵唔到email但搵到login=要登入；兩個都搵唔到=有問題（page未load/阻塞） page.goto("https://arena.ai/agent", wait_until="domcontentloaded", timeout=60000) # v0.93: URL 變動後過 checkbox 關（跟 CA URL-CHANGE 模式） ca_checkbox_gate(page, sid, "URL-CHANGE") log(sid, "先去 arena.ai/agent 登入頁，120s 內搵 slide menu 制（搵到=loaded）...") wait_for_sidebar_button(page, sid) # startup 登入檢查：等登入指標（電郵/Log In掣）出現先行 wait_for_login_indicators(page, sid) login_state = check_login_state(page, sid, cfg["account"], browser) log(sid, f"Startup 登入檢查（agent 頁）：{login_state}") if login_state == 'session_cleared_retry': log(sid, "Arena session missing 已清 arena，重開 browser 再試") try: browser.close() except: pass browser = None continue # 淨係搵到 email（登入成功）先至去指定 URL；其他狀態唔去，留返 first_load=True 重試 # 雙重確認：用 sidebar 明確 verify 一次 email，冇就唔去 URL if login_state == 'logged_in': verified_email = verify_email_via_sidebar(page, sid) if not verified_email: log(sid, "⚠️ sidebar 二次確認搵唔到 email → 唔去指定 URL，轉去 login") login_state = 'logged_out' else: log(sid, f"已登入（sidebar 確認有email），去指定 URL：{cfg['url']}") if sid in ROTATION_TAILS: _rot_cur = load_url_rotation(sid) if _rot_cur.get("url"): cfg["url"] = _rot_cur["url"] page.goto(cfg["url"], wait_until="domcontentloaded", timeout=60000) enforce_window_geometry(sid, stage="TARGET-URL") ensure_zoom_100(page, sid, "TARGET-URL", profile_dir) # v0.93: URL 變動後過 checkbox 關（跟 CA URL-CHANGE 模式） ca_checkbox_gate(page, sid, "URL-CHANGE") log(sid, "指定 URL 載入中，120s 內搵 slide menu 制...") wait_for_sidebar_button(page, sid) first_load = False elif login_state in ('logged_out', 'login_failed'): log(sid, "⚠️ 搵唔到email但搵到login→要登入，登入失敗，等人手處理（唔去指定URL，重試）") else: # unknown log(sid, "⚠️ 搵唔到email+搵唔到login=有問題（page未load/阻塞），唔去指定URL，重試") else: # 非首次：唔 goto，直接檢查當前 DOM pass # 先檢查頁面係咪 hang 咗（唔好俾假 DOM 呃） # 注意：首次載入後等夠耐先 check，唔好誤判 # owner 2026-10-05 14:50-14:51 下令：is_page_hung 只准用喺 Arena URL # （指定 URL＋agent 頁＋Arena 主頁），其他 URL（Google 登入等）唔好用， # 會誤判閂 browser _cur_url = page.url if hasattr(page, 'url') else '' if "arena.ai" in _cur_url and is_page_hung(page): log(sid, "⚠️ 頁面 JS hang 咗（DOM 可能係假狀態），reload 緊...") page.reload(wait_until="domcontentloaded", timeout=60000) time.sleep(3) # reload 後再 check 一次，如果仲 hang 就當 browser 死重開 if is_page_hung(page): log(sid, "Reload 後仲 hang，懷疑 renderer 死咗，重開 browser") try: browser.close() except: pass browser = None continue else: log(sid, "Reload 後恢復正常") # 檢查 DOM 狀態 _zok_c, _zmeas_c = ensure_zoom_100(page, sid, "CYCLE", profile_dir) if not _zok_c: log(sid, f"🚨 [CAUSAL-NOTE:Z2-HOLD] 本 cycle 因 Zoom 校唔實 ({_zmeas_c}) 跳過所有點擊；即刻收 Browser -> 清磁碟 Preferences -> 下輪重開") try: browser.close() except Exception: pass browser = None _sanitize_profile_zoom_prefs(profile_dir, sid) if once: break time.sleep(CHECK_INTERVAL) continue content = page.content() # 心跳記錄（防假成功）：每次 cycle 記錄 page 狀態 heartbeat = get_page_heartbeat(page, sid) log(sid, f"心跳: {heartbeat}") # 強制 window 尺寸（2026-10-05 19:49 HKT owner 下令：硬性規定） # 唔係 1408x900 就唔准運行，跳過呢個 cycle if not enforce_window_geometry(sid): log(sid, "🚨 尺寸唔啱，呢個 cycle 跳過，等 30 秒再試") time.sleep(CHECK_INTERVAL) continue # v3：一個 cycle 做晒（do_send_cycle 處理：Stop掣→skip；Keep working→撳+5s+reload；send；10s+reload驗證） result = do_send_cycle(page, sid) log(sid, f"Cycle 結果: {result}") # 檢測對話超長 → AI 已完結，URL 無效，交 Muse AI handle if is_conversation_too_long(page, sid): if sid in ROTATION_TAILS: _rot_nxt = rotation_advance(sid) if _rot_nxt.get("url"): cfg["url"] = _rot_nxt["url"] first_load = True log(sid, f"🚨 對話超長確認：{sid} 唔報 Muse，即刻經 first_load 轉去 URL（{_rot_nxt.get('label')}: {cfg['url']}）") continue log(sid, "🚨 對話超長確認：呢個 AI 已完結，URL 無效 → 交 Muse AI handle") # 寫 bus 訊息俾 host（經 SSH 去 server /root/muse-mesh/inbox/host/，host 會掃到） try: import json as _json, time as _t, subprocess as _sp msg = { "from": f"watcher-{sid.lower()}", "to": "host", "timestamp": int(_t.time()), "priority": "urgent", "type": "conversation_too_long", "body": f"{sid} 對話超長，AI 已完結，URL 無效：{cfg['url']}" } ts = int(_t.time()) # 經 SSH 寫去 server bus（host inbox） _sp.run([ "/home/hatch/workspace/ssh-newserver.sh", f"cat > /root/muse-mesh/inbox/host/{ts}-watcher-{sid.lower()}.json && chmod 600 /root/muse-mesh/inbox/host/{ts}-watcher-{sid.lower()}.json" ], input=_json.dumps(msg).encode(), timeout=30, capture_output=True) log(sid, "已寫 bus 訊息去 server：conversation_too_long（host 會掃到）") except Exception as e: log(sid, f"寫 bus 訊息出錯: {e}") # 停呢個 watcher，等 Muse AI 開新 URL log(sid, "停 watcher，等 Muse AI 開新對話") break if result in ("HOLD_ERROR", "ERROR", "NO_INPUT", "NO_INPUT_2", "KW_CLICK_ERROR"): log(sid, f"⚠️ {result} → 需要人手處理，{CHECK_INTERVAL}秒後重試") # （is_login_flow_stuck_on_session_missing 只准喺 Google login 期間用，唔喺主 loop 用） elif result == "LOGIN_FAILED": log(sid, "⚠️ 登入失敗，需要人手登入！") # 每 120 個 cycle（約 1 小時）做一次登入檢查 cycle_count += 1 if cycle_count % 120 == 0: login_state = check_login_state(page, sid, cfg["account"], browser) log(sid, f"定時登入檢查：{login_state}") if login_state == 'session_cleared_retry': log(sid, "Arena session missing 已清 arena，重開 browser 再試") try: browser.close() except: pass browser = None continue if login_state in ('logged_out', 'login_failed'): log(sid, "⚠️ 檢測到登出/登入失敗，需要人手登入！") elif login_state == 'unknown': log(sid, "⚠️ 登入狀態不明，已截圖留證") if once: break # SUCCESS_NEXT_CYCLE：即刻下一個循環，唔等 30 秒 if result == "SUCCESS_NEXT_CYCLE": log(sid, "即刻開始下一個循環（唔等 30 秒）") continue time.sleep(CHECK_INTERVAL) except Exception as e: log(sid, f"檢查出錯: {e}，{CHECK_INTERVAL}秒後重試") # 嘗試關閉壞掉的 browser，下輪重開 try: if browser: browser.close() except: pass browser = None if once: break time.sleep(CHECK_INTERVAL) finally: if browser: try: browser.close() except: pass return 0 if __name__ == "__main__": if len(sys.argv) < 2: print(f"用法: {sys.argv[0]} <席位ID> [--once]") sys.exit(1) sid = sys.argv[1].upper() once = "--once" in sys.argv sys.exit(run_watcher(sid, once))