codex-console-automation

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

codex-console Automation Skill

codex-console 自动化技能

Skill by ara.so — Codex Skills collection.

由ara.so开发的技能——Codex Skills合集。

Overview

概述

codex-console is a comprehensive automation platform for managing OpenAI accounts at scale. It handles registration, login, token extraction, subscription management, payment binding, auto-replenishment, and integration with API gateways like CPA, Sub2API, Team Manager, and New-API.

This is a maintained fork that fixes compatibility issues with OpenAI's evolving authentication flow, including Sentinel POW solving, split registration/login flows, and improved OTP handling.

Key capabilities:

Web UI for task management and monitoring
Batch account registration with email service integration
Semi-automated payment card binding with 3DS support
Automated token refresh and upload to API gateways
Self-check and repair system
Auto-replenishment based on inventory levels
Tag-based account pooling and team management
SQLite or PostgreSQL backend

codex-console是一个用于大规模管理OpenAI账户的综合性自动化平台。它支持注册、登录、令牌提取、订阅管理、支付绑定、自动补货，以及与CPA、Sub2API、Team Manager和New-API等API网关集成。

这是一个维护版本的分支，修复了与OpenAI不断演变的认证流程的兼容性问题，包括Sentinel POW求解、拆分式注册/登录流程优化，以及改进的OTP处理机制。

核心功能：

用于任务管理和监控的Web UI
集成邮件服务的批量账户注册
支持3DS验证的半自动化支付卡绑定
自动化令牌刷新及API网关上传
自检与修复系统
基于库存水平的自动补货
基于标签的账户池化与团队管理
支持SQLite或PostgreSQL后端

Installation

安装

Using Python (Recommended for Development)

使用Python（推荐用于开发环境）

bash

undefined

bash

undefined

Clone the repository

git clone https://github.com/dou-jiang/codex-console.git cd codex-console

Install dependencies with uv (recommended)

uv sync

Or use pip

pip install -r requirements.txt

undefined

pip install -r requirements.txt

undefined

Using Docker

使用Docker

bash

undefined

bash

undefined

With docker-compose

docker-compose up -d

Or with docker run

docker run -d
-p 1455:1455
-p 6080:6080
-e WEBUI_HOST=0.0.0.0
-e WEBUI_PORT=1455
-e WEBUI_ACCESS_PASSWORD=your_secure_password
-v $(pwd)/data:/app/data
--name codex-console
ghcr.io/dou-jiang/codex-console:latest

undefined

undefined

Building Standalone Executable

构建独立可执行文件

bash

undefined

bash

undefined

Windows

build.bat

Linux/macOS

bash build.sh


The executable will be in `dist/codex-console-windows-X64.exe` or equivalent.

bash build.sh


可执行文件将生成在`dist/codex-console-windows-X64.exe`或对应平台的目录下。

Configuration

配置

Environment Variables

环境变量

Create

.env

from template:

bash

cp .env.example .env

Key variables:

bash

undefined

从模板创建

.env

文件：

bash

cp .env.example .env

关键变量：

bash

undefined

Server

APP_HOST=0.0.0.0 APP_PORT=8000 APP_ACCESS_PASSWORD=admin123

Database (SQLite by default)

APP_DATABASE_URL=data/database.db

Or PostgreSQL

APP_DATABASE_URL=postgresql://user:password@host:5432/dbname

Logging

LOG_LEVEL=info

Debug mode

DEBUG=false


**Priority:** CLI args > `.env` > database settings > defaults

DEBUG=false


**优先级：** CLI参数 > `.env`文件 > 数据库设置 > 默认值

First-Time Setup

首次设置

Start the web UI:

bash

python webui.py --access-password mypassword

Access at
```
http://127.0.0.1:8000
```
Configure in Settings page:
- Email service credentials
- Proxy settings
- Upload targets (CPA/Sub2API/New-API)
- Payment automation options

启动Web UI：

bash

python webui.py --access-password mypassword

访问地址：
```
http://127.0.0.1:8000
```
在设置页面配置：
- 邮件服务凭证
- 代理设置
- 上传目标（CPA/Sub2API/New-API）
- 支付自动化选项

Core Usage Patterns

核心使用模式

Starting the Web UI

启动Web UI

python

undefined

python

undefined

webui.py - main entry point

import uvicorn from src.web.app import app from src.utils.settings import get_settings

if name == "main": settings = get_settings() uvicorn.run( app, host=settings.app_host, port=settings.app_port, log_level=settings.log_level.lower() )


CLI options:

```bash

import uvicorn from src.web.app import app from src.utils.settings import get_settings

if name == "main": settings = get_settings() uvicorn.run( app, host=settings.app_host, port=settings.app_port, log_level=settings.log_level.lower() )


CLI选项：

```bash

Default

python webui.py

Custom host/port

python webui.py --host 0.0.0.0 --port 8080

Set password

python webui.py --access-password mypassword

Debug mode (hot reload)

python webui.py --debug

undefined

python webui.py --debug

undefined

Email Service Integration

邮件服务集成

codex-console supports multiple email providers:

python

undefined

codex-console支持多种邮件服务商：

python

undefined

src/services/email_service.py

from src.services.email_service import EmailServiceFactory

Configure in database or code

email_config = { "service_type": "cloudmail", # or luckmail, yydsmail, outlook "api_key": None, # from env: CLOUDMAIL_API_KEY "config": { "base_url": "https://api.cloudmail.com", "timeout": 30 } }

Factory creates appropriate service

service = EmailServiceFactory.create(email_config)

Fetch verification code

otp = await service.get_verification_code( email="test@example.com", timeout=60 )


**Supported services:**
- CloudMail (API-based)
- LuckMail (API-based)
- YYDS Mail (API-based)
- Outlook (self-hosted accounts)

otp = await service.get_verification_code( email="test@example.com", timeout=60 )


**支持的服务：**
- CloudMail（基于API）
- LuckMail（基于API）
- YYDS Mail（基于API）
- Outlook（自托管账户）

Registration Flow

注册流程

python

undefined

python

undefined

src/core/register.py

from src.core.register import RegisterService from src.models.task import RegisterTask

async def register_account(email: str, password: str, proxy: str): """Register a new OpenAI account"""

task = RegisterTask(
    email=email,
    password=password,
    proxy=proxy,
    status="pending"
)

service = RegisterService(task)
result = await service.execute()

return result
# Returns: {"success": True, "account_id": 123, "token": "..."}


**Key steps handled:**
1. Sentinel POW solving
2. Email verification (auto-fetch from email service)
3. Split registration/login flow
4. Token extraction
5. Workspace caching

from src.core.register import RegisterService from src.models.task import RegisterTask

async def register_account(email: str, password: str, proxy: str): """Register a new OpenAI account"""

task = RegisterTask(
    email=email,
    password=password,
    proxy=proxy,
    status="pending"
)

service = RegisterService(task)
result = await service.execute()

return result
# Returns: {"success": True, "account_id": 123, "token": "..."}


**处理的关键步骤：**
1. Sentinel POW求解
2. 邮件验证（从邮件服务自动获取验证码）
3. 拆分式注册/登录流程
4. 令牌提取
5. 工作区缓存

Batch Registration

批量注册

python

undefined

python

undefined

src/core/auto_register.py

from src.core.auto_register import AutoRegisterService

async def batch_register(count: int): """Register multiple accounts"""

service = AutoRegisterService()

# Configure
await service.configure({
    "target_count": count,
    "email_service": "cloudmail",
    "proxy_pool": "residential",
    "auto_upload": True,
    "upload_target": "newapi"
})

# Start registration
await service.start()

# Monitor progress
status = await service.get_status()
# Returns: {"completed": 50, "failed": 2, "running": True}

undefined

from src.core.auto_register import AutoRegisterService

async def batch_register(count: int): """Register multiple accounts"""

service = AutoRegisterService()

# Configure
await service.configure({
    "target_count": count,
    "email_service": "cloudmail",
    "proxy_pool": "residential",
    "auto_upload": True,
    "upload_target": "newapi"
})

# Start registration
await service.start()

# Monitor progress
status = await service.get_status()
# Returns: {"completed": 50, "failed": 2, "running": True}

undefined

Payment Binding

支付绑定

python

undefined

python

undefined

src/core/payment.py

from src.core.payment import PaymentService from src.models.task import BindCardTask

async def bind_payment_card(account_id: int, card_info: dict): """Bind payment card to account (semi-automated)"""

task = BindCardTask(
    account_id=account_id,
    card_number=card_info["number"],  # Store encrypted
    expiry=card_info["expiry"],
    cvv=card_info["cvv"],
    billing_address={
        "street": "auto-generated",  # Random US address
        "city": "New York",
        "state": "NY",
        "zip": "10001",
        "country": "US"
    },
    status="pending"
)

service = PaymentService(task)
result = await service.execute()

# Note: 3DS verification requires manual intervention
# Browser window will open for 3DS flow

return result

undefined

from src.core.payment import PaymentService from src.models.task import BindCardTask

async def bind_payment_card(account_id: int, card_info: dict): """Bind payment card to account (semi-automated)"""

task = BindCardTask(
    account_id=account_id,
    card_number=card_info["number"],  # Store encrypted
    expiry=card_info["expiry"],
    cvv=card_info["cvv"],
    billing_address={
        "street": "auto-generated",  # Random US address
        "city": "New York",
        "state": "NY",
        "zip": "10001",
        "country": "US"
    },
    status="pending"
)

service = PaymentService(task)
result = await service.execute()

# Note: 3DS verification requires manual intervention
# Browser window will open for 3DS flow

return result

undefined

Token Management

令牌管理

python

undefined

python

undefined

src/core/account_manager.py

from src.core.account_manager import AccountManager from src.models.account import Account

async def refresh_account_token(account_id: int): """Refresh access token for an account"""

manager = AccountManager()
account = await manager.get_account(account_id)

# Refresh using refresh_token
new_token = await manager.refresh_token(account)

# Auto-upload to configured targets
await manager.upload_to_targets(account)

return new_token

undefined

from src.core.account_manager import AccountManager from src.models.account import Account

async def refresh_account_token(account_id: int): """Refresh access token for an account"""

manager = AccountManager()
account = await manager.get_account(account_id)

# Refresh using refresh_token
new_token = await manager.refresh_token(account)

# Auto-upload to configured targets
await manager.upload_to_targets(account)

return new_token

undefined

Auto-Upload to API Gateways

自动上传至API网关

python

undefined

python

undefined

src/services/upload_service.py

from src.services.upload_service import UploadService

async def upload_account(account_id: int, target: str): """Upload account to API gateway"""

service = UploadService()

# Supported targets: cpa, sub2api, team_manager, newapi
result = await service.upload(
    account_id=account_id,
    target=target
)

return result
# Returns: {"success": True, "external_id": "...", "quota": 20}


**New-API configuration example:**

```python

from src.services.upload_service import UploadService

async def upload_account(account_id: int, target: str): """Upload account to API gateway"""

service = UploadService()

# Supported targets: cpa, sub2api, team_manager, newapi
result = await service.upload(
    account_id=account_id,
    target=target
)

return result
# Returns: {"success": True, "external_id": "...", "quota": 20}


**New-API配置示例：**

```python

In settings or database

newapi_config = { "base_url": "https://api.example.com", "api_key": None, # from env: NEWAPI_API_KEY "auto_upload": True, "upload_on_register": True, "quota_per_account": 20 }

undefined

newapi_config = { "base_url": "https://api.example.com", "api_key": None, # from env: NEWAPI_API_KEY "auto_upload": True, "upload_on_register": True, "quota_per_account": 20 }

undefined

Task Management

任务管理

python

undefined

python

undefined

src/services/task_service.py

from src.services.task_service import TaskService

async def manage_tasks(): """Unified task management"""

service = TaskService()

# Create registration task
task_id = await service.create_task(
    task_type="register",
    count=10,
    config={"email_service": "cloudmail"}
)

# Pause task
await service.pause_task(task_id)

# Resume task
await service.resume_task(task_id)

# Cancel task
await service.cancel_task(task_id)

# Retry failed items
await service.retry_task(task_id)

# Get status
status = await service.get_task_status(task_id)

undefined

from src.services.task_service import TaskService

async def manage_tasks(): """Unified task management"""

service = TaskService()

# Create registration task
task_id = await service.create_task(
    task_type="register",
    count=10,
    config={"email_service": "cloudmail"}
)

# Pause task
await service.pause_task(task_id)

# Resume task
await service.resume_task(task_id)

# Cancel task
await service.cancel_task(task_id)

# Retry failed items
await service.retry_task(task_id)

# Get status
status = await service.get_task_status(task_id)

undefined

Auto-Replenishment

自动补货

python

undefined

python

undefined

src/core/auto_replenishment.py

from src.core.auto_replenishment import AutoReplenishmentService

async def configure_auto_replenish(): """Configure automatic account replenishment"""

service = AutoReplenishmentService()

await service.configure({
    "enabled": True,
    "min_threshold": 10,  # Start when inventory < 10
    "target_count": 50,   # Replenish to 50 accounts
    "check_interval": 3600,  # Check every hour
    "max_daily_registers": 100
})

# Start monitoring
await service.start()

undefined

from src.core.auto_replenishment import AutoReplenishmentService

async def configure_auto_replenish(): """Configure automatic account replenishment"""

service = AutoReplenishmentService()

await service.configure({
    "enabled": True,
    "min_threshold": 10,  # Start when inventory < 10
    "target_count": 50,   # Replenish to 50 accounts
    "check_interval": 3600,  # Check every hour
    "max_daily_registers": 100
})

# Start monitoring
await service.start()

undefined

Self-Check and Repair

自检与修复

python

undefined

python

undefined

src/services/selfcheck.py

from src.services.selfcheck import SelfCheckService

async def run_system_selfcheck(): """Run comprehensive system self-check"""

service = SelfCheckService()

# Run checks
results = await service.run_all_checks()

# Results include:
# - Database connectivity
# - Email service status
# - Proxy availability
# - Account token validity
# - Upload target connectivity

# Auto-repair if enabled
if results["has_issues"]:
    await service.auto_repair(results)

return results

undefined

from src.services.selfcheck import SelfCheckService

async def run_system_selfcheck(): """Run comprehensive system self-check"""

service = SelfCheckService()

# Run checks
results = await service.run_all_checks()

# Results include:
# - Database connectivity
# - Email service status
# - Proxy availability
# - Account token validity
# - Upload target connectivity

# Auto-repair if enabled
if results["has_issues"]:
    await service.auto_repair(results)

return results

undefined

Account Pooling and Tags

账户池化与标签管理

python

undefined

python

undefined

src/models/account.py

from src.core.account_manager import AccountManager

async def manage_account_pools(): """Organize accounts with tags and pools"""

manager = AccountManager()

# Add account to team pool
await manager.update_account(
    account_id=123,
    updates={
        "role_tag": "team_member",
        "biz_tag": "project_alpha",
        "pool_state": "team_pool",
        "priority": 5
    }
)

# Query by pool
team_accounts = await manager.get_accounts_by_pool("team_pool")

# Query by tag
project_accounts = await manager.get_accounts_by_tag("project_alpha")

undefined

from src.core.account_manager import AccountManager

async def manage_account_pools(): """Organize accounts with tags and pools"""

manager = AccountManager()

# Add account to team pool
await manager.update_account(
    account_id=123,
    updates={
        "role_tag": "team_member",
        "biz_tag": "project_alpha",
        "pool_state": "team_pool",
        "priority": 5
    }
)

# Query by pool
team_accounts = await manager.get_accounts_by_pool("team_pool")

# Query by tag
project_accounts = await manager.get_accounts_by_tag("project_alpha")

undefined

Database Models

数据库模型

Account Model

账户模型

python

undefined

python

undefined

src/models/account.py

from sqlalchemy import Column, Integer, String, DateTime, JSON from src.database import Base

class Account(Base): tablename = "accounts"

id = Column(Integer, primary_key=True)
email = Column(String, unique=True, nullable=False)
password = Column(String, nullable=False)

# Auth tokens
access_token = Column(String)
refresh_token = Column(String)
session_token = Column(String)

# Status
status = Column(String, default="active")  # active, suspended, expired
subscription_status = Column(String)  # none, trial, plus, team

# Pooling and tags
role_tag = Column(String)  # team_member, candidate, etc.
biz_tag = Column(String)   # project/client identifier
pool_state = Column(String)  # team_pool, candidate_pool, blocked_pool
priority = Column(Integer, default=0)

# Metadata
quota = Column(Integer, default=0)
workspace_id = Column(String)
last_used_at = Column(DateTime)
created_at = Column(DateTime)
updated_at = Column(DateTime)

# Upload tracking
upload_status = Column(JSON)  # {"cpa": true, "newapi": true}

undefined

from sqlalchemy import Column, Integer, String, DateTime, JSON from src.database import Base

class Account(Base): tablename = "accounts"

id = Column(Integer, primary_key=True)
email = Column(String, unique=True, nullable=False)
password = Column(String, nullable=False)

# Auth tokens
access_token = Column(String)
refresh_token = Column(String)
session_token = Column(String)

# Status
status = Column(String, default="active")  # active, suspended, expired
subscription_status = Column(String)  # none, trial, plus, team

# Pooling and tags
role_tag = Column(String)  # team_member, candidate, etc.
biz_tag = Column(String)   # project/client identifier
pool_state = Column(String)  # team_pool, candidate_pool, blocked_pool
priority = Column(Integer, default=0)

# Metadata
quota = Column(Integer, default=0)
workspace_id = Column(String)
last_used_at = Column(DateTime)
created_at = Column(DateTime)
updated_at = Column(DateTime)

# Upload tracking
upload_status = Column(JSON)  # {"cpa": true, "newapi": true}

undefined

Task Model

任务模型

python

undefined

python

undefined

src/models/task.py

from sqlalchemy import Column, Integer, String, DateTime, JSON from src.database import Base

class RegisterTask(Base): tablename = "register_tasks"

id = Column(Integer, primary_key=True)
email = Column(String, nullable=False)
password = Column(String, nullable=False)
proxy = Column(String)

# Task state
status = Column(String, default="pending")  # pending, running, completed, failed, paused
progress = Column(Integer, default=0)
error_message = Column(String)

# Result
account_id = Column(Integer)  # Reference to created account
access_token = Column(String)

# Metadata
created_at = Column(DateTime)
started_at = Column(DateTime)
completed_at = Column(DateTime)
retry_count = Column(Integer, default=0)

undefined

from sqlalchemy import Column, Integer, String, DateTime, JSON from src.database import Base

class RegisterTask(Base): tablename = "register_tasks"

id = Column(Integer, primary_key=True)
email = Column(String, nullable=False)
password = Column(String, nullable=False)
proxy = Column(String)

# Task state
status = Column(String, default="pending")  # pending, running, completed, failed, paused
progress = Column(Integer, default=0)
error_message = Column(String)

# Result
account_id = Column(Integer)  # Reference to created account
access_token = Column(String)

# Metadata
created_at = Column(DateTime)
started_at = Column(DateTime)
completed_at = Column(DateTime)
retry_count = Column(Integer, default=0)

undefined

API Routes

API路由

Account Management

账户管理

python

undefined

python

undefined

GET /api/accounts

List all accounts with filters

GET /api/accounts?status=active&pool=team_pool&limit=50

GET /api/accounts/{id}

Get account details

GET /api/accounts/123

POST /api/accounts/{id}/refresh

Refresh account token

POST /api/accounts/123/refresh

POST /api/accounts/{id}/upload

Upload to API gateway

POST /api/accounts/123/upload { "target": "newapi" }

DELETE /api/accounts/{id}

Delete account

DELETE /api/accounts/123

undefined

DELETE /api/accounts/123

undefined

Task Management

任务管理

python

undefined

python

undefined

POST /api/tasks/register

Create registration task

POST /api/tasks/register { "count": 10, "email_service": "cloudmail", "auto_upload": true, "upload_target": "newapi" }

POST /api/tasks/{id}/pause

Pause task

POST /api/tasks/123/pause

POST /api/tasks/{id}/resume

Resume task

POST /api/tasks/123/resume

POST /api/tasks/{id}/cancel

Cancel task

POST /api/tasks/123/cancel

POST /api/tasks/{id}/retry

Retry failed items

POST /api/tasks/123/retry

undefined

POST /api/tasks/123/retry

undefined

Export and Import

导出与导入

python

undefined

python

undefined

GET /api/export/accounts

Export accounts in various formats

GET /api/export/accounts?format=codex&pool=team_pool

Supported formats: codex, json, csv, newapi, cpa

POST /api/import/accounts

Import accounts from file

POST /api/import/accounts Content-Type: multipart/form-data file: accounts.json

undefined

POST /api/import/accounts Content-Type: multipart/form-data file: accounts.json

undefined

Common Workflows

常见工作流

Complete Registration Pipeline

完整注册流水线

python

import asyncio
from src.core.register import RegisterService
from src.services.upload_service import UploadService
from src.models.task import RegisterTask

async def full_registration_workflow():
    """End-to-end: register account, verify, bind card, upload"""
    
    # Step 1: Register
    task = RegisterTask(
        email="auto-generated@cloudmail.com",
        password="SecurePass123!",
        proxy="http://proxy.example.com:8080"
    )
    
    reg_service = RegisterService(task)
    result = await reg_service.execute()
    
    if not result["success"]:
        raise Exception(f"Registration failed: {result['error']}")
    
    account_id = result["account_id"]
    
    # Step 2: Bind payment (semi-automated)
    # Note: This opens browser for 3DS - requires monitoring
    from src.core.payment import PaymentService
    from src.models.task import BindCardTask
    
    bind_task = BindCardTask(
        account_id=account_id,
        card_number="encrypted_card_data",
        expiry="12/25",
        cvv="123",
        billing_address={"auto": True}  # Auto-generate random US address
    )
    
    payment_service = PaymentService(bind_task)
    payment_result = await payment_service.execute()
    
    # Step 3: Upload to API gateway
    upload_service = UploadService()
    upload_result = await upload_service.upload(
        account_id=account_id,
        target="newapi"
    )
    
    return {
        "account_id": account_id,
        "email": task.email,
        "uploaded": upload_result["success"]
    }

python

import asyncio
from src.core.register import RegisterService
from src.services.upload_service import UploadService
from src.models.task import RegisterTask

async def full_registration_workflow():
    """End-to-end: register account, verify, bind card, upload"""
    
    # Step 1: Register
    task = RegisterTask(
        email="auto-generated@cloudmail.com",
        password="SecurePass123!",
        proxy="http://proxy.example.com:8080"
    )
    
    reg_service = RegisterService(task)
    result = await reg_service.execute()
    
    if not result["success"]:
        raise Exception(f"Registration failed: {result['error']}")
    
    account_id = result["account_id"]
    
    # Step 2: Bind payment (semi-automated)
    # Note: This opens browser for 3DS - requires monitoring
    from src.core.payment import PaymentService
    from src.models.task import BindCardTask
    
    bind_task = BindCardTask(
        account_id=account_id,
        card_number="encrypted_card_data",
        expiry="12/25",
        cvv="123",
        billing_address={"auto": True}  # Auto-generate random US address
    )
    
    payment_service = PaymentService(bind_task)
    payment_result = await payment_service.execute()
    
    # Step 3: Upload to API gateway
    upload_service = UploadService()
    upload_result = await upload_service.upload(
        account_id=account_id,
        target="newapi"
    )
    
    return {
        "account_id": account_id,
        "email": task.email,
        "uploaded": upload_result["success"]
    }

Run

asyncio.run(full_registration_workflow())

undefined

asyncio.run(full_registration_workflow())

undefined

Batch Account Refresh

批量账户令牌刷新

python

from src.core.account_manager import AccountManager
import asyncio

async def batch_refresh_tokens():
    """Refresh tokens for all active accounts"""
    
    manager = AccountManager()
    
    # Get all active accounts
    accounts = await manager.get_accounts(status="active")
    
    results = []
    for account in accounts:
        try:
            new_token = await manager.refresh_token(account)
            
            # Auto-upload if configured
            if manager.settings.auto_upload_on_refresh:
                await manager.upload_to_targets(account)
            
            results.append({
                "account_id": account.id,
                "success": True,
                "token": new_token
            })
        except Exception as e:
            results.append({
                "account_id": account.id,
                "success": False,
                "error": str(e)
            })
    
    return results

python

from src.core.account_manager import AccountManager
import asyncio

async def batch_refresh_tokens():
    """Refresh tokens for all active accounts"""
    
    manager = AccountManager()
    
    # Get all active accounts
    accounts = await manager.get_accounts(status="active")
    
    results = []
    for account in accounts:
        try:
            new_token = await manager.refresh_token(account)
            
            # Auto-upload if configured
            if manager.settings.auto_upload_on_refresh:
                await manager.upload_to_targets(account)
            
            results.append({
                "account_id": account.id,
                "success": True,
                "token": new_token
            })
        except Exception as e:
            results.append({
                "account_id": account.id,
                "success": False,
                "error": str(e)
            })
    
    return results

Run

asyncio.run(batch_refresh_tokens())

undefined

asyncio.run(batch_refresh_tokens())

undefined

Scheduled Auto-Replenishment

定时自动补货

python

undefined

python

undefined

src/schedulers/auto_replenish_scheduler.py

from apscheduler.schedulers.asyncio import AsyncIOScheduler from src.core.auto_replenishment import AutoReplenishmentService

scheduler = AsyncIOScheduler()

async def check_and_replenish(): """Check inventory and trigger replenishment if needed"""

service = AutoReplenishmentService()
status = await service.check_inventory()

if status["current_count"] < status["min_threshold"]:
    needed = status["target_count"] - status["current_count"]
    await service.start_replenishment(count=needed)

from apscheduler.schedulers.asyncio import AsyncIOScheduler from src.core.auto_replenishment import AutoReplenishmentService

scheduler = AsyncIOScheduler()

async def check_and_replenish(): """Check inventory and trigger replenishment if needed"""

service = AutoReplenishmentService()
status = await service.check_inventory()

if status["current_count"] < status["min_threshold"]:
    needed = status["target_count"] - status["current_count"]
    await service.start_replenishment(count=needed)

Schedule every hour

scheduler.add_job( check_and_replenish, 'interval', hours=1, id='auto_replenish' )

scheduler.start()

undefined

scheduler.add_job( check_and_replenish, 'interval', hours=1, id='auto_replenish' )

scheduler.start()

undefined

Troubleshooting

故障排查

Sentinel POW Solving Fails

Sentinel POW求解失败

Issue: Registration fails with "Sentinel POW required" error.

Solution: Ensure you're using latest version (v1.1.2+) which includes POW solving:

python

undefined

问题： 注册失败，提示“Sentinel POW required”错误。

解决方案： 确保使用最新版本（v1.1.2+），该版本已集成POW求解功能：

python

undefined

src/core/register.py automatically handles this

If issues persist, check proxy quality:

from src.utils.proxy import test_proxy

result = await test_proxy("http://proxy.example.com:8080") if not result["sentinel_pass"]: print("Proxy blocked by Sentinel - use residential proxy")

undefined

from src.utils.proxy import test_proxy

result = await test_proxy("http://proxy.example.com:8080") if not result["sentinel_pass"]: print("Proxy blocked by Sentinel - use residential proxy")

undefined

Email OTP Not Received

邮件OTP未收到

Issue: Registration hangs waiting for verification code.

Solution:

Check email service status in Settings
Verify API key is set correctly
Test email service manually:

python

from src.services.email_service import EmailServiceFactory

service = EmailServiceFactory.create({
    "service_type": "cloudmail",
    "api_key": "test_key"
})

问题： 注册过程中卡在等待验证码环节。

解决方案：

在设置页面检查邮件服务状态
确认API密钥配置正确
手动测试邮件服务：

python

from src.services.email_service import EmailServiceFactory

service = EmailServiceFactory.create({
    "service_type": "cloudmail",
    "api_key": "test_key"
})

Test connection

status = await service.test_connection() print(status) # Should return {"success": True}

undefined

status = await service.test_connection() print(status) # Should return {"success": True}

undefined

Token Refresh Fails

令牌刷新失败

Issue:

refresh_token

returns 401 Unauthorized.

Solution:

Check if refresh token is expired (30 days typically)
Force re-login instead:

python

from src.core.account_manager import AccountManager

manager = AccountManager()
account = await manager.get_account(account_id)

问题：

refresh_token

返回401 Unauthorized错误。

解决方案：

检查刷新令牌是否过期（通常有效期为30天）
强制重新登录：

python

from src.core.account_manager import AccountManager

manager = AccountManager()
account = await manager.get_account(account_id)

Re-login to get fresh tokens

await manager.re_login(account)

undefined

await manager.re_login(account)

undefined

Payment Binding Stuck on 3DS

支付绑定卡在3DS验证环节

Issue: Browser opens for 3DS but verification never completes.

Solution:

3DS cannot be automated. Monitor the browser window and:

Complete 3DS challenge manually
Wait for callback
codex-console will detect completion and continue

问题： 浏览器打开3DS验证窗口，但验证始终无法完成。

解决方案：

3DS验证无法自动化，需监控浏览器窗口并：

手动完成3DS验证挑战
等待回调
codex-console会自动检测完成状态并继续流程

Database Migration Issues

数据库迁移问题

Issue: Alembic migration fails after update.

Solution:

bash

undefined

问题： 更新后Alembic迁移失败。

解决方案：

bash

undefined

Reset to latest migration

alembic stamp head

Or manually migrate

alembic upgrade head

If corrupted, backup and recreate:

cp data/database.db data/database.db.backup rm data/database.db python webui.py # Auto-creates with latest schema

undefined

cp data/database.db data/database.db.backup rm data/database.db python webui.py # Auto-creates with latest schema

undefined

Auto-Upload Not Working

自动上传功能失效

Issue: Accounts registered but not uploaded to New-API/CPA.

Solution:

Verify upload target is configured in Settings
Test connection:

python

from src.services.upload_service import UploadService

service = UploadService()
test_result = await service.test_target("newapi")
print(test_result)  # Check connectivity

Enable auto-upload in task config:

python

{
    "auto_upload": true,
    "upload_target": "newapi",
    "upload_on_register": true
}

问题： 账户已注册但未上传至New-API/CPA。

解决方案：

在设置页面确认上传目标已配置
测试连接：

python

from src.services.upload_service import UploadService

service = UploadService()
test_result = await service.test_target("newapi")
print(test_result)  # Check connectivity

在任务配置中启用自动上传：

python

{
    "auto_upload": true,
    "upload_target": "newapi",
    "upload_on_register": true
}

High Registration Failure Rate

注册失败率高

Issue: Most registration attempts fail.

Solution:

Use residential proxies - datacenter IPs are often blocked
Slow down - add delays between attempts
Check email service quota - may be rate-limited
Review logs - look for specific error patterns

bash

undefined

问题： 大多数注册尝试失败。

解决方案：

使用住宅代理 - 数据中心IP常被封禁
降低操作频率 - 在尝试之间添加延迟
检查邮件服务配额 - 可能存在速率限制
查看日志 - 查找特定错误模式

bash

undefined

Enable debug logging

export LOG_LEVEL=debug python webui.py

Check logs in data/logs/

tail -f data/logs/register.log

undefined

tail -f data/logs/register.log

undefined

Port Already in Use

端口已被占用

Issue:

Address already in use

error on startup.

Solution:

codex-console auto-switches ports if 8000 is taken. To force a specific port:

bash

python webui.py --port 8080

Or kill the process using port 8000:

bash

undefined

问题： 启动时提示“Address already in use”错误。

解决方案：

codex-console会在8000端口被占用时自动切换端口。如需指定特定端口：

bash

python webui.py --port 8080

或终止占用8000端口的进程：

bash

undefined

Linux/Mac

lsof -ti:8000 | xargs kill -9

Windows

netstat -ano | findstr :8000 taskkill /PID <PID> /F

undefined

netstat -ano | findstr :8000 taskkill /PID <PID> /F

undefined

Advanced Configuration

高级配置

Custom Email Service

自定义邮件服务

Implement custom email provider:

python

undefined

实现自定义邮件服务商：

python

undefined

src/services/custom_email.py

from src.services.email_service import BaseEmailService

class CustomEmailService(BaseEmailService): async def get_verification_code(self, email: str, timeout: int = 60) -> str: """Fetch OTP from custom provider""" # Your implementation pass

async def test_connection(self) -> dict:
    """Test API connectivity"""
    # Your implementation
    pass

from src.services.email_service import BaseEmailService

class CustomEmailService(BaseEmailService): async def get_verification_code(self, email: str, timeout: int = 60) -> str: """Fetch OTP from custom provider""" # Your implementation pass

async def test_connection(self) -> dict:
    """Test API connectivity"""
    # Your implementation
    pass

Register in factory

src/services/email_service.py

EMAIL_SERVICES["custom"] = CustomEmailService

undefined

EMAIL_SERVICES["custom"] = CustomEmailService

undefined

Custom Upload Target

自定义上传目标

Add new API gateway integration:

python

undefined

添加新的API网关集成：

python

undefined

src/services/custom_upload.py

from src.services.upload_service import BaseUploadTarget

class CustomUploadTarget(BaseUploadTarget): async def upload_account(self, account: Account) -> dict: """Upload account to custom gateway""" # Your implementation pass

from src.services.upload_service import BaseUploadTarget

class CustomUploadTarget(BaseUploadTarget): async def upload_account(self, account: Account) -> dict: """Upload account to custom gateway""" # Your implementation pass

Register

src/services/upload_service.py

UPLOAD_TARGETS["custom"] = CustomUploadTarget

undefined

UPLOAD_TARGETS["custom"] = CustomUploadTarget

undefined

Security Best Practices

安全最佳实践

Never commit secrets - use environment variables:

bash

export CLOUDMAIL_API_KEY=your_key_here
export NEWAPI_API_KEY=your_key_here

Change default password immediately:

bash

python webui.py --access-password strong_password_here

Use PostgreSQL for production - SQLite is development-only:

bash

export APP_DATABASE_URL=postgresql://user:pass@host:5432/db

Encrypt sensitive fields - card data, passwords are encrypted at rest
Enable audit logging - track all operations:

python

from src.models.audit import OperationAuditLog

切勿提交敏感信息 - 使用环境变量：

bash

export CLOUDMAIL_API_KEY=your_key_here
export NEWAPI_API_KEY=your_key_here

立即修改默认密码：

bash

python webui.py --access-password strong_password_here

生产环境使用PostgreSQL - SQLite仅适用于开发环境：

bash

export APP_DATABASE_URL=postgresql://user:pass@host:5432/db

加密敏感字段 - 卡片数据、密码在存储时已加密
启用审计日志 - 跟踪所有操作：

python

from src.models.audit import OperationAuditLog

Auto-logged for sensitive operations

undefined

undefined

Resources

资源

Official Repo: https://github.com/dou-jiang/codex-console
Blog/Docs: https://blog.cysq8.cn/
QQ Group: 291638849
Telegram: https://t.me/codex_console
License: MIT

官方仓库： https://github.com/dou-jiang/codex-console
博客/文档： https://blog.cysq8.cn/
QQ群： 291638849
Telegram： https://t.me/codex_console
许可证： MIT