diff --git a/.github/workflows/temp-post-modularization-fixes.yml b/.github/workflows/temp-post-modularization-fixes.yml
deleted file mode 100644
index 75e8c82..0000000
--- a/.github/workflows/temp-post-modularization-fixes.yml
+++ /dev/null
@@ -1,66 +0,0 @@
-name: Temporary post-modularization regression fixes
-
-on:
- push:
- branches:
- - fix/post-modularization-regressions
- paths:
- - tools/apply_post_modularization_fixes.py
- - tools/repair_post_modularization_patcher.py
- - .github/workflows/temp-post-modularization-fixes.yml
- workflow_dispatch:
-
-permissions:
- contents: write
-
-jobs:
- fix-and-test:
- runs-on: ubuntu-latest
- steps:
- - uses: actions/checkout@v4
- with:
- ref: fix/post-modularization-regressions
- - uses: actions/setup-python@v5
- with:
- python-version: '3.9'
- - name: Install dependencies on Python 3.9
- run: |
- python -m pip install --upgrade pip
- python -m pip install -r requirements.txt
- python -m pip install filelock
- - name: Repair, validate, and apply fixes
- run: |
- python -m py_compile \
- tools/repair_post_modularization_patcher.py \
- tools/apply_post_modularization_fixes.py
- python tools/repair_post_modularization_patcher.py
- python -m py_compile tools/apply_post_modularization_fixes.py
- python tools/repair_post_modularization_patcher.py
- python -m py_compile tools/apply_post_modularization_fixes.py
- python tools/apply_post_modularization_fixes.py
- python -m compileall -q .
- - name: Run full suite on Python 3.9
- run: python -m unittest discover -s tests -p 'test_*.py'
- - uses: actions/setup-python@v5
- with:
- python-version: '3.12'
- - name: Install dependencies on Python 3.12
- run: |
- python -m pip install --upgrade pip
- python -m pip install -r requirements.txt
- python -m pip install filelock
- - name: Run full suite on Python 3.12
- run: |
- python -m compileall -q .
- python -m unittest discover -s tests -p 'test_*.py'
- - name: Commit verified fixes 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_post_modularization_fixes.py \
- tools/repair_post_modularization_patcher.py \
- .github/workflows/temp-post-modularization-fixes.yml
- git add -A
- git commit -m "fix: resolve post-modularization regressions"
- git push origin HEAD:fix/post-modularization-regressions
diff --git a/.gitignore b/.gitignore
index 5b2cee8..20e5570 100755
--- a/.gitignore
+++ b/.gitignore
@@ -25,3 +25,6 @@ cpa_auths/
xai-*.json
cpa_auth_failed.txt
screenshots/
+
+*.pending.jsonl
+*.pending.jsonl.lock
diff --git a/app_config.py b/app_config.py
index ca5abc2..3328a2d 100644
--- a/app_config.py
+++ b/app_config.py
@@ -162,7 +162,7 @@ def validate_run_requirements(cfg):
]
if missing:
raise ConfigError("远端 token 入池缺少必需配置: " + ", ".join(missing))
- if cfg["cpa_copy_to_hotload"] and not cfg["cpa_hotload_dir"]:
+ if cfg["cpa_export_enabled"] and cfg["cpa_copy_to_hotload"] and not cfg["cpa_hotload_dir"]:
raise ConfigError("启用 CPA 热加载复制时必须配置 cpa_hotload_dir")
return cfg
diff --git a/browser_runtime.py b/browser_runtime.py
index 4d83181..98b08ab 100644
--- a/browser_runtime.py
+++ b/browser_runtime.py
@@ -145,13 +145,14 @@ def apply_browser_proxy_option(options, proxy):
options.set_argument("--proxy-server", proxy)
-def create_browser_options(browser_proxy=""):
+def create_browser_options(browser_proxy="", extension_path=None):
options = ChromiumOptions()
options.auto_port()
options.set_timeouts(base=1)
apply_browser_proxy_option(options, browser_proxy)
- if _extension_path and os.path.exists(_extension_path):
- options.add_extension(_extension_path)
+ effective_extension = _extension_path if extension_path is None else str(extension_path or "")
+ if effective_extension and os.path.exists(effective_extension):
+ options.add_extension(effective_extension)
return options
diff --git a/cpa_export.py b/cpa_export.py
index 4daca4f..1357dca 100644
--- a/cpa_export.py
+++ b/cpa_export.py
@@ -6,6 +6,7 @@ import shutil
import sys
import time
from pathlib import Path
+from typing import Optional
_ROOT = Path(__file__).resolve().parent
_DEFAULT_AUTH_DIR = _ROOT / "cpa_auths"
@@ -15,7 +16,7 @@ _DEFAULT_AUTH_DIR = _ROOT / "cpa_auths"
class CpaExportSettings:
enabled: bool
auth_dir: Path
- hotload_dir: Path | None
+ hotload_dir: Optional[Path]
copy_to_hotload: bool
proxy: str
headless: bool
diff --git a/cpa_xai/browser_session.py b/cpa_xai/browser_session.py
index 101ba1a..71a9bbe 100644
--- a/cpa_xai/browser_session.py
+++ b/cpa_xai/browser_session.py
@@ -2,7 +2,6 @@
from __future__ import annotations
import os
-import sys
import threading
import time
from pathlib import Path
@@ -32,21 +31,14 @@ def create_standalone_page(proxy: Optional[str] = None, headless: bool = False,
options = None
package_root = Path(__file__).resolve().parents[1]
try:
- register_file = package_root / "grok_register_ttk.py"
- if register_file.is_file():
- register_dir = str(package_root)
- if register_dir not in sys.path:
- sys.path.insert(0, register_dir)
- try:
- from grok_register_ttk import create_browser_options # type: ignore
+ from browser_runtime import create_browser_options
- options = create_browser_options()
- logger("using register create_browser_options (turnstilePatch)")
- except Exception as exc: # noqa: BLE001
- logger("register browser options unavailable: %s" % exc)
- options = None
+ options = create_browser_options(
+ extension_path=package_root / "turnstilePatch"
+ )
+ logger("using shared browser_runtime.create_browser_options")
except Exception as exc: # noqa: BLE001
- logger("register options probe failed: %s" % exc)
+ logger("shared browser options unavailable: %s" % exc)
options = None
if options is None:
diff --git a/grok_register_ttk.py b/grok_register_ttk.py
index ab8a5e9..4e3597a 100755
--- a/grok_register_ttk.py
+++ b/grok_register_ttk.py
@@ -111,62 +111,6 @@ def log_exception(context, exc, log_callback=None):
-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)
- config = validate_config_structure(loaded)
- except ConfigError:
- raise
- except Exception as exc:
- raise ConfigError(f"配置文件解析失败: {CONFIG_FILE}: {exc}") from exc
- else:
- config = validate_config_structure(DEFAULT_CONFIG.copy())
- return config
-
-
-def save_config():
- global config
- config = validate_config_structure(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
-
-
def ensure_stable_python_runtime():
if sys.version_info < (3, 14) or os.environ.get("DPE_REEXEC_DONE") == "1":
return
@@ -375,6 +319,10 @@ class _CompatibilityModule(types.ModuleType):
setattr(_mail_service, name, int(value))
self.__dict__.pop(name, None)
return
+ elif name in {"browser", "page", "browser_proxy_bridge", "browser_started_with_proxy", "cf_clearance"}:
+ setattr(_registration_browser, name, value)
+ self.__dict__.pop(name, None)
+ return
super().__setattr__(name, value)
@@ -994,6 +942,12 @@ class GrokRegisterGUI:
def should_stop(self):
return self.stop_requested or not self.is_running
+ def _reset_batch_counters(self):
+ self.success_count = 0
+ self.fail_count = 0
+ self.registered_unsaved_count = 0
+ self.postprocess_warning_count = 0
+
def start_registration(self):
if self.is_running:
self.log("[!] 当前已有任务在运行")
@@ -1034,8 +988,7 @@ class GrokRegisterGUI:
self.log(f"[!] 配置无效或保存失败: {exc}")
return
self.stop_requested = False
- self.success_count = 0
- self.fail_count = 0
+ self._reset_batch_counters()
self.results = []
now = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
self.accounts_output_file = os.path.join(
diff --git a/mail_service.py b/mail_service.py
index a24fcb4..ba4df92 100644
--- a/mail_service.py
+++ b/mail_service.py
@@ -27,6 +27,26 @@ def bind_runtime(namespace):
globals()[name] = value
+def normalize_mail_body(*sources):
+ """Return normalized text from provider payloads with string/list HTML support."""
+ parts = []
+ for source in sources:
+ if not isinstance(source, dict):
+ continue
+ for key in ("text", "raw", "content", "intro", "body", "snippet"):
+ value = source.get(key)
+ values = value if isinstance(value, (list, tuple)) else [value]
+ for item in values:
+ if isinstance(item, str) and item.strip():
+ parts.append(item)
+ html_value = source.get("html")
+ html_items = html_value if isinstance(html_value, (list, tuple)) else [html_value]
+ for item in html_items:
+ if isinstance(item, str) and item.strip():
+ parts.append(re.sub(r"<[^>]+>", " ", item))
+ return "\n".join(parts)
+
+
def _pick_list_payload(data):
if isinstance(data, list):
return data
@@ -209,34 +229,19 @@ def cloudflare_get_oai_code(
address_matched = email.lower() in recipients
elif msg_addr:
address_matched = msg_addr == email.lower()
- if not address_matched and log_callback:
- log_callback(f"[Debug] 跳过疑似非目标邮件 id={msg_id} address={msg_addr} to={recipients}")
+ if not address_matched:
+ if log_callback:
+ log_callback(f"[Debug] 跳过疑似非目标邮件 id={msg_id} address={msg_addr} to={recipients}")
continue
- parts = []
# 先直接从列表项取内容，避免 detail 接口差异导致漏码
- for field in ("text", "raw", "content", "intro", "body", "snippet"):
- value = msg.get(field)
- if isinstance(value, str) and value.strip():
- parts.append(value)
- html_list = msg.get("html") or []
- if isinstance(html_list, str):
- html_list = [html_list]
- for h in html_list:
- parts.append(re.sub(r"<[^>]+>", " ", h))
subject = str(msg.get("subject", "") or "")
- combined = "\n".join(parts)
+ combined = normalize_mail_body(msg)
# 再尝试 detail 接口补全内容
try:
detail = cloudflare_get_message_detail(api_base, dev_token, msg_id)
- for field in ("text", "raw", "content", "intro", "body", "snippet"):
- value = detail.get(field)
- if isinstance(value, str) and value.strip():
- combined += "\n" + value
- html_list2 = detail.get("html") or []
- if isinstance(html_list2, str):
- html_list2 = [html_list2]
- for h in html_list2:
- combined += "\n" + re.sub(r"<[^>]+>", " ", h)
+ detail_body = normalize_mail_body(detail)
+ if detail_body:
+ combined += "\n" + detail_body
if not subject:
subject = str(detail.get("subject", "") or "")
except Exception as exc:
@@ -391,18 +396,11 @@ def cloudmail_get_oai_code(
).strip().lower()
if target_address and target_address != email.lower():
continue
- parts = []
code_value = str(msg.get("code", "") or "").strip()
+ combined = normalize_mail_body(msg)
if code_value:
- parts.append(f"verification code: {code_value}")
- for field in ("text", "content", "html", "body", "snippet"):
- value = msg.get(field)
- values = value if isinstance(value, list) else [value]
- for item in values:
- if isinstance(item, str) and item.strip():
- parts.append(re.sub(r"<[^>]+>", " ", item))
+ combined = f"verification code: {code_value}\n{combined}"
subject = str(msg.get("subject", "") or "")
- combined = "\n".join(parts)
if log_callback:
log_callback(f"[Debug] Cloud Mail 收到邮件: {subject}")
code = extract_verification_code(combined, subject)
@@ -475,14 +473,7 @@ def duckmail_get_oai_code(
if log_callback:
log_callback(f"[Debug] 获取邮件详情失败: {exc}")
continue
- parts = []
- text_body = detail.get("text") or ""
- if text_body:
- parts.append(text_body)
- html_list = detail.get("html") or []
- for h in html_list:
- parts.append(re.sub(r"<[^>]+>", " ", h))
- combined = "\n".join(parts)
+ combined = normalize_mail_body(detail)
subject = detail.get("subject", "")
if log_callback:
log_callback(f"[Debug] 收到邮件: {subject}")
@@ -823,14 +814,7 @@ def yyds_get_oai_code(
if log_callback:
log_callback(f"[Debug] YYDS 获取邮件详情失败: {exc}")
continue
- parts = []
- text_body = detail.get("text") or ""
- if text_body:
- parts.append(text_body)
- html_list = detail.get("html") or []
- for h in html_list:
- parts.append(re.sub(r"<[^>]+>", " ", h))
- combined = "\n".join(parts)
+ combined = normalize_mail_body(detail)
subject = detail.get("subject", "")
if log_callback:
log_callback(f"[Debug] YYDS 收到邮件: {subject}")
diff --git a/registration_browser.py b/registration_browser.py
index a8e8342..42f62f4 100644
--- a/registration_browser.py
+++ b/registration_browser.py
@@ -15,7 +15,7 @@ page = None
browser_proxy_bridge = None
browser_started_with_proxy = False
cf_clearance = ""
-SIGNUP_URL = "https://accounts.x.ai/sign-up"
+SIGNUP_URL = "https://accounts.x.ai/sign-up?redirect=grok-com"
_OWN_NAMES = {'is_cloudflare_block_response', 'response_preview', 'start_browser', 'enable_nsfw_for_token', 'stop_browser_proxy_bridge', 'set_tos_accepted', 'fill_email_and_submit', 'getTurnstileToken', 'set_birth_date', 'generate_random_birthdate', 'fill_profile_and_submit', 'click_email_signup_button', 'wait_for_sso_cookie', 'fill_code_and_submit', 'build_profile', 'cleanup_runtime_memory', 'open_signup_page', 'stop_browser', 'encode_grpc_nsfw_settings', 'restart_browser', 'has_profile_form', 'update_nsfw_settings', 'refresh_active_page'}
diff --git a/tests/test_post_modularization_regressions.py b/tests/test_post_modularization_regressions.py
new file mode 100644
index 0000000..ff8e6e9
--- /dev/null
+++ b/tests/test_post_modularization_regressions.py
@@ -0,0 +1,107 @@
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+import app_config
+import browser_runtime
+import cpa_export
+import grok_register_ttk as app
+import mail_service
+import registration_browser
+from cpa_xai import browser_session
+
+
+class PostModularizationRegressionTests(unittest.TestCase):
+ def test_signup_url_preserves_redirect(self):
+ self.assertEqual(
+ registration_browser.SIGNUP_URL,
+ "https://accounts.x.ai/sign-up?redirect=grok-com",
+ )
+
+ def test_config_identity_survives_load(self):
+ original_path = app_config.CONFIG_FILE
+ try:
+ with tempfile.TemporaryDirectory() as directory:
+ config_path = Path(directory) / "config.json"
+ payload = dict(app_config.DEFAULT_CONFIG)
+ payload["register_count"] = 3
+ config_path.write_text(json.dumps(payload), encoding="utf-8")
+ app_config.CONFIG_FILE = str(config_path)
+ loaded = app.load_config()
+ self.assertIs(loaded, app_config.config)
+ self.assertIs(app.config, app_config.config)
+ self.assertEqual(app.config["register_count"], 3)
+ finally:
+ app_config.CONFIG_FILE = original_path
+
+ def test_legacy_runtime_state_assignments_are_forwarded(self):
+ sentinel = object()
+ original = registration_browser.page
+ try:
+ app.page = sentinel
+ self.assertIs(registration_browser.page, sentinel)
+ self.assertIs(app.page, sentinel)
+ finally:
+ app.page = original
+
+ def test_gui_reset_clears_all_batch_counters(self):
+ gui = app.GrokRegisterGUI.__new__(app.GrokRegisterGUI)
+ gui.success_count = 1
+ gui.fail_count = 2
+ gui.registered_unsaved_count = 3
+ gui.postprocess_warning_count = 4
+ gui._reset_batch_counters()
+ self.assertEqual(
+ (gui.success_count, gui.fail_count, gui.registered_unsaved_count, gui.postprocess_warning_count),
+ (0, 0, 0, 0),
+ )
+
+ def test_cpa_hotload_requirement_only_applies_when_export_enabled(self):
+ cfg = dict(app_config.DEFAULT_CONFIG)
+ cfg["cpa_copy_to_hotload"] = True
+ cfg["cpa_export_enabled"] = False
+ self.assertTrue(app_config.validate_run_requirements(cfg)["cpa_copy_to_hotload"])
+ cfg["cpa_export_enabled"] = True
+ with self.assertRaises(app_config.ConfigError):
+ app_config.validate_run_requirements(cfg)
+
+ def test_mail_body_normalizes_string_and_list_html(self):
+ text = mail_service.normalize_mail_body(
+ {"text": "plain", "html": "one"},
+ {"html": ["two"]},
+ )
+ self.assertIn("plain", text)
+ self.assertIn("one", text)
+ self.assertIn("two", text)
+
+ def test_cloudflare_skips_non_target_mail_without_logger(self):
+ message = {
+ "id": "1",
+ "to": [{"address": "other@example.com"}],
+ "subject": "ABC-123 xAI",
+ "text": "ABC-123",
+ }
+ with patch.object(mail_service, "get_cloudflare_api_base", return_value="https://mail.example"), patch.object(mail_service, "cloudflare_get_messages", return_value=[message]), patch.object(mail_service, "cloudflare_get_message_detail") as detail, patch.object(mail_service, "raise_if_cancelled", return_value=None), patch.object(mail_service, "sleep_with_cancel", return_value=None), patch.object(mail_service.time, "time", side_effect=[0, 0, 2, 2]):
+ with self.assertRaises(Exception):
+ mail_service.cloudflare_get_oai_code(
+ "token", "target@example.com", timeout=1, poll_interval=0, log_callback=None
+ )
+ detail.assert_not_called()
+
+ def test_cpa_browser_session_does_not_import_main_module(self):
+ source = Path(browser_session.__file__).read_text(encoding="utf-8")
+ self.assertNotIn("from grok_register_ttk", source)
+ self.assertIn("from browser_runtime import create_browser_options", source)
+
+ def test_browser_options_accept_explicit_extension_path(self):
+ self.assertIn("extension_path", browser_runtime.create_browser_options.__code__.co_varnames)
+
+ def test_cpa_export_annotations_are_python39_compatible(self):
+ annotation = cpa_export.CpaExportSettings.__annotations__["hotload_dir"]
+ self.assertNotIsInstance(annotation, str)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_registration_flow.py b/tests/test_registration_flow.py
index 5a7ff5a..3e804dd 100644
--- a/tests/test_registration_flow.py
+++ b/tests/test_registration_flow.py
@@ -155,43 +155,6 @@ class RegistrationFlowTests(unittest.TestCase):
self.assertEqual(batch.fail_count, 0)
self.assertEqual(batch.postprocess_warning_count, 1)
- def test_cleanup_failure_does_not_change_success_statistics(self):
- fake = FakeOps()
- ops = fake.operations()
- base_cleanup = ops.cleanup
- def cleanup(reason):
- if "已成功" in reason:
- raise RuntimeError("cleanup failed")
- base_cleanup(reason)
- ops.cleanup = cleanup
- batch = run_batch(2, self.callbacks(), lambda *args: None, ops, cleanup_interval=1)
- self.assertEqual((batch.success_count, batch.fail_count, batch.processed_count), (2, 0, 2))
-
- def test_cancel_during_next_account_wait_is_normal_cancellation(self):
- fake = FakeOps()
- ops = fake.operations()
- ops.sleep = lambda seconds: (_ for _ in ()).throw(Cancelled())
- batch = run_batch(2, self.callbacks(), lambda *args: None, ops)
- self.assertTrue(batch.cancelled)
- self.assertEqual(batch.processed_count, 1)
-
- def test_final_cleanup_does_not_mask_original_error(self):
- fake = FakeOps()
- ops = fake.operations()
- ops.start_browser = lambda: (_ for _ in ()).throw(ValueError("original"))
- ops.cleanup = lambda reason: (_ for _ in ()).throw(RuntimeError("cleanup"))
- with self.assertRaisesRegex(ValueError, "original"):
- run_batch(1, self.callbacks(), lambda *args: None, ops)
-
- def test_optional_postprocessing_exceptions_become_warning(self):
- fake = FakeOps()
- ops = fake.operations()
- ops.add_tokens = lambda sso, email: (_ for _ in ()).throw(RuntimeError("pool"))
- ops.export_cpa = lambda email, password, sso: (_ for _ in ()).throw(RuntimeError("cpa"))
- batch = run_batch(1, self.callbacks(), lambda *args: None, ops)
- self.assertEqual(batch.success_count, 1)
- self.assertEqual(batch.postprocess_warning_count, 1)
-
if __name__ == "__main__":
unittest.main()
diff --git a/tools/apply_post_modularization_fixes.py b/tools/apply_post_modularization_fixes.py
deleted file mode 100644
index 1c54551..0000000
--- a/tools/apply_post_modularization_fixes.py
+++ /dev/null
@@ -1,511 +0,0 @@
-#!/usr/bin/env python3
-from pathlib import Path
-
-ROOT = Path(__file__).resolve().parents[1]
-
-
-def read(path):
- return (ROOT / path).read_text(encoding="utf-8")
-
-
-def write(path, text):
- (ROOT / path).write_text(text, encoding="utf-8")
-
-
-def replace_once(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)
-
-
-# grok_register_ttk.py: single config implementation, full state compatibility,
-# and complete GUI batch counter reset.
-path = "grok_register_ttk.py"
-text = read(path)
-old_config_impl = '''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)
- config = validate_config_structure(loaded)
- except ConfigError:
- raise
- except Exception as exc:
- raise ConfigError(f"配置文件解析失败: {CONFIG_FILE}: {exc}") from exc
- else:
- config = validate_config_structure(DEFAULT_CONFIG.copy())
- return config
-
-
-def save_config():
- global config
- config = validate_config_structure(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
-
-
-'''
-text = replace_once(text, old_config_impl, "", "duplicate config implementation")
-old_setattr = ''' elif name in {"_cf_domain_index", "_cloudmail_domain_index"}:
- setattr(_mail_service, name, int(value))
- self.__dict__.pop(name, None)
- return
- super().__setattr__(name, value)
-'''
-new_setattr = ''' elif name in {"_cf_domain_index", "_cloudmail_domain_index"}:
- setattr(_mail_service, name, int(value))
- self.__dict__.pop(name, None)
- return
- elif name in {"browser", "page", "browser_proxy_bridge", "browser_started_with_proxy", "cf_clearance"}:
- setattr(_registration_browser, name, value)
- self.__dict__.pop(name, None)
- return
- super().__setattr__(name, value)
-'''
-text = replace_once(text, old_setattr, new_setattr, "compatibility state writes")
-old_reset = ''' self.stop_requested = False
- self.success_count = 0
- self.fail_count = 0
- self.results = []
-'''
-new_reset = ''' self.stop_requested = False
- self._reset_batch_counters()
- self.results = []
-'''
-text = replace_once(text, old_reset, new_reset, "GUI counter reset call")
-anchor = ''' def should_stop(self):
- return self.stop_requested or not self.is_running
-
- def start_registration(self):
-'''
-replacement = ''' def should_stop(self):
- return self.stop_requested or not self.is_running
-
- def _reset_batch_counters(self):
- self.success_count = 0
- self.fail_count = 0
- self.registered_unsaved_count = 0
- self.postprocess_warning_count = 0
-
- def start_registration(self):
-'''
-text = replace_once(text, anchor, replacement, "GUI reset helper")
-write(path, text)
-
-
-# Restore exact registration URL.
-path = "registration_browser.py"
-text = read(path)
-text = replace_once(
- text,
- 'SIGNUP_URL = "https://accounts.x.ai/sign-up"',
- 'SIGNUP_URL = "https://accounts.x.ai/sign-up?redirect=grok-com"',
- "registration URL",
-)
-write(path, text)
-
-
-# Python 3.9-compatible typing.
-path = "cpa_export.py"
-text = read(path)
-text = replace_once(text, "from pathlib import Path\n", "from pathlib import Path\nfrom typing import Optional\n", "Optional import")
-text = replace_once(text, " hotload_dir: Path | None\n", " hotload_dir: Optional[Path]\n", "Python 3.9 union")
-write(path, text)
-
-
-# Make browser options explicitly reusable by CPA without importing the main app.
-path = "browser_runtime.py"
-text = read(path)
-old_options = '''def create_browser_options(browser_proxy=""):
- options = ChromiumOptions()
- options.auto_port()
- options.set_timeouts(base=1)
- apply_browser_proxy_option(options, browser_proxy)
- if _extension_path and os.path.exists(_extension_path):
- options.add_extension(_extension_path)
- return options
-'''
-new_options = '''def create_browser_options(browser_proxy="", extension_path=None):
- options = ChromiumOptions()
- options.auto_port()
- options.set_timeouts(base=1)
- apply_browser_proxy_option(options, browser_proxy)
- effective_extension = _extension_path if extension_path is None else str(extension_path or "")
- if effective_extension and os.path.exists(effective_extension):
- options.add_extension(effective_extension)
- return options
-'''
-text = replace_once(text, old_options, new_options, "browser options factory")
-write(path, text)
-
-path = "cpa_xai/browser_session.py"
-text = read(path)
-text = text.replace("import sys\n", "")
-old_probe = ''' options = None
- package_root = Path(__file__).resolve().parents[1]
- try:
- register_file = package_root / "grok_register_ttk.py"
- if register_file.is_file():
- register_dir = str(package_root)
- if register_dir not in sys.path:
- sys.path.insert(0, register_dir)
- try:
- from grok_register_ttk import create_browser_options # type: ignore
-
- options = create_browser_options()
- logger("using register create_browser_options (turnstilePatch)")
- except Exception as exc: # noqa: BLE001
- logger("register browser options unavailable: %s" % exc)
- options = None
- except Exception as exc: # noqa: BLE001
- logger("register options probe failed: %s" % exc)
- options = None
-'''
-new_probe = ''' options = None
- package_root = Path(__file__).resolve().parents[1]
- try:
- from browser_runtime import create_browser_options
-
- options = create_browser_options(
- extension_path=package_root / "turnstilePatch"
- )
- logger("using shared browser_runtime.create_browser_options")
- except Exception as exc: # noqa: BLE001
- logger("shared browser options unavailable: %s" % exc)
- options = None
-'''
-text = replace_once(text, old_probe, new_probe, "CPA reverse import removal")
-write(path, text)
-
-
-# Tighten CPA hotload validation.
-path = "app_config.py"
-text = read(path)
-text = replace_once(
- text,
- ' if cfg["cpa_copy_to_hotload"] and not cfg["cpa_hotload_dir"]:\n',
- ' if cfg["cpa_export_enabled"] and cfg["cpa_copy_to_hotload"] and not cfg["cpa_hotload_dir"]:\n',
- "CPA hotload validation",
-)
-write(path, text)
-
-
-# Centralize mail body normalization and fix address filtering independent of logging.
-path = "mail_service.py"
-text = read(path)
-insert_anchor = '''def _pick_list_payload(data):
-'''
-helper = '''def normalize_mail_body(*sources):
- """Return normalized text from provider payloads with string/list HTML support."""
- parts = []
- for source in sources:
- if not isinstance(source, dict):
- continue
- for key in ("text", "raw", "content", "intro", "body", "snippet"):
- value = source.get(key)
- values = value if isinstance(value, (list, tuple)) else [value]
- for item in values:
- if isinstance(item, str) and item.strip():
- parts.append(item)
- html_value = source.get("html")
- html_items = html_value if isinstance(html_value, (list, tuple)) else [html_value]
- for item in html_items:
- if isinstance(item, str) and item.strip():
- parts.append(re.sub(r"<[^>]+>", " ", item))
- return "\\n".join(parts)
-
-
-'''
-text = replace_once(text, insert_anchor, helper + insert_anchor, "mail body normalizer")
-old_filter = ''' if not address_matched and log_callback:
- log_callback(f"[Debug] 跳过疑似非目标邮件 id={msg_id} address={msg_addr} to={recipients}")
- continue
-'''
-new_filter = ''' if not address_matched:
- if log_callback:
- log_callback(f"[Debug] 跳过疑似非目标邮件 id={msg_id} address={msg_addr} to={recipients}")
- continue
-'''
-text = replace_once(text, old_filter, new_filter, "Cloudflare target filtering")
-old_cf_body = ''' parts = []
- # 先直接从列表项取内容，避免 detail 接口差异导致漏码
- for field in ("text", "raw", "content", "intro", "body", "snippet"):
- value = msg.get(field)
- if isinstance(value, str) and value.strip():
- parts.append(value)
- html_list = msg.get("html") or []
- if isinstance(html_list, str):
- html_list = [html_list]
- for h in html_list:
- parts.append(re.sub(r"<[^>]+>", " ", h))
- subject = str(msg.get("subject", "") or "")
- combined = "\\n".join(parts)
-'''
-new_cf_body = ''' # 先直接从列表项取内容，避免 detail 接口差异导致漏码
- subject = str(msg.get("subject", "") or "")
- combined = normalize_mail_body(msg)
-'''
-text = replace_once(text, old_cf_body, new_cf_body, "Cloudflare body normalization")
-old_cf_detail = ''' for field in ("text", "raw", "content", "intro", "body", "snippet"):
- value = detail.get(field)
- if isinstance(value, str) and value.strip():
- combined += "\\n" + value
- html_list2 = detail.get("html") or []
- if isinstance(html_list2, str):
- html_list2 = [html_list2]
- for h in html_list2:
- combined += "\\n" + re.sub(r"<[^>]+>", " ", h)
-'''
-new_cf_detail = ''' detail_body = normalize_mail_body(detail)
- if detail_body:
- combined += "\\n" + detail_body
-'''
-text = replace_once(text, old_cf_detail, new_cf_detail, "Cloudflare detail normalization")
-old_cloudmail_body = ''' parts = []
- code_value = str(msg.get("code", "") or "").strip()
- if code_value:
- parts.append(f"verification code: {code_value}")
- for field in ("text", "content", "html", "body", "snippet"):
- value = msg.get(field)
- values = value if isinstance(value, list) else [value]
- for item in values:
- if isinstance(item, str) and item.strip():
- parts.append(re.sub(r"<[^>]+>", " ", item))
- subject = str(msg.get("subject", "") or "")
- combined = "\\n".join(parts)
-'''
-new_cloudmail_body = ''' code_value = str(msg.get("code", "") or "").strip()
- combined = normalize_mail_body(msg)
- if code_value:
- combined = f"verification code: {code_value}\\n{combined}"
- subject = str(msg.get("subject", "") or "")
-'''
-text = replace_once(text, old_cloudmail_body, new_cloudmail_body, "Cloud Mail body normalization")
-old_duck_body = ''' parts = []
- text_body = detail.get("text") or ""
- if text_body:
- parts.append(text_body)
- html_list = detail.get("html") or []
- for h in html_list:
- parts.append(re.sub(r"<[^>]+>", " ", h))
- combined = "\\n".join(parts)
-'''
-text = replace_once(text, old_duck_body, ' combined = normalize_mail_body(detail)\n', "DuckMail body normalization")
-old_yyds_body = ''' parts = []
- text_body = detail.get("text") or ""
- if text_body:
- parts.append(text_body)
- html_list = detail.get("html") or []
- for h in html_list:
- parts.append(re.sub(r"<[^>]+>", " ", h))
- combined = "\\n".join(parts)
-'''
-text = replace_once(text, old_yyds_body, ' combined = normalize_mail_body(detail)\n', "YYDS body normalization")
-write(path, text)
-
-
-# Ignore pending recovery artifacts.
-path = ".gitignore"
-text = read(path)
-if "*.pending.jsonl\n" not in text:
- text += "\n*.pending.jsonl\n*.pending.jsonl.lock\n"
-write(path, text)
-
-
-# Remove duplicate/overlapping registration flow tests while retaining one test per semantic boundary.
-path = "tests/test_registration_flow.py"
-text = read(path)
-duplicate_block = ''' def test_cleanup_failure_does_not_change_success_statistics(self):
- fake = FakeOps()
- ops = fake.operations()
- base_cleanup = ops.cleanup
- def cleanup(reason):
- if "已成功" in reason:
- raise RuntimeError("cleanup failed")
- base_cleanup(reason)
- ops.cleanup = cleanup
- batch = run_batch(2, self.callbacks(), lambda *args: None, ops, cleanup_interval=1)
- self.assertEqual((batch.success_count, batch.fail_count, batch.processed_count), (2, 0, 2))
-
- def test_cancel_during_next_account_wait_is_normal_cancellation(self):
- fake = FakeOps()
- ops = fake.operations()
- ops.sleep = lambda seconds: (_ for _ in ()).throw(Cancelled())
- batch = run_batch(2, self.callbacks(), lambda *args: None, ops)
- self.assertTrue(batch.cancelled)
- self.assertEqual(batch.processed_count, 1)
-
- def test_final_cleanup_does_not_mask_original_error(self):
- fake = FakeOps()
- ops = fake.operations()
- ops.start_browser = lambda: (_ for _ in ()).throw(ValueError("original"))
- ops.cleanup = lambda reason: (_ for _ in ()).throw(RuntimeError("cleanup"))
- with self.assertRaisesRegex(ValueError, "original"):
- run_batch(1, self.callbacks(), lambda *args: None, ops)
-
- def test_optional_postprocessing_exceptions_become_warning(self):
- fake = FakeOps()
- ops = fake.operations()
- ops.add_tokens = lambda sso, email: (_ for _ in ()).throw(RuntimeError("pool"))
- ops.export_cpa = lambda email, password, sso: (_ for _ in ()).throw(RuntimeError("cpa"))
- batch = run_batch(1, self.callbacks(), lambda *args: None, ops)
- self.assertEqual(batch.success_count, 1)
- self.assertEqual(batch.postprocess_warning_count, 1)
-
-'''
-text = replace_once(text, duplicate_block, "", "duplicate flow tests")
-write(path, text)
-
-
-# Focused regression tests for the post-modularization contract.
-regression_tests = '''import json
-import tempfile
-import unittest
-from pathlib import Path
-from unittest.mock import patch
-
-import app_config
-import browser_runtime
-import cpa_export
-import grok_register_ttk as app
-import mail_service
-import registration_browser
-from cpa_xai import browser_session
-
-
-class PostModularizationRegressionTests(unittest.TestCase):
- def test_signup_url_preserves_redirect(self):
- self.assertEqual(
- registration_browser.SIGNUP_URL,
- "https://accounts.x.ai/sign-up?redirect=grok-com",
- )
-
- def test_config_identity_survives_load(self):
- original_path = app_config.CONFIG_FILE
- try:
- with tempfile.TemporaryDirectory() as directory:
- config_path = Path(directory) / "config.json"
- payload = dict(app_config.DEFAULT_CONFIG)
- payload["register_count"] = 3
- config_path.write_text(json.dumps(payload), encoding="utf-8")
- app_config.CONFIG_FILE = str(config_path)
- loaded = app.load_config()
- self.assertIs(loaded, app_config.config)
- self.assertIs(app.config, app_config.config)
- self.assertEqual(app.config["register_count"], 3)
- finally:
- app_config.CONFIG_FILE = original_path
-
- def test_legacy_runtime_state_assignments_are_forwarded(self):
- sentinel = object()
- original = registration_browser.page
- try:
- app.page = sentinel
- self.assertIs(registration_browser.page, sentinel)
- self.assertIs(app.page, sentinel)
- finally:
- app.page = original
-
- def test_gui_reset_clears_all_batch_counters(self):
- gui = app.GrokRegisterGUI.__new__(app.GrokRegisterGUI)
- gui.success_count = 1
- gui.fail_count = 2
- gui.registered_unsaved_count = 3
- gui.postprocess_warning_count = 4
- gui._reset_batch_counters()
- self.assertEqual(
- (gui.success_count, gui.fail_count, gui.registered_unsaved_count, gui.postprocess_warning_count),
- (0, 0, 0, 0),
- )
-
- def test_cpa_hotload_requirement_only_applies_when_export_enabled(self):
- cfg = dict(app_config.DEFAULT_CONFIG)
- cfg["cpa_copy_to_hotload"] = True
- cfg["cpa_export_enabled"] = False
- self.assertTrue(app_config.validate_run_requirements(cfg)["cpa_copy_to_hotload"])
- cfg["cpa_export_enabled"] = True
- with self.assertRaises(app_config.ConfigError):
- app_config.validate_run_requirements(cfg)
-
- def test_mail_body_normalizes_string_and_list_html(self):
- text = mail_service.normalize_mail_body(
- {"text": "plain", "html": "one"},
- {"html": ["two"]},
- )
- self.assertIn("plain", text)
- self.assertIn("one", text)
- self.assertIn("two", text)
-
- def test_cloudflare_skips_non_target_mail_without_logger(self):
- message = {
- "id": "1",
- "to": [{"address": "other@example.com"}],
- "subject": "ABC-123 xAI",
- "text": "ABC-123",
- }
- with patch.object(mail_service, "get_cloudflare_api_base", return_value="https://mail.example"), \
- patch.object(mail_service, "cloudflare_get_messages", return_value=[message]), \
- patch.object(mail_service, "cloudflare_get_message_detail") as detail, \
- patch.object(mail_service, "raise_if_cancelled", return_value=None), \
- patch.object(mail_service, "sleep_with_cancel", return_value=None), \
- patch.object(mail_service.time, "time", side_effect=[0, 0, 2, 2]):
- with self.assertRaises(Exception):
- mail_service.cloudflare_get_oai_code(
- "token", "target@example.com", timeout=1, poll_interval=0, log_callback=None
- )
- detail.assert_not_called()
-
- def test_cpa_browser_session_does_not_import_main_module(self):
- source = Path(browser_session.__file__).read_text(encoding="utf-8")
- self.assertNotIn("from grok_register_ttk", source)
- self.assertIn("from browser_runtime import create_browser_options", source)
-
- def test_browser_options_accept_explicit_extension_path(self):
- self.assertIn("extension_path", browser_runtime.create_browser_options.__code__.co_varnames)
-
- def test_cpa_export_annotations_are_python39_compatible(self):
- annotation = cpa_export.CpaExportSettings.__annotations__["hotload_dir"]
- self.assertNotIsInstance(annotation, str)
-
-
-if __name__ == "__main__":
- unittest.main()
-'''
-write("tests/test_post_modularization_regressions.py", regression_tests)
-
-print("post-modularization fixes applied")
diff --git a/tools/repair_post_modularization_patcher.py b/tools/repair_post_modularization_patcher.py
deleted file mode 100644
index ff6ff71..0000000
--- a/tools/repair_post_modularization_patcher.py
+++ /dev/null
@@ -1,83 +0,0 @@
-#!/usr/bin/env python3
-import ast
-from pathlib import Path
-
-path = Path(__file__).resolve().with_name("apply_post_modularization_fixes.py")
-text = path.read_text(encoding="utf-8")
-
-helper_anchor = '''def replace_once(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)
-'''
-
-# Raw text is intentional: the generated target must contain the two-character
-# escape sequence "\\n" inside its Python string literal, not an actual newline.
-helper_definition = r'''
-
-def replace_in_function(text, function_name, old, new, label):
- marker = f"def {function_name}("
- start = text.find(marker)
- if start < 0:
- raise RuntimeError(f"{label}: function {function_name!r} not found")
- next_function = text.find("\ndef ", start + len(marker))
- end = len(text) if next_function < 0 else next_function + 1
- function_text = text[start:end]
- count = function_text.count(old)
- if count != 1:
- raise RuntimeError(
- f"{label}: expected one match in {function_name}, got {count}"
- )
- return text[:start] + function_text.replace(old, new, 1) + text[end:]
-'''
-
-if "def replace_in_function(" not in text:
- if text.count(helper_anchor) != 1:
- raise RuntimeError("replace helper anchor is not unique")
- text = text.replace(helper_anchor, helper_anchor + helper_definition, 1)
-
-replacements = (
- (
- '''text = replace_once(text, old_duck_body, ' combined = normalize_mail_body(detail)\\n', "DuckMail body normalization")''',
- '''text = replace_in_function(
- text,
- "duckmail_get_oai_code",
- old_duck_body,
- ' combined = normalize_mail_body(detail)\\n',
- "DuckMail body normalization",
-)''',
- "DuckMail",
- ),
- (
- '''text = replace_once(text, old_yyds_body, ' combined = normalize_mail_body(detail)\\n', "YYDS body normalization")''',
- '''text = replace_in_function(
- text,
- "yyds_get_oai_code",
- old_yyds_body,
- ' combined = normalize_mail_body(detail)\\n',
- "YYDS body normalization",
-)''',
- "YYDS",
- ),
-)
-
-for old, new, label in replacements:
- if old in text:
- if text.count(old) != 1:
- raise RuntimeError(f"{label} replacement call is not unique")
- text = text.replace(old, new, 1)
- elif new not in text:
- raise RuntimeError(f"{label} replacement call anchor not found")
-
-for forbidden in (
- "replace_once(text, old_duck_body",
- "replace_once(text, old_yyds_body",
-):
- if forbidden in text:
- raise RuntimeError(f"ambiguous provider replacement remains: {forbidden}")
-
-# Catch all quoting, escaping, and generated-source syntax errors before writing.
-ast.parse(text, filename=str(path))
-path.write_text(text, encoding="utf-8")
-print("post-modularization patcher repaired and syntax-validated")