kyson/grok-register
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: address review follow-up hardening

Browse Source
This commit is contained in:
github-actions[bot]
2026-07-14 14:53:49 +00:00
parent 800248d646
commit 6867124b17
7 changed files with 148 additions and 434 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/temp-review-fixes.yml
-29
View File
@@ -1,29 +0,0 @@
name: Temporary review fixes
on:
workflow_dispatch:
permissions:
contents: write
jobs:
apply:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Repair review patch GUI replacement
run: python tools/fix_review_patch_gui_replace.py
- name: Apply review fixes
run: python tools/apply_review_fixes.py
- name: Commit and cleanup
run: |
git config user.name "github-actions[bot]"
git config user.email "41898282+github-actions[bot]@users.noreply.github.com"
git rm -f tools/apply_review_fixes.py tools/fix_review_patch_gui_replace.py .github/workflows/temp-review-fixes.yml
git add -A
if git diff --cached --quiet; then
echo "No changes to commit"
else
git commit -m "fix: address review follow-up hardening"
git push
fi
.gitignore
-1
View File
@@ -25,4 +25,3 @@ cpa_auths/
xai-*.json
cpa_auth_failed.txt
screenshots/
*.png
cpa_xai/browser_confirm.py
+23 -15
View File
@@ -194,22 +194,30 @@ def create_standalone_page(proxy: Optional[str] = None, headless: bool = False,
resolved = resolve_proxy(proxy)
proxy_bridge = None
chrome_proxy, proxy_bridge = prepare_chromium_proxy(resolved, log=logger)
if chrome_proxy:
options.set_argument("--proxy-server=%s" % chrome_proxy)
logger("browser proxy=%s (chromium %s)" % (proxy_log_label(resolved), chrome_proxy))
else:
logger("browser proxy=(none)")
try:
if chrome_proxy:
options.set_argument("--proxy-server=%s" % chrome_proxy)
logger("browser proxy=%s (chromium %s)" % (proxy_log_label(resolved), chrome_proxy))
else:
logger("browser proxy=(none)")
browser = Chromium(options)
if proxy_bridge is not None:
try:
setattr(browser, "_cpa_proxy_bridge", proxy_bridge)
except Exception:
pass
_register_mint_browser(browser)
page = browser.latest_tab
logger("standalone chromium started")
return browser, page
browser = Chromium(options)
if proxy_bridge is not None:
try:
setattr(browser, "_cpa_proxy_bridge", proxy_bridge)
except Exception:
pass
_register_mint_browser(browser)
page = browser.latest_tab
logger("standalone chromium started")
return browser, page
except Exception:
if proxy_bridge is not None:
try:
proxy_bridge.stop()
except Exception:
pass
raise
def close_standalone(browser: Any) -> None:
cpa_xai/oauth_device.py
+2 -2
View File
@@ -251,9 +251,9 @@ def poll_device_token(
"device_code": str(device_code).strip(),
"client_id": client_id,
},
timeout=timeout,
timeout=min(float(timeout), 5.0),
proxy=proxy,
retries=2,
retries=0,
retry_sleep=1.0,
)
net_streak = 0
grok_register_ttk.py
+123 -24
View File
@@ -36,6 +36,7 @@ import socket
import socketserver
import ssl
import urllib.parse
import tempfile
os.environ.setdefault("TK_SILENCE_DEPRECATION", "1")
@@ -104,6 +105,10 @@ class AccountRetryNeeded(Exception):
pass
class ConfigError(RuntimeError):
pass
def load_config():
global config
if os.path.exists(CONFIG_FILE):
@@ -114,20 +119,48 @@ def load_config():
raise ValueError("config root must be a JSON object")
config = {**DEFAULT_CONFIG, **loaded}
except Exception as exc:
message = f"配置文件解析失败: {CONFIG_FILE}: {exc}"
print(f"[!] {message}", file=sys.stderr)
raise SystemExit(message)
raise ConfigError(f"配置文件解析失败: {CONFIG_FILE}: {exc}") from exc
else:
config = DEFAULT_CONFIG.copy()
return config
def save_config():
config_dir = os.path.dirname(os.path.abspath(CONFIG_FILE))
os.makedirs(config_dir, exist_ok=True)
fd = None
temp_path = None
try:
with open(CONFIG_FILE, "w", encoding="utf-8") as f:
fd, temp_path = tempfile.mkstemp(prefix=".config-", suffix=".json.tmp", dir=config_dir)
with os.fdopen(fd, "w", encoding="utf-8") as f:
fd = None
json.dump(config, f, indent=4, ensure_ascii=False)
except Exception as e:
print(f"保存配置失败: {e}")
f.write("\n")
f.flush()
os.fsync(f.fileno())
try:
os.chmod(temp_path, 0o600)
except Exception:
pass
os.replace(temp_path, CONFIG_FILE)
temp_path = None
try:
os.chmod(CONFIG_FILE, 0o600)
except Exception:
pass
except Exception as exc:
raise ConfigError(f"保存配置失败: {exc}") from exc
finally:
if fd is not None:
try:
os.close(fd)
except Exception:
pass
if temp_path and os.path.exists(temp_path):
try:
os.unlink(temp_path)
except Exception:
pass
def ensure_stable_python_runtime():
@@ -668,6 +701,12 @@ def add_token_to_grok2api_local_pool(raw_token, email="", log_callback=None):
parent = os.path.dirname(token_file)
os.makedirs(parent, exist_ok=True)
lock_path = token_file + ".lock"
try:
with open(lock_path, "a", encoding="utf-8"):
pass
os.chmod(lock_path, 0o600)
except Exception:
pass
try:
from filelock import FileLock
except Exception as exc:
@@ -688,8 +727,10 @@ def add_token_to_grok2api_local_pool(raw_token, email="", log_callback=None):
if not isinstance(data, dict):
raise RuntimeError("本地 token 文件根节点不是 JSON object，拒绝覆盖")
pool = data.get(pool_name)
if not isinstance(pool, list):
if pool is None:
pool = []
elif not isinstance(pool, list):
raise RuntimeError(f"本地 token 池 {pool_name} 不是列表，拒绝覆盖")
existing = set()
for item in pool:
if isinstance(item, str):
@@ -709,18 +750,31 @@ def add_token_to_grok2api_local_pool(raw_token, email="", log_callback=None):
dst.write(src.read())
dst.flush()
os.fsync(dst.fileno())
try:
os.chmod(backup_path, 0o600)
except Exception:
pass
except Exception as exc:
raise RuntimeError(f"创建本地 token 备份失败，拒绝继续写入: {exc}")
temp_path = token_file + ".tmp"
fd, temp_path = tempfile.mkstemp(prefix=".token-", suffix=".tmp", dir=parent)
try:
with open(temp_path, "w", encoding="utf-8") as f:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
f.write("\n")
f.flush()
os.fsync(f.fileno())
try:
os.chmod(temp_path, 0o600)
except Exception:
pass
os.replace(temp_path, token_file)
temp_path = None
try:
os.chmod(token_file, 0o600)
except Exception:
pass
finally:
if os.path.exists(temp_path):
if temp_path and os.path.exists(temp_path):
try:
os.unlink(temp_path)
except Exception:
@@ -822,8 +876,10 @@ def add_token_to_grok2api_remote_pool(raw_token, email="", log_callback=None):
if not loaded_remote_state:
raise RuntimeError("无法安全读取远端 token 池，拒绝执行全量覆盖: " + "; ".join(load_errors))
pool = current.get(pool_name)
if not isinstance(pool, list):
if pool is None:
pool = []
elif not isinstance(pool, list):
raise RuntimeError(f"远端 token 池 {pool_name} 不是列表，拒绝全量覆盖")
existing = set()
for item in pool:
if isinstance(item, str):
@@ -1978,6 +2034,12 @@ def cleanup_runtime_memory(log_callback=None, reason="定期清理"):
if log_callback:
log_callback(f"[*] {reason}: 关闭浏览器并清理内存")
stop_browser()
try:
from cpa_xai.browser_confirm import shutdown_mint_browsers
shutdown_mint_browsers()
except Exception as exc:
if log_callback:
log_callback(f"[Debug] CPA 浏览器清理失败: {exc}")
collected = gc.collect()
if log_callback:
log_callback(f"[*] Python GC 已回收对象数: {collected}")
@@ -3071,6 +3133,7 @@ class GrokRegisterGUI:
self.results = []
self.stop_requested = False
self.ui_queue = queue.Queue()
self._ui_thread_id = threading.get_ident()
self.accounts_output_file = ""
self.setup_ui()
@@ -3288,25 +3351,40 @@ class GrokRegisterGUI:
self.log("[*] GUI 已就绪，配置已加载")
self.log(f"[*] 当前邮箱服务商: {self.email_provider_var.get()} | 注册数量: {self.count_var.get()}")
def _call_ui(self, func, *args):
if threading.get_ident() == self._ui_thread_id:
func(*args)
return
try:
self.root.after(0, lambda: func(*args))
except Exception:
pass
def _append_log_line(self, line):
self.log_text.insert(tk.END, f"{line}\n")
self.log_text.see(tk.END)
def log(self, message):
timestamp = datetime.datetime.now().strftime("%H:%M:%S")
line = f"[{timestamp}] {message}"
print(line, flush=True)
self.log_text.insert(tk.END, f"{line}\n")
self.log_text.see(tk.END)
self._call_ui(self._append_log_line, line)
def clear_log(self):
self.log_text.delete(1.0, tk.END)
self._call_ui(lambda: self.log_text.delete(1.0, tk.END))
def update_stats(self):
self.stats_var.set(f"成功: {self.success_count} | 失败: {self.fail_count}")
self._call_ui(lambda: self.stats_var.set(f"成功: {self.success_count} | 失败: {self.fail_count}"))
def _set_running_ui(self, running):
self.is_running = running
self.start_btn.config(state=tk.DISABLED if running else tk.NORMAL)
self.stop_btn.config(state=tk.NORMAL if running else tk.DISABLED)
self.status_var.set("运行中..." if running else "就绪")
self.status_label.config(foreground="blue" if running else "green")
def apply():
self.start_btn.config(state=tk.DISABLED if running else tk.NORMAL)
self.stop_btn.config(state=tk.NORMAL if running else tk.DISABLED)
self.status_var.set("运行中..." if running else "就绪")
self.status_label.config(foreground="blue" if running else "green")
self._call_ui(apply)
def should_stop(self):
return self.stop_requested or not self.is_running
@@ -3340,7 +3418,11 @@ class GrokRegisterGUI:
config["cloudflare_path_accounts"] = raw_paths[1] if raw_paths[1].startswith("/") else ("/" + raw_paths[1])
config["cloudflare_path_token"] = raw_paths[2] if raw_paths[2].startswith("/") else ("/" + raw_paths[2])
config["cloudflare_path_messages"] = raw_paths[3] if raw_paths[3].startswith("/") else ("/" + raw_paths[3])
save_config()
try:
save_config()
except ConfigError as exc:
self.log(f"[!] 配置保存失败: {exc}")
return
if config["email_provider"] == "cloudflare" and not config["cloudflare_api_base"]:
self.log("[!] Cloudflare 模式需要先填写 Cloudflare API Base")
return
@@ -3361,7 +3443,11 @@ class GrokRegisterGUI:
self.log("[!] 注册数量无效")
return
config["register_count"] = count
save_config()
try:
save_config()
except ConfigError as exc:
self.log(f"[!] 配置保存失败: {exc}")
return
self.stop_requested = False
self.success_count = 0
self.fail_count = 0
@@ -3522,7 +3608,7 @@ class GrokRegisterGUI:
except Exception as exc:
self.log(f"[!] 任务异常: {exc}")
finally:
stop_browser()
cleanup_runtime_memory(log_callback=self.log, reason="任务结束")
self._set_running_ui(False)
self.log("[*] 任务结束")
@@ -3691,7 +3777,11 @@ def run_registration_cli(count):
def main_cli():
load_config()
try:
load_config()
except ConfigError as exc:
cli_log(f"[!] {exc}")
return
count = int(config.get("register_count", 1) or 1)
cli_log("[*] CLI 已加载配置")
cli_log(f"[*] 当前邮箱服务商: {config.get('email_provider', 'duckmail')} | 注册数量: {count}")
@@ -3717,7 +3807,16 @@ def main():
return
root = tk.Tk()
setup_light_theme(root)
app = GrokRegisterGUI(root)
try:
app = GrokRegisterGUI(root)
except ConfigError as exc:
print(f"[!] {exc}", file=sys.stderr)
try:
messagebox.showerror("配置错误", str(exc))
except Exception:
pass
root.destroy()
return
root.mainloop()
tools/apply_review_fixes.py
-302
View File
@@ -1,302 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from pathlib import Path
import ast
ROOT = Path(__file__).resolve().parents[1]
APP = ROOT / "grok_register_ttk.py"
BROWSER = ROOT / "cpa_xai" / "browser_confirm.py"
OAUTH = ROOT / "cpa_xai" / "oauth_device.py"
GITIGNORE = ROOT / ".gitignore"
def read(path, encoding="utf-8"):
return path.read_text(encoding=encoding)
def write(path, content, encoding="utf-8"):
path.write_text(content, encoding=encoding)
def one(text, old, new, label):
count = text.count(old)
if count != 1:
raise RuntimeError(f"{label}: expected one match, got {count}")
return text.replace(old, new, 1)
def between(text, start, end, new, label):
i = text.find(start)
if i < 0:
raise RuntimeError(f"{label}: start not found")
j = text.find(end, i + len(start))
if j < 0:
raise RuntimeError(f"{label}: end not found")
return text[:i] + new + text[j:]
app = read(APP, "utf-8-sig")
if "import tempfile\n" not in app:
app = one(app, "import urllib.parse\n", "import urllib.parse\nimport tempfile\n", "import tempfile")
if "class ConfigError(RuntimeError):" not in app:
app = one(
app,
"class AccountRetryNeeded(Exception):\n pass\n\n\ndef load_config():",
"class AccountRetryNeeded(Exception):\n pass\n\n\nclass ConfigError(RuntimeError):\n pass\n\n\ndef load_config():",
"ConfigError",
)
app = between(app, "def load_config():\n", "def save_config():\n", '''def load_config():
global config
if os.path.exists(CONFIG_FILE):
try:
with open(CONFIG_FILE, "r", encoding="utf-8") as f:
loaded = json.load(f)
if not isinstance(loaded, dict):
raise ValueError("config root must be a JSON object")
config = {**DEFAULT_CONFIG, **loaded}
except Exception as exc:
raise ConfigError(f"配置文件解析失败: {CONFIG_FILE}: {exc}") from exc
else:
config = DEFAULT_CONFIG.copy()
return config
''', "load_config")
app = between(app, "def save_config():\n", "def ensure_stable_python_runtime():\n", '''def save_config():
config_dir = os.path.dirname(os.path.abspath(CONFIG_FILE))
os.makedirs(config_dir, exist_ok=True)
fd = None
temp_path = None
try:
fd, temp_path = tempfile.mkstemp(prefix=".config-", suffix=".json.tmp", dir=config_dir)
with os.fdopen(fd, "w", encoding="utf-8") as f:
fd = None
json.dump(config, f, indent=4, ensure_ascii=False)
f.write("\\n")
f.flush()
os.fsync(f.fileno())
try:
os.chmod(temp_path, 0o600)
except Exception:
pass
os.replace(temp_path, CONFIG_FILE)
temp_path = None
try:
os.chmod(CONFIG_FILE, 0o600)
except Exception:
pass
except Exception as exc:
raise ConfigError(f"保存配置失败: {exc}") from exc
finally:
if fd is not None:
try:
os.close(fd)
except Exception:
pass
if temp_path and os.path.exists(temp_path):
try:
os.unlink(temp_path)
except Exception:
pass
''', "save_config")
app = one(app, " lock_path = token_file + \".lock\"\n try:\n from filelock import FileLock\n", " lock_path = token_file + \".lock\"\n try:\n with open(lock_path, \"a\", encoding=\"utf-8\"):\n pass\n os.chmod(lock_path, 0o600)\n except Exception:\n pass\n try:\n from filelock import FileLock\n", "lock chmod")
app = one(app, " pool = data.get(pool_name)\n if not isinstance(pool, list):\n pool = []\n", " pool = data.get(pool_name)\n if pool is None:\n pool = []\n elif not isinstance(pool, list):\n raise RuntimeError(f\"本地 token 池 {pool_name} 不是列表，拒绝覆盖\")\n", "local pool guard")
app = one(app, " os.fsync(dst.fileno())\n except Exception as exc:\n", " os.fsync(dst.fileno())\n try:\n os.chmod(backup_path, 0o600)\n except Exception:\n pass\n except Exception as exc:\n", "backup chmod")
app = one(app, ''' temp_path = token_file + ".tmp"
try:
with open(temp_path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
f.write("\\n")
f.flush()
os.fsync(f.fileno())
os.replace(temp_path, token_file)
finally:
if os.path.exists(temp_path):
try:
os.unlink(temp_path)
except Exception:
pass
''', ''' fd, temp_path = tempfile.mkstemp(prefix=".token-", suffix=".tmp", dir=parent)
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
f.write("\\n")
f.flush()
os.fsync(f.fileno())
try:
os.chmod(temp_path, 0o600)
except Exception:
pass
os.replace(temp_path, token_file)
temp_path = None
try:
os.chmod(token_file, 0o600)
except Exception:
pass
finally:
if temp_path and os.path.exists(temp_path):
try:
os.unlink(temp_path)
except Exception:
pass
''', "token mkstemp")
remote_pool_new = " pool = current.get(pool_name)\n if pool is None:\n pool = []\n elif not isinstance(pool, list):\n raise RuntimeError(f\"远端 token 池 {pool_name} 不是列表，拒绝全量覆盖\")\n"
if remote_pool_new not in app:
app = one(app, " pool = current.get(pool_name)\n if not isinstance(pool, list):\n pool = []\n", remote_pool_new, "remote pool guard")
app = between(app, "def cleanup_runtime_memory(log_callback=None, reason=\"定期清理\"):\n", "def refresh_active_page():\n", '''def cleanup_runtime_memory(log_callback=None, reason="定期清理"):
if log_callback:
log_callback(f"[*] {reason}: 关闭浏览器并清理内存")
stop_browser()
try:
from cpa_xai.browser_confirm import shutdown_mint_browsers
shutdown_mint_browsers()
except Exception as exc:
if log_callback:
log_callback(f"[Debug] CPA 浏览器清理失败: {exc}")
collected = gc.collect()
if log_callback:
log_callback(f"[*] Python GC 已回收对象数: {collected}")
''', "cleanup")
app = one(app, " self.ui_queue = queue.Queue()\n self.accounts_output_file = \"\"\n", " self.ui_queue = queue.Queue()\n self._ui_thread_id = threading.get_ident()\n self.accounts_output_file = \"\"\n", "ui thread id")
app = one(app, ''' def log(self, message):
timestamp = datetime.datetime.now().strftime("%H:%M:%S")
line = f"[{timestamp}] {message}"
print(line, flush=True)
self.log_text.insert(tk.END, f"{line}\n")
self.log_text.see(tk.END)
def clear_log(self):
self.log_text.delete(1.0, tk.END)
def update_stats(self):
self.stats_var.set(f"成功: {self.success_count} | 失败: {self.fail_count}")
def _set_running_ui(self, running):
self.is_running = running
self.start_btn.config(state=tk.DISABLED if running else tk.NORMAL)
self.stop_btn.config(state=tk.NORMAL if running else tk.DISABLED)
self.status_var.set("运行中..." if running else "就绪")
self.status_label.config(foreground="blue" if running else "green")
''', ''' def _call_ui(self, func, *args):
if threading.get_ident() == self._ui_thread_id:
func(*args)
return
try:
self.root.after(0, lambda: func(*args))
except Exception:
pass
def _append_log_line(self, line):
self.log_text.insert(tk.END, f"{line}\\n")
self.log_text.see(tk.END)
def log(self, message):
timestamp = datetime.datetime.now().strftime("%H:%M:%S")
line = f"[{timestamp}] {message}"
print(line, flush=True)
self._call_ui(self._append_log_line, line)
def clear_log(self):
self._call_ui(lambda: self.log_text.delete(1.0, tk.END))
def update_stats(self):
self._call_ui(lambda: self.stats_var.set(f"成功: {self.success_count} | 失败: {self.fail_count}"))
def _set_running_ui(self, running):
self.is_running = running
def apply():
self.start_btn.config(state=tk.DISABLED if running else tk.NORMAL)
self.stop_btn.config(state=tk.NORMAL if running else tk.DISABLED)
self.status_var.set("运行中..." if running else "就绪")
self.status_label.config(foreground="blue" if running else "green")
self._call_ui(apply)
''', "GUI UI calls")
app = one(app, " save_config()\n if config[\"email_provider\"] == \"cloudflare\" and not config[\"cloudflare_api_base\"]:\n", " try:\n save_config()\n except ConfigError as exc:\n self.log(f\"[!] 配置保存失败: {exc}\")\n return\n if config[\"email_provider\"] == \"cloudflare\" and not config[\"cloudflare_api_base\"]:\n", "first save guard")
app = one(app, " config[\"register_count\"] = count\n save_config()\n self.stop_requested = False\n", " config[\"register_count\"] = count\n try:\n save_config()\n except ConfigError as exc:\n self.log(f\"[!] 配置保存失败: {exc}\")\n return\n self.stop_requested = False\n", "second save guard")
app = one(app, " finally:\n stop_browser()\n self._set_running_ui(False)\n self.log(\"[*] 任务结束\")\n", " finally:\n cleanup_runtime_memory(log_callback=self.log, reason=\"任务结束\")\n self._set_running_ui(False)\n self.log(\"[*] 任务结束\")\n", "GUI final cleanup")
app = one(app, "def main_cli():\n load_config()\n count = int(config.get(\"register_count\", 1) or 1)\n", "def main_cli():\n try:\n load_config()\n except ConfigError as exc:\n cli_log(f\"[!] {exc}\")\n return\n count = int(config.get(\"register_count\", 1) or 1)\n", "CLI config handling")
app = one(app, " root = tk.Tk()\n setup_light_theme(root)\n app = GrokRegisterGUI(root)\n root.mainloop()\n", " root = tk.Tk()\n setup_light_theme(root)\n try:\n app = GrokRegisterGUI(root)\n except ConfigError as exc:\n print(f\"[!] {exc}\", file=sys.stderr)\n try:\n messagebox.showerror(\"配置错误\", str(exc))\n except Exception:\n pass\n root.destroy()\n return\n root.mainloop()\n", "GUI config handling")
ast.parse(app)
write(APP, app, "utf-8")
browser = read(BROWSER)
browser = one(browser, ''' resolved = resolve_proxy(proxy)
proxy_bridge = None
chrome_proxy, proxy_bridge = prepare_chromium_proxy(resolved, log=logger)
if chrome_proxy:
options.set_argument("--proxy-server=%s" % chrome_proxy)
logger("browser proxy=%s (chromium %s)" % (proxy_log_label(resolved), chrome_proxy))
else:
logger("browser proxy=(none)")
browser = Chromium(options)
if proxy_bridge is not None:
try:
setattr(browser, "_cpa_proxy_bridge", proxy_bridge)
except Exception:
pass
_register_mint_browser(browser)
page = browser.latest_tab
logger("standalone chromium started")
return browser, page
''', ''' resolved = resolve_proxy(proxy)
proxy_bridge = None
chrome_proxy, proxy_bridge = prepare_chromium_proxy(resolved, log=logger)
try:
if chrome_proxy:
options.set_argument("--proxy-server=%s" % chrome_proxy)
logger("browser proxy=%s (chromium %s)" % (proxy_log_label(resolved), chrome_proxy))
else:
logger("browser proxy=(none)")
browser = Chromium(options)
if proxy_bridge is not None:
try:
setattr(browser, "_cpa_proxy_bridge", proxy_bridge)
except Exception:
pass
_register_mint_browser(browser)
page = browser.latest_tab
logger("standalone chromium started")
return browser, page
except Exception:
if proxy_bridge is not None:
try:
proxy_bridge.stop()
except Exception:
pass
raise
''', "CPA bridge cleanup")
ast.parse(browser)
write(BROWSER, browser)
oauth = read(OAUTH)
oauth = one(oauth, " timeout=timeout,\n proxy=proxy,\n retries=2,\n retry_sleep=1.0,\n )\n net_streak = 0\n", " timeout=min(float(timeout), 5.0),\n proxy=proxy,\n retries=0,\n retry_sleep=1.0,\n )\n net_streak = 0\n", "oauth poll timeout")
ast.parse(oauth)
write(OAUTH, oauth)
gitignore = read(GITIGNORE)
write(GITIGNORE, "\n".join(line for line in gitignore.splitlines() if line.strip() != "*.png").rstrip() + "\n")
for path in (APP, BROWSER, OAUTH):
ast.parse(read(path, "utf-8"))
print("review fixes applied")
tools/fix_review_patch_gui_replace.py
-61
View File
@@ -1,61 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from pathlib import Path
script = Path(__file__).resolve().with_name("apply_review_fixes.py")
text = script.read_text(encoding="utf-8")
needle = "app = one(app, ''' def log(self, message):"
start = text.find(needle)
if start < 0:
if "app = between(app, \" def log(self, message):\\n\"" in text:
print("GUI replacement already robust")
raise SystemExit(0)
raise RuntimeError("GUI UI replacement block start not found")
end_marker = "\n\napp = one(app, \" save_config()\\n if config[\\\"email_provider\\\"] == \\\"cloudflare\\\""
end = text.find(end_marker, start)
if end < 0:
raise RuntimeError("GUI UI replacement block end not found")
replacement = r'''app = between(app, " def log(self, message):\n", " def should_stop(self):\n", ''' + "'''" + r''' def _call_ui(self, func, *args):
if threading.get_ident() == self._ui_thread_id:
func(*args)
return
try:
self.root.after(0, lambda: func(*args))
except Exception:
pass
def _append_log_line(self, line):
self.log_text.insert(tk.END, f"{line}\\n")
self.log_text.see(tk.END)
def log(self, message):
timestamp = datetime.datetime.now().strftime("%H:%M:%S")
line = f"[{timestamp}] {message}"
print(line, flush=True)
self._call_ui(self._append_log_line, line)
def clear_log(self):
self._call_ui(lambda: self.log_text.delete(1.0, tk.END))
def update_stats(self):
self._call_ui(lambda: self.stats_var.set(f"成功: {self.success_count} | 失败: {self.fail_count}"))
def _set_running_ui(self, running):
self.is_running = running
def apply():
self.start_btn.config(state=tk.DISABLED if running else tk.NORMAL)
self.stop_btn.config(state=tk.NORMAL if running else tk.DISABLED)
self.status_var.set("运行中..." if running else "就绪")
self.status_label.config(foreground="blue" if running else "green")
self._call_ui(apply)
''' + "'''" + r''', "GUI UI calls")
'''
text = text[:start] + replacement + text[end:]
script.write_text(text, encoding="utf-8")
print("review patch GUI replacement made robust")