"""Dashboard payload builders and runtime helpers for zhuce6.""" from __future__ import annotations from collections import deque from dataclasses import replace from datetime import date, datetime, time as datetime_time import json import math import os from pathlib import Path import sys import time from typing import Any from urllib.parse import urlsplit, urlunsplit try: from fastapi import FastAPI, HTTPException except ModuleNotFoundError: FastAPI = Any # type: ignore[assignment] class HTTPException(Exception): def __init__(self, status_code: int, detail: str = "") -> None: super().__init__(detail) self.status_code = status_code self.detail = detail from core.paths import DEFAULT_DASHBOARD_LOG_FILE from core.registry import list_platforms from core.settings import AppSettings from ops.account_survival import account_survival_once, load_account_survival_state, print_account_survival_summary from ops.common import CpaClient, create_backend_client from ops.responses_survival import ( load_responses_survival_state, print_responses_survival_summary, responses_survival_once, ) from ops.d1_cleanup import d1_cleanup_once from ops.rotate_log import rotate_log_tail as _rotate_log_tail from ops.service import RepeatedTask FREE_ACCOUNT_WEEKLY_TOKENS = max( 1, int(str(os.getenv("ZHUCE6_FREE_ACCOUNT_WEEKLY_TOKENS", "5000000")).strip() or "5000000"), ) OVERVIEW_CACHE_TTL_SECONDS = 30.0 def _cleanup_once(*args, **kwargs): # type: ignore[no-untyped-def] from ops.cleanup import cleanup_once return cleanup_once(*args, **kwargs) def _validate_once(*args, **kwargs): # type: ignore[no-untyped-def] from ops.validate import validate_once return validate_once(*args, **kwargs) def _print_validate_summary(*args, **kwargs): # type: ignore[no-untyped-def] from ops.validate import print_validate_summary return print_validate_summary(*args, **kwargs) def _rotate_once(*args, **kwargs): # type: ignore[no-untyped-def] from ops.rotate import rotate_once return rotate_once(*args, **kwargs) def _print_rotate_summary(*args, **kwargs): # type: ignore[no-untyped-def] from ops.rotate import print_rotate_summary return print_rotate_summary(*args, **kwargs) def _fetch_validate_management_auth_files(*args, **kwargs): # type: ignore[no-untyped-def] from ops import validate as validate_ops return validate_ops._fetch_management_auth_files(*args, **kwargs) # type: ignore[attr-defined] 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) def _invoke_count_cpa_files(fn: object, settings: AppSettings) -> int: return int(fn(settings)) # type: ignore[misc] def _build_background_tasks(settings: AppSettings) -> list[RepeatedTask]: tasks: list[RepeatedTask] = [] if settings.cleanup_enabled: tasks.append( RepeatedTask( "cleanup", lambda: _cleanup_once( client=create_backend_client(settings), proxy=settings.cleanup_proxy, management_base_url=settings.cpa_management_base_url, management_key=settings.cpa_management_key, pool_dir=settings.pool_dir, ), settings.cleanup_interval, ) ) if settings.d1_cleanup_enabled: tasks.append( RepeatedTask( "d1_cleanup", lambda: d1_cleanup_once( database_id=settings.d1_database_id, mail_retention_hours=settings.d1_mail_retention_hours, address_retention_hours=settings.d1_address_retention_hours, ), settings.d1_cleanup_interval, ) ) if settings.validate_enabled: tasks.append( RepeatedTask( "validate", lambda: _print_validate_summary( _validate_once( client=create_backend_client(settings), proxy=settings.validate_proxy, dry_run=False, max_workers=settings.validate_max_workers, pool_dir=settings.pool_dir, scope=settings.validate_scope, management_base_url=settings.cpa_management_base_url, management_key=settings.cpa_management_key, ) ), settings.validate_interval, ) ) if settings.rotate_enabled: tasks.append( RepeatedTask( "rotate", lambda: _print_rotate_summary( _rotate_once( pool_dir=settings.pool_dir, client=create_backend_client(settings), management_base_url=settings.cpa_management_base_url, cpa_management_key=settings.cpa_management_key, rotate_probe_workers=settings.rotate_probe_workers, fresh_grace_seconds=settings.rotate_fresh_grace_seconds, cpa_runtime_reconcile_enabled=settings.cpa_runtime_reconcile_enabled, cpa_runtime_reconcile_cooldown_seconds=settings.cpa_runtime_reconcile_cooldown_seconds, cpa_runtime_reconcile_restart_enabled=settings.cpa_runtime_reconcile_restart_enabled, ) ), settings.rotate_interval, ) ) if settings.account_survival_enabled: tasks.append( RepeatedTask( "account_survival", lambda: print_responses_survival_summary( responses_survival_once( pool_dir=settings.pool_dir, state_file=settings.responses_survival_state_file, cohort_size=settings.account_survival_cohort_size, proxy=settings.account_survival_proxy, timeout_seconds=settings.account_survival_timeout_seconds, settings=settings, require_provenance=settings.responses_survival_require_provenance, recent_window_seconds=settings.responses_survival_recent_window_seconds, warmup_min_age_seconds=settings.warmup_min_age_seconds, warmup_min_successful_probes=settings.warmup_min_successful_probes, ) ), settings.account_survival_interval, ) ) return tasks def _count_pool_files(pool_dir: Path) -> int: if not pool_dir.is_dir(): return 0 try: return sum(1 for path in pool_dir.iterdir() if path.is_file() and path.suffix == ".json") except Exception: return 0 def _count_cpa_files(settings: AppSettings) -> int: try: client = create_backend_client(settings) return len( [ entry for entry in getattr(client, "list_auth_files")() if "@" in str(entry.get("name") or "").strip() ] ) except Exception: return 0 def _fetch_management_auth_files(settings: AppSettings) -> tuple[bool, list[dict[str, object]]]: if settings.runtime_mode == "lite": return False, [] try: client = create_backend_client(settings) if not getattr(client, "health_check")(): return False, [] files = [ item for item in getattr(client, "list_auth_files")() if isinstance(item, dict) ] except Exception: return False, [] return True, files def _is_regular_free_account(item: dict[str, object]) -> bool: name = str(item.get("name") or "") if "@" not in name: return False id_token = item.get("id_token") or {} if isinstance(id_token, dict): plan_type = str(id_token.get("plan_type") or "").strip().lower() if plan_type: return plan_type == "free" return True def _classify_regular_account_status(item: dict[str, object]) -> str | None: if not _is_regular_free_account(item): return None status_message = str(item.get("status_message") or "") unavailable = bool(item.get("unavailable")) lowered_status = status_message.lower() if "unauthorized" in lowered_status or "invalidated" in lowered_status: return "invalid" if unavailable: if "usage_limit_reached" in lowered_status or item.get("next_retry_after"): return "waiting_reset" return "other" return "available" def _classify_regular_accounts(files: list[dict[str, object]], *, source_available: bool) -> dict[str, object]: stats: dict[str, object] = { "total": 0, "available": 0, "waiting_reset": 0, "invalid": 0, "other": 0, "source": "management", "source_available": source_available, "source_error": None if source_available else "management_data_unavailable", } if not source_available: return stats for item in files: status = _classify_regular_account_status(item) if status is None: continue stats["total"] = int(stats["total"]) + 1 stats[status] = int(stats[status]) + 1 return stats def _estimate_tokens(regular_accounts: dict[str, object]) -> dict[str, object]: available = int(regular_accounts.get("available") or 0) waiting_reset = int(regular_accounts.get("waiting_reset") or 0) relevant_accounts = available + waiting_reset source_available = bool(regular_accounts.get("source_available")) return { "per_account": FREE_ACCOUNT_WEEKLY_TOKENS, "available_now": available * FREE_ACCOUNT_WEEKLY_TOKENS, "available_with_reset": relevant_accounts * FREE_ACCOUNT_WEEKLY_TOKENS, "period": "weekly", "estimation_mode": "count_based", "baseline_source": "configured", "relevant_accounts": relevant_accounts, "matched_accounts": 0, "weighted_accounts": 0, "fallback_accounts": relevant_accounts, "fallback_reason": None if source_available else "missing_management_inventory", "snapshot_timestamp": None, "snapshot_age_seconds": None, "snapshot_fresh": False, } def _count_today_new(pool_dir: Path) -> int: if not pool_dir.is_dir(): return 0 try: today_start = datetime.combine(date.today(), datetime_time.min).timestamp() return sum( 1 for path in pool_dir.iterdir() if path.is_file() and path.suffix == ".json" and path.stat().st_mtime >= today_start ) except Exception: return 0 def _dashboard_overview_payload(app: FastAPI) -> dict[str, object]: cache = getattr(app.state, "dashboard_overview_cache", None) now_monotonic = time.monotonic() if isinstance(cache, dict): created_at = float(cache.get("created_at") or 0.0) cached_payload = cache.get("payload") if now_monotonic - created_at <= OVERVIEW_CACHE_TTL_SECONDS and isinstance(cached_payload, dict): return cached_payload settings: AppSettings = app.state.settings runtime = _runtime_payload(app) register_task = next((task for task in runtime["task_states"] if task.get("name") == "register"), {}) if settings.runtime_mode == "lite": cpa_count = None regular_accounts = None tokens = None observed_loss = None cpa_inventory = { "management_available": False, "count_source": "lite_mode", "auth_file_count": None, } else: fetch_management_auth_files = _compat_main_attr("_fetch_management_auth_files", _fetch_management_auth_files) count_cpa_files = _compat_main_attr("_count_cpa_files", _count_cpa_files) management_ok, auth_files = fetch_management_auth_files(settings) # type: ignore[misc] cpa_count = len(auth_files) if management_ok else _invoke_count_cpa_files(count_cpa_files, settings) regular_accounts = _classify_regular_accounts(auth_files, source_available=management_ok) tokens = _estimate_tokens(regular_accounts) observed_loss = int(regular_accounts.get("waiting_reset") or 0) + int(regular_accounts.get("invalid") or 0) cpa_inventory = { "management_available": management_ok, "count_source": "backend_api" if management_ok else "api_unavailable", "auth_file_count": cpa_count, } total_attempts = int(register_task.get("total_attempts") or 0) registered_success_total = int(register_task.get("total_success_registered") or register_task.get("total_success") or 0) cpa_sync_success_total = int(register_task.get("total_cpa_sync_success") or 0) cpa_sync_failure_total = int(register_task.get("total_cpa_sync_failure") or 0) payload = { "generated_at": datetime.now().isoformat(timespec="seconds"), "pool_count": runtime["pool_count"], "cpa_count": cpa_count, "cpa_inventory": cpa_inventory, "regular_accounts": regular_accounts, "tokens": tokens, "today_new": _compat_main_attr("_count_today_new", _count_today_new)(settings.pool_dir), # type: ignore[misc] "success_rate": register_task.get("success_rate") if total_attempts > 0 else None, "registered_success_total": registered_success_total, "cpa_sync_success_total": cpa_sync_success_total, "cpa_sync_failure_total": cpa_sync_failure_total, "registered_success_rate": round(registered_success_total / max(total_attempts, 1) * 100, 1) if total_attempts > 0 else None, "cpa_sync_success_rate": round(cpa_sync_success_total / max(total_attempts, 1) * 100, 1) if total_attempts > 0 else None, "burn_rate": None, "observed_loss": observed_loss, } app.state.dashboard_overview_cache = { "created_at": now_monotonic, "payload": payload, } return payload def _recent_pool_files(pool_dir: Path, limit: int = 8) -> list[dict[str, object]]: if not pool_dir.is_dir(): return [] try: normalized_limit = max(1, int(limit)) files = [ path for path in pool_dir.iterdir() if path.is_file() and path.suffix == ".json" ] files.sort(key=lambda item: item.stat().st_mtime, reverse=True) except Exception: return [] out: list[dict[str, object]] = [] for path in files[:normalized_limit]: try: stat = path.stat() out.append({ "name": path.name, "path": str(path), "size_bytes": stat.st_size, "modified_at": stat.st_mtime, "modified_at_iso": datetime.fromtimestamp(stat.st_mtime).isoformat(timespec="seconds"), }) except OSError: continue return out def _register_log_tail(settings: AppSettings, limit: int = 80) -> dict[str, object]: log_path_raw = str(settings.register_log_file or "").strip() if not log_path_raw: return { "available": False, "path": "", "updated_at": None, "updated_at_iso": None, "error": "register log file not configured", "lines": [], } log_path = Path(log_path_raw).expanduser() if not log_path.exists(): return { "available": False, "path": str(log_path), "updated_at": None, "updated_at_iso": None, "error": "register log file not found", "lines": [], } try: with log_path.open("r", encoding="utf-8", errors="replace") as fh: lines = deque((line.rstrip("\r\n") for line in fh), maxlen=limit) stat = log_path.stat() except OSError as exc: return { "available": False, "path": str(log_path), "updated_at": None, "updated_at_iso": None, "error": str(exc), "lines": [], } return { "available": True, "path": str(log_path), "updated_at": stat.st_mtime, "updated_at_iso": datetime.fromtimestamp(stat.st_mtime).isoformat(timespec="seconds"), "error": None, "lines": list(lines), } def _runtime_state_file_meta(settings: AppSettings) -> dict[str, object]: state_file = Path(settings.runtime_state_file) if not state_file.exists(): return { "exists": False, "path": str(state_file), "updated_at": None, "updated_at_iso": None, } stat = state_file.stat() return { "exists": True, "path": str(state_file), "updated_at": stat.st_mtime, "updated_at_iso": datetime.fromtimestamp(stat.st_mtime).isoformat(timespec="seconds"), } def _format_duration_hms(value: object) -> str | None: try: total = int(value) # type: ignore[arg-type] except Exception: return None if total < 0: total = 0 hours, remainder = divmod(total, 3600) minutes, seconds = divmod(remainder, 60) parts: list[str] = [] if hours > 0: parts.append(f"{hours}h") if hours > 0 or minutes > 0: parts.append(f"{minutes}m") parts.append(f"{seconds}s") return " ".join(parts) def _attach_survival_duration_fields(payload: dict[str, object]) -> dict[str, object]: result = dict(payload) members = result.get("members") if isinstance(members, list): enriched_members: list[dict[str, object]] = [] for item in members: if not isinstance(item, dict): continue member = dict(item) survival_text = _format_duration_hms(member.get("survival_seconds")) if survival_text is not None: member["survival_text"] = survival_text enriched_members.append(member) result["members"] = enriched_members changes = result.get("changes") if isinstance(changes, list): enriched_changes: list[dict[str, object]] = [] for item in changes: if not isinstance(item, dict): continue change = dict(item) survival_text = _format_duration_hms(change.get("survival_seconds")) if survival_text is not None: change["survival_text"] = survival_text enriched_changes.append(change) result["changes"] = enriched_changes return result def _latest_fresh_unauthorized_state(state_dir: Path) -> dict[str, object]: candidates = sorted( state_dir.glob("track_new8_unauthorized*.json"), key=lambda path: path.stat().st_mtime, reverse=True, ) for path in candidates: try: payload = json.loads(path.read_text(encoding="utf-8")) except Exception: continue if isinstance(payload, dict): payload = dict(payload) payload["path"] = str(path) payload["updated_at_iso"] = datetime.fromtimestamp(path.stat().st_mtime).isoformat(timespec="seconds") return payload return {} def _fresh_unauthorized_experiment_payload(settings: AppSettings) -> dict[str, object]: payload = _latest_fresh_unauthorized_state(settings.state_dir) if not payload: return { "available": False, "path": "", "summary": { "tracked": 0, "first_401_count": 0, "completed": 0, "pending": 0, }, "members": [], } members_raw = payload.get("members") enriched_members: list[dict[str, object]] = [] first_401_count = 0 completed = 0 if isinstance(members_raw, list): for item in members_raw: if not isinstance(item, dict): continue member = dict(item) first_401_text = _format_duration_hms(member.get("first_401_seconds")) if first_401_text is not None: member["first_401_text"] = first_401_text if str(member.get("first_401_at") or "").strip(): first_401_count += 1 completed += 1 enriched_members.append(member) payload["members"] = enriched_members payload["available"] = True payload["summary"] = { "tracked": len(enriched_members), "first_401_count": first_401_count, "completed": completed, "pending": max(0, len(enriched_members) - completed), } return payload def _derive_survival_promotion_stats(payload: dict[str, object]) -> dict[str, int]: members = payload.get("members") if not isinstance(members, list): return { "promoted_success_total": 0, "promoted_failure_total": 0, } success_total = 0 failure_total = 0 for item in members: if not isinstance(item, dict): continue path_raw = str(item.get("path") or "").strip() if not path_raw: continue try: record = json.loads(Path(path_raw).read_text(encoding="utf-8")) except Exception: continue if not isinstance(record, dict) or not bool(record.get("warmup_required")): continue status = str(record.get("cpa_sync_status") or "").strip().lower() if status == "synced": success_total += 1 elif status == "failed": failure_total += 1 return { "promoted_success_total": success_total, "promoted_failure_total": failure_total, } def _account_survival_payload(settings: AppSettings) -> dict[str, object]: responses_state_file = Path(settings.responses_survival_state_file) responses_payload = load_responses_survival_state(responses_state_file) if responses_payload: payload = _attach_survival_duration_fields(dict(responses_payload)) payload["promotion_stats"] = _derive_survival_promotion_stats(payload) payload["fresh_unauthorized_experiment"] = _fresh_unauthorized_experiment_payload(settings) payload["enabled"] = settings.account_survival_enabled payload["available"] = True payload["path"] = str(responses_state_file) payload.setdefault("probe_mode", "responses") return payload state_file = Path(settings.account_survival_state_file) payload = load_account_survival_state(state_file) if not payload: return { "enabled": settings.account_survival_enabled, "available": False, "path": str(state_file), "error": "account survival state file not found", } payload = _attach_survival_duration_fields(dict(payload)) payload["fresh_unauthorized_experiment"] = _fresh_unauthorized_experiment_payload(settings) payload["enabled"] = settings.account_survival_enabled payload["available"] = True payload["path"] = str(state_file) return payload def _responses_survival_promotion_stats(settings: AppSettings) -> dict[str, int]: payload = load_responses_survival_state(Path(settings.responses_survival_state_file)) stats = payload.get("promotion_stats") if isinstance(payload, dict) else None if not isinstance(stats, dict): return { "promoted_success_total": 0, "promoted_failure_total": 0, } return { "promoted_success_total": int(stats.get("promoted_success_total") or 0), "promoted_failure_total": int(stats.get("promoted_failure_total") or 0), } def _parse_runtime_timestamp(value: object) -> datetime | None: raw = str(value or "").strip() if not raw: return None try: parsed = datetime.fromisoformat(raw) except Exception: return None if parsed.tzinfo is None: return parsed.astimezone() return parsed def _count_runtime_warmup_promotions(settings: AppSettings, *, runtime_started_at: object) -> int: started_at = _parse_runtime_timestamp(runtime_started_at) if started_at is None or not settings.pool_dir.is_dir(): return 0 total = 0 for path in settings.pool_dir.iterdir(): if not path.is_file() or path.suffix != ".json": continue try: payload = json.loads(path.read_text(encoding="utf-8")) except Exception: continue if not isinstance(payload, dict): continue if not bool(payload.get("warmup_required")): continue if str(payload.get("cpa_sync_status") or "").strip().lower() != "synced": continue created_at = _parse_runtime_timestamp(payload.get("created_at")) if created_at is None: try: created_at = datetime.fromtimestamp(path.stat().st_mtime).astimezone() except Exception: continue if created_at >= started_at: total += 1 return total def _count_runtime_current_warmup_backlog(settings: AppSettings, *, runtime_started_at: object) -> int: started_at = _parse_runtime_timestamp(runtime_started_at) if started_at is None or not settings.pool_dir.is_dir(): return 0 total = 0 for path in settings.pool_dir.iterdir(): if not path.is_file() or path.suffix != ".json": continue try: payload = json.loads(path.read_text(encoding="utf-8")) except Exception: continue if not isinstance(payload, dict): continue if str(payload.get("cpa_sync_status") or "").strip().lower() != "warmup_pending": continue created_at = _parse_runtime_timestamp(payload.get("created_at")) if created_at is None: try: created_at = datetime.fromtimestamp(path.stat().st_mtime).astimezone() except Exception: continue if created_at >= started_at: total += 1 return total def _apply_warmup_promotion_metrics(task_snapshots: list[dict[str, object]], settings: AppSettings) -> list[dict[str, object]]: updated_snapshots: list[dict[str, object]] = [] for snapshot in task_snapshots: if not isinstance(snapshot, dict) or snapshot.get("name") != "register": updated_snapshots.append(snapshot) continue current = dict(snapshot) promoted_success_total = _count_runtime_warmup_promotions( settings, runtime_started_at=current.get("last_started_at"), ) total_attempts = int(current.get("total_attempts") or 0) direct_success_total = int(current.get("total_success_direct") or current.get("total_success_registered") or current.get("total_success") or 0) direct_cpa_sync_total = int(current.get("total_cpa_sync_success_direct") or current.get("total_cpa_sync_success") or 0) effective_success_total = direct_success_total + promoted_success_total effective_cpa_sync_total = direct_cpa_sync_total + promoted_success_total threads_total = int(current.get("threads_total") or 0) retry_sidecar_threads = 1 if threads_total > 0 and isinstance(current.get("pending_token_queue"), dict) else 0 register_worker_threads = max(0, threads_total - retry_sidecar_threads) current_warmup_backlog = _count_runtime_current_warmup_backlog( settings, runtime_started_at=current.get("last_started_at"), ) current["total_success_direct"] = direct_success_total current["total_success_promoted"] = promoted_success_total current["total_success"] = effective_success_total current["total_success_registered"] = effective_success_total current["total_cpa_sync_success_direct"] = direct_cpa_sync_total current["total_cpa_sync_success"] = effective_cpa_sync_total current["register_worker_threads"] = register_worker_threads current["retry_sidecar_threads"] = retry_sidecar_threads current["current_warmup_backlog"] = current_warmup_backlog current["success_rate"] = round(effective_success_total / max(total_attempts, 1) * 100, 1) if total_attempts > 0 else 0.0 current["registered_success_rate"] = round(effective_success_total / max(total_attempts, 1) * 100, 1) if total_attempts > 0 else 0.0 current["cpa_sync_success_rate"] = round(effective_cpa_sync_total / max(total_attempts, 1) * 100, 1) if total_attempts > 0 else 0.0 updated_snapshots.append(current) return updated_snapshots def _task_snapshots(background_tasks: list[RepeatedTask], registration_loop: RegistrationLoop | None = None) -> list[dict[str, object]]: snapshots = [task.snapshot() for task in background_tasks] if registration_loop: snapshots.append(registration_loop.snapshot()) return snapshots def _external_runtime_state(settings: AppSettings) -> dict[str, object] | None: state_file = Path(settings.runtime_state_file) if not state_file.is_file(): return None try: payload = json.loads(state_file.read_text(encoding="utf-8")) except Exception: return None if not isinstance(payload, dict): return None return payload def _proxy_pool_payload( settings: AppSettings, registration_loop: RegistrationLoop | None = None, ) -> dict[str, object]: pool = getattr(registration_loop, "_proxy_pool", None) if registration_loop is not None else None if pool is None: external = _external_runtime_state(settings) proxy_pool = external.get("proxy_pool") if isinstance(external, dict) else None if isinstance(proxy_pool, dict): return 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(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 _runtime_payload(app: FastAPI) -> dict[str, object]: runtime_settings: AppSettings = app.state.settings background_tasks = getattr(app.state, "background_tasks", []) registration_loop = getattr(app.state, "registration_loop", None) task_snapshots = _task_snapshots(background_tasks, registration_loop) if registration_loop is None: external = _external_runtime_state(runtime_settings) register_snapshot = external.get("register_snapshot") if isinstance(external, dict) else None if isinstance(register_snapshot, dict): task_snapshots.append(register_snapshot) task_snapshots = _apply_warmup_promotion_metrics(task_snapshots, runtime_settings) return { "runtime_mode": runtime_settings.runtime_mode, "architecture": "single-process-fastapi" if registration_loop is not None else "split-runtime-fastapi+loop", "cleanup_enabled": runtime_settings.cleanup_enabled, "validate_enabled": runtime_settings.validate_enabled, "cleanup_interval": runtime_settings.cleanup_interval, "validate_interval": runtime_settings.validate_interval, "validate_scope": runtime_settings.validate_scope, "pool_dir": str(runtime_settings.pool_dir), "pool_count": _count_pool_files(runtime_settings.pool_dir), "backend": runtime_settings.backend, "cpa_management_base_url": runtime_settings.cpa_management_base_url, "account_survival_enabled": runtime_settings.account_survival_enabled, "account_survival_interval": runtime_settings.account_survival_interval, "account_survival_cohort_size": runtime_settings.account_survival_cohort_size, "account_survival_state_file": str(runtime_settings.account_survival_state_file), "rotate_enabled": runtime_settings.rotate_enabled, "rotate_interval": runtime_settings.rotate_interval, "rotate_fresh_grace_seconds": runtime_settings.rotate_fresh_grace_seconds, "register_fresh_proxy_regions": list(runtime_settings.register_fresh_proxy_regions), "responses_survival_recent_window_seconds": runtime_settings.responses_survival_recent_window_seconds, "responses_survival_require_provenance": runtime_settings.responses_survival_require_provenance, "warmup_min_age_seconds": runtime_settings.warmup_min_age_seconds, "warmup_min_successful_probes": runtime_settings.warmup_min_successful_probes, "registered_tasks": [task["name"] for task in task_snapshots], "task_states": task_snapshots, "proxy_pool": _proxy_pool_payload(runtime_settings, registration_loop), } def _register_burst_plan_payload(settings: AppSettings) -> dict[str, object]: interval_seconds = max(60, int(settings.register_batch_interval_seconds)) target_count = max(1, int(settings.register_batch_target_count)) batches_per_day = max(1, math.floor(86400 / interval_seconds)) accounts_per_day = target_count * batches_per_day return { "mode": "burst", "threads": max(1, int(settings.register_batch_threads)), "target_count": target_count, "interval_seconds": interval_seconds, "accounts_per_day": accounts_per_day, "accounts_needed_for_one_day_target": target_count, "accounts_needed_for_sustained_daily_target": max(accounts_per_day - target_count, 0), } def _summary_payload(app: FastAPI) -> dict[str, object]: runtime = _runtime_payload(app) settings: AppSettings = app.state.settings overview = _dashboard_overview_payload(app) register_task = next((task for task in runtime["task_states"] if task.get("name") == "register"), {}) rotate_task = next((task for task in runtime["task_states"] if task.get("name") == "rotate"), {}) account_survival = _account_survival_payload(settings) rotate_log_tail = _compat_main_attr("_rotate_log_tail", _rotate_log_tail)() return { "project": "zhuce6", "generated_at": overview["generated_at"], "runtime": runtime, "platforms": list_platforms(), "pool_count": overview["pool_count"], "cpa_count": overview["cpa_count"], "cpa_inventory": overview["cpa_inventory"], "regular_accounts": overview["regular_accounts"], "tokens": overview["tokens"], "today_new": overview["today_new"], "success_rate": overview["success_rate"], "registered_success_total": overview["registered_success_total"], "cpa_sync_success_total": overview["cpa_sync_success_total"], "cpa_sync_failure_total": overview["cpa_sync_failure_total"], "registered_success_rate": overview["registered_success_rate"], "cpa_sync_success_rate": overview["cpa_sync_success_rate"], "burn_rate": overview["burn_rate"], "observed_loss": overview["observed_loss"], "register_failure_by_stage": register_task.get("failure_by_stage") or {}, "register_failure_signals": register_task.get("failure_signals") or {}, "register_recent_failure_hotspots": register_task.get("recent_failure_hotspots") or [], "register_recent_attempts": register_task.get("recent_attempts") or [], "register_cfmail_domain_pool": register_task.get("cfmail_domain_pool") or {}, "register_cfmail_add_phone_stoploss": register_task.get("cfmail_add_phone_stoploss") or {}, "register_cfmail_wait_otp_stoploss": register_task.get("cfmail_wait_otp_stoploss") or {}, "register_burst_plan": _register_burst_plan_payload(settings), "rotate_task": rotate_task, "rotate_log_tail": rotate_log_tail, "rotate_latest_summary": rotate_log_tail.get("latest_summary"), "rotate_current_summary": rotate_log_tail.get("current_summary"), "account_survival": account_survival, "runtime_state_file": _runtime_state_file_meta(settings), "recent_pool_files": _recent_pool_files(Path(str(runtime["pool_dir"]))), "register_log_tail": _register_log_tail(settings), "routes": { "healthz": "/healthz", "platforms": "/api/platforms", "runtime": "/api/runtime", "summary": "/api/summary", "settings": "/api/settings", "health_dependencies": "/api/health/dependencies", "register_control": "/api/control/register", "account_survival": "/api/account-survival", "chatgpt_preflight": "/api/register/chatgpt/preflight", "chatgpt_register_once": "/api/register/chatgpt/run", "chatgpt_callback_exchange": "/api/register/chatgpt/callback-exchange", "zhuce6": "/zhuce6", }, "commands": { "start": "uv run python main.py --mode full", "chatgpt_preflight": "uv run python scripts/chatgpt_preflight.py --json", "chatgpt_register_once": "uv run python scripts/chatgpt_register_once.py --json --mail-provider cfmail", "chatgpt_callback_exchange": "uv run python scripts/chatgpt_exchange_callback.py --json --callback-url '' --state '' --code-verifier ''", "cleanup_once": "uv run python -m ops.cleanup --once", "validate_used_dry_run": "uv run python -m ops.validate --scope used --dry-run --once", "validate_all_dry_run": "uv run python -m ops.validate --scope all --dry-run --once --limit 20", "scan_local_pool": "uv run python -m ops.scan --limit 20", "update_priority_dry_run": "uv run python -m ops.update_priority --dry-run --limit 20", }, "manual_test": [ "Start the service and visit /zhuce6.", "Run scripts/chatgpt_preflight.py with a working mailbox provider and network.", "Run scripts/chatgpt_register_once.py with a working mailbox provider, proxy, and upstream availability if you want a full attempt.", "Complete the OAuth login in a browser, then run scripts/chatgpt_exchange_callback.py to write a pool file.", "Run ops.cleanup / ops.validate only when backend API is reachable.", "Live CPA invalid account cleanup remains manual_test and should be checked via quota probe plus rotate summary.", ], } def _cpa_management_root(settings: AppSettings) -> str: parsed = urlsplit(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 _settings_payload(app: FastAPI) -> dict[str, object]: settings: AppSettings = app.state.settings registration_loop = getattr(app.state, "registration_loop", None) missing_cfmail = settings.validate_cfmail_env() return { "mode": settings.runtime_mode, "register": { "enabled": bool(registration_loop is not None or settings.register_enabled), "threads": settings.register_threads, "batch_target_count": settings.register_batch_target_count, "batch_interval_seconds": settings.register_batch_interval_seconds, "mail_provider": settings.register_mail_provider, "proxy": settings.register_proxy, "fresh_proxy_regions": ",".join(settings.register_fresh_proxy_regions), }, "proxy_pool": { "enabled": settings.enable_proxy_pool, "size": settings.proxy_pool_size, "config_path": str(settings.proxy_pool_config) if settings.proxy_pool_config else "", "direct_urls": settings.proxy_pool_direct_urls, "regions": ",".join(settings.proxy_pool_regions), }, "cfmail": { "configured": len(missing_cfmail) == 0, "zone_name": str(os.getenv("ZHUCE6_CFMAIL_ZONE_NAME", "")).strip(), "worker_name": str(os.getenv("ZHUCE6_CFMAIL_WORKER_NAME", "")).strip(), "rotation_window": settings.cfmail_rotation_window, "rotation_blacklist_threshold": settings.cfmail_rotation_blacklist_threshold, }, "cpa": { "configured": settings.runtime_mode != "lite" and settings.backend == "cpa", "backend": settings.backend, "management_url": _cpa_management_root(settings), "rotate_enabled": settings.rotate_enabled, "rotate_interval": settings.rotate_interval, "rotate_fresh_grace_seconds": settings.rotate_fresh_grace_seconds, }, "survival": { "recent_window_seconds": settings.responses_survival_recent_window_seconds, "require_provenance": settings.responses_survival_require_provenance, "warmup_min_age_seconds": settings.warmup_min_age_seconds, "warmup_min_successful_probes": settings.warmup_min_successful_probes, }, } def _encode_env_value(value: object) -> str: text = "" if value is None else str(value) if not text: return "" if any(ch.isspace() for ch in text) or "#" in text: return json.dumps(text) return text def _persist_env_updates(path: Path, updates: dict[str, object]) -> None: existing_lines = path.read_text(encoding="utf-8").splitlines() if path.exists() else [] normalized_updates = {key: _encode_env_value(value) for key, value in updates.items()} handled: set[str] = set() output_lines: list[str] = [] for line in existing_lines: stripped = line.strip() candidate = stripped[7:] if stripped.startswith("export ") else stripped key, sep, _value = candidate.partition("=") if sep and key in normalized_updates: if key in handled: continue output_lines.append(f"{key}={normalized_updates[key]}") handled.add(key) continue output_lines.append(line) for key, value in normalized_updates.items(): if key not in handled: output_lines.append(f"{key}={value}") path.parent.mkdir(parents=True, exist_ok=True) path.write_text("\n".join(output_lines).rstrip() + "\n", encoding="utf-8") def _parse_settings_patch(changes: dict[str, object]) -> tuple[dict[str, object], dict[str, object]]: updates: dict[str, object] = {} env_updates: dict[str, object] = {} def parse_regions(value: object) -> tuple[str, ...]: return tuple(part.strip().lower() for part in str(value or "").split(",") if part.strip()) allowed: dict[str, tuple[str, str, object]] = { "register.threads": ("register_threads", "ZHUCE6_REGISTER_THREADS", lambda value: max(1, int(value))), "register.batch_target_count": ( "register_batch_target_count", "ZHUCE6_REGISTER_BATCH_TARGET_COUNT", lambda value: max(1, int(value)), ), "register.batch_interval_seconds": ( "register_batch_interval_seconds", "ZHUCE6_REGISTER_BATCH_INTERVAL_SECONDS", lambda value: max(60, int(value)), ), "register.mail_provider": ( "register_mail_provider", "ZHUCE6_REGISTER_MAIL_PROVIDER", lambda value: str(value or "").strip() or "cfmail", ), "register.proxy": ("register_proxy", "ZHUCE6_REGISTER_PROXY", lambda value: str(value or "").strip()), "register.fresh_proxy_regions": ( "register_fresh_proxy_regions", "ZHUCE6_REGISTER_FRESH_PROXY_REGIONS", parse_regions, ), "proxy_pool.size": ("proxy_pool_size", "ZHUCE6_PROXY_POOL_SIZE", lambda value: max(1, int(value))), "proxy_pool.direct_urls": ( "proxy_pool_direct_urls", "ZHUCE6_PROXY_POOL_DIRECT_URLS", lambda value: str(value or "").strip(), ), "proxy_pool.regions": ("proxy_pool_regions", "ZHUCE6_PROXY_POOL_REGIONS", parse_regions), "cpa.rotate_interval": ("rotate_interval", "ZHUCE6_ROTATE_INTERVAL", lambda value: max(1, int(value))), "cpa.rotate_fresh_grace_seconds": ( "rotate_fresh_grace_seconds", "ZHUCE6_ROTATE_FRESH_GRACE_SECONDS", lambda value: max(0, int(value)), ), "survival.recent_window_seconds": ( "responses_survival_recent_window_seconds", "ZHUCE6_RESPONSES_SURVIVAL_RECENT_WINDOW_SECONDS", lambda value: max(0, int(value)), ), "survival.require_provenance": ( "responses_survival_require_provenance", "ZHUCE6_RESPONSES_SURVIVAL_REQUIRE_PROVENANCE", lambda value: str(value or "").strip().lower() in {"1", "true", "yes", "on"}, ), "survival.warmup_min_age_seconds": ( "warmup_min_age_seconds", "ZHUCE6_WARMUP_MIN_AGE_SECONDS", lambda value: max(0, int(value)), ), "survival.warmup_min_successful_probes": ( "warmup_min_successful_probes", "ZHUCE6_WARMUP_MIN_SUCCESSFUL_PROBES", lambda value: max(1, int(value)), ), } for key, value in changes.items(): spec = allowed.get(key) if spec is None: raise HTTPException(status_code=400, detail=f"unsupported setting: {key}") field_name, env_name, parser = spec parsed_value = parser(value) updates[field_name] = parsed_value if isinstance(parsed_value, tuple): env_updates[env_name] = ",".join(str(item) for item in parsed_value) else: env_updates[env_name] = parsed_value return updates, env_updates def _cfmail_dependency_payload(settings: AppSettings) -> dict[str, object]: if "cfmail" not in {part.strip() for part in settings.register_mail_provider.split(",") if part.strip()}: return {"status": "unconfigured", "detail": "register_mail_provider_not_cfmail"} missing = settings.validate_cfmail_env() if missing: return {"status": "unconfigured", "detail": f"missing: {', '.join(missing)}"} return {"status": "ok", "detail": "configuration_present"} def _proxy_pool_dependency_payload(app: FastAPI) -> dict[str, object]: settings: AppSettings = app.state.settings if not settings.enable_proxy_pool: return {"status": "unconfigured", "detail": "proxy_pool_disabled", "active_nodes": 0, "total_nodes": 0} if not settings.proxy_pool_configured: return {"status": "unconfigured", "detail": "proxy_pool_not_configured", "active_nodes": 0, "total_nodes": 0} proxy_pool = _proxy_pool_payload(settings, getattr(app.state, "registration_loop", None)) total_nodes = int(proxy_pool.get("node_count") or 0) active_nodes = max(0, total_nodes - int(proxy_pool.get("disabled_count") or 0)) snapshot_error = str(proxy_pool.get("snapshot_error") or "").strip() if snapshot_error: return { "status": "error", "detail": snapshot_error, "active_nodes": active_nodes, "total_nodes": total_nodes, } return { "status": "ok" if total_nodes > 0 else "error", "detail": "ok" if total_nodes > 0 else "no_proxy_nodes", "active_nodes": active_nodes, "total_nodes": total_nodes, } def _cpa_dependency_payload(settings: AppSettings) -> dict[str, object]: if settings.runtime_mode == "lite": return { "status": "unconfigured", "management_reachable": False, } if settings.backend == "sub2api": return { "status": "unconfigured", "management_reachable": False, } management_reachable = False try: management_reachable = CpaClient.from_settings(settings).health_check() except Exception: management_reachable = False return { "status": "ok" if management_reachable else "error", "management_reachable": management_reachable, } def _sub2api_dependency_payload(settings: AppSettings) -> dict[str, object]: if settings.runtime_mode == "lite": return {"status": "unconfigured", "error": "lite_mode", "auth_configured": False} if settings.backend != "sub2api": return {"status": "unconfigured", "error": "backend_cpa", "auth_configured": False} auth_configured = bool(settings.sub2api_api_key or (settings.sub2api_admin_email and settings.sub2api_admin_password)) if not auth_configured: return {"status": "error", "error": "missing_auth", "auth_configured": False} reachable = False try: reachable = bool(create_backend_client(settings).health_check()) except Exception: reachable = False return { "status": "ok" if reachable else "error", "error": None if reachable else "unreachable", "auth_configured": True, "base_url": settings.sub2api_base_url, } build_background_tasks = _build_background_tasks count_pool_files = _count_pool_files count_cpa_files = _count_cpa_files fetch_management_auth_files = _fetch_management_auth_files is_regular_free_account = _is_regular_free_account classify_regular_account_status = _classify_regular_account_status classify_regular_accounts = _classify_regular_accounts estimate_tokens = _estimate_tokens count_today_new = _count_today_new dashboard_overview_payload = _dashboard_overview_payload recent_pool_files = _recent_pool_files register_log_tail = _register_log_tail runtime_state_file_meta = _runtime_state_file_meta account_survival_payload = _account_survival_payload task_snapshots = _task_snapshots external_runtime_state = _external_runtime_state proxy_pool_payload = _proxy_pool_payload runtime_payload = _runtime_payload register_burst_plan_payload = _register_burst_plan_payload summary_payload = _summary_payload settings_payload = _settings_payload cpa_management_root = _cpa_management_root encode_env_value = _encode_env_value persist_env_updates = _persist_env_updates parse_settings_patch = _parse_settings_patch cfmail_dependency_payload = _cfmail_dependency_payload proxy_pool_dependency_payload = _proxy_pool_dependency_payload cpa_dependency_payload = _cpa_dependency_payload sub2api_dependency_payload = _sub2api_dependency_payload