autoChatGPTPlus-Paypal/providers.py

"""卡号、地址、短信三个外部数据源。"""
from __future__ import annotations

import json
import random
import re
import time
import urllib.error
import urllib.request


CARD_API = "https://api2.suijidaquan.com/api/v2/random-credit-card"
ADDR_API = "https://www.meiguodizhi.com/api/v1/dz"
SMS_API = "http://a.62-us.com/api/get_sms?key=a5d3262e05efaba982aba7cfae20b8bc"


# VISA 起 4，16 位；Mastercard 起 51-55 或 2221-2720，16 位
CARD_BIN_POOLS = {
    "visa": [str(random.randint(4, 4)) + "".join(str(random.randint(0, 9)) for _ in range(5)) for _ in range(0)],
}


def _luhn_check_digit(number_without_check: str) -> str:
    digits = [int(c) for c in number_without_check]
    # 从右往左、每隔一位（即偶数索引位）×2
    parity = (len(digits) + 1) % 2  # 让最后一位 parity=0 才需要×2
    total = 0
    for i, d in enumerate(digits):
        if i % 2 == parity:
            d *= 2
            if d > 9:
                d -= 9
        total += d
    return str((10 - total % 10) % 10)


def _gen_visa_pan() -> str:
    # 起 4，再补 14 位随机，最后 1 位 Luhn
    body = "4" + "".join(str(random.randint(0, 9)) for _ in range(14))
    return body + _luhn_check_digit(body)


def _gen_mastercard_pan() -> str:
    # 起 51-55，再补 13 位随机，最后 1 位 Luhn
    prefix = str(random.randint(51, 55))
    body = prefix + "".join(str(random.randint(0, 9)) for _ in range(13))
    return body + _luhn_check_digit(body)


def generate_local_card(brand: str = "visa") -> dict:
    """本地随机生成一张 Luhn 合规的 VISA / Mastercard 测试卡。
    expiry 取未来 1-4 年的随机月份，CVV 三位随机。
    """
    brand = (brand or "visa").lower()
    if brand == "mastercard":
        pan = _gen_mastercard_pan()
    else:
        pan = _gen_visa_pan()
        brand = "visa"
    now = time.localtime()
    exp_year = (now.tm_year + random.randint(1, 4)) % 100
    exp_month = random.randint(1, 12)
    expiry = f"{exp_month:02d} / {exp_year:02d}"
    cvv = "".join(str(random.randint(0, 9)) for _ in range(3))
    return {
        "number": pan,
        "expiry": expiry,
        "cvv": cvv,
        "brand": brand,
    }


DEFAULT_BROWSER_HEADERS = {
    "Accept": "application/json, text/plain, */*",
    "Accept-Language": "zh-CN,zh;q=0.9",
    "Cache-Control": "no-cache",
    "Pragma": "no-cache",
    "DNT": "1",
    "Priority": "u=1, i",
    "Sec-Ch-Ua": '"Chromium";v="148", "Google Chrome";v="148", "Not/A)Brand";v="99"',
    "Sec-Ch-Ua-Mobile": "?0",
    "Sec-Ch-Ua-Platform": '"macOS"',
    "Sec-Fetch-Dest": "empty",
    "Sec-Fetch-Mode": "cors",
    "Sec-Fetch-Site": "same-site",
    "User-Agent": (
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/148.0.0.0 Safari/537.36"
    ),
}


def _post_json(url: str, body: dict, headers: dict | None = None, timeout: int = 15, log=print) -> tuple[int, str, dict]:
    data = json.dumps(body).encode("utf-8")
    req = urllib.request.Request(url, data=data, method="POST")
    merged = {**DEFAULT_BROWSER_HEADERS, "Content-Type": "application/json;charset=UTF-8"}
    merged.update(headers or {})
    for k, v in merged.items():
        req.add_header(k, v)
    started = time.time()
    log(f"[http] POST {url} body={json.dumps(body, ensure_ascii=False)}")
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            status = resp.status
            text = resp.read().decode("utf-8", errors="replace")
    except urllib.error.HTTPError as exc:
        text = exc.read().decode("utf-8", errors="replace")
        log(f"[http] {url} HTTP {exc.code} 耗时{int((time.time()-started)*1000)}ms 返回={text[:300]}")
        raise
    log(f"[http] {url} HTTP {status} 耗时{int((time.time()-started)*1000)}ms 返回={text[:300]}")
    return status, text, json.loads(text or "{}")


def _normalize_expiry(expires: str) -> str:
    m = re.match(r"^\s*(\d{1,2})\s*/\s*(\d{2,4})\s*$", expires or "")
    if not m:
        return expires
    mm = m.group(1).zfill(2)
    yy = m.group(2)
    if len(yy) == 4:
        yy = yy[2:]
    return f"{mm} / {yy}"


def fetch_visa_card(max_attempts: int = 8, log=print, *, prefer_local: bool = True) -> dict:
    """获取一张可用的卡。默认本地随机生成（避免接口返回重复卡导致 PayPal CC_LINKED_TO_FULL_ACCOUNT），
    若 prefer_local=False 则走旧的接口逻辑。
    """
    if prefer_local:
        brand = random.choice(["visa", "mastercard"])
        card = generate_local_card(brand=brand)
        log(f"[card] 本地生成 {card['brand'].upper()} 卡 尾号 {card['number'][-4:]} 有效期 {card['expiry']} CVV {card['cvv']}")
        return card

    log(f"[card] 开始通过接口获取 VISA 卡，最多重试 {max_attempts} 次")
    for attempt in range(1, max_attempts + 1):
        log(f"[card] 第 {attempt}/{max_attempts} 次请求 {CARD_API}")
        try:
            _, _, data = _post_json(
                CARD_API,
                {"count": 4, "method": "random_credit_card"},
                headers={
                    "Origin": "https://www.suijidaquan.com",
                    "Referer": "https://www.suijidaquan.com/",
                },
                log=log,
            )
        except (urllib.error.URLError, json.JSONDecodeError) as exc:
            log(f"[card] 请求异常: {exc!r}")
            time.sleep(0.6)
            continue

        cards = data.get("data") or []
        types = [c.get("Credit_Card_Type") for c in cards]
        log(f"[card] 本次返回 {len(cards)} 张卡，类型 = {types}")
        for c in cards:
            if (c.get("Credit_Card_Type") or "").lower() == "visa":
                card = {
                    "number": c["Credit_Card_Number"],
                    "expiry": _normalize_expiry(c["Expires"]),
                    "cvv": c["CVV2"],
                }
                log(f"[card] 命中 VISA 尾号 {card['number'][-4:]} 有效期 {card['expiry']} CVV {card['cvv']}")
                return card
        log("[card] 本次未拿到 VISA，准备重试")
        time.sleep(0.4)

    raise RuntimeError(f"已重试 {max_attempts} 次，仍未取到 VISA 卡")


def fetch_us_address(log=print) -> dict:
    log(f"[addr] 请求随机美国地址 {ADDR_API}")
    try:
        _, _, data = _post_json(ADDR_API, {"path": "/", "method": "address"}, log=log)
        a = data.get("address") or data
        addr = {
            "street": a.get("Address") or a.get("street") or "123 Main St",
            "city": a.get("City") or a.get("city") or "New York",
            "state": a.get("State_Full") or a.get("State") or a.get("state") or "New York",
            "zip": (a.get("Zip_Code") or a.get("zip") or "10001")[:5],
        }
    except Exception as exc:
        log(f"[addr] 取地址失败，使用兜底: {exc!r}")
        addr = {"street": "123 Main St", "city": "New York", "state": "New York", "zip": "10001"}
    log(f"[addr] 解析结果: {addr}")
    return addr


def fetch_sms_code(timeout: int = 180, interval: int = 5, log=print) -> str:
    log(f"[sms] 开始轮询验证码，超时 {timeout}s，间隔 {interval}s")
    deadline = time.time() + timeout
    last_text = ""
    polls = 0
    while time.time() < deadline:
        polls += 1
        try:
            req = urllib.request.Request(SMS_API)
            req.add_header("Accept", "*/*")
            req.add_header("Accept-Language", "zh-CN,zh;q=0.9")
            req.add_header(
                "User-Agent",
                "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
                "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36",
            )
            with urllib.request.urlopen(req, timeout=10) as resp:
                text = resp.read().decode("utf-8", errors="replace").strip()
        except Exception as exc:
            log(f"[sms] 第{polls}次轮询请求异常: {exc!r}")
            time.sleep(interval)
            continue

        if text != last_text:
            log(f"[sms] 第{polls}次轮询新返回: {text}")
            last_text = text
        else:
            log(f"[sms] 第{polls}次轮询无变化")

        parts = text.split("|")
        status = parts[0].lower() if parts else ""
        content = parts[1] if len(parts) > 1 else ""

        if status == "yes":
            m = re.search(r"\b(\d{4,8})\b", content)
            if m:
                code = m.group(1)
                log(f"[sms] 命中验证码: {code}")
                return code
            log(f"[sms] status=yes 但未能从内容中匹配到数字: {content!r}")
        time.sleep(interval)

    raise TimeoutError(f"等待短信验证码超时 {timeout}s（共轮询 {polls} 次，最后返回={last_text!r}）")