4399 注册机：ddddocr 验证码识别 + 多线程并发 + 代理池

pip install ddddocr faker requests

import ddddocr, requests, base64

ocr = ddddocr.DdddOcr(show_ad=False)

def get_captcha_text(session: requests.Session) -> tuple[str, str]:
    # 获取验证码图片并识别，返回 (验证码文本, 验证码ID)
    # 1. 请求验证码图片
    r = session.get(
        "https://ptlogin.4399.com/captcha/get",
        params={"r": str(time.time())},
        timeout=10
    )
    data = r.json()
    captcha_id = data.get("captchaId", "")
    img_b64 = data.get("img", "")  # base64 图片

    # 2. 识别
    img_bytes = base64.b64decode(img_b64)
    text = ocr.classification(img_bytes)
    return text.strip(), captcha_id

from faker import Faker
import random, string

fake = Faker("zh_CN")

def gen_username() -> str:
    # 生成随机用户名（中文拼音 + 数字混合）
    prefix = fake.user_name()[:8]
    suffix = "".join(random.choices(string.digits, k=4))
    return f"{prefix}{suffix}"

def gen_password(length: int = 12) -> str:
    # 生成符合规则的随机密码
    chars = string.ascii_letters + string.digits + "!@#$"
    while True:
        pwd = "".join(random.choices(chars, k=length))
        # 确保包含大写、小写、数字、特殊字符
        if (any(c.isupper() for c in pwd) and
            any(c.islower() for c in pwd) and
            any(c.isdigit() for c in pwd)):
            return pwd

import requests, time

REGISTER_URL = "https://ptlogin.4399.com/register"

def register_one(proxy: str = None) -> dict:
    # 注册一个账号，返回结果
    session = requests.Session()
    if proxy:
        session.proxies = {
            "http": f"http://{proxy}",
            "https": f"http://{proxy}"
        }

    username = gen_username()
    password = gen_password()

    # 1. 获取并识别验证码
    try:
        captcha_text, captcha_id = get_captcha_text(session)
    except Exception as e:
        return {"status": "captcha_error", "error": str(e)}

    # 2. 提交注册
    payload = {
        "username": username,
        "password": password,
        "password2": password,
        "captcha": captcha_text,
        "captchaId": captcha_id,
        "agreement": "1",
    }
    try:
        r = session.post(REGISTER_URL, data=payload, timeout=15)
        data = r.json()
        if data.get("code") == 0:
            return {"status": "ok", "username": username, "password": password}
        else:
            return {"status": "fail", "msg": data.get("msg"), "username": username}
    except Exception as e:
        return {"status": "error", "error": str(e)}

from concurrent.futures import ThreadPoolExecutor, as_completed
import csv, threading

MAX_WORKERS = 30
results_lock = threading.Lock()
success_count = 0
fail_count = 0

def batch_register(target: int = 100, proxy_pool=None):
    global success_count, fail_count
    rows = []

    def task():
        proxy = proxy_pool.get() if proxy_pool else None
        result = register_one(proxy)
        if proxy_pool and result["status"] != "ok" and proxy:
            proxy_pool.remove(proxy)  # 代理失效则移除
        return result

    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
        futures = [pool.submit(task) for _ in range(target)]
        for i, f in enumerate(as_completed(futures), 1):
            result = f.result()
            with results_lock:
                if result["status"] == "ok":
                    success_count += 1
                    rows.append(result)
                    print(f"[{i}/{target}] ✓ {result['username']}")
                else:
                    fail_count += 1
                    print(f"[{i}/{target}] ✗ {result.get('msg', result.get('error', ''))}")

    # 保存成功账号
    if rows:
        with open("accounts.csv", "a", newline="", encoding="utf-8-sig") as f:
            writer = csv.DictWriter(f, fieldnames=["username", "password", "status"])
            writer.writerows(rows)
    print(f"
完成：成功 {success_count}，失败 {fail_count}")

from collections import Counter

def robust_ocr(img_bytes: bytes, n: int = 3) -> str:
    results = [ocr.classification(img_bytes) for _ in range(n)]
    return Counter(results).most_common(1)[0][0]

def register_with_retry(proxy=None, max_retry=3) -> dict:
    for _ in range(max_retry):
        result = register_one(proxy)
        if result["status"] == "ok":
            return result
        if "验证码" in result.get("msg", ""):
            time.sleep(0.5)
            continue
        break
    return result