"""Registration runtime loops for zhuce6.""" from __future__ import annotations from collections import deque from dataclasses import replace from datetime import datetime import json import logging import os from pathlib import Path import random import sys import threading import time from typing import Any from urllib.parse import urlsplit, urlunsplit from core.registry import load_all from core.settings import AppSettings from dashboard.api import _count_cpa_files, _fetch_management_auth_files, _is_regular_free_account from ops.common import create_backend_client, get_management_key from ops.rotate_runtime import _maybe_reconcile_cpa_runtime from platforms.chatgpt.fingerprint import build_registration_provenance, infer_proxy_region from platforms.chatgpt.pool import is_warmup_pending_record, now_iso, update_token_record from core.chatgpt_flow_runner import run_chatgpt_register_once DEFAULT_ADD_PHONE_STOPLOSS_WINDOW = 10 DEFAULT_ADD_PHONE_STOPLOSS_THRESHOLD = 3 DEFAULT_ADD_PHONE_STOPLOSS_COOLDOWN_SECONDS = 300 DEFAULT_ADD_PHONE_STOPLOSS_MAX_SUCCESSES = 2 DEFAULT_WAIT_OTP_STOPLOSS_WINDOW = 6 DEFAULT_WAIT_OTP_STOPLOSS_THRESHOLD = 2 DEFAULT_WAIT_OTP_STOPLOSS_COOLDOWN_SECONDS = 300 DEFAULT_WAIT_OTP_LIVE_ABORT_THRESHOLD = 0 DEFAULT_WAIT_OTP_LIVE_ABORT_AGE_SECONDS = 90 DEFAULT_CFMAIL_FRESH_DOMAIN_ATTEMPT_BUDGET = 2 def _classify_token_file(*args, **kwargs): # type: ignore[no-untyped-def] from ops.scan import classify_token_file return classify_token_file(*args, **kwargs) def _compat_main_attr(name: str, default: object) -> object: main_module = sys.modules.get("main") if main_module is None: return default return getattr(main_module, name, default) class RegistrationLoop: """Multi-threaded continuous registration with fallback, target count, and logging.""" def __init__(self, settings: AppSettings) -> None: self.settings = settings self._threads: list[threading.Thread] = [] self._stop_event = threading.Event() self._target_reached = threading.Event() self._lock = threading.RLock() self._total_attempts = 0 self._total_success = 0 self._total_warmup_pending = 0 self._total_cpa_sync_success = 0 self._total_cpa_sync_failure = 0 self._total_failure = 0 self._last_error: str | None = None self._started_at: float | None = None self._failure_by_stage: dict[str, int] = {} self._failure_signals: dict[str, int] = {} self._recent_attempts: deque[dict[str, object]] = deque(maxlen=80) self._providers: list[str] = [] self._proxy_pool = None self._logger = self._setup_logger() self._cfmail_tracker = None self._cfmail_provisioner = None self._cfmail_manager: Any = None self._cfmail_rotation_lock = threading.Lock() self._cfmail_rotation_pause = threading.Event() self._cfmail_rotation_pause.set() self._cpa_management_key_cache: str | None | bool = False self._cfmail_add_phone_window = max( 1, int(str(os.getenv("ZHUCE6_CFMAIL_ADD_PHONE_WINDOW", DEFAULT_ADD_PHONE_STOPLOSS_WINDOW)).strip() or str(DEFAULT_ADD_PHONE_STOPLOSS_WINDOW)), ) self._cfmail_add_phone_threshold = max( 1, int( str( os.getenv( "ZHUCE6_CFMAIL_ADD_PHONE_THRESHOLD", DEFAULT_ADD_PHONE_STOPLOSS_THRESHOLD, ) ).strip() or str(DEFAULT_ADD_PHONE_STOPLOSS_THRESHOLD) ), ) self._cfmail_add_phone_cooldown_seconds = max( 1, int( str( os.getenv( "ZHUCE6_CFMAIL_ADD_PHONE_COOLDOWN_SECONDS", DEFAULT_ADD_PHONE_STOPLOSS_COOLDOWN_SECONDS, ) ).strip() or str(DEFAULT_ADD_PHONE_STOPLOSS_COOLDOWN_SECONDS) ), ) self._cfmail_add_phone_max_successes = max( 0, int( str( os.getenv( "ZHUCE6_CFMAIL_ADD_PHONE_MAX_SUCCESSES", DEFAULT_ADD_PHONE_STOPLOSS_MAX_SUCCESSES, ) ).strip() or str(DEFAULT_ADD_PHONE_STOPLOSS_MAX_SUCCESSES) ), ) self._cfmail_add_phone_events: dict[str, deque[dict[str, object]]] = {} self._cfmail_add_phone_state: dict[str, object] = { "active_domain": "", "in_cooldown": False, "cooldown_until": 0.0, "last_triggered_at": "", "last_rotation_attempted_at": "", "last_reason": "", "last_add_phone_failures": 0, "last_successes": 0, "last_window_size": 0, "last_logged_at": 0.0, } self._cfmail_wait_otp_window = max( 1, int(str(os.getenv("ZHUCE6_CFMAIL_WAIT_OTP_WINDOW", DEFAULT_WAIT_OTP_STOPLOSS_WINDOW)).strip() or str(DEFAULT_WAIT_OTP_STOPLOSS_WINDOW)), ) self._cfmail_wait_otp_threshold = max( 1, int( str( os.getenv( "ZHUCE6_CFMAIL_WAIT_OTP_THRESHOLD", DEFAULT_WAIT_OTP_STOPLOSS_THRESHOLD, ) ).strip() or str(DEFAULT_WAIT_OTP_STOPLOSS_THRESHOLD) ), ) self._cfmail_wait_otp_cooldown_seconds = max( 0, int( str( os.getenv( "ZHUCE6_CFMAIL_WAIT_OTP_COOLDOWN_SECONDS", DEFAULT_WAIT_OTP_STOPLOSS_COOLDOWN_SECONDS, ) ).strip() or str(DEFAULT_WAIT_OTP_STOPLOSS_COOLDOWN_SECONDS) ), ) self._cfmail_wait_otp_events: dict[str, deque[dict[str, object]]] = {} self._cfmail_wait_otp_state: dict[str, object] = { "active_domain": "", "in_cooldown": False, "cooldown_until": 0.0, "last_triggered_at": "", "last_rotation_attempted_at": "", "last_reason": "", "last_no_message_timeouts": 0, "last_window_size": 0, "last_logged_at": 0.0, } try: self._cfmail_wait_otp_live_threshold = max( 0, int( str( os.getenv( "ZHUCE6_CFMAIL_WAIT_OTP_LIVE_ABORT_THRESHOLD", DEFAULT_WAIT_OTP_LIVE_ABORT_THRESHOLD, ) ).strip() or str(DEFAULT_WAIT_OTP_LIVE_ABORT_THRESHOLD) ), ) except Exception: self._cfmail_wait_otp_live_threshold = DEFAULT_WAIT_OTP_LIVE_ABORT_THRESHOLD try: self._cfmail_wait_otp_live_age_seconds = max( 30, int( str( os.getenv( "ZHUCE6_CFMAIL_WAIT_OTP_LIVE_ABORT_AGE_SECONDS", DEFAULT_WAIT_OTP_LIVE_ABORT_AGE_SECONDS, ) ).strip() or str(DEFAULT_WAIT_OTP_LIVE_ABORT_AGE_SECONDS) ), ) except Exception: self._cfmail_wait_otp_live_age_seconds = DEFAULT_WAIT_OTP_LIVE_ABORT_AGE_SECONDS self._cfmail_wait_otp_live_lock = threading.RLock() self._cfmail_wait_otp_live_progress: dict[str, dict[str, dict[str, object]]] = {} self._cfmail_canary_state: dict[str, object] = { "active_domain": "", "pending": False, "owner_thread_id": 0, "attempt_started_at": 0.0, "last_ready_at": "", "last_ready_reason": "", "last_logged_at": 0.0, } try: self._cfmail_start_interval_seconds = max( 0, int(str(os.getenv("ZHUCE6_CFMAIL_START_INTERVAL_SECONDS", "8")).strip() or "8"), ) except Exception: self._cfmail_start_interval_seconds = 8 try: self._cfmail_max_inflight = max( 1, int(str(os.getenv("ZHUCE6_CFMAIL_MAX_INFLIGHT", "4")).strip() or "4"), ) except Exception: self._cfmail_max_inflight = 4 self._cfmail_flow_state: dict[str, object] = { "inflight_by_thread": {}, "last_started_by_domain": {}, "selected_profile_by_thread": {}, "last_logged_at": 0.0, } try: self._cfmail_active_domain_count = max( 1, int(str(os.getenv("ZHUCE6_CFMAIL_ACTIVE_DOMAIN_COUNT", "3")).strip() or "3"), ) except Exception: self._cfmail_active_domain_count = 3 try: self._cfmail_fresh_domain_attempt_budget = max( 0, int( str( os.getenv( "ZHUCE6_CFMAIL_FRESH_DOMAIN_ATTEMPT_BUDGET", DEFAULT_CFMAIL_FRESH_DOMAIN_ATTEMPT_BUDGET, ) ).strip() or str(DEFAULT_CFMAIL_FRESH_DOMAIN_ATTEMPT_BUDGET) ), ) except Exception: self._cfmail_fresh_domain_attempt_budget = DEFAULT_CFMAIL_FRESH_DOMAIN_ATTEMPT_BUDGET self._cfmail_fresh_domain_state: dict[str, object] = { "active_domain": "", "completed_attempts": 0, "mail_seen_attempts": 0, "successes": 0, "last_triggered_at": "", "last_rotation_attempted_at": "", "last_reason": "", } # Solution B: deferred retry queue for add_phone_gate accounts self._pending_token_queue: list[dict[str, Any]] = [] self._pending_token_lock = threading.Lock() try: raw_pending_retry_delay = int( str(os.getenv("ZHUCE6_PENDING_TOKEN_RETRY_DELAY_SECONDS", "600")).strip() or "600" ) except Exception: raw_pending_retry_delay = 600 # add_phone accounts are only useful if the deferred retry happens inside the # same short observation window as the registration loop. Cap the first retry # base delay so a large env value cannot postpone every retry past 5 minutes. self._pending_token_retry_delay_seconds = max(60, min(raw_pending_retry_delay, 60)) try: self._pending_token_max_retries = max( 1, int(str(os.getenv("ZHUCE6_PENDING_TOKEN_MAX_RETRIES", "3")).strip() or "3"), ) except Exception: self._pending_token_max_retries = 3 self._pending_token_total_enqueued = 0 self._pending_token_total_success = 0 self._pending_token_total_failed = 0 self._cfmail_replenish_thread: threading.Thread | None = None self._cfmail_replenish_reason = "" def _write_runtime_state(self) -> None: state_file = Path(self.settings.runtime_state_file) try: state_file.parent.mkdir(parents=True, exist_ok=True) payload = { "updated_at": datetime.now().isoformat(timespec="seconds"), "register_snapshot": self.snapshot(), "proxy_pool": self._proxy_pool_snapshot(), } tmp_file = state_file.with_name( f"{state_file.name}.{os.getpid()}.{threading.get_ident()}.tmp" ) tmp_file.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") tmp_file.replace(state_file) except Exception as exc: self._log(f"[zhuce6:register] runtime state write failed: {exc}") def _cfmail_canary_snapshot(self) -> dict[str, object]: return { "active_domain": "", "pending": False, "owner_thread_id": 0, "attempt_started_at": 0.0, "last_ready_at": "", "last_ready_reason": "disabled", } def _arm_cfmail_canary(self, domain: str, *, pending: bool = True) -> None: del domain, pending return def _mark_cfmail_canary_ready(self, domain: str, *, reason: str) -> None: del domain, reason return def _update_cfmail_canary_after_result(self, *, thread_id: int, result: dict[str, object]) -> None: del thread_id, result return def _wait_if_cfmail_canary_pending(self, thread_id: int, provider: str) -> bool: del thread_id, provider return False def _release_cfmail_flow_slot(self, thread_id: int) -> None: with self._lock: inflight_by_thread = self._cfmail_flow_state.setdefault("inflight_by_thread", {}) if isinstance(inflight_by_thread, dict): inflight_by_thread.pop(thread_id, None) selected_profile_by_thread = self._cfmail_flow_state.setdefault("selected_profile_by_thread", {}) if isinstance(selected_profile_by_thread, dict): selected_profile_by_thread.pop(thread_id, None) def _cfmail_active_domain_set(self) -> set[str]: domains = { str(item.get("domain") or "").strip().lower() for item in self._current_cfmail_active_accounts() if str(item.get("domain") or "").strip() } if domains: return domains current_domain = str(self._current_cfmail_active_domain() or "").strip().lower() return {current_domain} if current_domain else set() def _is_cfmail_active_domain(self, domain: str) -> bool: domain_key = str(domain or "").strip().lower() if not domain_key: return False return domain_key in self._cfmail_active_domain_set() def _schedule_cfmail_domain_pool_replenish(self, *, trigger_thread_id: int, reason: str) -> None: if self._cfmail_provisioner is None: return with self._lock: thread = self._cfmail_replenish_thread if thread is not None and thread.is_alive(): return self._cfmail_replenish_reason = str(reason or "").strip() thread = threading.Thread( target=self._cfmail_domain_pool_replenish_worker, kwargs={ "trigger_thread_id": trigger_thread_id, "reason": self._cfmail_replenish_reason, }, daemon=True, name="zhuce6-cfmail-replenish", ) self._cfmail_replenish_thread = thread thread.start() def _ensure_cfmail_domain_pool_target(self, *, trigger_thread_id: int, reason: str) -> None: if self._cfmail_provisioner is None: return if len(self._current_cfmail_active_accounts()) >= self._cfmail_active_domain_count: return self._schedule_cfmail_domain_pool_replenish( trigger_thread_id=trigger_thread_id, reason=reason, ) def _cfmail_domain_pool_replenish_worker(self, *, trigger_thread_id: int, reason: str) -> None: provisioner = self._cfmail_provisioner if provisioner is None: return while not self._stop_event.is_set(): active_accounts = self._current_cfmail_active_accounts() if len(active_accounts) >= self._cfmail_active_domain_count: return result = provisioner.provision_additional_domain() if not result.success: self._log( f"[zhuce6:register] [thread-{trigger_thread_id}] [cfmail] replenish failed after {reason}: " f"{result.error}" ) if self._stop_event.wait(5.0): return continue self._reload_cfmail_manager_after_rotation() if self._cfmail_tracker is not None: try: self._cfmail_tracker.mark_rotation_completed("", result.new_domain) except Exception: pass self._log( f"[zhuce6:register] [thread-{trigger_thread_id}] [cfmail] replenished domain pool " f"after {reason}: +{result.new_domain}" ) def _clear_cfmail_domain_state(self, domain: str) -> None: domain_key = str(domain or "").strip().lower() if not domain_key: return with self._lock: self._cfmail_add_phone_events.pop(domain_key, None) if str(self._cfmail_add_phone_state.get("active_domain") or "").strip().lower() == domain_key: self._cfmail_add_phone_state = { "active_domain": "", "in_cooldown": False, "cooldown_until": 0.0, "last_triggered_at": "", "last_rotation_attempted_at": "", "last_reason": "", "last_add_phone_failures": 0, "last_successes": 0, "last_window_size": 0, "last_logged_at": 0.0, } self._cfmail_wait_otp_events.pop(domain_key, None) if str(self._cfmail_wait_otp_state.get("active_domain") or "").strip().lower() == domain_key: self._cfmail_wait_otp_state = { "active_domain": "", "in_cooldown": False, "cooldown_until": 0.0, "last_triggered_at": "", "last_rotation_attempted_at": "", "last_reason": "", "last_no_message_timeouts": 0, "last_successes": 0, "last_message_seen": 0, "last_window_size": 0, "last_logged_at": 0.0, } self._cfmail_wait_otp_live_progress.pop(domain_key, None) if str(self._cfmail_fresh_domain_state.get("active_domain") or "").strip().lower() == domain_key: self._cfmail_fresh_domain_state = { "active_domain": "", "completed_attempts": 0, "mail_seen_attempts": 0, "successes": 0, "last_triggered_at": "", "last_rotation_attempted_at": "", "last_reason": "", } def _replace_cfmail_domain( self, *, thread_id: int, domain: str, reason_label: str, ) -> bool: domain_key = str(domain or "").strip().lower() if not domain_key or self._cfmail_provisioner is None: return False active_accounts = self._current_cfmail_active_accounts() if not active_accounts: active_accounts = [{"name": "", "domain": domain_key}] if not any(item["domain"] == domain_key for item in active_accounts): self._clear_cfmail_domain_state(domain_key) return False if not self._cfmail_rotation_lock.acquire(blocking=False): return False self._cfmail_rotation_pause.clear() try: if self._cfmail_tracker is not None: self._cfmail_tracker.mark_rotation_started(domain_key, reason_label) if len(active_accounts) <= 1: provision_result = self._cfmail_provisioner.rotate_active_domain() if not provision_result.success: if self._cfmail_tracker is not None: self._cfmail_tracker.mark_rotation_failed(domain_key, provision_result.error) self._log( f"[zhuce6:register] [thread-{thread_id}] [cfmail] {reason_label} rotation failed: " f"{provision_result.error}" ) return False self._reload_cfmail_manager_after_rotation() self._clear_cfmail_domain_state(provision_result.old_domain or domain_key) self._reset_cfmail_add_phone_stoploss(provision_result.new_domain) self._reset_cfmail_wait_otp_stoploss(provision_result.new_domain) self._reset_cfmail_fresh_domain_budget(provision_result.new_domain) if self._cfmail_tracker is not None: self._cfmail_tracker.mark_rotation_completed( provision_result.old_domain, provision_result.new_domain, ) self._log( f"[zhuce6:register] [thread-{thread_id}] [cfmail] {reason_label} rotation completed: " f"{provision_result.old_domain} -> {provision_result.new_domain}" ) return True retire_result = self._cfmail_provisioner.retire_domain(domain_key) if not retire_result.success: if self._cfmail_tracker is not None: self._cfmail_tracker.mark_rotation_failed(domain_key, retire_result.error) self._log( f"[zhuce6:register] [thread-{thread_id}] [cfmail] {reason_label} retire failed: " f"{retire_result.error}" ) return False self._reload_cfmail_manager_after_rotation() self._clear_cfmail_domain_state(domain_key) self._log( f"[zhuce6:register] [thread-{thread_id}] [cfmail] retired domain {domain_key} " f"because {reason_label}; scheduling replenish" ) self._schedule_cfmail_domain_pool_replenish( trigger_thread_id=thread_id, reason=reason_label, ) return True finally: self._cfmail_rotation_pause.set() self._cfmail_rotation_lock.release() def _wait_if_cfmail_flow_throttled(self, thread_id: int, provider: str) -> bool: if provider != "cfmail": return False self._ensure_cfmail_domain_pool_target( trigger_thread_id=thread_id, reason="usable domain pool below target", ) active_accounts = self._current_cfmail_active_accounts() if not active_accounts: return False with self._lock: inflight_by_thread = self._cfmail_flow_state.setdefault("inflight_by_thread", {}) if not isinstance(inflight_by_thread, dict): inflight_by_thread = {} self._cfmail_flow_state["inflight_by_thread"] = inflight_by_thread last_started_by_domain = self._cfmail_flow_state.setdefault("last_started_by_domain", {}) if not isinstance(last_started_by_domain, dict): last_started_by_domain = {} self._cfmail_flow_state["last_started_by_domain"] = last_started_by_domain selected_profile_by_thread = self._cfmail_flow_state.setdefault("selected_profile_by_thread", {}) if not isinstance(selected_profile_by_thread, dict): selected_profile_by_thread = {} self._cfmail_flow_state["selected_profile_by_thread"] = selected_profile_by_thread tracked = inflight_by_thread.get(thread_id) if isinstance(tracked, dict): tracked_domain = str(tracked.get("domain") or "").strip().lower() tracked_profile = str(tracked.get("profile_name") or "").strip() if tracked_domain and tracked_profile: selected_profile_by_thread[thread_id] = tracked_profile return False now = time.time() candidates: list[tuple[int, float, str, str]] = [] min_wait_seconds = 2.0 wait_domain = "" wait_reason = "inflight_limit" for account in active_accounts: domain = account["domain"] profile_name = account["name"] active_inflight = sum( 1 for value in inflight_by_thread.values() if isinstance(value, dict) and str(value.get("domain") or "").strip().lower() == domain ) if active_inflight >= self._cfmail_max_inflight: wait_domain = wait_domain or domain continue last_started_at = float(last_started_by_domain.get(domain) or 0.0) remaining = 0.0 if self._cfmail_start_interval_seconds > 0 and last_started_at > 0.0: remaining = self._cfmail_start_interval_seconds - (now - last_started_at) if remaining > 0.0: wait_domain = wait_domain or domain wait_reason = "start_interval" min_wait_seconds = min(min_wait_seconds, min(2.0, max(0.5, remaining))) continue candidates.append((active_inflight, last_started_at, profile_name, domain)) if candidates: candidates.sort(key=lambda item: (item[0], item[1], item[3], item[2])) _active_inflight, _last_started_at, profile_name, domain = candidates[0] inflight_by_thread[thread_id] = { "domain": domain, "profile_name": profile_name, "started_at": now, } last_started_by_domain[domain] = now selected_profile_by_thread[thread_id] = profile_name return False last_logged_at = float(self._cfmail_flow_state.get("last_logged_at") or 0.0) should_log = now - last_logged_at >= 15.0 if should_log: self._cfmail_flow_state["last_logged_at"] = now if should_log: if wait_reason == "inflight_limit": self._log( f"[zhuce6:register] [thread-{thread_id}] [cfmail] flow throttle for {wait_domain or '-'}: " f"inflight_limit reached on all active domains" ) else: self._log( f"[zhuce6:register] [thread-{thread_id}] [cfmail] flow throttle for {wait_domain or '-'}: " f"start_interval_remaining={min_wait_seconds:.1f}s" ) self._stop_event.wait(min_wait_seconds) return True def _proxy_pool_snapshot(self) -> dict[str, object]: pool = self._proxy_pool nodes: list[dict[str, object]] = [] snapshot_error: str | None = None if pool is not None: try: snapshot = pool.snapshot() except Exception as exc: snapshot_error = str(exc) else: if isinstance(snapshot, list): nodes = [item for item in snapshot if isinstance(item, dict)] return { "configured": bool(self.settings.proxy_pool_configured or pool is not None), "enabled": pool is not None, "snapshot_error": snapshot_error, "node_count": len(nodes), "in_use_count": sum(1 for item in nodes if item.get("in_use")), "disabled_count": sum(1 for item in nodes if item.get("disabled")), "nodes": nodes, } def _setup_logger(self) -> Any: import logging from logging.handlers import RotatingFileHandler logger = logging.getLogger("zhuce6.register") logger.setLevel(logging.INFO) logger.propagate = False if not logger.handlers: console = logging.StreamHandler() console.setFormatter(logging.Formatter("%(message)s")) logger.addHandler(console) if self.settings.register_log_file: fh = RotatingFileHandler( self.settings.register_log_file, maxBytes=2 * 1024 * 1024, # 2MB backupCount=5, encoding="utf-8", ) fh.setFormatter(logging.Formatter("%(asctime)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S")) logger.addHandler(fh) return logger def _log(self, msg: str) -> None: self._logger.info(msg) def _record_attempt( self, *, success: bool, stage: str, error_message: str, metadata: dict[str, object] | None = None, proxy_key: str = "", email: str = "", ) -> None: meta = metadata if isinstance(metadata, dict) else {} stage_key = str(stage or "?").strip() or "?" signal = self._classify_failure_signal(stage=stage_key, metadata=meta) timestamp = datetime.now().isoformat(timespec="seconds") event = { "timestamp": timestamp, "success": success, "stage": stage_key if not success else "completed", "signal": signal, "error_message": str(error_message or "").strip(), "email_domain": str(meta.get("email_domain") or "").strip(), "post_create_gate": str(meta.get("post_create_gate") or "").strip(), "create_account_error_code": str(meta.get("create_account_error_code") or "").strip(), "signup_error_code": str(meta.get("signup_error_code") or "").strip(), "signup_http_status": meta.get("signup_http_status"), "mailbox_error_kind": str(meta.get("mailbox_error_kind") or "").strip(), "mailbox_error_stage": str(meta.get("mailbox_error_stage") or "").strip(), "proxy_key": proxy_key, "email": email, } self._recent_attempts.append(event) if success or stage_key == "warmup_pending": return self._failure_by_stage[stage_key] = self._failure_by_stage.get(stage_key, 0) + 1 if signal: self._failure_signals[signal] = self._failure_signals.get(signal, 0) + 1 def _classify_failure_signal(self, *, stage: str, metadata: dict[str, object]) -> str: code = str(metadata.get("create_account_error_code") or "").strip().lower() signup_code = str(metadata.get("signup_error_code") or "").strip().lower() post_gate = str(metadata.get("post_create_gate") or "").strip().lower() if stage == "cpa_sync": return "cpa_sync_failed" if stage == "signup" and signup_code: return signup_code if stage == "add_phone_gate" or post_gate == "add_phone": return "add_phone_gate" if stage == "create_account" and code == "user_already_exists": return "mailbox_reused" if stage == "create_account" and code in {"registration_disallowed", "unsupported_email"}: return code if stage == "mailbox": provider = str(metadata.get("mail_provider") or "").strip().lower() if provider == "cfmail": mailbox_error_kind = str(metadata.get("mailbox_error_kind") or "").strip().lower() mailbox_error_stage = str(metadata.get("mailbox_error_stage") or "").strip().lower() if mailbox_error_stage == "create_email": mailbox_error_stage = "create" elif mailbox_error_stage == "fetch_email": mailbox_error_stage = "fetch" if mailbox_error_kind == "transport_error": suffix = mailbox_error_stage or "backend" return f"mailbox_{suffix}_transport_error" if mailbox_error_kind == "provider_error": suffix = mailbox_error_stage or "backend" return f"mailbox_{suffix}_provider_error" return "mailbox_backend_failure" return "mailbox_failure" return "" def _recent_failure_hotspots(self, limit: int = 5) -> list[dict[str, object]]: return self._recent_failure_hotspots_from_attempts(self._recent_attempts, limit=limit) def _recent_failure_hotspots_from_attempts( self, attempts: list[dict[str, object]] | deque[dict[str, object]], *, limit: int = 5, ) -> list[dict[str, object]]: counts: dict[tuple[str, str], int] = {} for item in attempts: if item.get("success"): continue stage = str(item.get("stage") or "?").strip() or "?" signal = str(item.get("signal") or "").strip() key = (signal or stage, stage) counts[key] = counts.get(key, 0) + 1 ordered = sorted(counts.items(), key=lambda kv: (-kv[1], kv[0][0], kv[0][1])) return [ {"key": key, "stage": stage, "count": count} for (key, stage), count in ordered[:limit] ] def _failure_counts_from_attempts( self, attempts: list[dict[str, object]] | deque[dict[str, object]], ) -> tuple[dict[str, int], dict[str, int]]: failure_by_stage: dict[str, int] = {} failure_signals: dict[str, int] = {} for item in attempts: if item.get("success"): continue stage = str(item.get("stage") or "?").strip() or "?" failure_by_stage[stage] = failure_by_stage.get(stage, 0) + 1 signal = str(item.get("signal") or "").strip() if signal: failure_signals[signal] = failure_signals.get(signal, 0) + 1 return ( dict(sorted(failure_by_stage.items(), key=lambda item: (-item[1], item[0]))), dict(sorted(failure_signals.items(), key=lambda item: (-item[1], item[0]))), ) def _active_domain_attempts( self, recent_attempts: list[dict[str, object]], active_domain: str, ) -> list[dict[str, object]]: domain = str(active_domain or "").strip().lower() if not domain: return list(recent_attempts) filtered = [ item for item in recent_attempts if str(item.get("email_domain") or "").strip().lower() == domain ] return filtered def _infer_active_domain( self, recent_attempts: list[dict[str, object]], cfmail_rotation: dict[str, object] | None, stoploss: dict[str, object], ) -> str: if isinstance(cfmail_rotation, dict): domain = str(cfmail_rotation.get("active_domain") or "").strip().lower() if domain: return domain domain = str(stoploss.get("active_domain") or "").strip().lower() if domain: return domain for item in reversed(recent_attempts): domain = str(item.get("email_domain") or "").strip().lower() if domain: return domain return "" def _extract_email_domain(self, result: dict[str, object]) -> str: metadata = result.get("metadata") if isinstance(result.get("metadata"), dict) else {} domain = str(metadata.get("email_domain") or "").strip().lower() if domain: return domain email = str(result.get("email") or "").strip().lower() if "@" not in email: return "" return email.rsplit("@", 1)[-1].strip().lower() def _update_cfmail_add_phone_stoploss(self, result: dict[str, object]) -> None: metadata = result.get("metadata") if isinstance(result.get("metadata"), dict) else {} provider = str(metadata.get("mail_provider") or result.get("mail_provider") or "").strip().lower() if provider not in {"", "cfmail"}: return domain = self._extract_email_domain(result) if not domain: return if not self._is_cfmail_active_domain(domain): return success = bool(result.get("success")) stage = str(result.get("stage") or "").strip().lower() post_gate = str(metadata.get("post_create_gate") or "").strip().lower() is_add_phone = stage == "add_phone_gate" or post_gate == "add_phone" with self._lock: events = self._cfmail_add_phone_events.setdefault( domain, deque(maxlen=self._cfmail_add_phone_window), ) events.append( { "success": success, "is_add_phone": is_add_phone, } ) state = self._cfmail_add_phone_state state["active_domain"] = domain if self._cfmail_add_phone_cooldown_seconds <= 0: state["in_cooldown"] = False state["cooldown_until"] = 0.0 return cooldown_until = float(state.get("cooldown_until") or 0.0) if time.time() < cooldown_until: state["in_cooldown"] = True return state["in_cooldown"] = False if len(events) < self._cfmail_add_phone_window: return add_phone_failures = sum(1 for item in events if item.get("is_add_phone")) successes = sum(1 for item in events if item.get("success")) if ( add_phone_failures >= self._cfmail_add_phone_threshold and successes <= self._cfmail_add_phone_max_successes ): state["in_cooldown"] = True state["cooldown_until"] = time.time() + self._cfmail_add_phone_cooldown_seconds state["last_triggered_at"] = datetime.now().isoformat(timespec="seconds") state["last_rotation_attempted_at"] = "" state["last_reason"] = "add_phone threshold reached" state["last_add_phone_failures"] = add_phone_failures state["last_successes"] = successes state["last_window_size"] = len(events) self._log( f"[zhuce6:register] [cfmail] add_phone stoploss activated for {domain} " f"(add_phone_failures={add_phone_failures}, successes={successes}, window={len(events)})" ) def _is_cfmail_wait_otp_no_message_timeout(self, result: dict[str, object]) -> bool: metadata = result.get("metadata") if isinstance(result.get("metadata"), dict) else {} provider = str(metadata.get("mail_provider") or result.get("mail_provider") or "").strip().lower() if provider not in {"", "cfmail"}: return False stage = str(result.get("stage") or "").strip().lower() if stage != "wait_otp": return False failure_reason = str(metadata.get("otp_wait_failure_reason") or "").strip().lower() if failure_reason: return failure_reason == "mailbox_timeout_no_message" try: return int(metadata.get("otp_mailbox_message_scan_count") or 0) <= 0 except Exception: return False def _is_cfmail_invalid_domain_mailbox_failure(self, result: dict[str, object]) -> bool: metadata = result.get("metadata") if isinstance(result.get("metadata"), dict) else {} if str(result.get("stage") or "").strip().lower() != "mailbox": return False provider = str(metadata.get("mail_provider") or result.get("mail_provider") or "").strip().lower() if provider not in {"", "cfmail"}: return False haystacks = [str(result.get("error_message") or "")] haystacks.extend(str(item or "") for item in (result.get("logs") or [])) text = "\n".join(haystacks).lower() return "invalid domain" in text or "无效的域名" in text def _on_cfmail_wait_progress(self, account: object, diagnostics: dict[str, object]) -> None: if self._cfmail_wait_otp_live_threshold <= 0: return email = str(getattr(account, "email", "") or "").strip().lower() extra = getattr(account, "extra", {}) or {} domain = str(extra.get("email_domain") or "").strip().lower() if not domain and "@" in email: domain = email.rsplit("@", 1)[-1].strip().lower() if not domain: return with self._cfmail_wait_otp_live_lock: now = time.time() for tracked_domain in list(self._cfmail_wait_otp_live_progress.keys()): entries = self._cfmail_wait_otp_live_progress.get(tracked_domain) or {} fresh_entries = { key: value for key, value in entries.items() if now - float(value.get("updated_at") or 0.0) <= 15.0 } if fresh_entries: self._cfmail_wait_otp_live_progress[tracked_domain] = fresh_entries else: self._cfmail_wait_otp_live_progress.pop(tracked_domain, None) if not self._is_cfmail_active_domain(domain): self._cfmail_wait_otp_live_progress.pop(domain, None) return key = str(getattr(account, "account_id", "") or email or id(account)) domain_entries = self._cfmail_wait_otp_live_progress.setdefault(domain, {}) domain_entries[key] = { "scan_count": int(diagnostics.get("message_scan_count") or 0), "elapsed_seconds": float(diagnostics.get("elapsed_seconds") or 0.0), "updated_at": now, } if int(diagnostics.get("message_scan_count") or 0) > 0: self._mark_cfmail_canary_ready(domain, reason="live_mailbox_message_seen") state = self._cfmail_wait_otp_state if bool(state.get("in_cooldown")) and str(state.get("active_domain") or "").strip().lower() == domain: return stalled = [ value for value in domain_entries.values() if int(value.get("scan_count") or 0) <= 0 and float(value.get("elapsed_seconds") or 0.0) >= self._cfmail_wait_otp_live_age_seconds ] if len(stalled) < self._cfmail_wait_otp_live_threshold: return with self._lock: state = self._cfmail_wait_otp_state if bool(state.get("in_cooldown")) and str(state.get("active_domain") or "").strip().lower() == domain: return self._cfmail_wait_otp_state = { "active_domain": domain, "in_cooldown": True, "cooldown_until": time.time() + self._cfmail_wait_otp_cooldown_seconds, "last_triggered_at": now_iso(), "last_rotation_attempted_at": "", "last_reason": "live wait_otp no-message threshold reached", "last_no_message_timeouts": len(stalled), "last_window_size": len(stalled), "last_logged_at": 0.0, } self._log( f"[zhuce6:register] [cfmail] wait_otp live stoploss activated for {domain} " f"(stalled_waits={len(stalled)}, age>={self._cfmail_wait_otp_live_age_seconds}s)" ) def _update_cfmail_wait_otp_stoploss(self, result: dict[str, object]) -> None: metadata = result.get("metadata") if isinstance(result.get("metadata"), dict) else {} provider = str(metadata.get("mail_provider") or result.get("mail_provider") or "").strip().lower() if provider not in {"", "cfmail"}: return domain = self._extract_email_domain(result) if not domain: return if not self._is_cfmail_active_domain(domain): return is_no_message_timeout = self._is_cfmail_wait_otp_no_message_timeout(result) try: message_scan_count = int(metadata.get("otp_mailbox_message_scan_count") or 0) except Exception: message_scan_count = 0 has_message_seen = message_scan_count > 0 with self._lock: events = self._cfmail_wait_otp_events.setdefault( domain, deque(maxlen=self._cfmail_wait_otp_window), ) events.append( { "success": bool(result.get("success")), "is_no_message_timeout": is_no_message_timeout, "has_message_seen": has_message_seen, } ) state = self._cfmail_wait_otp_state state["active_domain"] = domain if self._cfmail_wait_otp_cooldown_seconds <= 0: state["in_cooldown"] = False state["cooldown_until"] = 0.0 return cooldown_until = float(state.get("cooldown_until") or 0.0) if time.time() < cooldown_until: state["in_cooldown"] = True return state["in_cooldown"] = False if len(events) < self._cfmail_wait_otp_window: return no_message_timeouts = sum(1 for item in events if item.get("is_no_message_timeout")) successes = sum(1 for item in events if item.get("success")) message_seen = sum(1 for item in events if item.get("has_message_seen")) if ( no_message_timeouts >= self._cfmail_wait_otp_threshold and successes <= 0 and message_seen <= 0 ): state["in_cooldown"] = True state["cooldown_until"] = time.time() + self._cfmail_wait_otp_cooldown_seconds state["last_triggered_at"] = datetime.now().isoformat(timespec="seconds") state["last_rotation_attempted_at"] = "" state["last_reason"] = "wait_otp no-message threshold reached" state["last_no_message_timeouts"] = no_message_timeouts state["last_successes"] = successes state["last_message_seen"] = message_seen state["last_window_size"] = len(events) self._log( f"[zhuce6:register] [cfmail] wait_otp stoploss activated for {domain} " f"(no_message_timeouts={no_message_timeouts}, successes={successes}, " f"message_seen={message_seen}, window={len(events)})" ) def _update_cfmail_fresh_domain_budget(self, result: dict[str, object]) -> None: if self._cfmail_fresh_domain_attempt_budget <= 0: return metadata = result.get("metadata") if isinstance(result.get("metadata"), dict) else {} provider = str(metadata.get("mail_provider") or result.get("mail_provider") or "").strip().lower() if provider not in {"", "cfmail"}: return domain = self._extract_email_domain(result) if not domain: return if not self._is_cfmail_active_domain(domain): return try: message_scan_count = int(metadata.get("otp_mailbox_message_scan_count") or 0) except Exception: message_scan_count = 0 success = bool(result.get("success")) with self._lock: state = self._cfmail_fresh_domain_state tracked_domain = str(state.get("active_domain") or "").strip().lower() if tracked_domain != domain: self._cfmail_fresh_domain_state = { "active_domain": domain, "completed_attempts": 0, "mail_seen_attempts": 0, "successes": 0, "last_triggered_at": "", "last_rotation_attempted_at": "", "last_reason": "", } state = self._cfmail_fresh_domain_state state["completed_attempts"] = int(state.get("completed_attempts") or 0) + 1 if message_scan_count > 0: state["mail_seen_attempts"] = int(state.get("mail_seen_attempts") or 0) + 1 if success: state["successes"] = int(state.get("successes") or 0) + 1 if ( int(state.get("mail_seen_attempts") or 0) > 0 and int(state.get("completed_attempts") or 0) >= self._cfmail_fresh_domain_attempt_budget and not str(state.get("last_rotation_attempted_at") or "").strip() ): state["last_triggered_at"] = datetime.now().isoformat(timespec="seconds") state["last_reason"] = "fresh_domain_attempt_budget_reached" self._log( f"[zhuce6:register] [cfmail] fresh-domain budget reached for {domain} " f"(completed_attempts={int(state.get('completed_attempts') or 0)}, " f"mail_seen_attempts={int(state.get('mail_seen_attempts') or 0)}, " f"budget={self._cfmail_fresh_domain_attempt_budget})" ) def _cfmail_add_phone_stoploss_snapshot(self) -> dict[str, object]: with self._lock: state = dict(self._cfmail_add_phone_state) if self._cfmail_add_phone_cooldown_seconds <= 0: return { "active_domain": str(state.get("active_domain") or ""), "in_cooldown": False, "cooldown_remaining_seconds": 0, "last_triggered_at": str(state.get("last_triggered_at") or ""), "last_reason": str(state.get("last_reason") or ""), "last_add_phone_failures": int(state.get("last_add_phone_failures") or 0), "last_successes": int(state.get("last_successes") or 0), "last_window_size": int(state.get("last_window_size") or 0), "window_size": self._cfmail_add_phone_window, "threshold": self._cfmail_add_phone_threshold, "max_successes_in_window": self._cfmail_add_phone_max_successes, } cooldown_until = float(state.get("cooldown_until") or 0.0) remaining = max(0, int(cooldown_until - time.time())) if remaining <= 0: state["in_cooldown"] = False return { "active_domain": str(state.get("active_domain") or ""), "in_cooldown": bool(state.get("in_cooldown")), "cooldown_remaining_seconds": remaining, "last_triggered_at": str(state.get("last_triggered_at") or ""), "last_reason": str(state.get("last_reason") or ""), "last_add_phone_failures": int(state.get("last_add_phone_failures") or 0), "last_successes": int(state.get("last_successes") or 0), "last_window_size": int(state.get("last_window_size") or 0), "window_size": self._cfmail_add_phone_window, "threshold": self._cfmail_add_phone_threshold, "max_successes_in_window": self._cfmail_add_phone_max_successes, } def _reset_cfmail_add_phone_stoploss(self, new_domain: str = "") -> None: with self._lock: self._cfmail_add_phone_state = { "active_domain": new_domain, "in_cooldown": False, "cooldown_until": 0.0, "last_triggered_at": "", "last_rotation_attempted_at": "", "last_reason": "", "last_add_phone_failures": 0, "last_successes": 0, "last_window_size": 0, "last_logged_at": 0.0, } def _cfmail_wait_otp_stoploss_snapshot(self) -> dict[str, object]: with self._lock: state = dict(self._cfmail_wait_otp_state) if self._cfmail_wait_otp_cooldown_seconds <= 0: return { "active_domain": str(state.get("active_domain") or ""), "in_cooldown": False, "cooldown_remaining_seconds": 0, "last_triggered_at": str(state.get("last_triggered_at") or ""), "last_reason": str(state.get("last_reason") or ""), "last_no_message_timeouts": int(state.get("last_no_message_timeouts") or 0), "last_successes": int(state.get("last_successes") or 0), "last_message_seen": int(state.get("last_message_seen") or 0), "last_window_size": int(state.get("last_window_size") or 0), "window_size": self._cfmail_wait_otp_window, "threshold": self._cfmail_wait_otp_threshold, } cooldown_until = float(state.get("cooldown_until") or 0.0) remaining = max(0, int(cooldown_until - time.time())) if remaining <= 0: state["in_cooldown"] = False return { "active_domain": str(state.get("active_domain") or ""), "in_cooldown": bool(state.get("in_cooldown")), "cooldown_remaining_seconds": remaining, "last_triggered_at": str(state.get("last_triggered_at") or ""), "last_reason": str(state.get("last_reason") or ""), "last_no_message_timeouts": int(state.get("last_no_message_timeouts") or 0), "last_successes": int(state.get("last_successes") or 0), "last_message_seen": int(state.get("last_message_seen") or 0), "last_window_size": int(state.get("last_window_size") or 0), "window_size": self._cfmail_wait_otp_window, "threshold": self._cfmail_wait_otp_threshold, } def _cfmail_fresh_domain_budget_snapshot(self) -> dict[str, object]: with self._lock: state = dict(self._cfmail_fresh_domain_state) return { "active_domain": str(state.get("active_domain") or ""), "completed_attempts": int(state.get("completed_attempts") or 0), "mail_seen_attempts": int(state.get("mail_seen_attempts") or 0), "successes": int(state.get("successes") or 0), "last_triggered_at": str(state.get("last_triggered_at") or ""), "last_rotation_attempted_at": str(state.get("last_rotation_attempted_at") or ""), "last_reason": str(state.get("last_reason") or ""), "attempt_budget": self._cfmail_fresh_domain_attempt_budget, } def _reset_cfmail_wait_otp_stoploss(self, new_domain: str = "") -> None: with self._lock: self._cfmail_wait_otp_state = { "active_domain": new_domain, "in_cooldown": False, "cooldown_until": 0.0, "last_triggered_at": "", "last_rotation_attempted_at": "", "last_reason": "", "last_no_message_timeouts": 0, "last_successes": 0, "last_message_seen": 0, "last_window_size": 0, "last_logged_at": 0.0, } with self._cfmail_wait_otp_live_lock: if new_domain: self._cfmail_wait_otp_live_progress = { str(new_domain).strip().lower(): {} } else: self._cfmail_wait_otp_live_progress = {} def _reset_cfmail_fresh_domain_budget(self, new_domain: str = "") -> None: with self._lock: self._cfmail_fresh_domain_state = { "active_domain": str(new_domain or "").strip().lower(), "completed_attempts": 0, "mail_seen_attempts": 0, "successes": 0, "last_triggered_at": "", "last_rotation_attempted_at": "", "last_reason": "", } def _reload_cfmail_manager_after_rotation(self) -> None: manager = self._cfmail_manager if manager is None: return try: reload_if_needed = getattr(manager, "reload_if_needed", None) if callable(reload_if_needed): try: reload_if_needed(force=True) except TypeError: reload_if_needed() except Exception as exc: self._log(f"[zhuce6:register] [cfmail] manager reload after rotation failed: {exc}") def _ensure_cfmail_active_domain_ready(self) -> bool: if self._cfmail_provisioner is None or self._cfmail_manager is None: return False active_accounts = self._current_cfmail_active_accounts() active_domains = [item["domain"] for item in active_accounts] if not active_domains: return False self._log( f"[zhuce6:register] [cfmail] active-domain pool ready: " f"{', '.join(active_domains)}" ) if self._cfmail_tracker is not None: for domain in active_domains: try: self._cfmail_tracker.active_domain = domain except Exception: pass return True def _force_rotate_cfmail_for_invalid_mailbox(self, thread_id: int, result: dict[str, object]) -> bool: if not self._is_cfmail_invalid_domain_mailbox_failure(result): return False if self._cfmail_tracker is None or self._cfmail_provisioner is None: return False current_domain = self._extract_email_domain(result) or self._current_cfmail_active_domain() if not current_domain: return False self._log( f"[zhuce6:register] [thread-{thread_id}] [cfmail] rotating invalid domain " f"{current_domain} after mailbox bootstrap failure" ) return self._replace_cfmail_domain( thread_id=thread_id, domain=current_domain, reason_label="mailbox invalid domain", ) def _rotate_cfmail_for_failed_canary(self, thread_id: int, result: dict[str, object]) -> bool: del thread_id, result return False def _rotate_cfmail_for_fresh_domain_budget(self, thread_id: int) -> bool: if self._cfmail_fresh_domain_attempt_budget <= 0: return False if self._cfmail_tracker is None or self._cfmail_provisioner is None: return False with self._lock: state = dict(self._cfmail_fresh_domain_state) domain = str(state.get("active_domain") or "").strip().lower() completed_attempts = int(state.get("completed_attempts") or 0) mail_seen_attempts = int(state.get("mail_seen_attempts") or 0) if ( not domain or mail_seen_attempts <= 0 or completed_attempts < self._cfmail_fresh_domain_attempt_budget or str(state.get("last_rotation_attempted_at") or "").strip() ): return False self._cfmail_fresh_domain_state["last_rotation_attempted_at"] = datetime.now().isoformat(timespec="seconds") if not self._is_cfmail_active_domain(domain): self._clear_cfmail_domain_state(domain) return False self._log( f"[zhuce6:register] [thread-{thread_id}] [cfmail] rotating domain {domain} " f"because fresh domain budget reached" ) if self._replace_cfmail_domain( thread_id=thread_id, domain=domain, reason_label="fresh domain budget reached", ): return True with self._lock: self._cfmail_fresh_domain_state["last_rotation_attempted_at"] = "" return False def _rotate_cfmail_for_stoploss( self, *, thread_id: int, state_attr: str, reason_label: str, ) -> bool: if self._cfmail_tracker is None or self._cfmail_provisioner is None: return False with self._lock: state_obj = getattr(self, state_attr, None) if not isinstance(state_obj, dict): return False domain = str(state_obj.get("active_domain") or "").strip().lower() if not state_obj.get("in_cooldown") or not domain: return False if str(state_obj.get("last_rotation_attempted_at") or "").strip(): return False state_obj["last_rotation_attempted_at"] = datetime.now().isoformat(timespec="seconds") return self._replace_cfmail_domain( thread_id=thread_id, domain=domain, reason_label=reason_label, ) def _wait_if_cfmail_add_phone_stopped(self, thread_id: int, provider: str) -> bool: if provider != "cfmail": return False if self._cfmail_add_phone_cooldown_seconds <= 0: return False state = self._cfmail_add_phone_stoploss_snapshot() if not state.get("in_cooldown"): return False had_spare_domains = len(self._cfmail_active_domain_set()) > 1 domain = str(state.get("active_domain") or "").strip().lower() if not self._is_cfmail_active_domain(domain): self._clear_cfmail_domain_state(domain) return False if self._rotate_cfmail_for_stoploss( thread_id=thread_id, state_attr="_cfmail_add_phone_state", reason_label="add_phone stoploss", ): return not had_spare_domains remaining = int(state.get("cooldown_remaining_seconds") or 0) should_log = False with self._lock: last_logged_at = float(self._cfmail_add_phone_state.get("last_logged_at") or 0.0) now = time.time() if now - last_logged_at >= 15: self._cfmail_add_phone_state["last_logged_at"] = now should_log = True if should_log: self._log( f"[zhuce6:register] [thread-{thread_id}] [cfmail] add_phone stoploss active for {domain}, " f"remaining={remaining}s" ) wait_seconds = min(max(remaining, 1), 5) self._stop_event.wait(wait_seconds) return True def _wait_if_cfmail_wait_otp_stopped(self, thread_id: int, provider: str) -> bool: if provider != "cfmail": return False if self._cfmail_wait_otp_cooldown_seconds <= 0: return False state = self._cfmail_wait_otp_stoploss_snapshot() if not state.get("in_cooldown"): return False had_spare_domains = len(self._cfmail_active_domain_set()) > 1 domain = str(state.get("active_domain") or "").strip().lower() if not self._is_cfmail_active_domain(domain): self._clear_cfmail_domain_state(domain) return False if self._rotate_cfmail_for_stoploss( thread_id=thread_id, state_attr="_cfmail_wait_otp_state", reason_label="wait_otp stoploss", ): return not had_spare_domains remaining = int(state.get("cooldown_remaining_seconds") or 0) should_log = False with self._lock: last_logged_at = float(self._cfmail_wait_otp_state.get("last_logged_at") or 0.0) now = time.time() if now - last_logged_at >= 15: self._cfmail_wait_otp_state["last_logged_at"] = now should_log = True if should_log: self._log( f"[zhuce6:register] [thread-{thread_id}] [cfmail] wait_otp stoploss active for {domain}, " f"remaining={remaining}s" ) wait_seconds = min(max(remaining, 1), 5) self._stop_event.wait(wait_seconds) return True def start(self) -> None: if self._threads: return _compat_main_attr("load_all", load_all)() self._stop_event.clear() self._target_reached.clear() self._started_at = time.time() num = self.settings.register_threads self._providers = [p.strip() for p in self.settings.register_mail_provider.split(",") if p.strip()] if not self._providers: self._providers = ["cfmail"] if "cfmail" in self._providers: from core.cfmail_domain_rotation import DomainHealthTracker import core.cfmail as cfmail_module from core.cfmail import DEFAULT_CFMAIL_MANAGER from core.cfmail_provisioner import CfmailProvisioner self._cfmail_tracker = DomainHealthTracker() self._cfmail_provisioner = CfmailProvisioner(proxy_url=self.settings.register_proxy) self._cfmail_manager = DEFAULT_CFMAIL_MANAGER cfmail_module.CFMAIL_WAIT_ABORT_PREDICATE = self._should_abort_cfmail_wait cfmail_module.CFMAIL_WAIT_PROGRESS_CALLBACK = self._on_cfmail_wait_progress try: normalize_result = self._cfmail_provisioner.normalize_to_domain_pool(self._cfmail_active_domain_count) self._cfmail_manager.reload_if_needed(force=True) provisioned_domains = list(normalize_result.get("provisioned_domains") or []) retired_domains = list(normalize_result.get("retired_domains") or []) active_domains = list(normalize_result.get("active_domains") or []) if provisioned_domains or retired_domains: self._log( "[zhuce6:register] [cfmail] normalized active domain pool: " f"active={','.join(active_domains) or '-'} " f"provisioned={','.join(provisioned_domains) or '-'} " f"retired={','.join(retired_domains) or '-'}" ) except Exception as exc: self._log(f"[zhuce6:register] [cfmail] normalize active domain pool failed: {exc}") try: self._ensure_cfmail_active_domain_ready() self._ensure_cfmail_domain_pool_target( trigger_thread_id=0, reason="startup", ) except Exception as exc: self._log(f"[zhuce6:register] [cfmail] startup active-domain preflight failed: {exc}") if self.settings.backend == "cpa" and self.settings.cpa_runtime_reconcile_enabled: try: _maybe_reconcile_cpa_runtime( pool_dir=self.settings.pool_dir, management_base_url=self.settings.cpa_management_base_url, enabled=True, cooldown_seconds=self.settings.cpa_runtime_reconcile_cooldown_seconds, restart_enabled=self.settings.cpa_runtime_reconcile_restart_enabled, state_file=self.settings.pool_dir / "cpa_runtime_reconcile_state.json", client=create_backend_client(self.settings), management_key=self.settings.cpa_management_key, ) except Exception as exc: self._log(f"[zhuce6:register] startup reconcile failed: {exc}") if self.settings.proxy_pool_configured: from core.proxy_pool import ProxyPool self._proxy_pool = ProxyPool.from_settings(self.settings) if self._proxy_pool is not None: self._proxy_pool.start() target_msg = f", target={self.settings.register_target_count}" if self.settings.register_target_count > 0 else "" self._log( f"[zhuce6:register] starting {num} threads, " f"providers={','.join(self._providers)}, " f"proxy={self.settings.register_proxy or 'none'}, " f"sleep={self.settings.register_sleep_min}-{self.settings.register_sleep_max}s" f"{target_msg}" ) threading_module = _compat_main_attr("threading", threading) for i in range(num): provider = self._providers[i % len(self._providers)] t = threading_module.Thread( target=self._worker, args=(i + 1, provider), daemon=True, name=f"zhuce6-register-{i + 1}", ) t.start() self._threads.append(t) # Solution B: start deferred retry worker thread pending_t = threading_module.Thread( target=self._pending_token_retry_worker, daemon=True, name="zhuce6-pending-token-retry", ) pending_t.start() self._threads.append(pending_t) self._write_runtime_state() def stop(self) -> None: self._stop_event.set() for t in self._threads: t.join(timeout=2) self._threads.clear() try: import core.cfmail as cfmail_module cfmail_module.CFMAIL_WAIT_ABORT_PREDICATE = None cfmail_module.CFMAIL_WAIT_PROGRESS_CALLBACK = None except Exception: pass if self._proxy_pool is not None: self._proxy_pool.close() self._proxy_pool = None self._write_runtime_state() def _should_stop(self) -> bool: return self._stop_event.is_set() or self._target_reached.is_set() def _check_target(self) -> bool: """Return True if target reached and threads should stop.""" if self.settings.register_target_count <= 0: return False with self._lock: if self._total_success >= self.settings.register_target_count: self._target_reached.set() return True return False def _cpa_api_root(self) -> str: parsed = urlsplit(self.settings.cpa_management_base_url) path = parsed.path or "" suffix = "/v0/management" if path.endswith(suffix): path = path[: -len(suffix)] return urlunsplit((parsed.scheme, parsed.netloc, path, "", "")).rstrip("/") def _get_cpa_management_key(self) -> str | None: cached = self._cpa_management_key_cache if cached is not False: return str(cached or "") or None key = str(get_management_key() or "").strip() or None self._cpa_management_key_cache = key or None return key def _sync_cpa_from_success(self, result: dict[str, object], thread_id: int) -> tuple[bool, str, str]: """Persist a registered account to CPA immediately while keeping pool as backup.""" pool_file_raw = str(result.get("pool_file") or "").strip() if not pool_file_raw: return False, "missing pool file", "" pool_file = Path(pool_file_raw) if not pool_file.is_file(): return False, f"pool file missing: {pool_file.name}", "" sync_started_at = now_iso() key = self._get_cpa_management_key() if not key: update_token_record( pool_file, backup_written=True, cpa_sync_status="failed", last_cpa_sync_at=sync_started_at, last_cpa_sync_error="CPA management key unavailable", ) return False, "CPA management key unavailable", "" try: from platforms.chatgpt.pool import load_token_record token_data = load_token_record(pool_file) except Exception as exc: return False, f"invalid pool file {pool_file.name}: {exc}", "" metadata = result.get("metadata") if isinstance(result.get("metadata"), dict) else {} post_create_gate = str((metadata or {}).get("post_create_gate") or token_data.get("registration_post_create_gate") or "").strip().lower() if post_create_gate == "add_phone" and not bool(token_data.get("warmup_required")): token_data = update_token_record( pool_file, warmup_required=True, warmup_state="pending", warmup_passed=False, registration_post_create_gate="add_phone", ) if not isinstance(token_data, dict) or not str(token_data.get("email") or "").strip(): update_token_record( pool_file, backup_written=True, cpa_sync_status="failed", last_cpa_sync_at=sync_started_at, last_cpa_sync_error="missing email in pool record", ) return False, f"missing email in {pool_file.name}", "" if is_warmup_pending_record(token_data): update_token_record( pool_file, backup_written=True, cpa_sync_status="warmup_pending", last_cpa_sync_at=sync_started_at, last_cpa_sync_error="warmup pending", ) return ( False, "warmup pending", str(token_data.get("email") or pool_file.name).strip() or pool_file.name, ) from platforms.chatgpt.cpa_upload import upload_to_cpa ok, message = upload_to_cpa( token_data, api_url=self._cpa_api_root(), api_key=key, proxy=None, ) email = str(token_data.get("email") or pool_file.name).strip() or pool_file.name if ok: update_token_record( pool_file, health_status="good", backup_written=True, cpa_sync_status="synced", last_cpa_sync_at=sync_started_at, last_cpa_sync_error="", ) return True, "", email else: update_token_record( pool_file, backup_written=True, cpa_sync_status="failed", last_cpa_sync_at=sync_started_at, last_cpa_sync_error=message, ) return False, message, email # ── Solution B: Deferred retry queue ────────────────────────── def _enqueue_pending_token( self, result: dict[str, object], thread_id: int, *, proxy_key: str = "", proxy_url: str = "", ) -> None: """Save an add_phone_gate account for deferred token acquisition retry.""" metadata = result.get("metadata") if isinstance(result.get("metadata"), dict) else {} deferred = metadata.get("deferred_credentials") if not isinstance(deferred, dict): return email = str(deferred.get("email") or "").strip() password = str(deferred.get("password") or "").strip() if not email or not password: return entry = { "email": email, "password": password, "mailbox_jwt": str(deferred.get("mailbox_jwt") or "").strip(), "mailbox_extra": dict(deferred.get("mailbox_extra") or {}), "registration_proxy_key": str(deferred.get("registration_proxy_key") or proxy_key or "").strip(), "registration_proxy_region": str( deferred.get("registration_proxy_region") or infer_proxy_region( str(deferred.get("registration_proxy_key") or proxy_key or "").strip() or str(deferred.get("registration_proxy_url") or proxy_url or "").strip() ) or "" ).strip(), "registration_proxy_url": str(deferred.get("registration_proxy_url") or proxy_url or "").strip(), "registration_fingerprint_profile": str( deferred.get("registration_fingerprint_profile") or "chrome120_win" ).strip(), "cfmail_profile_name": str( deferred.get("cfmail_profile_name") or metadata.get("cfmail_profile_name") or "" ).strip(), "add_phone_trace_path": str( deferred.get("add_phone_trace_path") or metadata.get("add_phone_trace_path") or "" ).strip(), "created_at": time.time(), "retry_count": 0, "last_retry_at": 0.0, } with self._pending_token_lock: self._pending_token_queue.append(entry) self._pending_token_total_enqueued += 1 origin_proxy = str(entry.get("registration_proxy_key") or entry.get("registration_proxy_region") or entry.get("registration_proxy_url") or "").strip() self._log( f"[zhuce6:register] [thread-{thread_id}] 📥 deferred token retry enqueued: {email} " f"(origin_proxy={origin_proxy or '-'}, queue_size={len(self._pending_token_queue)})" ) def _pending_token_retry_worker(self) -> None: """Background thread that retries token acquisition for queued add_phone_gate accounts.""" while not self._should_stop(): if self._stop_event.wait(30): break batch: list[dict[str, Any]] = [] now = time.time() with self._pending_token_lock: remaining: list[dict[str, Any]] = [] for entry in self._pending_token_queue: created_at = float(entry.get("created_at") or 0) retry_count = int(entry.get("retry_count") or 0) last_retry = float(entry.get("last_retry_at") or 0) age = now - created_at since_last = now - last_retry if last_retry > 0 else age delay = self._pending_token_retry_delay_seconds * (retry_count + 1) if retry_count >= self._pending_token_max_retries: self._pending_token_total_failed += 1 self._log( f"[zhuce6:register] [pending] ❌ exhausted retries for " f"{entry.get('email')}, discarding" ) continue if since_last >= delay: batch.append(entry) else: remaining.append(entry) self._pending_token_queue = remaining if not batch: continue for entry in batch: if self._should_stop(): break self._retry_pending_token(entry) def _retry_pending_token(self, entry: dict[str, Any]) -> None: """Attempt token acquisition for a single deferred account.""" email = str(entry.get("email") or "").strip() password = str(entry.get("password") or "").strip() mailbox_jwt = str(entry.get("mailbox_jwt") or "").strip() mailbox_extra = dict(entry.get("mailbox_extra") or {}) registration_proxy_key = str(entry.get("registration_proxy_key") or "").strip() registration_proxy_region = str(entry.get("registration_proxy_region") or "").strip() registration_proxy_url = str(entry.get("registration_proxy_url") or "").strip() registration_fingerprint_profile = str( entry.get("registration_fingerprint_profile") or "" ).strip() cfmail_profile_name = str(entry.get("cfmail_profile_name") or "").strip() add_phone_trace_path = str(entry.get("add_phone_trace_path") or "").strip() retry_count = int(entry.get("retry_count") or 0) + 1 self._log( f"[zhuce6:register] [pending] 🔄 retrying token acquisition " f"for {email} (attempt {retry_count}/{self._pending_token_max_retries})" ) proxy_lease = None proxy_url = registration_proxy_url or self.settings.register_proxy proxy_release_success = False try: from core.base_mailbox import MailboxAccount from core.cfmail import CfMailMailbox, DEFAULT_CFMAIL_MANAGER from platforms.chatgpt.plugin import MailboxEmailServiceAdapter from platforms.chatgpt.register import RegistrationEngine from platforms.chatgpt.pool import write_token_record if self._proxy_pool is not None: try: proxy_lease = self._proxy_pool.acquire( timeout=5.0, preferred_name=registration_proxy_key or None, preferred_regions=(registration_proxy_region,) if registration_proxy_region else (), ) proxy_url = str(proxy_lease.proxy_url or "").strip() or proxy_url self._log( "[zhuce6:register] [pending] " f"using retry proxy {proxy_lease.name} for {email} " f"(origin={registration_proxy_key or registration_proxy_region or registration_proxy_url or '-'})" ) except Exception as exc: self._log( "[zhuce6:register] [pending] " f"proxy acquire failed for {email}: {exc}" ) mailbox = CfMailMailbox(manager=DEFAULT_CFMAIL_MANAGER) adapter = MailboxEmailServiceAdapter(mailbox) # Reconstruct the mailbox account so _wait_for_mailbox_code can poll if mailbox_jwt and mailbox_extra: adapter._account = MailboxAccount( email=email, account_id=mailbox_jwt, extra=dict(mailbox_extra), ) engine = RegistrationEngine( email_service=adapter, proxy_url=proxy_url, ) engine.email = email engine.password = password token_info = engine._login_for_token() if token_info: proxy_release_success = True self._log(f"[zhuce6:register] [pending] ✅ deferred token acquired for {email}") # Write to pool token_data = { "type": "codex", "email": email, "password": password, "mail_provider": "cfmail", "expired": str(token_info.get("expired") or ""), "id_token": str(token_info.get("id_token") or ""), "account_id": str(token_info.get("account_id") or ""), "access_token": str(token_info.get("access_token") or ""), "last_refresh": str(token_info.get("last_refresh") or ""), "refresh_token": str(token_info.get("refresh_token") or ""), "source": "deferred_retry", "add_phone_trace_path": add_phone_trace_path, } token_data.update( build_registration_provenance( { "location": "", "mail_provider": "cfmail", "post_create_gate": "add_phone", }, proxy_url=registration_proxy_url or proxy_url, proxy_key=registration_proxy_key or getattr(proxy_lease, "name", ""), proxy_region=registration_proxy_region, cfmail_profile_name=cfmail_profile_name, ) ) if registration_fingerprint_profile: token_data["registration_fingerprint_profile"] = registration_fingerprint_profile pool_file = write_token_record(token_data, self.settings.pool_dir) sync_ok, sync_error, synced_email = self._sync_cpa_from_success( {"pool_file": str(pool_file), "success": True, "stage": "deferred_retry"}, thread_id=0, ) with self._lock: if sync_ok: self._total_success += 1 self._total_cpa_sync_success += 1 self._pending_token_total_success += 1 elif sync_error == "warmup pending": self._total_warmup_pending += 1 else: self._total_failure += 1 self._total_cpa_sync_failure += 1 self._pending_token_total_failed += 1 if sync_ok: self._log(f"[zhuce6:register] [pending] ✅ CPA sync success: {synced_email or email}") elif sync_error == "warmup pending": self._log(f"[zhuce6:register] [pending] ⏳ warmup pending: {synced_email or email}") else: self._log(f"[zhuce6:register] [pending] ❌ failed [stage=cpa_sync]: {sync_error or 'unknown'}") return else: self._log(f"[zhuce6:register] [pending] ⏳ deferred retry failed for {email}") except Exception as exc: self._log(f"[zhuce6:register] [pending] ⚠️ deferred retry error for {email}: {exc}") finally: if proxy_lease is not None and self._proxy_pool is not None: try: self._proxy_pool.release( proxy_lease, success=proxy_release_success, stage="deferred_retry", ) except Exception as exc: self._log(f"[zhuce6:register] [pending] proxy release failed for {email}: {exc}") # Re-enqueue with incremented retry count entry["retry_count"] = retry_count entry["last_retry_at"] = time.time() with self._pending_token_lock: self._pending_token_queue.append(entry) def _worker(self, thread_id: int, initial_provider: str) -> None: provider = initial_provider consecutive_failures = 0 max_failures = self.settings.register_max_consecutive_failures while not self._should_stop(): if provider == "cfmail" and self._cfmail_tracker is not None: self._cfmail_rotation_pause.wait() if self._wait_if_cfmail_add_phone_stopped(thread_id, provider): continue if self._wait_if_cfmail_wait_otp_stopped(thread_id, provider): continue if self._wait_if_cfmail_canary_pending(thread_id, provider): continue if provider == "cfmail" and self._cfmail_manager is not None: if self._check_cfmail_all_cooldown_rotation(thread_id): continue if self._cfmail_all_accounts_in_cooldown(): self._stop_event.wait(10) continue if self._wait_if_cfmail_flow_throttled(thread_id, provider): continue try: result: dict[str, object] = {} proxy_key = "" proxy_outcome: bool | None = False result_metadata: dict[str, object] = {} result_stage = "?" result_error = "" result_email = "" max_proxy_attempts = 2 if self._proxy_pool is not None else 1 for proxy_attempt in range(1, max_proxy_attempts + 1): proxy_lease = None proxy_url = self.settings.register_proxy proxy_key = proxy_url or "" proxy_outcome = False release_stage = "exception" try: if self._proxy_pool is not None: proxy_lease = self._proxy_pool.acquire( timeout=5, preferred_regions=tuple(self.settings.register_fresh_proxy_regions or ()), ) proxy_url = proxy_lease.proxy_url proxy_key = proxy_lease.name or proxy_url or "" self._log( f"[zhuce6:register] [thread-{thread_id}] acquired proxy {proxy_lease.local_port} ({proxy_lease.name})" ) self._log(f"[zhuce6:register] [thread-{thread_id}] attempting (provider={provider})") cfmail_profile_name = self._selected_cfmail_profile(thread_id) if provider == "cfmail" else "auto" result = _compat_main_attr("run_chatgpt_register_once", run_chatgpt_register_once)( email=None, password=None, mail_provider=provider, cfmail_profile_name=cfmail_profile_name, proxy=proxy_url, write_pool=True, pool_dir=self.settings.pool_dir, ) proxy_outcome = self._classify_proxy_outcome(result) result_metadata = result.get("metadata") if isinstance(result.get("metadata"), dict) else {} result_stage = str(result.get("stage") or "?").strip() or "?" if bool(result.get("success")) and result_stage == "?": result_stage = "completed" result_error = str(result.get("error_message") or "").strip() result_email = str(result.get("email") or "").strip() release_stage = result_stage should_retry_device_id = ( self._proxy_pool is not None and proxy_attempt < max_proxy_attempts and not bool(result.get("success")) and result_stage == "device_id" ) if should_retry_device_id: self._log( f"[zhuce6:register] [thread-{thread_id}] device_id failed on proxy {proxy_key}; " "rotating proxy and retrying once" ) continue break finally: if proxy_lease is not None and self._proxy_pool is not None: try: self._proxy_pool.release( proxy_lease, success=proxy_outcome, stage=release_stage, ) except Exception as exc: self._log(f"[zhuce6:register] [thread-{thread_id}] proxy release failed: {exc}") success = bool(result.get("success")) should_break_after_iteration = False with self._lock: self._total_attempts += 1 if success: pool_file_raw = str(result.get("pool_file") or "").strip() if pool_file_raw: try: update_token_record( Path(pool_file_raw), **build_registration_provenance( result_metadata, proxy_url=proxy_url, proxy_key=proxy_key, proxy_region=infer_proxy_region(proxy_key or proxy_url), cfmail_profile_name=str(result_metadata.get("cfmail_profile_name") or ""), ), ) except Exception as exc: self._log( f"[zhuce6:register] [thread-{thread_id}] provenance update failed: {exc}" ) sync_ok, sync_error, synced_email = self._sync_cpa_from_success(result, thread_id) with self._lock: if sync_ok: self._total_success += 1 self._total_cpa_sync_success += 1 self._last_error = None consecutive_failures = 0 self._record_attempt( success=True, stage=result_stage, error_message="", metadata=result_metadata, proxy_key=proxy_key, email=result_email or synced_email, ) email = result_email or synced_email or "?" self._log(f"[zhuce6:register] [thread-{thread_id}] \u2705 success: {email}") self._log(f"[zhuce6:register] [thread-{thread_id}] ✅ CPA sync success: {email}") if self._check_target(): self._log( f"[zhuce6:register] target reached ({self.settings.register_target_count}), stopping" ) should_break_after_iteration = True elif sync_error == "warmup pending": self._total_warmup_pending += 1 self._last_error = None consecutive_failures = 0 self._record_attempt( success=False, stage="warmup_pending", error_message="warmup pending", metadata=result_metadata, proxy_key=proxy_key, email=result_email or synced_email, ) email = result_email or synced_email or "?" self._log(f"[zhuce6:register] [thread-{thread_id}] ⏳ warmup pending: {email}") else: self._total_failure += 1 self._total_cpa_sync_failure += 1 consecutive_failures += 1 err = sync_error or "CPA sync failed" self._last_error = err self._record_attempt( success=False, stage="cpa_sync", error_message=err, metadata=result_metadata, proxy_key=proxy_key, email=result_email or synced_email, ) self._log( f"[zhuce6:register] [thread-{thread_id}] \u274c failed ({consecutive_failures}/{max_failures}) " f"[stage=cpa_sync]: {err}" ) else: with self._lock: self._total_failure += 1 consecutive_failures += 1 err = result_error or "unknown" stage = result_stage self._last_error = err self._record_attempt( success=False, stage=stage, error_message=err, metadata=result_metadata, proxy_key=proxy_key, email=result_email, ) self._log(f"[zhuce6:register] [thread-{thread_id}] \u274c failed ({consecutive_failures}/{max_failures}) [stage={stage}]: {err}") for log_line in result.get("logs", []): self._log(f"[zhuce6:register] [thread-{thread_id}] \u21b3 {log_line}") # Solution B: enqueue add_phone_gate accounts for deferred retry if result_stage == "add_phone_gate": self._enqueue_pending_token( result, thread_id, proxy_key=proxy_key, proxy_url=proxy_url, ) invalid_mailbox_rotation = self._force_rotate_cfmail_for_invalid_mailbox( thread_id, result, ) canary_rotation = self._rotate_cfmail_for_failed_canary(thread_id, result) rotation_success = self._handle_cfmail_rotation( thread_id=thread_id, result=result, proxy_key=proxy_key, ) self._update_cfmail_add_phone_stoploss(result) self._update_cfmail_wait_otp_stoploss(result) self._update_cfmail_canary_after_result(thread_id=thread_id, result=result) self._update_cfmail_fresh_domain_budget(result) fresh_domain_rotation = self._rotate_cfmail_for_fresh_domain_budget(thread_id) if invalid_mailbox_rotation or canary_rotation or rotation_success or fresh_domain_rotation: consecutive_failures = 0 self._write_runtime_state() if should_break_after_iteration: break except Exception as exc: with self._lock: self._total_attempts += 1 self._total_failure += 1 self._last_error = str(exc) self._record_attempt( success=False, stage="exception", error_message=str(exc), metadata={"mail_provider": provider}, proxy_key=proxy_key, email="", ) consecutive_failures += 1 self._log(f"[zhuce6:register] [thread-{thread_id}] \u274c exception ({consecutive_failures}/{max_failures}): {exc}") self._write_runtime_state() finally: self._release_cfmail_flow_slot(thread_id) # Fallback: switch provider after N consecutive failures if consecutive_failures >= max_failures and len(self._providers) > 1: old_provider = provider current_idx = self._providers.index(provider) if provider in self._providers else 0 provider = self._providers[(current_idx + 1) % len(self._providers)] consecutive_failures = 0 self._log( f"[zhuce6:register] [thread-{thread_id}] [fallback] switching {old_provider} -> {provider}" ) # Random sleep between attempts sleep_sec = random.randint( self.settings.register_sleep_min, max(self.settings.register_sleep_min, self.settings.register_sleep_max), ) if self._stop_event.wait(sleep_sec) or self._target_reached.is_set(): break def _cfmail_all_accounts_in_cooldown(self) -> bool: manager = self._cfmail_manager if manager is None: return False try: manager.reload_if_needed() except Exception: pass return manager.select_account() is None def _current_cfmail_active_domain(self) -> str: accounts = self._current_cfmail_active_accounts() if accounts: return str(accounts[0]["domain"]).strip().lower() return "" def _should_abort_cfmail_wait(self, account: Any) -> bool: try: extra = account.extra if hasattr(account, "extra") and isinstance(account.extra, dict) else {} domain = str(extra.get("email_domain") or "").strip().lower() if not domain: email = str(getattr(account, "email", "") or "").strip().lower() if "@" in email: domain = email.rsplit("@", 1)[-1].strip().lower() if not domain: return False stoploss = self._cfmail_wait_otp_stoploss_snapshot() if not bool(stoploss.get("in_cooldown")): return False if str(stoploss.get("active_domain") or "").strip().lower() != domain: return False wait_started_at = float(extra.get("otp_wait_started_at") or 0.0) triggered_at_raw = str(stoploss.get("last_triggered_at") or "").strip() if wait_started_at > 0.0 and triggered_at_raw: try: triggered_at = datetime.fromisoformat(triggered_at_raw).timestamp() except Exception: triggered_at = 0.0 if triggered_at > 0.0 and wait_started_at < triggered_at: return False return True except Exception: return False def _check_cfmail_all_cooldown_rotation(self, thread_id: int) -> bool: """When all cfmail accounts are in cooldown, proactively trigger domain rotation.""" if self._cfmail_tracker is None or self._cfmail_provisioner is None: return False if not self._cfmail_all_accounts_in_cooldown(): return False if not self._cfmail_rotation_lock.acquire(blocking=False): return False self._cfmail_rotation_pause.clear() try: self._log( f"[zhuce6:register] [thread-{thread_id}] [cfmail] all accounts in cooldown, " "forcing domain rotation to break deadlock" ) provision_result = self._cfmail_provisioner.rotate_active_domain() if not provision_result.success: if provision_result.old_domain: self._cfmail_tracker.mark_rotation_failed( provision_result.old_domain, provision_result.error, ) self._log( f"[zhuce6:register] [thread-{thread_id}] [cfmail] deadlock rotation failed: " f"{provision_result.error}" ) return False self._cfmail_tracker.mark_rotation_completed( provision_result.old_domain, provision_result.new_domain, ) self._reload_cfmail_manager_after_rotation() self._reset_cfmail_add_phone_stoploss(provision_result.new_domain) self._reset_cfmail_wait_otp_stoploss(provision_result.new_domain) self._reset_cfmail_fresh_domain_budget(provision_result.new_domain) self._arm_cfmail_canary(provision_result.new_domain) self._log( f"[zhuce6:register] [thread-{thread_id}] [cfmail] deadlock rotation completed: " f"{provision_result.old_domain} -> {provision_result.new_domain}" ) return True finally: self._cfmail_rotation_pause.set() self._cfmail_rotation_lock.release() def _current_cfmail_active_accounts(self) -> list[dict[str, str]]: manager = self._cfmail_manager accounts = list(getattr(manager, "accounts", []) or []) active_accounts: list[dict[str, str]] = [] for item in accounts: if not bool(getattr(item, "email_domain", "") or ""): continue if manager is not None and callable(getattr(manager, "skip_remaining_seconds", None)): try: if int(manager.skip_remaining_seconds(getattr(item, "name", "")) or 0) > 0: continue except Exception: pass active_accounts.append( { "name": str(getattr(item, "name", "") or "").strip(), "domain": str(getattr(item, "email_domain", "") or "").strip().lower(), } ) return [item for item in active_accounts if item["name"] and item["domain"]] def _cfmail_domain_pool_snapshot( self, recent_attempts: list[dict[str, object]], ) -> dict[str, object]: active_accounts = self._current_cfmail_active_accounts() with self._lock: inflight_by_thread = dict(self._cfmail_flow_state.get("inflight_by_thread") or {}) last_started_by_domain = dict(self._cfmail_flow_state.get("last_started_by_domain") or {}) replenish_thread = self._cfmail_replenish_thread replenish_reason = self._cfmail_replenish_reason domains: list[dict[str, object]] = [] now_ts = time.time() manager = self._cfmail_manager for account in active_accounts: domain = account["domain"] profile_name = account["name"] domain_attempts = self._active_domain_attempts(recent_attempts, domain) failure_by_stage, failure_signals = self._failure_counts_from_attempts(domain_attempts) recent_success = sum(1 for item in domain_attempts if bool(item.get("success"))) recent_failure = sum(1 for item in domain_attempts if not bool(item.get("success"))) inflight = sum( 1 for value in inflight_by_thread.values() if isinstance(value, dict) and str(value.get("domain") or "").strip().lower() == domain ) last_started_at = float(last_started_by_domain.get(domain) or 0.0) start_interval_remaining = 0.0 if self._cfmail_start_interval_seconds > 0 and last_started_at > 0: start_interval_remaining = max( 0.0, (last_started_at + self._cfmail_start_interval_seconds) - now_ts, ) skip_remaining_seconds = 0 if manager is not None and callable(getattr(manager, "skip_remaining_seconds", None)): try: skip_remaining_seconds = int(manager.skip_remaining_seconds(profile_name) or 0) except Exception: skip_remaining_seconds = 0 add_phone_state = self._cfmail_add_phone_stoploss_snapshot() wait_otp_state = self._cfmail_wait_otp_stoploss_snapshot() domains.append( { "name": profile_name, "domain": domain, "inflight": inflight, "recent_attempts": len(domain_attempts), "recent_success": recent_success, "recent_failure": recent_failure, "failure_by_stage": failure_by_stage, "failure_signals": failure_signals, "recent_failure_hotspots": self._recent_failure_hotspots_from_attempts(domain_attempts), "last_started_at": datetime.fromtimestamp(last_started_at).isoformat(timespec="seconds") if last_started_at > 0 else "", "start_interval_remaining_seconds": round(start_interval_remaining, 1), "skip_remaining_seconds": skip_remaining_seconds, "add_phone_cooldown": bool( add_phone_state.get("in_cooldown") and str(add_phone_state.get("active_domain") or "").strip().lower() == domain ), "wait_otp_cooldown": bool( wait_otp_state.get("in_cooldown") and str(wait_otp_state.get("active_domain") or "").strip().lower() == domain ), } ) return { "target_count": self._cfmail_active_domain_count, "active_count": len(domains), "active_domains": domains, "replenishing": bool(replenish_thread is not None and replenish_thread.is_alive()), "replenish_reason": str(replenish_reason or ""), } def _selected_cfmail_profile(self, thread_id: int) -> str: with self._lock: selected_profile_by_thread = self._cfmail_flow_state.setdefault("selected_profile_by_thread", {}) if not isinstance(selected_profile_by_thread, dict): return "auto" value = str(selected_profile_by_thread.get(thread_id) or "").strip() return value or "auto" def _handle_cfmail_rotation( self, *, thread_id: int, result: dict[str, object], proxy_key: str, ) -> bool: if self._cfmail_tracker is None or self._cfmail_provisioner is None: return False metadata = result.get("metadata") if not isinstance(metadata, dict): metadata = {} if str(metadata.get("mail_provider") or result.get("mail_provider") or "").strip().lower() not in {"", "cfmail"}: return False from core.cfmail_domain_rotation import classify_domain_attempt attempt = classify_domain_attempt(result, proxy_key=proxy_key) if attempt is None: return False decision = self._cfmail_tracker.record(attempt) if attempt.backend_failure and not decision.should_rotate: self._log( f"[zhuce6:register] [thread-{thread_id}] [cfmail] backend unhealthy for domain={attempt.domain}; rotation skipped" ) return False if not decision.should_rotate: return False self._log( f"[zhuce6:register] [thread-{thread_id}] [cfmail] rotating domain {decision.domain} " f"(reason={decision.reason}, failures={decision.blacklist_failures}, window={decision.window_size})" ) return self._replace_cfmail_domain( thread_id=thread_id, domain=decision.domain, reason_label=decision.reason, ) def _classify_proxy_outcome(self, result: dict[str, object]) -> bool | None: if bool(result.get("success")): return True stage = str(result.get("stage") or "").strip().lower() metadata = result.get("metadata") if isinstance(result.get("metadata"), dict) else {} code = str(metadata.get("create_account_error_code") or "").strip().lower() post_create_gate = str(metadata.get("post_create_gate") or "").strip().lower() if stage == "create_account" and code in {"registration_disallowed", "unsupported_email"}: return None if stage == "add_phone_gate" or post_create_gate == "add_phone": return None if stage in {"mailbox", "device_id", "password", "token_acquisition"}: return False return False def snapshot(self) -> dict[str, object]: with self._lock: alive = sum(1 for t in self._threads if t.is_alive()) target = self.settings.register_target_count cfmail_rotation = self._cfmail_tracker.snapshot() if self._cfmail_tracker is not None else None recent_attempts = list(self._recent_attempts) add_phone_stoploss = self._cfmail_add_phone_stoploss_snapshot() wait_otp_stoploss = self._cfmail_wait_otp_stoploss_snapshot() cfmail_canary = self._cfmail_canary_snapshot() cfmail_fresh_domain_budget = self._cfmail_fresh_domain_budget_snapshot() cfmail_domain_pool = self._cfmail_domain_pool_snapshot(recent_attempts) active_domain = self._infer_active_domain(recent_attempts, cfmail_rotation, add_phone_stoploss) active_domain_attempts = self._active_domain_attempts(recent_attempts, active_domain) active_failure_by_stage, active_failure_signals = self._failure_counts_from_attempts(active_domain_attempts) return { "name": "register", "status": "running" if alive > 0 else ("pending" if self._total_attempts == 0 else "stopped"), "threads_alive": alive, "threads_total": len(self._threads), "total_attempts": self._total_attempts, "total_success": self._total_success, "total_success_registered": self._total_success, "total_success_direct": self._total_success, "total_warmup_pending": self._total_warmup_pending, "total_cpa_sync_success": self._total_cpa_sync_success, "total_cpa_sync_success_direct": self._total_cpa_sync_success, "total_cpa_sync_failure": self._total_cpa_sync_failure, "total_failure": self._total_failure, "success_rate": round(self._total_success / max(self._total_attempts, 1) * 100, 1), "registered_success_rate": round(self._total_success / max(self._total_attempts, 1) * 100, 1), "cpa_sync_success_rate": round(self._total_cpa_sync_success / max(self._total_attempts, 1) * 100, 1), "target_count": target if target > 0 else None, "target_reached": self._target_reached.is_set(), "last_error": self._last_error, "proxy": self.settings.register_proxy, "proxy_pool_enabled": self._proxy_pool is not None, "mail_provider": self.settings.register_mail_provider, "interval_seconds": self.settings.register_interval, "run_count": self._total_attempts, "success_count": self._total_success, "failure_count": self._total_failure, "is_running": alive > 0, "last_started_at": datetime.fromtimestamp(self._started_at).isoformat(timespec="seconds") if self._started_at else None, "last_finished_at": None, "last_duration_seconds": None, "next_run_at": None, "failure_by_stage": dict(sorted(self._failure_by_stage.items(), key=lambda item: (-item[1], item[0]))), "failure_signals": dict(sorted(self._failure_signals.items(), key=lambda item: (-item[1], item[0]))), "recent_failure_hotspots": self._recent_failure_hotspots(), "recent_attempts": recent_attempts, "active_domain_recent_attempts": active_domain_attempts, "active_domain_failure_by_stage": active_failure_by_stage, "active_domain_failure_signals": active_failure_signals, "active_domain_recent_failure_hotspots": self._recent_failure_hotspots_from_attempts(active_domain_attempts), "cfmail_rotation": cfmail_rotation, "cfmail_add_phone_stoploss": add_phone_stoploss, "cfmail_wait_otp_stoploss": wait_otp_stoploss, "cfmail_canary": cfmail_canary, "cfmail_fresh_domain_budget": cfmail_fresh_domain_budget, "cfmail_domain_pool": cfmail_domain_pool, "pending_token_queue": { "queue_size": len(self._pending_token_queue), "total_enqueued": self._pending_token_total_enqueued, "total_success": self._pending_token_total_success, "total_failed": self._pending_token_total_failed, "retry_delay_seconds": self._pending_token_retry_delay_seconds, "max_retries": self._pending_token_max_retries, }, } class RegistrationBurstScheduler: """Run registration in timed batches and expose scheduler state via runtime_state.json.""" def __init__(self, settings: AppSettings) -> None: self.settings = settings self._stop_event = threading.Event() self._lock = threading.RLock() self._active_loop: RegistrationLoop | None = None self._active_started_at: float | None = None self._next_run_at_ts: float | None = None self._run_count = 0 self._total_attempts = 0 self._total_success = 0 self._total_cpa_sync_success = 0 self._total_cpa_sync_failure = 0 self._total_failure = 0 self._last_error: str | None = None self._recent_attempts: deque[dict[str, object]] = deque(maxlen=80) self._failure_by_stage: dict[str, int] = {} self._failure_signals: dict[str, int] = {} self._last_batch_started_at: str | None = None self._last_batch_finished_at: str | None = None self._last_batch_duration_seconds: float | None = None self._last_cfmail_add_phone_stoploss: dict[str, object] | None = None self._last_cfmail_wait_otp_stoploss: dict[str, object] | None = None self._last_proxy_pool_snapshot: dict[str, object] = { "configured": bool(self.settings.proxy_pool_configured), "enabled": False, "snapshot_error": None, "node_count": 0, "in_use_count": 0, "disabled_count": 0, "nodes": [], } self._logger = self._setup_logger() def _setup_logger(self) -> Any: from logging.handlers import RotatingFileHandler logger = logging.getLogger("zhuce6.register") logger.setLevel(logging.INFO) logger.propagate = False if not logger.handlers: console = logging.StreamHandler() console.setFormatter(logging.Formatter("%(message)s")) logger.addHandler(console) if self.settings.register_log_file: fh = RotatingFileHandler( self.settings.register_log_file, maxBytes=2 * 1024 * 1024, backupCount=5, encoding="utf-8", ) fh.setFormatter(logging.Formatter("%(asctime)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S")) logger.addHandler(fh) return logger def _log(self, msg: str) -> None: self._logger.info(msg) def _merge_counts(self, target: dict[str, int], incoming: dict[str, object] | None) -> None: if not isinstance(incoming, dict): return for key, value in incoming.items(): try: inc = int(value or 0) except Exception: continue target[str(key)] = target.get(str(key), 0) + inc def _write_runtime_state(self) -> None: state_file = Path(self.settings.runtime_state_file) try: state_file.parent.mkdir(parents=True, exist_ok=True) payload = { "updated_at": datetime.now().isoformat(timespec="seconds"), "register_snapshot": self.snapshot(), "proxy_pool": self._current_proxy_pool_snapshot(), } tmp_file = state_file.with_name( f"{state_file.name}.{os.getpid()}.{threading.get_ident()}.tmp" ) tmp_file.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") tmp_file.replace(state_file) except Exception as exc: self._log(f"[zhuce6:register] burst runtime state write failed: {exc}") def _current_proxy_pool_snapshot(self) -> dict[str, object]: with self._lock: active_loop = self._active_loop last_snapshot = dict(self._last_proxy_pool_snapshot) if active_loop is not None: try: return active_loop._proxy_pool_snapshot() except Exception as exc: last_snapshot["snapshot_error"] = str(exc) return last_snapshot return last_snapshot def _absorb_batch_snapshot(self, snapshot: dict[str, object], *, duration_seconds: float) -> None: with self._lock: self._run_count += 1 self._total_attempts += int(snapshot.get("total_attempts") or 0) self._total_success += int(snapshot.get("total_success") or 0) self._total_cpa_sync_success += int(snapshot.get("total_cpa_sync_success") or 0) self._total_cpa_sync_failure += int(snapshot.get("total_cpa_sync_failure") or 0) self._total_failure += int(snapshot.get("total_failure") or 0) self._last_error = str(snapshot.get("last_error") or "").strip() or None self._last_batch_started_at = snapshot.get("last_started_at") if isinstance(snapshot.get("last_started_at"), str) else None self._last_batch_finished_at = datetime.now().isoformat(timespec="seconds") self._last_batch_duration_seconds = round(duration_seconds, 3) self._merge_counts(self._failure_by_stage, snapshot.get("failure_by_stage") if isinstance(snapshot.get("failure_by_stage"), dict) else None) self._merge_counts(self._failure_signals, snapshot.get("failure_signals") if isinstance(snapshot.get("failure_signals"), dict) else None) attempts = snapshot.get("recent_attempts") if isinstance(attempts, list): for item in attempts: if isinstance(item, dict): self._recent_attempts.append(item) if isinstance(snapshot.get("cfmail_add_phone_stoploss"), dict): self._last_cfmail_add_phone_stoploss = dict(snapshot.get("cfmail_add_phone_stoploss") or {}) if isinstance(snapshot.get("cfmail_wait_otp_stoploss"), dict): self._last_cfmail_wait_otp_stoploss = dict(snapshot.get("cfmail_wait_otp_stoploss") or {}) def snapshot(self) -> dict[str, object]: with self._lock: active_loop = self._active_loop next_run_at_ts = self._next_run_at_ts total_attempts = self._total_attempts total_success = self._total_success total_cpa_sync_success = self._total_cpa_sync_success total_cpa_sync_failure = self._total_cpa_sync_failure total_failure = self._total_failure run_count = self._run_count failure_by_stage = dict(sorted(self._failure_by_stage.items(), key=lambda item: (-item[1], item[0]))) failure_signals = dict(sorted(self._failure_signals.items(), key=lambda item: (-item[1], item[0]))) recent_attempts = list(self._recent_attempts) last_error = self._last_error last_batch_started_at = self._last_batch_started_at last_batch_finished_at = self._last_batch_finished_at last_batch_duration_seconds = self._last_batch_duration_seconds add_phone_stoploss = dict(self._last_cfmail_add_phone_stoploss or {}) wait_otp_stoploss = dict(self._last_cfmail_wait_otp_stoploss or {}) if active_loop is not None: current = dict(active_loop.snapshot()) current.update( { "scheduler_mode": "burst", "batch_threads": self.settings.register_batch_threads, "batch_target_count": self.settings.register_batch_target_count, "batch_interval_seconds": self.settings.register_batch_interval_seconds, "run_count": run_count, "next_run_at": None, } ) return current status = "scheduled" if next_run_at_ts and not self._stop_event.is_set() else ("stopped" if run_count > 0 or self._stop_event.is_set() else "pending") counts: dict[tuple[str, str], int] = {} for item in recent_attempts: if item.get("success"): continue stage = str(item.get("stage") or "?").strip() or "?" signal = str(item.get("signal") or "").strip() key = (signal or stage, stage) counts[key] = counts.get(key, 0) + 1 recent_failure_hotspots = [ {"key": key, "stage": stage, "count": count} for (key, stage), count in sorted( counts.items(), key=lambda kv: (-kv[1], kv[0][0], kv[0][1]), )[:5] ] return { "name": "register", "status": status, "scheduler_mode": "burst", "threads_alive": 0, "threads_total": self.settings.register_batch_threads, "total_attempts": total_attempts, "total_success": total_success, "total_success_registered": total_success, "total_success_direct": total_success, "total_cpa_sync_success": total_cpa_sync_success, "total_cpa_sync_success_direct": total_cpa_sync_success, "total_cpa_sync_failure": total_cpa_sync_failure, "total_failure": total_failure, "success_rate": round(total_success / max(total_attempts, 1) * 100, 1), "registered_success_rate": round(total_success / max(total_attempts, 1) * 100, 1), "cpa_sync_success_rate": round(total_cpa_sync_success / max(total_attempts, 1) * 100, 1), "target_count": self.settings.register_batch_target_count, "target_reached": False, "last_error": last_error, "proxy": self.settings.register_proxy, "proxy_pool_enabled": bool(self.settings.proxy_pool_configured), "mail_provider": self.settings.register_mail_provider, "interval_seconds": self.settings.register_interval, "run_count": run_count, "success_count": total_success, "failure_count": total_failure, "is_running": False, "last_started_at": last_batch_started_at, "last_finished_at": last_batch_finished_at, "last_duration_seconds": last_batch_duration_seconds, "next_run_at": datetime.fromtimestamp(next_run_at_ts).isoformat(timespec="seconds") if next_run_at_ts else None, "failure_by_stage": failure_by_stage, "failure_signals": failure_signals, "recent_failure_hotspots": recent_failure_hotspots, "recent_attempts": recent_attempts, "active_domain_recent_attempts": [], "active_domain_failure_by_stage": {}, "active_domain_failure_signals": {}, "active_domain_recent_failure_hotspots": [], "cfmail_rotation": None, "cfmail_add_phone_stoploss": add_phone_stoploss, "cfmail_wait_otp_stoploss": wait_otp_stoploss, "batch_threads": self.settings.register_batch_threads, "batch_target_count": self.settings.register_batch_target_count, "batch_interval_seconds": self.settings.register_batch_interval_seconds, } def stop(self) -> None: self._stop_event.set() with self._lock: active_loop = self._active_loop if active_loop is not None: active_loop.stop() self._write_runtime_state() def run(self) -> None: self._next_run_at_ts = time.time() self._write_runtime_state() while not self._stop_event.is_set(): now = time.time() next_run_at_ts = self._next_run_at_ts or now if now < next_run_at_ts: self._write_runtime_state() self._stop_event.wait(min(max(next_run_at_ts - now, 1), 5)) continue batch_settings = replace( self.settings, register_enabled=True, register_threads=self.settings.register_batch_threads, register_target_count=self.settings.register_batch_target_count, ) loop = _compat_main_attr("RegistrationLoop", RegistrationLoop)(batch_settings) started_at = time.time() with self._lock: self._active_loop = loop self._active_started_at = started_at self._write_runtime_state() loop.start() try: while not self._stop_event.is_set(): snapshot = loop.snapshot() if int(snapshot.get("threads_alive") or 0) <= 0: break self._write_runtime_state() self._stop_event.wait(1) finally: loop.stop() batch_snapshot = loop.snapshot() with self._lock: self._active_loop = None self._active_started_at = None proxy_snapshot_fn = getattr(loop, "_proxy_pool_snapshot", None) if callable(proxy_snapshot_fn): self._last_proxy_pool_snapshot = proxy_snapshot_fn() self._absorb_batch_snapshot(batch_snapshot, duration_seconds=time.time() - started_at) self._next_run_at_ts = started_at + self.settings.register_batch_interval_seconds self._write_runtime_state() if self._stop_event.is_set(): break