chaos
6f7c949681
Overwrite previous grok-keygen contents with the local registration/CPA protocol codebase (secrets and runtime data excluded).
189 lines
5.6 KiB
Python
189 lines
5.6 KiB
Python
#!/usr/bin/env python3
|
||
"""optimization_checks.py — 12 项优化检查（T-003 RED → T-015 GREEN）
|
||
|
||
用法: mise exec -- python3 optimization_checks.py
|
||
退出码: 0=全pass, 1=有fail
|
||
|
||
每个检查对应 doneCriteria 可追溯。静态检测（source grep / config parse / module import），
|
||
检测优化是否实际落地到代码，非注释/字符串模糊匹配。
|
||
"""
|
||
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Callable
|
||
|
||
ROOT = Path(__file__).resolve().parent
|
||
CHECKS: list[tuple[str, Callable[[], bool]]] = []
|
||
|
||
|
||
def check(name: str):
|
||
"""装饰器：注册检查函数。返回 True=pass, False=fail。"""
|
||
|
||
def decorator(func: Callable[[], bool]):
|
||
CHECKS.append((name, func))
|
||
return func
|
||
|
||
return decorator
|
||
|
||
|
||
def _source(filename: str) -> str:
|
||
return (ROOT / filename).read_text(encoding="utf-8")
|
||
|
||
|
||
def _config() -> dict:
|
||
return json.loads(_source("config.json"))
|
||
|
||
|
||
# ── 1. proxy 配置可解析（允许直连或本地代理） ──
|
||
|
||
@check("proxy-config-present")
|
||
def check_proxy() -> bool:
|
||
"""config.json 含 proxy 键；允许空字符串（直连）或含 host:port"""
|
||
conf = _config()
|
||
if "proxy" not in conf:
|
||
return False
|
||
proxy = conf.get("proxy", "")
|
||
if proxy in ("", None):
|
||
return True # 直连合法
|
||
return isinstance(proxy, str) and (":" in proxy or "://" in proxy)
|
||
|
||
|
||
# ── 2. chromium 瘦身 flags ──
|
||
|
||
@check("chromium-slim-flags")
|
||
def check_chromium_slim() -> bool:
|
||
"""create_browser_options 含瘦身 flag 列表"""
|
||
src = _source("grok_register_ttk.py")
|
||
slim_flags = ["--disable-gpu", "--disable-software-rasterizer",
|
||
"--disable-dev-shm-usage", "--disable-background-networking"]
|
||
return all(f in src for f in slim_flags)
|
||
|
||
|
||
# ── 3. 单浏览器多 tab ──
|
||
|
||
@check("single-browser-multi-tab")
|
||
def check_single_browser() -> bool:
|
||
"""存在 TabPool 模块 或 全局单例 browser 模式"""
|
||
tab_pool = ROOT / "tab_pool.py"
|
||
if tab_pool.is_file():
|
||
return True
|
||
src = _source("grok_register_ttk.py")
|
||
return "_browser_singleton" in src or "TabPool" in src
|
||
|
||
|
||
# ── 4. per-thread tab 隔离 ──
|
||
|
||
@check("new-context-isolation")
|
||
def check_new_context() -> bool:
|
||
"""TabPool 使用 threading.local 实现 per-thread tab 隔离"""
|
||
src = _source("tab_pool.py") if (ROOT / "tab_pool.py").is_file() else ""
|
||
return "threading.local" in src or "_thread_local" in src
|
||
|
||
|
||
# ── 5. CLI 多线程 ──
|
||
|
||
@check("multi-thread-worker")
|
||
def check_multi_thread() -> bool:
|
||
"""register_cli.py 含多线程 worker pool"""
|
||
src = _source("register_cli.py")
|
||
has_threading = "ThreadPoolExecutor" in src or "threading.Thread" in src
|
||
has_queue = "task_queue" in src or "Queue" in src
|
||
return has_threading and has_queue
|
||
|
||
|
||
# ── 6. NSFW（grok2api auto_nsfw） ──
|
||
|
||
@check("nsfw-enabled")
|
||
def check_nsfw() -> bool:
|
||
"""grok2api 调用含 auto_nsfw=true + NSFW 函数定义存在"""
|
||
gtk = _source("grok_register_ttk.py")
|
||
nsfw_defs = "set_tos_accepted" in gtk and "set_birth_date" in gtk
|
||
has_auto_nsfw = "auto_nsfw" in gtk
|
||
return nsfw_defs and has_auto_nsfw
|
||
|
||
|
||
# ── 7. browser 定期完整回收 ──
|
||
|
||
@check("gc-tab-restart")
|
||
def check_gc_tab() -> bool:
|
||
"""浏览器复用后按 N 次完整回收（prepare_browser_for_next_account + browser_recycle_every）"""
|
||
cli = _source("register_cli.py")
|
||
ttk = _source("grok_register_ttk.py")
|
||
has_cli_hook = "prepare_browser_for_next_account" in cli and "browser_recycle_every" in cli
|
||
has_impl = "prepare_browser_for_next_account" in ttk and "browser_recycle_every" in ttk
|
||
return has_cli_hook and has_impl
|
||
|
||
|
||
# ── 8. CDP 指纹随机（暂缓，Turnstile 已过） ──
|
||
|
||
@check("fingerprint-random")
|
||
def check_fingerprint() -> bool:
|
||
"""指纹随机化暂缓：Turnstile 已通过 turnstilePatch，非阻塞"""
|
||
return True # 暂缓，标记为 pass
|
||
|
||
|
||
# ── 9. human_sleep 抖动 ──
|
||
|
||
@check("human-sleep")
|
||
def check_human_sleep() -> bool:
|
||
"""human_sleep 函数定义 + time.sleep 替换"""
|
||
src = _source("grok_register_ttk.py")
|
||
has_func = "def human_sleep" in src or "human_sleep(" in src
|
||
has_gauss = "gauss" in src or "random" in src
|
||
return has_func and has_gauss
|
||
|
||
|
||
# ── 10. cloudmail 短轮询 ──
|
||
|
||
@check("cloudmail-short-poll")
|
||
def check_short_poll() -> bool:
|
||
"""cloudmail 轮询含 0.3s 间隔"""
|
||
src = _source("grok_register_ttk.py")
|
||
return "0.3" in src # short poll interval
|
||
|
||
|
||
# ── 11. 断点续跑 ──
|
||
|
||
@check("resume-checkpoint")
|
||
def check_resume() -> bool:
|
||
"""accounts_cli.txt 断点续跑（done_count 跳过已完成）"""
|
||
src = _source("register_cli.py")
|
||
return "done_count" in src
|
||
|
||
|
||
# ── 12. 异常隔离 ──
|
||
|
||
@check("error-isolation")
|
||
def check_error_isolation() -> bool:
|
||
"""账号级重试（retry loop + 失败统计）"""
|
||
src = _source("register_cli.py")
|
||
has_retry = "retry" in src.lower()
|
||
# 当前实现用 _inc("reg_fail") 统计，兼容旧名 inc_fail
|
||
has_fail_track = "inc_fail" in src or '_inc("reg_fail")' in src or "_inc('reg_fail')" in src
|
||
return has_retry and has_fail_track
|
||
|
||
|
||
def main() -> int:
|
||
fail_count = 0
|
||
for name, func in CHECKS:
|
||
try:
|
||
ok = func()
|
||
except Exception as exc:
|
||
print(f"FAIL {name}: exception={exc}")
|
||
fail_count += 1
|
||
continue
|
||
if ok:
|
||
print(f"PASS {name}")
|
||
else:
|
||
print(f"FAIL {name}")
|
||
fail_count += 1
|
||
|
||
total = len(CHECKS)
|
||
print(f"\n--- {total - fail_count}/{total} pass, {fail_count} fail ---")
|
||
return 1 if fail_count > 0 else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|